utils-shell/lotabout.skim
1
0
Fork 0
mirror of https://github.com/lotabout/skim.git synced 2026-09-10 07:16:23 -04:00
Code Issues Actions Packages Projects Releases Wiki Activity

feat: improve matcher & reader performance (#1020)

Browse source 
* chore: migrate bench.py to rust to remove python deps

* feat: replace rayon with a custom thread pool manager

* wip: insert into item_list processed_items directly from matcher

* wip: perf optimizations

* wip: perf optimizations

* wip: reader perf optimizations

* fix: skip --bench injected in bench args

* chore: add ARCHITECTURE.md

* Update src/helper/item_reader.rs

Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com>

* Update src/matcher.rs

Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com>

* feat: use the same pool between reader and matcher

* chore: misc

* fix: tests

---------

Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com>
This commit is contained in:
LoricAndre 2026-03-31 18:05:20 +02:00 committed by GitHub
parent 94c132a94f
commit 1a10e405f1
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
21 changed files with 3696 additions and 1155 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
.github/workflows/test.yml vendored
View file

@ -81,7 +81,6 @@ jobs:
echo "old: $base"
cat "$base"
diff "$base" "$new_snap"
else
fi
done
shell: bash

16
AGENTS.md
View file

@ -3,7 +3,7 @@
## Build/Test/Lint Commands
- Build: `cargo build [--release]`
- Run: `cargo run [--release]`
- Test (all): `cargo nextest`
- Test (all): `cargo nextest run`
- Test (single): `cargo nextest test_name`
- Integration/E2E tests: `cargo nextest --tests` (will need tmux under the hood)
- Memory leak detection: `cargo nextest run --profile valgrind`
@ -24,10 +24,16 @@
- Follow the existing structure for new modules (see src/engine/ or src/model/)
- Implement relevant traits (SkimItem, etc.) for new types when needed
## Project Structure
- Core functionality in `skim/src/`
- Common utilities in `skim-common/`
- Task automation in `xtask/`
## Architecture Documentation
- `ARCHITECTURE.md` documents the full architecture: data flow, operating modes, subsystems, threading model, and public API.
- **Update `ARCHITECTURE.md` whenever you make structural changes**, including:
- Adding, removing, or renaming modules, structs, or traits
- Changing the data flow between subsystems (reader → pool → matcher → TUI)
- Adding new operating modes or modifying existing ones
- Changing the threading model or synchronization primitives
- Adding or removing public API surface (`SkimItem`, `SkimOptions`, `SkimOutput`, etc.)
- Changing the event/action system or key binding infrastructure
- Keep call-site line numbers in the cross-reference table up to date when the referenced functions move.
## Testing

1197
ARCHITECTURE.md Normal file
View file

File diff suppressed because it is too large Load diff

1
CLAUDE.md Symbolic link
View file

@ -0,0 +1 @@
AGENTS.md

2
Cargo.lock generated
View file

@ -2292,11 +2292,11 @@ dependencies = [
"portable-pty",
"rand 0.10.0",
"ratatui",
"rayon",
"regex",
"roff",
"ron",
"serde",
"serde_json",
"shell-quote",
"shlex",
"tempfile",

9
Cargo.toml
View file

@ -11,6 +11,7 @@ keywords = ["fuzzy", "menu", "util"]
license = "MIT"
edition = "2024"
rust-version = "1.91"
default-run = "sk"
[package.metadata.wix]
upgrade-guid = "6DDAED06-EBE2-41C4-94F5-CB03F2A4B92E"
@ -54,7 +55,7 @@ indexmap = "=2.13.0"
log = "=0.4.29"
nix = { version = "=0.31.2", features = ["fs", "poll"] }
rand = "=0.10.0"
rayon = "=1.11.0"
regex = "=1.12.3"
shell-quote = "=0.7.2"
shlex = { version = "=1.3.0", optional = true }
@ -95,6 +96,7 @@ gungraun = ["dep:gungraun"]
[dev-dependencies]
criterion = { version = "0.8.2", features = ["async_tokio"] }
insta = "1.46"
serde_json = { version = "=1.0.149" }
[lints.rust]
unexpected_cfgs = { level = "warn", check-cfg = ['cfg(coverage, coverage_nightly)'] }
@ -119,3 +121,8 @@ harness = false
name = "gungraun"
harness = false
required-features = ["gungraun"]
[[bench]]
name = "cli"
harness = false
bench = false

21
README.md
View file

@ -743,15 +743,28 @@ export TERMINFO=/data/data/com.termux/files/usr/share/terminfo
# Benchmarks
## Shell script
## Interactive benchmark (`cli`)
The `bench.py` script is available to benchmark the code against other versions or fzf using `tmux` and querying the output. This is by no means a precise or foolproof way of running benchmarks, but it has the added benefit of allowing us to benchmark against `fzf` and of giving us resource metrics.
The `cli` bench benchmarks skim (or any compatible binary) against other versions or fzf by running the interactive interface inside a tmux session and polling the status line until the matched count stabilises. This is by no means a precise or foolproof measurement, but it has the added benefit of benchmarking against `fzf` and of providing resource metrics (peak RSS and CPU).
You can use it directly using `./bench.py <binary> -n <number of items> -r <number of runs>`, or generate the data using `./bench.py -g <output file> -n <number of items>`, then `./bench.py <binary> -f <file> -r <number of runs>`
```sh
cargo bench --bench cli # defaults: sk, 1 M items, query "test"
cargo bench --bench cli -- sk -n 500000 -q foo # bare name resolved via $PATH
cargo bench --bench cli -- ./old/sk ./new/sk -r 5 # compare two binaries, 5 runs each
cargo bench --bench cli -- sk -r 5 # 5 runs, show average
cargo bench --bench cli -- sk -f input.txt -q search # use an existing file
cargo bench --bench cli -- -g testdata.txt -n 2000000 # generate input file and exit
cargo bench --bench cli -- sk -p # record perf data (auto-named file)
cargo bench --bench cli -- sk -p perf.data # record perf data to perf.data
cargo bench --bench cli -- sk -j # JSON output
cargo bench --bench cli -- sk -r 3 -- --tiebreak=index # pass extra flags to sk
```
Binary names are resolved to absolute paths via `which` before use, so bare names like `sk` or `fzf` work as long as they are on `$PATH`.
### Criterion benchmarks
Criterion benchmarks are available to measure skim's performance more precisely.
To run them, you need to generate input data using `./bench.py -g benches/fixtures/10M.txt -n 10000000 && ./bench.py -g benches/fixtures/1M.txt -n 1000000`, then run `cargo bench -j 1`.
To run them, you need to generate input data using `cargo bench --bench cli -- -g benches/fixtures/10M.txt -n 10000000 && cargo bench --bench cli -- -g benches/fixtures/1M.txt -n 1000000`, then run `cargo bench -j 1`.
These will run for several minutes.

890
bench.py
View file

@ -1,890 +0,0 @@
#!/usr/bin/env python3
"""
Benchmark script to measure ingestion + matching rate in skim interactive mode.
This measures how fast skim can ingest items and display matched results.
Usage: bench.py [BINARY_PATH ...] [-n|--num-items NUM] [-q|--query QUERY]
[-r|--runs RUNS] [-w|--warmup N] [-f|--file FILE]
[-g|--generate-file FILE] [-j|--json] [-p|--perf [FILE]]
[-- EXTRA_ARGS...]
Arguments:
BINARY_PATH ... One or more paths to binaries (default: ./target/release/sk)
When multiple are given they run in round-robin and the
first is used as the baseline for +/- comparisons.
-n, --num-items NUM Number of items to generate (default: 1000000)
-q, --query QUERY Query string to search (default: "test")
-r, --runs RUNS Number of benchmark runs per binary (default: 1)
-w, --warmup N Number of warmup runs per binary (default: 1)
-f, --file FILE Use existing file as input instead of generating
-g, --generate-file FILE Generate test data to file and exit
-j, --json Output results as JSON
-- Pass remaining arguments to the binary
Examples:
./bench.py # Use defaults
./bench.py ./target/release/sk -n 500000 -q foo
./bench.py ./old/sk ./new/sk -r 5 # Compare two binaries
./bench.py -r 5 # Run 5 times and show average
./bench.py -f input.txt -q search # Use existing file
./bench.py -g testdata.txt -n 2000000 # Generate file and exit
./bench.py -p # Record perf data (auto-named file)
./bench.py -p perf.data # Record perf data to perf.data
"""
import json
import math
import os
import random
import subprocess
import sys
import tempfile
import time
# ---------------------------------------------------------------------------
# Defaults
# ---------------------------------------------------------------------------
DEFAULT_BINARY = "./target/release/sk"
DEFAULT_NUM_ITEMS = 1_000_000
DEFAULT_QUERY = "test"
DEFAULT_RUNS = 1
DEFAULT_WARMUP = 1
# Stability / timeout tuning (mirrors bench.sh values)
REQUIRED_STABLE_S = 5.0 # seconds the matched count must be unchanged
MAX_WAIT_S = 60.0 # hard timeout per run
CHECK_INTERVAL_S = 0.05 # polling interval
# ---------------------------------------------------------------------------
# Test-data generation
# ---------------------------------------------------------------------------
WORDS = [
"home",
"usr",
"etc",
"var",
"opt",
"tmp",
"dev",
"proc",
"sys",
"lib",
"bin",
"sbin",
"boot",
"mnt",
"media",
"src",
"test",
"config",
"data",
"logs",
"cache",
"backup",
"docs",
"images",
"videos",
"audio",
"downloads",
"uploads",
"temp",
"shared",
]
def generate_test_data(output_file: str, num_items: int) -> None:
rng = random.Random()
with open(output_file, "w") as fh:
for i in range(1, num_items + 1):
depth = rng.randint(2, 10)
parts = [rng.choice(WORDS) for _ in range(depth)]
fh.write("/".join(parts) + f"_{i}\n")
# ---------------------------------------------------------------------------
# Argument parsing (no argparse – stdlib only, but argparse IS stdlib…
# however, to keep the spirit of "no dependencies" we use manual parsing
# since argparse is a stdlib module, not a third-party dep. We use it.)
# ---------------------------------------------------------------------------
def parse_args(argv):
"""Return (binaries, opts, extra_args)."""
import argparse
# Split off everything after "--"
extra_args = []
if "--" in argv:
sep_idx = argv.index("--")
extra_args = argv[sep_idx + 1 :]
argv = argv[:sep_idx]
parser = argparse.ArgumentParser(
description="Skim benchmark script",
add_help=True,
)
parser.add_argument(
"binaries",
nargs="*",
metavar="BINARY_PATH",
help="Path(s) to binary (default: ./target/release/sk)",
)
parser.add_argument("-n", "--num-items", type=int, default=DEFAULT_NUM_ITEMS)
parser.add_argument("-q", "--query", default=DEFAULT_QUERY)
parser.add_argument("-r", "--runs", type=int, default=DEFAULT_RUNS)
parser.add_argument("-w", "--warmup", type=int, default=DEFAULT_WARMUP)
parser.add_argument("-f", "--file", default="")
parser.add_argument("-g", "--generate-file", default="")
parser.add_argument("-j", "--json", action="store_true")
parser.add_argument(
"-p",
"--perf",
nargs="?",
const="", # flag present but no value → auto-name
default=None, # flag absent
metavar="FILE",
help="Record perf data for the final benchmark run. "
"Optionally specify output file path (default: auto-named perf-<binary>-<ts>.data).",
)
opts = parser.parse_args(argv)
if not opts.binaries:
opts.binaries = [DEFAULT_BINARY]
if opts.file and opts.generate_file:
parser.error("Cannot use both --file and --generate-file")
return opts.binaries, opts, extra_args
# ---------------------------------------------------------------------------
# Resource monitor (background thread)
# ---------------------------------------------------------------------------
import threading
class ResourceMonitor(threading.Thread):
"""Sample CPU and RSS of *pid* every 50 ms until the process exits."""
def __init__(self, pid: int):
super().__init__(daemon=True)
self.pid = pid
self.peak_mem_kb: int = 0 # RSS in kB
self.peak_cpu: float = 0.0 # %CPU
def run(self):
while True:
try:
result = subprocess.run(
["ps", "-p", str(self.pid), "-o", "rss=,%cpu="],
capture_output=True,
text=True,
)
line = result.stdout.strip()
if not line:
break
parts = line.split()
if len(parts) >= 2:
try:
mem = int(parts[0])
cpu = float(parts[1])
if mem > self.peak_mem_kb:
self.peak_mem_kb = mem
if cpu > self.peak_cpu:
self.peak_cpu = cpu
except ValueError:
pass
except Exception:
break
time.sleep(0.05)
# ---------------------------------------------------------------------------
# Single run
# ---------------------------------------------------------------------------
def _find_sk_pid(pane_pid: int, binary_path: str) -> int:
"""Try for up to 5 s to find the sk child PID under *pane_pid*."""
for _ in range(50):
time.sleep(0.1)
try:
result = subprocess.run(
["pgrep", "-P", str(pane_pid), "-f", binary_path],
capture_output=True,
text=True,
)
pids = result.stdout.strip().splitlines()
if pids:
return int(pids[0])
except Exception:
pass
return 0
def run_once(
binary_path: str,
query: str,
tmp_file: str,
num_items: int,
extra_args: list,
run_index: int,
session_suffix: str,
perf_output: str | None = None,
) -> dict:
"""
Execute one benchmark run against *binary_path*.
Returns a dict with keys: elapsed_s, rate, matched, peak_mem_kb, peak_cpu,
completed, perf_file (path or None).
If *perf_output* is a non-empty string, ``perf record`` is attached to the
sk process and data written to that path.
"""
session_name = f"skim_bench_{os.getpid()}_{session_suffix}_{run_index}"
status_fd, status_file = tempfile.mkstemp(prefix="skim_bench_status_")
os.close(status_fd)
env = os.environ.copy()
env["SHELL"] = "/bin/sh"
env.pop("HISTFILE", None)
env.pop("FZF_DEFAULT_OPTS", None)
env.pop("SKIM_DEFAULT_OPTIONS", None)
try:
# Create tmux session
subprocess.run(
["tmux", "new-session", "-s", session_name, "-d"],
check=True,
env=env,
capture_output=True,
)
# Clear env vars inside the session
for cmd in [
"unset HISTFILE",
"unset FZF_DEFAULT_OPTS",
"unset SKIM_DEFAULT_OPTIONS",
]:
subprocess.run(
["tmux", "send-keys", "-t", session_name, cmd, "Enter"],
check=True,
capture_output=True,
)
time.sleep(0.1)
# Build the command string
extra_str = " ".join(extra_args)
perf_prefix = f"perf record -o {perf_output} -- " if perf_output else ""
cmd_str = f"cat {tmp_file} | {perf_prefix}{binary_path} --query '{query}' {extra_str}"
subprocess.run(
["tmux", "send-keys", "-t", session_name, cmd_str, "Enter"],
check=True,
capture_output=True,
)
start_ns = time.perf_counter_ns()
# Locate sk PID for resource monitoring
pane_pid = 0
try:
r = subprocess.run(
["tmux", "list-panes", "-t", session_name, "-F", "#{pane_pid}"],
capture_output=True,
text=True,
)
pane_pid = int(r.stdout.strip().splitlines()[0])
except Exception:
pass
sk_pid = 0
monitor = None
if pane_pid:
sk_pid = _find_sk_pid(pane_pid, binary_path)
if sk_pid:
monitor = ResourceMonitor(sk_pid)
monitor.start()
# Poll for matcher completion
completed = False
matched_count = 0
prev_matched_count = -1
stable_start: float = 0.0
end_ns = 0
loop_start = time.monotonic()
while True:
time.sleep(CHECK_INTERVAL_S)
now = time.monotonic()
if now - loop_start >= MAX_WAIT_S:
break
# Early exit if sk process is gone
if sk_pid:
try:
os.kill(sk_pid, 0)
except ProcessLookupError:
break
# Capture tmux pane
try:
subprocess.run(
[
"tmux",
"capture-pane",
"-b",
f"status-{session_name}",
"-t",
session_name,
],
capture_output=True,
)
subprocess.run(
[
"tmux",
"save-buffer",
"-b",
f"status-{session_name}",
status_file,
],
capture_output=True,
)
except Exception:
continue
# Parse "matched/total" from status line
try:
with open(status_file) as fh:
content = fh.read()
except OSError:
continue
import re
m = re.search(r"(\d+)/(\d+)", content)
if not m:
continue
mc = int(m.group(1))
total = int(m.group(2))
if total == num_items:
if mc != prev_matched_count:
prev_matched_count = mc
matched_count = mc
stable_start = time.monotonic()
end_ns = time.perf_counter_ns()
elif stable_start > 0:
if time.monotonic() - stable_start >= REQUIRED_STABLE_S:
completed = True
break
if end_ns == 0:
end_ns = time.perf_counter_ns()
# Exit skim
subprocess.run(
["tmux", "send-keys", "-t", session_name, "Escape"],
capture_output=True,
)
time.sleep(0.1)
# If perf is recording, wait for it to exit and flush data before we
# kill the tmux session. perf record is the parent of sk in the shell
# pipeline, so it exits on its own once sk does – we just need to give
# it enough time to finish writing.
if perf_output and pane_pid:
perf_wait_start = time.monotonic()
while time.monotonic() - perf_wait_start < 15.0:
result = subprocess.run(
["pgrep", "-P", str(pane_pid), "-f", "perf record"],
capture_output=True,
)
if result.returncode != 0:
# perf has exited
break
time.sleep(0.1)
else:
print(
"Warning: perf record did not exit within 15 s; perf data may be incomplete.",
file=sys.stderr,
)
# Wait for monitor
if monitor is not None:
monitor.join(timeout=2.0)
elapsed_s = (end_ns - start_ns) / 1e9
rate = num_items / elapsed_s if elapsed_s > 0 else 0
peak_mem_kb = monitor.peak_mem_kb if monitor and monitor.peak_mem_kb else 0
peak_cpu = monitor.peak_cpu if monitor and monitor.peak_cpu else 0.0
recorded_perf = perf_output if perf_output else None
return {
"elapsed_s": elapsed_s,
"rate": rate,
"matched": matched_count,
"peak_mem_kb": peak_mem_kb if peak_mem_kb else None,
"peak_cpu": peak_cpu if peak_cpu else None,
"completed": completed,
"perf_file": recorded_perf,
}
finally:
subprocess.run(
["tmux", "kill-session", "-t", session_name],
capture_output=True,
)
try:
os.unlink(status_file)
except OSError:
pass
# ---------------------------------------------------------------------------
# Aggregate statistics
# ---------------------------------------------------------------------------
def _avg(values):
vals = [v for v in values if v is not None]
return sum(vals) / len(vals) if vals else None
def _min(values):
vals = [v for v in values if v is not None]
return min(vals) if vals else None
def _max(values):
vals = [v for v in values if v is not None]
return max(vals) if vals else None
def aggregate(results: list) -> dict:
completed_results = [r for r in results if r["completed"]]
times = [r["elapsed_s"] for r in completed_results]
rates = [r["rate"] for r in completed_results]
matched = [r["matched"] for r in completed_results]
mems = [r["peak_mem_kb"] for r in completed_results]
cpus = [r["peak_cpu"] for r in completed_results]
completed = len(completed_results)
return {
"completed": completed,
"runs": len(results),
"avg_time": _avg(times),
"min_time": _min(times),
"max_time": _max(times),
"avg_rate": _avg(rates),
"min_rate": _min(rates),
"max_rate": _max(rates),
"avg_matched": _avg(matched),
"min_matched": _min(matched),
"max_matched": _max(matched),
"avg_mem": _avg(mems),
"min_mem": _min(mems),
"max_mem": _max(mems),
"avg_cpu": _avg(cpus),
"min_cpu": _min(cpus),
"max_cpu": _max(cpus),
}
# ---------------------------------------------------------------------------
# Formatting helpers
# ---------------------------------------------------------------------------
def _pct(baseline, value):
"""Return a +/-XX.X% string comparing *value* to *baseline*."""
if baseline is None or value is None or baseline == 0:
return ""
diff = (value - baseline) / abs(baseline) * 100
sign = "+" if diff >= 0 else ""
return f"{sign}{diff:.1f}%"
def _fmt_mem(kb):
if kb is None:
return None
return kb / 1024 # MB
def _fmt_optional(value, fmt):
if value is None:
return "N/A"
return fmt.format(value)
# ---------------------------------------------------------------------------
# Output
# ---------------------------------------------------------------------------
def print_human(
binary_label: str,
agg: dict,
num_items: int,
baseline: dict | None = None,
is_baseline: bool = False,
):
tag = " [baseline]" if is_baseline else ""
print(f"\n=== Results: {binary_label}{tag} ===")
print(f"Completed runs: {agg['completed']} / {agg['runs']}")
def cmp(key, baseline_key=None):
"""Format comparison vs baseline for a given aggregate key."""
bk = baseline_key or key
if baseline is None or is_baseline:
return ""
return " " + _pct(baseline.get(bk), agg.get(key))
# Items matched
avg_m = _fmt_optional(agg["avg_matched"], "{:.0f}")
min_m = _fmt_optional(agg["min_matched"], "{:.0f}")
max_m = _fmt_optional(agg["max_matched"], "{:.0f}")
print(
f"Average items matched: {avg_m} / {num_items}"
f" (min: {min_m}, max: {max_m})"
f"{cmp('avg_matched')}"
)
# Time
avg_t = _fmt_optional(agg["avg_time"], "{:.3f}s")
min_t = _fmt_optional(agg["min_time"], "{:.3f}s")
max_t = _fmt_optional(agg["max_time"], "{:.3f}s")
# Lower time is better, so flip sign for display
time_cmp = ""
if (
baseline
and not is_baseline
and baseline.get("avg_time")
and agg.get("avg_time")
):
diff = (
(agg["avg_time"] - baseline["avg_time"]) / abs(baseline["avg_time"]) * 100
)
sign = "+" if diff >= 0 else ""
time_cmp = f" {sign}{diff:.1f}%"
print(f"Average time: {avg_t} (min: {min_t}, max: {max_t}){time_cmp}")
# Rate
avg_r = _fmt_optional(agg["avg_rate"], "{:.0f}")
min_r = _fmt_optional(agg["min_rate"], "{:.0f}")
max_r = _fmt_optional(agg["max_rate"], "{:.0f}")
print(
f"Average items/second: {avg_r} (min: {min_r}, max: {max_r}){cmp('avg_rate')}"
)
# Memory
if agg["avg_mem"] is not None:
avg_mb = _fmt_mem(agg["avg_mem"])
min_mb = _fmt_mem(agg["min_mem"])
max_mb = _fmt_mem(agg["max_mem"])
print(
f"Average peak memory usage: {avg_mb:.1f} MB"
f" (min: {min_mb:.1f} MB, max: {max_mb:.1f} MB)"
f"{cmp('avg_mem')}"
)
# CPU
if agg["avg_cpu"] is not None:
avg_c = _fmt_optional(agg["avg_cpu"], "{:.1f}%")
min_c = _fmt_optional(agg["min_cpu"], "{:.1f}%")
max_c = _fmt_optional(agg["max_cpu"], "{:.1f}%")
print(
f"Average peak CPU usage: {avg_c}"
f" (min: {min_c}, max: {max_c})"
f"{cmp('avg_cpu')}"
)
def print_json_multi(binaries: list, aggregates: list, num_items: int, runs: int):
output = []
for binary, agg in zip(binaries, aggregates):
entry = {
"binary": binary,
"num_items": num_items,
"runs": runs,
"completed_runs": agg["completed"],
"items_matched": {
"avg": agg["avg_matched"],
"min": agg["min_matched"],
"max": agg["max_matched"],
},
"time_s": {
"avg": agg["avg_time"],
"min": agg["min_time"],
"max": agg["max_time"],
},
"items_per_second": {
"avg": agg["avg_rate"],
"min": agg["min_rate"],
"max": agg["max_rate"],
},
"peak_memory_kb": {
"avg": agg["avg_mem"],
"min": agg["min_mem"],
"max": agg["max_mem"],
},
"peak_cpu": {
"avg": agg["avg_cpu"],
"min": agg["min_cpu"],
"max": agg["max_cpu"],
},
}
output.append(entry)
print(json.dumps(output if len(output) > 1 else output[0]))
# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
def _perf_path_for(binary: str, explicit: str) -> str:
"""Return the perf output file path.
If *explicit* is a non-empty string use it directly; otherwise build an
auto-named path of the form ``perf-<basename>-<timestamp>.data`` in the
current working directory.
"""
if explicit:
return explicit
ts = int(time.time())
base = os.path.basename(binary).replace(" ", "_") or "sk"
return f"perf-{base}-{ts}.data"
def main():
import re # ensure import at top of main scope for run_once
binaries, opts, extra_args = parse_args(sys.argv[1:])
num_items = opts.num_items
query = opts.query
runs = opts.runs
warmup = opts.warmup
input_file = opts.file
generate_file = opts.generate_file
as_json = opts.json
record_perf = opts.perf is not None # True when -p/--perf was supplied
perf_explicit = opts.perf or "" # "" means auto-name
# ---- generate-file mode ------------------------------------------------
if generate_file:
print(f"Generating {num_items} items to {generate_file}...", file=sys.stderr)
generate_test_data(generate_file, num_items)
print(f"Generated {num_items} items successfully", file=sys.stderr)
return
# ---- prepare input data ------------------------------------------------
cleanup_input = False
if input_file:
if not os.path.isfile(input_file):
print(f"Error: Input file '{input_file}' not found", file=sys.stderr)
sys.exit(1)
tmp_file = input_file
with open(input_file) as fh:
num_items = sum(1 for _ in fh)
print(f"Using input file with {num_items} items", file=sys.stderr)
else:
fd, tmp_file = tempfile.mkstemp(prefix="skim_bench_input_")
os.close(fd)
cleanup_input = True
print("Generating test data...", file=sys.stderr)
generate_test_data(tmp_file, num_items)
try:
# ---- header ---------------------------------------------------------
binary_list = ", ".join(binaries)
print(f"=== Skim Ingestion + Matching Benchmark ===", file=sys.stderr)
print(
f"Binaries: {binary_list} | Items: {num_items} | "
f"Query: '{query}' | Warmup: {warmup} | Runs: {runs} (per binary)",
file=sys.stderr,
)
if input_file:
print(f"Input file: {input_file}", file=sys.stderr)
if extra_args:
print(f"Extra args: {' '.join(extra_args)}", file=sys.stderr)
if record_perf:
print("Perf recording: enabled (final measured run only)", file=sys.stderr)
# ---- warmup (results discarded) -------------------------------------
if warmup > 0:
print(f"\n=== Warmup ({warmup} run(s) per binary) ===", file=sys.stderr)
for bi, binary in enumerate(binaries):
for wu in range(1, warmup + 1):
print(
f" Warmup {wu}/{warmup} — {binary} ...",
file=sys.stderr,
)
run_once(
binary_path=binary,
query=query,
tmp_file=tmp_file,
num_items=num_items,
extra_args=extra_args,
run_index=wu,
session_suffix=f"warmup_b{bi}",
perf_output=None, # never record during warmup
)
# ---- run benchmark in round-robin -----------------------------------
# all_results[i] = list of per-run dicts for binaries[i]
all_results = [[] for _ in binaries]
# Determine which (run_num, bi) pairs get perf recording.
# We record only on the very last run for each binary to avoid
# overwriting data and to minimise measurement overhead.
perf_files: dict[int, str] = {} # bi -> path
if record_perf:
for bi, binary in enumerate(binaries):
perf_files[bi] = _perf_path_for(binary, perf_explicit if len(binaries) == 1 else "")
for run_num in range(1, runs + 1):
for bi, binary in enumerate(binaries):
label = f"[{os.path.basename(binary)}]"
if runs > 1 or len(binaries) > 1:
print(
f"\n=== Run {run_num}/{runs} — binary {bi + 1}/{len(binaries)}: {binary} ===",
file=sys.stderr,
)
# Attach perf only on the final run for this binary
this_perf = perf_files.get(bi) if run_num == runs else None
result = run_once(
binary_path=binary,
query=query,
tmp_file=tmp_file,
num_items=num_items,
extra_args=extra_args,
run_index=run_num,
session_suffix=f"b{bi}",
perf_output=this_perf,
)
all_results[bi].append(result)
if runs > 1 or len(binaries) > 1:
status = "COMPLETED" if result["completed"] else "TIMEOUT"
print(f"Status: {status}", file=sys.stderr)
print(
f"Items matched: {result['matched']} / {num_items}",
file=sys.stderr,
)
print(f"Total time: {result['elapsed_s']:.3f}s", file=sys.stderr)
print(f"Items/second: {result['rate']:.0f}", file=sys.stderr)
if result["peak_mem_kb"]:
print(
f"Peak memory usage: {result['peak_mem_kb'] / 1024:.1f} MB",
file=sys.stderr,
)
if result["peak_cpu"]:
print(
f"Peak CPU usage: {result['peak_cpu']:.1f}%",
file=sys.stderr,
)
if result.get("perf_file"):
print(
f"Perf data: {result['perf_file']}",
file=sys.stderr,
)
# ---- aggregate ------------------------------------------------------
aggregates = [aggregate(all_results[i]) for i in range(len(binaries))]
# ---- output ---------------------------------------------------------
if as_json:
print_json_multi(binaries, aggregates, num_items, runs)
else:
baseline_agg = aggregates[0]
for i, (binary, agg) in enumerate(zip(binaries, aggregates)):
print_human(
binary_label=binary,
agg=agg,
num_items=num_items,
baseline=baseline_agg if len(binaries) > 1 else None,
is_baseline=(i == 0),
)
# Summary comparison table when multiple binaries
if len(binaries) > 1:
print(f"\n=== Comparison Summary (vs baseline: {binaries[0]}) ===")
header = (
f"{'Binary':<40} {'Avg time':>12} {'Δ time':>10}"
f" {'Avg rate':>14} {'Δ rate':>10}"
f" {'Avg mem (MB)':>14} {'Δ mem':>10}"
f" {'Avg CPU (%)':>12} {'Δ CPU':>10}"
)
print(header)
print("-" * len(header))
for i, (binary, agg) in enumerate(zip(binaries, aggregates)):
t = (
f"{agg['avg_time']:.3f}s"
if agg["avg_time"] is not None
else "N/A"
)
r = (
f"{agg['avg_rate']:.0f}"
if agg["avg_rate"] is not None
else "N/A"
)
m = (
f"{agg['avg_mem'] / 1024:.1f}"
if agg["avg_mem"] is not None
else "N/A"
)
c = (
f"{agg['avg_cpu']:.1f}"
if agg["avg_cpu"] is not None
else "N/A"
)
if i == 0:
dt = "baseline"
dr = "baseline"
dm = "baseline"
dc = "baseline"
else:
dt = _pct(baseline_agg["avg_time"], agg["avg_time"])
dr = _pct(baseline_agg["avg_rate"], agg["avg_rate"])
dm = _pct(baseline_agg["avg_mem"], agg["avg_mem"])
dc = _pct(baseline_agg["avg_cpu"], agg["avg_cpu"])
name = os.path.basename(binary) if len(binary) > 40 else binary
print(
f"{name:<40} {t:>12} {dt:>10}"
f" {r:>14} {dr:>10}"
f" {m:>14} {dm:>10}"
f" {c:>12} {dc:>10}"
)
# ---- perf summary ---------------------------------------------------
if record_perf:
print("\n=== Perf recording output ===", file=sys.stderr)
for bi, binary in enumerate(binaries):
path = perf_files.get(bi, "")
if path and os.path.isfile(path):
print(f" [{binary}] perf data: {path}", file=sys.stderr)
else:
print(
f" [{binary}] perf data not found (perf may have failed)",
file=sys.stderr,
)
finally:
if cleanup_input:
try:
os.unlink(tmp_file)
except OSError:
pass
if __name__ == "__main__":
main()

1150
benches/cli.rs Normal file
View file

File diff suppressed because it is too large Load diff

57
flake.nix
View file

@ -3,25 +3,52 @@
inputs.nixpkgs.url = "https://channels.nixos.org/nixpkgs-unstable/nixexprs.tar.xz";
outputs = inputs: let
outputs =
inputs:
let
inherit (inputs.nixpkgs) lib;
systems = lib.systems.flakeExposed;
eachSystem = lib.genAttrs systems;
pkgsFor = system: import inputs.nixpkgs {
pkgsFor =
system:
import inputs.nixpkgs {
inherit system;
config.allowUnfreePredicate = pkg: builtins.elem (lib.getName pkg) [ "vagrant" ];
};
in {
devShells = eachSystem (system: let
in
{
devShells = eachSystem (
system:
let
pkgs = pkgsFor system;
# --- package groups -------------------------------------------------------
base = with pkgs; [ rustup just ];
tests = with pkgs; [ cargo-nextest cargo-insta cargo-llvm-cov tmux ];
utils = with pkgs; [ hyperfine cargo-edit cargo-public-api git-cliff cargo-dist ];
gungraun = with pkgs; [ valgrind libclang binutils ];
bench = with pkgs; [ uv python313Packages.matplotlib python313Packages.requests ];
vagrantDeps = with pkgs; [ vagrant rsync ];
base = with pkgs; [
rustup
just
];
tests = with pkgs; [
cargo-nextest
cargo-insta
cargo-llvm-cov
tmux
];
utils = with pkgs; [
hyperfine
cargo-edit
cargo-public-api
git-cliff
cargo-dist
];
gungraun = with pkgs; [
valgrind
libclang
binutils
];
vagrantDeps = with pkgs; [
vagrant
rsync
];
# --- shell hooks (only groups that need env vars) -------------------------
gungraunHook = ''
@ -33,16 +60,16 @@
'';
mkShell = packages: shellHook: pkgs.mkShellNoCC { inherit packages shellHook; };
in {
in
{
default = mkShell base "";
tests = mkShell (base ++ tests) "";
utils = mkShell (base ++ utils) "";
gungraun = mkShell (base ++ gungraun) gungraunHook;
bench = mkShell (base ++ bench) "";
vagrant = mkShell (base ++ vagrantDeps) vagrantHook;
full = mkShell (base ++ tests ++ utils ++ gungraun ++ bench ++ vagrantDeps)
(gungraunHook + vagrantHook);
});
full = mkShell (base ++ tests ++ utils ++ gungraun ++ vagrantDeps) (gungraunHook + vagrantHook);
}
);
formatter = eachSystem (system: (pkgsFor system).nixfmt);
};

227
src/helper/item_reader.rs
View file

@ -1,5 +1,6 @@
//! Helper utilities for converting input sources into skim item streams.
use std::collections::BTreeMap;
use std::error::Error;
use std::io::{BufRead, BufReader};
use std::process::{Child, Stdio};
@ -8,6 +9,11 @@ use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::thread;
use std::time::{Duration, Instant};
use crate::thread_pool::ThreadPool;
/// Size of the read buffer used by the parallel I/O reader thread.
const PARALLEL_READ_BUF_SIZE: usize = 256 * 1024;
use regex::Regex;
use crate::field::FieldRange;
@ -170,12 +176,26 @@ impl SkimItemReaderOption {
/// Reader for converting various input sources into streams of skim items
pub struct SkimItemReader {
option: Arc<SkimItemReaderOption>,
/// Thread pool used for chunk-processing jobs. Reader and matcher share
/// this pool so they compete for the same thread budget rather than each
/// spawning their own OS threads. Defaults to a private pool sized to the
/// number of logical CPUs; callers can replace it with a shared pool via
/// [`with_thread_pool`](Self::with_thread_pool) or
/// [`set_thread_pool`](Self::set_thread_pool).
thread_pool: Arc<ThreadPool>,
}
fn default_thread_pool() -> Arc<ThreadPool> {
let n = std::thread::available_parallelism().map_or(1, std::num::NonZero::get);
let (reader_threads, _) = crate::thread_pool::partition_threads(n);
Arc::new(ThreadPool::new(reader_threads))
}
impl Default for SkimItemReader {
fn default() -> Self {
Self {
option: Arc::new(Default::default()),
thread_pool: default_thread_pool(),
}
}
}
@ -186,6 +206,7 @@ impl SkimItemReader {
pub fn new(option: SkimItemReaderOption) -> Self {
Self {
option: Arc::new(option),
thread_pool: default_thread_pool(),
}
}
@ -195,6 +216,20 @@ impl SkimItemReader {
self.option = Arc::new(option);
self
}
/// Replaces the thread pool used for chunk-processing. Pass the matcher's
/// pool here so that reader and matcher share the same thread budget.
#[must_use]
pub fn with_thread_pool(mut self, pool: Arc<ThreadPool>) -> Self {
self.thread_pool = pool;
self
}
/// Like [`with_thread_pool`] but takes `&mut self` — useful when the pool
/// is only available after construction (e.g. injected from the app).
pub fn set_thread_pool(&mut self, pool: Arc<ThreadPool>) {
self.thread_pool = pool;
}
}
impl SkimItemReader {
@ -281,18 +316,187 @@ impl SkimItemReader {
}
}
/// helper: convert bufread into `SkimItemReceiver`
/// Parallel reader for the simple (no ANSI, no field transforms) case.
///
/// Pipeline:
///
/// 1. **I/O thread** (dedicated) — reads large byte chunks (~256 KB) from
/// `source`, splitting on line boundaries, and sends them tagged with
/// monotonic sequence numbers into a bounded channel.
/// 2. **Dispatcher thread** (dedicated, lightweight) — drains that channel
/// and submits one pool job per chunk. The bounded channel provides
/// natural back-pressure on the I/O thread when the pool is busy.
/// 3. **Pool jobs** — parse lines, validate UTF-8, and create
/// `DefaultSkimItem` + `Arc` per line. Because these jobs share the
/// same pool as the matcher, reader and matcher compete for the same
/// thread budget rather than over-subscribing available CPU cores.
/// 4. **Reorder thread** (dedicated) — collects `(seq, items)` from pool
/// jobs and emits them in sequence order so downstream index assignment
/// and `--tac` behaviour are correct.
fn raw_bufread(&self, source: impl BufRead + Send + 'static) -> SkimItemReceiver {
let (tx_item, rx_item): (SkimItemSender, SkimItemReceiver) = kanal::bounded(1024 * 1024);
let option = self.option.clone();
let pool = Arc::clone(&self.thread_pool);
let num_threads = pool.num_threads();
let (tx_chunks, rx_chunks) = kanal::bounded::<(usize, Vec<u8>)>(num_threads * 4);
let (tx_results, rx_results) = kanal::bounded::<(usize, Vec<Arc<dyn SkimItem>>)>(num_threads * 4);
let line_ending = option.line_ending;
// Stage 1: I/O thread.
Self::spawn_io_reader(source, tx_chunks, line_ending);
// Stage 2: dispatcher thread — bridges the bounded channel to the pool.
thread::spawn(move || {
Self::read_lines_into_items(source, &tx_item, &option, &[], &[]);
while let Ok((seq, chunk)) = rx_chunks.recv() {
let tx = tx_results.clone();
let opt = option.clone();
pool.spawn(move || {
let result = Self::process_chunk(seq, &chunk, &opt);
let _ = tx.send(result);
});
}
// rx_chunks closed → all chunks dispatched; tx_results dropped here
// so the reorder thread exits once the last pool job finishes.
});
// Stage 4: reorder thread.
Self::spawn_reorder_thread(rx_results, tx_item);
rx_item
}
/// Stage 1 of the parallel reader: reads large byte chunks from `source`,
/// splitting on line boundaries, and sends them to workers.
fn spawn_io_reader(
source: impl BufRead + Send + 'static,
tx_chunks: kanal::Sender<(usize, Vec<u8>)>,
line_ending: u8,
) {
thread::spawn(move || {
debug!("parallel reader: I/O thread start");
let mut source = source;
let mut leftover: Vec<u8> = Vec::new();
let mut seq = 0usize;
let mut read_buf = vec![0u8; PARALLEL_READ_BUF_SIZE];
loop {
let n = match std::io::Read::read(&mut source, &mut read_buf) {
Ok(0) => {
// EOF — flush any remaining leftover as the final chunk.
if !leftover.is_empty() {
let _ = tx_chunks.send((seq, std::mem::take(&mut leftover)));
}
break;
}
Ok(n) => n,
Err(ref e) if e.kind() == std::io::ErrorKind::Interrupted => continue,
Err(_) => {
// Flush any accumulated data before exiting on error.
if !leftover.is_empty() {
let _ = tx_chunks.send((seq, std::mem::take(&mut leftover)));
}
break;
}
};
// Combine leftover from previous iteration with fresh data.
let data = if leftover.is_empty() {
read_buf[..n].to_vec()
} else {
let mut combined = std::mem::take(&mut leftover);
combined.extend_from_slice(&read_buf[..n]);
combined
};
// Split at the last newline: everything up to it forms a
// complete-line chunk; the remainder carries over.
if let Some(last_nl) = memchr::memrchr(line_ending, &data) {
leftover = data[last_nl + 1..].to_vec();
let mut chunk = data;
chunk.truncate(last_nl + 1);
if tx_chunks.send((seq, chunk)).is_err() {
break;
}
seq += 1;
} else {
// No newline at all — accumulate for the next read.
leftover = data;
}
}
debug!("parallel reader: I/O thread stop (sent {seq} chunks)");
});
}
/// Parses a raw byte chunk into a tagged batch of items.
///
/// Shared by both the pool-based and dedicated-thread code paths.
fn process_chunk(seq: usize, chunk: &[u8], opt: &SkimItemReaderOption) -> (usize, Vec<Arc<dyn SkimItem>>) {
let mut items = Vec::new();
let line_ending = opt.line_ending;
// Chunks produced by the I/O thread end with the line-ending delimiter
// (except possibly the final leftover at EOF). `split()` would produce
// a spurious trailing empty segment in that case, so we trim the
// trailing delimiter first. After trimming, every segment — including
// empty ones — maps 1:1 to an input line.
let chunk_trimmed: &[u8] = if chunk.last() == Some(&line_ending) {
&chunk[..chunk.len() - 1]
} else {
chunk
};
for line_bytes in chunk_trimmed.split(|&b: &u8| b == line_ending) {
// Strip optional \r for \r\n endings.
let line_bytes: &[u8] = line_bytes.strip_suffix(b"\r").unwrap_or(line_bytes);
let Ok(line) = std::str::from_utf8(line_bytes) else {
continue;
};
// Use DefaultSkimItem::new to preserve ANSI-escape stripping
// behaviour even when --ansi is not set.
items.push(Arc::new(DefaultSkimItem::new(
line,
opt.use_ansi_color,
&opt.transform_fields,
&opt.matching_fields,
&opt.delimiter,
)) as Arc<dyn SkimItem>);
}
(seq, items)
}
/// Stage 3: receives item batches from workers and emits them through the
/// downstream channel in the original sequence order.
fn spawn_reorder_thread(rx_results: kanal::Receiver<(usize, Vec<Arc<dyn SkimItem>>)>, tx_item: SkimItemSender) {
thread::spawn(move || {
debug!("parallel reader: reorder thread start");
let mut expected = 0usize;
let mut pending: BTreeMap<usize, Vec<Arc<dyn SkimItem>>> = BTreeMap::new();
while let Ok((seq, items)) = rx_results.recv() {
pending.insert(seq, items);
// Flush consecutive batches starting from the expected seq.
while let Some(batch) = pending.remove(&expected) {
if tx_item.send(batch).is_err() {
return;
}
expected += 1;
}
}
// Drain anything left (shouldn't normally happen).
while let Some((&seq, _)) = pending.first_key_value() {
if pending.remove(&seq).is_some_and(|batch| tx_item.send(batch).is_err()) {
return;
}
}
debug!("parallel reader: reorder thread stop");
});
}
/// `components_to_stop` == 0 => all the threads have been stopped
/// return (`channel_for_receive_item`, `channel_to_stop_command`)
fn read_and_collect_from_command(
@ -359,29 +563,24 @@ impl SkimItemReader {
// busy waiting for the thread to start. (components_to_stop is added)
}
let started = Arc::new(AtomicBool::new(false));
let started_clone = started.clone();
let tx_interrupt_clone = tx_interrupt.clone();
let option = self.option.clone();
let transform_fields = option.transform_fields.clone();
let matching_fields = option.matching_fields.clone();
thread::spawn(move || {
debug!("collector: command collector start");
// Increment before submitting so components_to_stop is already non-zero
// when this function returns; no busy-wait needed.
components_to_stop.fetch_add(1, Ordering::SeqCst);
started_clone.store(true, Ordering::SeqCst); // notify parent that it is started
self.thread_pool.spawn(move || {
debug!("collector: command collector start");
Self::read_lines_into_items(source, &tx_item, &option, &transform_fields, &matching_fields);
let _ = tx_interrupt_clone.send(1); // ensure the waiting thread will exit
let _ = tx_interrupt_clone.send(1); // ensure the killer thread will exit
components_to_stop.fetch_sub(1, Ordering::SeqCst);
debug!("collector: command collector stop");
});
while !started.load(Ordering::SeqCst) {
// busy waiting for the thread to start. (components_to_stop is added)
}
(rx_item, tx_interrupt)
}
}
@ -394,6 +593,10 @@ impl CommandCollector for SkimItemReader {
) -> (SkimItemReceiver, crate::prelude::Sender<i32>) {
self.read_and_collect_from_command(components_to_stop, CollectorInput::Command(cmd.to_string()))
}
fn set_thread_pool(&mut self, pool: Arc<ThreadPool>) {
self.thread_pool = pool;
}
}
type CommandOutput = (Option<Child>, Box<dyn BufRead + Send>);

128
src/item.rs
View file

@ -118,10 +118,11 @@ pub struct MatchedItem {
pub item: Arc<dyn SkimItem>,
/// Raw match measurements
pub rank: Rank,
/// The tiebreak criteria used to derive sort order from `rank`
pub rank_builder: Arc<RankBuilder>,
/// Range of characters that matched the pattern
pub matched_range: Option<MatchRange>,
/// Sort key precomputed at construction time from `rank` and the tiebreak
/// criteria. Caching avoids recomputing it on every comparison during sort.
sort_key: [i32; 5],
}
impl std::fmt::Debug for MatchedItem {
@ -129,9 +130,8 @@ impl std::fmt::Debug for MatchedItem {
f.debug_struct("MatchedItem")
.field("item", &self.item.text())
.field("rank", &self.rank)
.field("sort_key", &self.rank.sort_key(self.rank_builder.criteria()))
.field("matched_range", &self.matched_range)
.finish()
.finish_non_exhaustive()
}
}
@ -151,6 +151,20 @@ impl Deref for MatchedItem {
}
impl MatchedItem {
/// Create a new `MatchedItem`, building the `sort_key` from the Rank and `RankBuilder`
pub fn new(
item: Arc<dyn SkimItem>,
rank: Rank,
matched_range: Option<MatchRange>,
rank_builder: &RankBuilder,
) -> Self {
Self {
item,
rank,
matched_range,
sort_key: rank.sort_key(rank_builder.criteria()),
}
}
/// Merge two sorted `Vec<MatchedItem>` lists into one, preserving sort order by rank.
///
/// Both input lists must already be sorted by the same tiebreak criteria (ascending).
@ -212,36 +226,109 @@ impl MatchedItem {
/// Merge `incoming` into an already-sorted `existing` vector in-place.
///
/// This function chooses between two strategies:
/// - If `incoming` is small (few items), insert them one-by-one using binary
/// search to find the insertion point. This is O(m log n) for m incoming
/// items and is faster when m << n.
/// - Otherwise, fall back to the linear two-way merge which is O(n+m).
/// - If `incoming` is small (≤ 256 items), insert them one-by-one using
/// binary search to find the insertion point. Each insert is O(n) due
/// to element shifting, giving O(m·n) overall, but the constant factor
/// is small for tiny m and avoids any extra allocation.
/// - Otherwise, perform a backwards in-place merge that writes the result
/// directly into `existing`'s buffer (after a single `reserve`). This
/// is O(n+m) time with **zero additional heap allocation** beyond the
/// amortised `Vec::reserve`.
///
/// `existing` must be sorted according to the same ordering used by
/// `MatchedItem::cmp`.
pub fn merge_into_sorted(existing: &mut Vec<MatchedItem>, incoming: Vec<MatchedItem>) {
// Heuristic threshold: for small incoming batches, prefer binary-insert.
// This avoids allocating a new vector and copying the entire existing
// list when we only need to insert a few new items.
const SMALL_INSERT_THRESHOLD: usize = 256;
if incoming.is_empty() {
return;
}
// When existing is empty, extend preserves any pre-allocated capacity
// (e.g. a caller that did `Vec::with_capacity(total)` before a fold).
if existing.is_empty() {
existing.extend(incoming);
return;
}
// Fast path: all existing ≤ first incoming — just append.
#[allow(clippy::missing_panics_doc)]
if existing.last().unwrap() <= incoming.first().unwrap() {
existing.extend(incoming);
return;
}
if incoming.len() <= SMALL_INSERT_THRESHOLD {
// Insert each incoming item into the existing sorted vector.
// For small m this is typically faster than allocating a new
// buffer and performing a full linear merge.
for item in incoming {
let pos = existing.binary_search_by(|e| e.cmp(&item)).unwrap_or_else(|p| p);
existing.insert(pos, item);
}
} else {
// For larger incoming batches, perform the linear two-way merge
// which is O(n+m) and avoids the O(n*m) cost of repeated inserts.
let old = std::mem::take(existing);
*existing = MatchedItem::sorted_merge(old, incoming);
Self::merge_backwards(existing, incoming);
}
}
/// Merges `incoming` into `existing` in-place using a right-to-left merge.
///
/// Both inputs must already be sorted. After `existing.reserve(b_len)`,
/// the buffer has room for all elements. We then merge from the rightmost
/// end of each run, writing the larger element at the write cursor which
/// starts at `new_len - 1` and moves left.
///
/// **Key invariant**: the write position is always strictly greater than
/// the read position in `existing` while both runs have remaining elements
/// (because `write - ai == remaining B elements > 0`), so
/// `copy_nonoverlapping` never aliases. Each element is moved exactly
/// once.
///
/// # Safety (internal)
///
/// Uses `unsafe` for raw-pointer moves. `MatchedItem::cmp` compares plain
/// integer fields and cannot panic, so no element is leaked or
/// double-dropped. `incoming`'s backing allocation is freed with length 0
/// after all its elements have been moved out.
fn merge_backwards(existing: &mut Vec<MatchedItem>, incoming: Vec<MatchedItem>) {
let a_len = existing.len();
let b_len = incoming.len();
let new_len = a_len + b_len;
existing.reserve(b_len);
// Decompose `incoming` so we can move elements out via raw pointers
// and free the allocation separately.
let (b_ptr, b_cap) = {
let mut v = std::mem::ManuallyDrop::new(incoming);
(v.as_mut_ptr(), v.capacity())
};
// SAFETY: see doc-comment above for the aliasing / move proof.
unsafe {
let a_ptr = existing.as_mut_ptr();
let mut write = new_len;
let mut ai = a_len;
let mut bi = b_len;
while ai > 0 && bi > 0 {
write -= 1;
if *a_ptr.add(ai - 1) >= *b_ptr.add(bi - 1) {
ai -= 1;
std::ptr::copy_nonoverlapping(a_ptr.add(ai), a_ptr.add(write), 1);
} else {
bi -= 1;
std::ptr::copy_nonoverlapping(b_ptr.add(bi), a_ptr.add(write), 1);
}
}
// Remaining B elements go to the front of existing.
// (Remaining A elements at 0..ai are already in their final positions.)
if bi > 0 {
std::ptr::copy_nonoverlapping(b_ptr, a_ptr, bi);
}
existing.set_len(new_len);
// Free incoming's backing allocation; all elements were moved out.
drop(Vec::from_raw_parts(b_ptr, 0, b_cap));
}
}
}
@ -272,8 +359,9 @@ impl PartialOrd for MatchedItem {
impl Ord for MatchedItem {
fn cmp(&self, other: &Self) -> CmpOrd {
let criteria = self.rank_builder.criteria();
self.rank.sort_key(criteria).cmp(&other.rank.sort_key(criteria))
self.sort_key
.cmp(&other.sort_key)
.then_with(|| self.rank.index.cmp(&other.rank.index))
}
}

1
src/lib.rs
View file

@ -67,6 +67,7 @@ mod skim;
mod skim_item;
pub mod spinlock;
pub mod theme;
pub mod thread_pool;
#[cfg(unix)]
pub mod tmux;
pub mod tui;

246
src/matcher.rs
View file

@ -1,16 +1,84 @@
//! This module contains the matching coordinator
use rayon::ThreadPool;
use crate::thread_pool::{self, ThreadPool};
use crate::tui::item_list::{MergeStrategy, ProcessedItems};
use std::rc::Rc;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use rayon::prelude::*;
use crate::engine::normalized::NormalizedEngineFactory;
use crate::engine::split::SplitMatchEngineFactory;
use crate::item::{ItemPool, MatchedItem, RankBuilder};
use crate::prelude::{AndOrEngineFactory, ExactOrFuzzyEngineFactory, RegexEngineFactory};
use crate::{CaseMatching, MatchEngineFactory, SkimOptions};
use crate::spinlock::SpinLock;
use crate::{CaseMatching, MatchEngineFactory, SkimItem, SkimOptions};
/// Merges per-worker match results and writes them into `processed_items`.
///
/// When `no_sort` is false, concatenates the pre-sorted worker results into a
/// single contiguous `Vec` and calls `sort()`. Rust's stable sort (driftsort
/// since 1.81, a `TimSort` variant before that) detects the k pre-sorted runs
/// and merges them in O(n log k) on contiguous memory — benchmarking shows
/// this consistently outperforms tree-based or fold-based merge strategies
/// due to driftsort's cache-friendly single-buffer merge passes.
///
/// When `no_sort` is true, the worker results are simply flattened.
///
/// Signals `needs_render` after writing so the UI picks up the new data.
fn merge_worker_results(
worker_results: Vec<Vec<MatchedItem>>,
no_sort: bool,
processed_items: &SpinLock<Option<ProcessedItems>>,
merge_strategy: MergeStrategy,
needs_render: &AtomicBool,
) {
let total_len: usize = worker_results.iter().map(Vec::len).sum();
let mut items = Vec::with_capacity(total_len);
for chunk in worker_results {
items.extend(chunk);
}
// Each worker's sub-list is already sorted by `prepare`, so the
// concatenated Vec consists of k sorted runs. Rust's stable sort
// (driftsort since 1.81, a TimSort variant before that) detects
// pre-existing runs and merges them in O(n log k) for k workers,
// all on contiguous memory with a single auxiliary buffer.
if !no_sort {
items.sort();
}
trace!("matcher stop, total matched: {}", items.len());
// Single lock, single write into processed_items.
let mut guard = processed_items.lock();
if matches!(merge_strategy, MergeStrategy::Replace) {
*guard = Some(ProcessedItems {
items,
merge: MergeStrategy::Replace,
});
drop(guard);
needs_render.store(true, Ordering::Relaxed);
return;
}
match &mut *guard {
Some(existing) => {
if no_sort {
existing.items.extend(items);
} else {
// Both sides are fully sorted — one O(n+m) merge.
MatchedItem::merge_into_sorted(&mut existing.items, items);
}
}
None => {
*guard = Some(ProcessedItems {
items,
merge: merge_strategy,
});
}
}
// Guard is dropped here, releasing the lock before we signal the render flag.
needs_render.store(true, Ordering::Relaxed);
}
//==============================================================================
/// Control handle for a running matcher operation.
@ -183,18 +251,23 @@ impl Matcher {
/// Runs the matcher on items from the pool in a background thread.
///
/// The callback is invoked when matching is complete with the matched items.
/// Returns a `MatcherControl` that can be used to monitor progress or stop the matcher.
pub fn run<C>(
/// When matching completes, the coordinator merges results directly into
/// `processed_items` according to `merge_strategy`, then signals
/// `needs_render` so the UI picks up the new data on its next tick.
///
/// Returns a `MatcherControl` that can be used to monitor progress or
/// stop the matcher.
#[allow(clippy::too_many_arguments)]
pub(crate) fn run(
&self,
query: &str,
item_pool: &Arc<ItemPool>,
thread_pool: &ThreadPool,
callback: C,
) -> MatcherControl
where
C: Fn(Vec<MatchedItem>) + Send + 'static,
{
thread_pool: &Arc<ThreadPool>,
processed_items: Arc<SpinLock<Option<ProcessedItems>>>,
merge_strategy: MergeStrategy,
no_sort: bool,
needs_render: Arc<AtomicBool>,
) -> MatcherControl {
let matcher_engine = self.engine_factory.create_engine_with_case(query, self.case_matching);
debug!("engine: {matcher_engine}");
let stopped = Arc::new(AtomicBool::new(false));
@ -216,85 +289,108 @@ impl Matcher {
let total = items.len();
trace!("matcher start, total: {total}");
thread_pool.spawn(move || {
// Process items in parallel using chunk-based accounting to minimize
// atomic contention. Each rayon work unit processes a chunk of items,
// updating the shared `processed` and `matched` counters only once per
// chunk instead of once per item. The interrupt flag is also checked
// only once per chunk to amortize the atomic load.
// The coordinator runs on a dedicated OS thread so it does not occupy
// a pool slot while waiting for workers. All pool threads are
// therefore available for the parallel matching work.
let num_workers = thread_pool.num_threads();
let pool_for_work = Arc::clone(thread_pool);
std::thread::spawn(move || {
// Process items in parallel using a shared work queue. Each worker
// thread atomically grabs the next available chunk, processes it,
// and immediately merges its partial results. This means threads
// that finish early automatically pick up more work, providing
// natural load balancing.
//
// `with_min_len` ensures rayon doesn't split work into chunks smaller
// than CHUNK_SIZE, keeping the overhead of the parallel iterator low
// relative to the actual matching work.
// The chunk size controls the granularity of work distribution and
// the frequency of atomic counter updates / interrupt checks.
const CHUNK_SIZE: usize = 512;
let matched_items: Vec<MatchedItem> = items
.into_par_iter()
.with_min_len(CHUNK_SIZE)
.enumerate()
.fold(
|| (Vec::new(), 0usize, 0usize), // (local_matches, local_processed, local_matched)
|(mut local_matches, mut local_processed, mut local_matched), (index, item)| {
// Check interrupt once at the start of each chunk boundary.
// The fold processes items sequentially within each rayon work unit,
// so checking every CHUNK_SIZE items amortizes the atomic load.
if local_processed % CHUNK_SIZE == 0 && interrupt.load(Ordering::Relaxed) {
return (local_matches, local_processed, local_matched);
// Convert items into an Arc slice so all workers can share them.
let shared_items: Arc<[Arc<dyn SkimItem>]> = items.into();
// Clones for the process_chunk closure.
let matcher_engine: Arc<dyn crate::MatchEngine> = Arc::from(matcher_engine);
let interrupt_for_work = Arc::clone(&interrupt);
let processed_for_work = Arc::clone(&processed);
let matched_for_work = Arc::clone(&matched);
let rank_builder_for_work = Arc::clone(&rank_builder);
thread_pool::parallel_work_queue(
&pool_for_work,
num_workers,
&shared_items,
CHUNK_SIZE,
// identity – seed value for each worker's local accumulator
Vec::<MatchedItem>::new,
// process_chunk – called for each chunk; returns a Vec of matches
move |chunk_start, chunk: &[Arc<dyn crate::SkimItem>]| {
// Check interrupt before processing this chunk.
if interrupt_for_work.load(Ordering::Relaxed) {
return Vec::new();
}
local_processed += 1;
let mut local_matches = Vec::new();
let mut chunk_matched: usize = 0;
for (i, item) in chunk.iter().enumerate() {
if let Some(match_result) = matcher_engine.match_item(item.as_ref()) {
local_matched += 1;
chunk_matched += 1;
let mut rank = match_result.rank;
rank.index = i32::try_from(index + start).unwrap_or(i32::MAX);
local_matches.push(MatchedItem {
item,
let index = chunk_start + i + start;
rank.index = i32::try_from(index).unwrap_or(i32::MAX);
local_matches.push(MatchedItem::new(
Arc::clone(item),
rank,
rank_builder: rank_builder.clone(),
matched_range: Some(match_result.matched_range),
});
}
// Flush counters periodically so the UI sees progress updates.
if local_processed % CHUNK_SIZE == 0 {
processed.fetch_add(CHUNK_SIZE, Ordering::Relaxed);
if local_matched > 0 {
matched.fetch_add(local_matched, Ordering::Relaxed);
local_matched = 0;
Some(match_result.matched_range),
&rank_builder_for_work,
));
}
}
(local_matches, local_processed, local_matched)
},
)
.map(|(local_matches, local_processed, local_matched)| {
// Flush any remaining counts that didn't hit a chunk boundary.
let remainder = local_processed % CHUNK_SIZE;
if remainder > 0 {
processed.fetch_add(remainder, Ordering::Relaxed);
}
if local_matched > 0 {
matched.fetch_add(local_matched, Ordering::Relaxed);
// Flush counters for this chunk so the UI sees progress.
processed_for_work.fetch_add(chunk.len(), Ordering::Relaxed);
if chunk_matched > 0 {
matched_for_work.fetch_add(chunk_matched, Ordering::Relaxed);
}
local_matches
})
.reduce(Vec::new, |mut a, mut b| {
// Merge per-thread result vectors. Always extend the larger one
// to avoid unnecessary reallocations.
if a.len() >= b.len() {
a.extend(b);
a
},
// reduce – accumulate chunk matches into the worker-local Vec.
// No sorting here — that would be O(m²/chunk_size) per worker.
|acc: &mut Vec<MatchedItem>, mut partial: Vec<MatchedItem>| {
if acc.len() >= partial.len() {
acc.extend(partial);
} else {
b.extend(a);
b
partial.append(acc);
*acc = partial;
}
},
// prepare – sort each worker's accumulator **on the worker
// thread** so that sorting runs in parallel across all workers.
// A single O((m/k)·log(m/k)) sort per worker is far cheaper
// than sorting during reduce.
// sort_unstable is used here because the worker's accumulator
// has no pre-existing sorted runs (items were appended in
// chunk order), so driftsort's run-detection overhead is pure
// cost. The final merge uses sort() so that driftsort can
// exploit the k sorted runs produced by the workers.
move |acc: &mut Vec<MatchedItem>| {
if !no_sort {
acc.sort_unstable();
}
},
// merge – concat pre-sorted worker results and sort().
// Rust's stable sort detects the k sorted runs and merges
// them in O(n log k), then writes into processed_items.
|worker_results: Vec<Vec<MatchedItem>>| {
if interrupt.load(Ordering::SeqCst) {
return;
}
});
if !interrupt.load(Ordering::SeqCst) {
trace!("matcher stop, total matched: {}", matched_items.len());
callback(matched_items);
}
merge_worker_results(worker_results, no_sort, &processed_items, merge_strategy, &needs_render);
},
);
stopped.store(true, Ordering::Relaxed);
});

16
src/reader.rs
View file

@ -5,6 +5,7 @@ use crate::item::ItemPool;
use crate::options::SkimOptions;
use crate::prelude::{Sender, SkimItemReader};
use crate::spinlock::SpinLock;
use crate::thread_pool::ThreadPool;
use crate::{SkimItem, SkimItemReceiver};
use std::cell::RefCell;
use std::rc::Rc;
@ -25,6 +26,12 @@ pub trait CommandCollector {
cmd: &str,
components_to_stop: Arc<AtomicUsize>,
) -> (SkimItemReceiver, crate::prelude::Sender<i32>);
/// Provides a shared thread pool so that chunk-processing work submitted
/// by this collector competes for the same threads as the matcher rather
/// than spawning additional OS threads. The default implementation is a
/// no-op; collectors that support pool-based I/O should override it.
fn set_thread_pool(&mut self, _pool: Arc<ThreadPool>) {}
}
/// Handle for controlling a running reader
@ -94,6 +101,13 @@ impl Reader {
self
}
/// Forwards a shared thread pool to the underlying [`CommandCollector`] so
/// that I/O work shares the matcher's thread budget instead of spawning
/// separate OS threads.
pub fn set_thread_pool(&mut self, pool: Arc<ThreadPool>) {
self.cmd_collector.borrow_mut().set_thread_pool(pool);
}
/// Starts the reader and returns a control handle
pub fn run(&mut self, app_tx: Sender<Vec<Arc<dyn SkimItem>>>, cmd: &str) -> ReaderControl {
let components_to_stop: Arc<AtomicUsize> = Arc::new(AtomicUsize::new(0));
@ -182,7 +196,7 @@ where
callback(items);
}
Ok(None) => {
std::thread::sleep(std::time::Duration::from_millis(10));
std::thread::sleep(std::time::Duration::from_millis(1));
}
Err(_) => {
break;

7
src/skim.rs
View file

@ -140,7 +140,7 @@ where
// application state
// Initialize theme from options
let theme = Arc::new(crate::theme::ColorTheme::init_from_options(&options));
let reader = Reader::from_options(&options).source(source);
let mut reader = Reader::from_options(&options).source(source);
let default_command = String::from(match env::var("SKIM_DEFAULT_COMMAND").as_deref() {
Err(_) | Ok("") => crate::SKIM_DEFAULT_COMMAND,
Ok(v) => v,
@ -149,6 +149,11 @@ where
let app = App::from_options(options, theme.clone(), cmd.clone());
// Give the reader its own dedicated pool (⌈N/3⌉ threads) so it never
// competes with the matcher's pool (⌊2N/3⌋ threads) for the same
// worker threads.
reader.set_thread_pool(Arc::clone(&app.reader_pool));
//------------------------------------------------------------------------------
// reader
// In interactive mode, expand all placeholders ({}, {q}, etc) with initial query (empty or from --query)

13
src/spinlock.rs
View file

@ -51,9 +51,11 @@ impl<T: ?Sized> SpinLock<T> {
pub fn lock(&self) -> SpinLockGuard<'_, T> {
while self
.locked
.compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
.compare_exchange_weak(false, true, Ordering::Acquire, Ordering::Relaxed)
.is_err()
{}
{
core::hint::spin_loop();
}
SpinLockGuard::new(self)
}
}
@ -75,12 +77,7 @@ impl<T: ?Sized> DerefMut for SpinLockGuard<'_, T> {
impl<T: ?Sized> Drop for SpinLockGuard<'_, T> {
#[inline]
fn drop(&mut self) {
while self
.__lock
.locked
.compare_exchange(true, false, Ordering::SeqCst, Ordering::SeqCst)
.is_err()
{}
self.__lock.locked.store(false, Ordering::Release);
}
}

659
src/thread_pool.rs Normal file
View file

@ -0,0 +1,659 @@
//! A lightweight thread pool with a shared work queue.
//!
//! Worker threads pick up the next available job as soon as they finish their
//! current one, giving natural load balancing without work-stealing.
use std::cell::UnsafeCell;
use std::collections::VecDeque;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Condvar, Mutex};
use std::thread;
// ---------------------------------------------------------------------------
// Thread-count partitioning
// ---------------------------------------------------------------------------
/// Splits `n` logical threads between the reader pipeline and the matcher.
///
/// Returns `(reader, matcher)` where:
/// - `reader` = ⌈n / 3⌉, minimum 1
/// - `matcher` = ⌊2n / 3⌋, minimum 1
///
/// On single-core machines (`n = 1`) both values are 1 so neither subsystem
/// starves; the OS time-slices between the two small pools as usual.
#[must_use]
pub fn partition_threads(n: usize) -> (usize, usize) {
let reader = n.div_ceil(3); // ⌈n/3⌉
let matcher = (2 * n) / 3; // ⌊2n/3⌋
(reader.max(1), matcher.max(1))
}
// ---------------------------------------------------------------------------
// Job type
// ---------------------------------------------------------------------------
type Job = Box<dyn FnOnce() + Send + 'static>;
// ---------------------------------------------------------------------------
// Shared state between the pool handle and workers
// ---------------------------------------------------------------------------
struct SharedState {
queue: Mutex<QueueInner>,
/// Wakes workers when a new job is enqueued or shutdown is requested.
job_available: Condvar,
}
struct QueueInner {
jobs: VecDeque<Job>,
shutdown: bool,
}
// ---------------------------------------------------------------------------
// ThreadPool
// ---------------------------------------------------------------------------
/// A simple thread pool backed by a FIFO work queue.
///
/// Each worker thread blocks on a shared condvar and picks up the next
/// available job as soon as it becomes idle. This means that when a thread
/// finishes a job (including any reduce/merge work that was part of that job)
/// it immediately checks for the next queued job without any extra
/// coordination.
pub struct ThreadPool {
shared: Arc<SharedState>,
workers: Vec<thread::JoinHandle<()>>,
num_threads: usize,
}
impl ThreadPool {
/// Creates a new pool with `num_threads` worker threads.
///
/// # Panics
///
/// Panics if `num_threads` is 0.
#[must_use]
pub fn new(num_threads: usize) -> Self {
assert!(num_threads > 0, "ThreadPool requires at least 1 thread");
let shared = Arc::new(SharedState {
queue: Mutex::new(QueueInner {
jobs: VecDeque::new(),
shutdown: false,
}),
job_available: Condvar::new(),
});
let mut workers = Vec::with_capacity(num_threads);
for _ in 0..num_threads {
let worker_shared = Arc::clone(&shared);
workers.push(thread::spawn(move || worker_loop(&worker_shared)));
}
Self {
shared,
workers,
num_threads,
}
}
/// Returns the number of worker threads in the pool.
#[inline]
#[must_use]
pub fn num_threads(&self) -> usize {
self.num_threads
}
/// Submits a closure to be executed by the next available worker.
///
/// The lock is dropped *before* notifying the condvar so that the woken
/// worker can acquire it immediately instead of blocking on the notifier.
///
/// # Panics
///
/// Panics if the internal job-queue mutex is poisoned.
pub fn spawn<F>(&self, f: F)
where
F: FnOnce() + Send + 'static,
{
{
let mut queue = self.shared.queue.lock().unwrap();
queue.jobs.push_back(Box::new(f));
} // lock dropped before notify
self.shared.job_available.notify_one();
}
/// Submits multiple closures in a single lock acquisition, then wakes all
/// workers. This is more efficient than calling [`spawn`](Self::spawn) in
/// a loop when you have several jobs ready at once.
///
/// # Panics
///
/// Panics if the internal job-queue mutex is poisoned.
pub fn spawn_batch<I>(&self, jobs: I)
where
I: IntoIterator<Item = Box<dyn FnOnce() + Send + 'static>>,
{
{
let mut queue = self.shared.queue.lock().unwrap();
for job in jobs {
queue.jobs.push_back(job);
}
} // lock dropped before notify
self.shared.job_available.notify_all();
}
}
impl Drop for ThreadPool {
fn drop(&mut self) {
// Signal shutdown.
{
let mut queue = self.shared.queue.lock().unwrap();
queue.shutdown = true;
}
self.shared.job_available.notify_all();
// Join all workers (ignore panics from individual threads).
for handle in self.workers.drain(..) {
let _ = handle.join();
}
}
}
// ---------------------------------------------------------------------------
// Worker loop
// ---------------------------------------------------------------------------
fn worker_loop(shared: &SharedState) {
loop {
let next_job = {
let mut queue = shared.queue.lock().unwrap();
loop {
if let Some(ready) = queue.jobs.pop_front() {
break Some(ready);
}
if queue.shutdown {
break None;
}
queue = shared.job_available.wait(queue).unwrap();
}
};
match next_job {
Some(runnable) => runnable(),
None => return, // shutdown
}
}
}
// ---------------------------------------------------------------------------
// Parallel work-queue helpers used by the matcher
// ---------------------------------------------------------------------------
// Cache-line–aligned result slot
// ---------------------------------------------------------------------------
/// A single worker's result slot, padded to a full cache line so that
/// concurrent writes to adjacent slots by different cores don't cause
/// false sharing.
///
/// Alignment is platform-dependant, this is taken from <https://docs.rs/crossbeam-utils/0.8.21/src/crossbeam_utils/cache_padded.rs.html>
///
// Starting from Intel's Sandy Bridge, spatial prefetcher is now pulling pairs of 64-byte cache
// lines at a time, so we have to align to 128 bytes rather than 64.
#[cfg_attr(
any(
target_arch = "x86_64",
target_arch = "aarch64",
target_arch = "arm64ec",
target_arch = "powerpc64",
),
repr(align(128))
)]
// arm, mips, mips64, sparc, and hexagon have 32-byte cache line size.
#[cfg_attr(
any(
target_arch = "arm",
target_arch = "mips",
target_arch = "mips32r6",
target_arch = "mips64",
target_arch = "mips64r6",
target_arch = "sparc",
target_arch = "hexagon",
),
repr(align(32))
)]
// m68k has 16-byte cache line size.
#[cfg_attr(target_arch = "m68k", repr(align(16)))]
// s390x has 256-byte cache line size.
#[cfg_attr(target_arch = "s390x", repr(align(256)))]
// x86, wasm, riscv, and sparc64 have 64-byte cache line size.
// All others are assumed to have 64-byte cache line size.
#[cfg_attr(
not(any(
target_arch = "x86_64",
target_arch = "aarch64",
target_arch = "arm64ec",
target_arch = "powerpc64",
target_arch = "arm",
target_arch = "mips",
target_arch = "mips32r6",
target_arch = "mips64",
target_arch = "mips64r6",
target_arch = "sparc",
target_arch = "hexagon",
target_arch = "m68k",
target_arch = "s390x",
)),
repr(align(64))
)]
struct Slot<R> {
value: UnsafeCell<Option<R>>,
}
// SAFETY: each slot is written by exactly one worker (unique `worker_id`)
// and read by the coordinator only after the barrier guarantees all writes
// are visible. No two threads ever access the same slot concurrently.
unsafe impl<R: Send> Send for Slot<R> {}
unsafe impl<R: Send> Sync for Slot<R> {}
impl<R> Slot<R> {
fn new() -> Self {
Self {
value: UnsafeCell::new(None),
}
}
}
// ---------------------------------------------------------------------------
/// Processes `items` in parallel across `num_workers` threads from the given
/// pool, then hands the per-worker results to `merge`.
///
/// The work is split into chunks of `chunk_size`. Each worker thread
/// repeatedly grabs the next available chunk (via an atomic counter), runs
/// `process_chunk` on it, and folds the partial result into a *local*
/// accumulator using `reduce`. When all chunks are consumed, each worker
/// calls `prepare` on its local accumulator (e.g. to sort it) — this step
/// runs **in parallel** across all workers — and then writes the prepared
/// result into its slot. The coordinator collects every worker's result and
/// passes them all to `merge` in a single call.
///
/// Because each worker picks up the *next* chunk as soon as it finishes the
/// previous one, faster threads naturally do more work without any explicit
/// work-stealing.
///
/// # Parameters
///
/// * `pool` – thread pool whose workers will execute the chunks.
/// * `num_workers` – how many workers to dispatch (capped to pool size internally by caller).
/// * `items` – the data to process; shared read-only across workers via `Arc`.
/// * `chunk_size` – number of items per chunk.
/// * `identity` – the identity/seed value for per-worker local accumulators (called once per worker).
/// * `process_chunk` – `(chunk_start_index, &[T]) -> R` – processes one chunk.
/// * `reduce` – folds a per-chunk result into a worker-local accumulator (`&mut acc, partial`).
/// * `prepare` – called on each worker's finished accumulator **on the worker thread** (runs in parallel). Use this for expensive per-worker work like sorting.
/// * `merge` – called once on the coordinator with all per-worker results.
#[allow(clippy::too_many_arguments)]
pub fn parallel_work_queue<T, R, P, M, I, W, G>(
pool: &ThreadPool,
num_workers: usize,
items: &Arc<[T]>,
chunk_size: usize,
identity: I,
process_chunk: P,
reduce: M,
prepare: W,
merge: G,
) where
T: Send + Sync + 'static,
R: Send + 'static,
P: Fn(usize, &[T]) -> R + Send + Sync + 'static,
M: Fn(&mut R, R) + Send + Sync + 'static,
I: Fn() -> R + Send + Sync + 'static,
W: Fn(&mut R) + Send + Sync + 'static,
G: FnOnce(Vec<R>),
{
let total = items.len();
if total == 0 {
merge(Vec::new());
return;
}
let num_chunks = total.div_ceil(chunk_size);
// Shared atomic counter – workers fetch-add to grab the next chunk index.
let next_chunk = Arc::new(AtomicUsize::new(0));
// Contiguous, cache-line-aligned per-worker result slots. Each worker
// writes only to its own slot (lock-free via UnsafeCell); the
// coordinator reads after the AtomicCounter barrier.
let slots: Arc<Vec<Slot<R>>> = Arc::new((0..num_workers).map(|_| Slot::new()).collect());
// Barrier: we wait until all workers have finished.
let remaining = Arc::new(AtomicCounter::new(num_workers));
let process_chunk = Arc::new(process_chunk);
let reduce = Arc::new(reduce);
let prepare = Arc::new(prepare);
let identity = Arc::new(identity);
// Build all jobs up-front and submit them in a single batch to minimise
// lock acquisitions on the work queue.
let jobs: Vec<Box<dyn FnOnce() + Send + 'static>> = (0..num_workers)
.map(|worker_id| {
let w_items = Arc::clone(items);
let w_next_chunk = Arc::clone(&next_chunk);
let w_slots: Arc<Vec<Slot<R>>> = Arc::clone(&slots);
let w_remaining = Arc::clone(&remaining);
let w_process_chunk = Arc::clone(&process_chunk);
let w_reduce = Arc::clone(&reduce);
let w_prepare = Arc::clone(&prepare);
let w_identity = Arc::clone(&identity);
let job: Box<dyn FnOnce() + Send + 'static> = Box::new(move || {
// Scope all Arc-holding work so clones are dropped before we
// signal completion. This lets the coordinator safely unwrap
// the outer Arcs.
let local_acc = {
let mut local_acc = w_identity();
loop {
let chunk_idx = w_next_chunk.fetch_add(1, Ordering::Relaxed);
if chunk_idx >= num_chunks {
break;
}
let start = chunk_idx * chunk_size;
let end = total.min(start + chunk_size);
let partial = w_process_chunk(start, &w_items[start..end]);
w_reduce(&mut local_acc, partial);
}
// Run prepare (e.g. sort) while still on the worker thread
// so that this work happens in parallel across workers.
w_prepare(&mut local_acc);
// w_items, w_next_chunk, w_process_chunk, w_reduce,
// w_prepare, w_identity are dropped when this block ends.
local_acc
};
// Write into our own slot – lock-free, no contention.
// SAFETY: each worker_id is unique; no other thread writes to
// this slot, and the coordinator reads only after the barrier.
unsafe { *w_slots[worker_id].value.get() = Some(local_acc) };
// Drop the slots Arc *before* signalling completion.
// The coordinator calls Arc::into_inner(slots) after wait_for_zero
// returns; that requires the strong count to be exactly 1. Without
// this explicit drop, w_slots would still be alive in the closure
// frame when dec_and_notify wakes the coordinator, causing
// Arc::into_inner to spuriously return None and silently lose results.
drop(w_slots);
// Signal completion.
w_remaining.dec_and_notify();
});
job
})
.collect();
pool.spawn_batch(jobs);
// Block until all workers are done.
remaining.wait_for_zero();
// Collect per-worker results and hand them to `merge` in one call.
// Workers dropped their `w_slots` Arc clone explicitly before signalling
// completion, so we are the sole owner here.
if let Some(slots) = Arc::into_inner(slots) {
let results: Vec<R> = slots.into_iter().filter_map(|slot| slot.value.into_inner()).collect();
merge(results);
} else {
log::error!("More than one ref to the slots remaining after workers exit. This SHOULD NOT happen.");
}
}
// ---------------------------------------------------------------------------
// AtomicCounter with parking (avoids spinning in the coordinator)
// ---------------------------------------------------------------------------
struct AtomicCounter {
state: Mutex<usize>,
done: Condvar,
}
impl AtomicCounter {
fn new(n: usize) -> Self {
Self {
state: Mutex::new(n),
done: Condvar::new(),
}
}
/// Decrements the counter by one and notifies waiters if it reaches zero.
///
/// # Panics (debug only)
///
/// Debug-asserts that the counter has not already reached zero, which
/// would indicate a double-decrement bug.
fn dec_and_notify(&self) {
let mut count = self.state.lock().unwrap();
debug_assert!(
*count > 0,
"AtomicCounter decremented below zero — double-decrement bug?"
);
*count -= 1;
if *count == 0 {
self.done.notify_all();
}
}
/// Blocks until the counter reaches zero.
fn wait_for_zero(&self) {
let mut count = self.state.lock().unwrap();
while *count > 0 {
count = self.done.wait(count).unwrap();
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn partition_threads_split() {
// Single-core: both pools get at least 1 thread.
assert_eq!(partition_threads(1), (1, 1));
// Two cores: 1 reader, 1 matcher.
assert_eq!(partition_threads(2), (1, 1));
// Three cores: 1 reader, 2 matcher.
assert_eq!(partition_threads(3), (1, 2));
// Six cores: 2 reader, 4 matcher.
assert_eq!(partition_threads(6), (2, 4));
// Eight cores: 3 reader, 5 matcher; sums to 8.
assert_eq!(partition_threads(8), (3, 5));
// Nine cores: 3 reader, 6 matcher; sums to 9.
assert_eq!(partition_threads(9), (3, 6));
// The two values always sum to n for n >= 3.
for n in 3..=64 {
let (r, m) = partition_threads(n);
assert_eq!(r + m, n, "partition_threads({n}) = ({r}, {m}) does not sum to {n}");
assert!(r >= 1);
assert!(m >= 1);
}
}
#[test]
fn spawn_runs_closure() {
let pool = ThreadPool::new(2);
let flag = Arc::new(AtomicUsize::new(0));
let flag2 = Arc::clone(&flag);
pool.spawn(move || {
flag2.store(42, Ordering::SeqCst);
});
// Give it a moment.
std::thread::sleep(std::time::Duration::from_millis(50));
assert_eq!(flag.load(Ordering::SeqCst), 42);
}
#[test]
fn spawn_batch_runs_all() {
let pool = ThreadPool::new(4);
let counter = Arc::new(AtomicUsize::new(0));
let jobs: Vec<Box<dyn FnOnce() + Send + 'static>> = (0..10)
.map(|_| {
let c = Arc::clone(&counter);
let job: Box<dyn FnOnce() + Send + 'static> = Box::new(move || {
c.fetch_add(1, Ordering::SeqCst);
});
job
})
.collect();
pool.spawn_batch(jobs);
std::thread::sleep(std::time::Duration::from_millis(100));
assert_eq!(counter.load(Ordering::SeqCst), 10);
}
#[test]
fn parallel_work_queue_sums() {
let pool = ThreadPool::new(4);
let items: Arc<[u64]> = (1..=1000u64).collect::<Vec<_>>().into();
let mut result = 0u64;
parallel_work_queue(
&pool,
4,
&items,
64,
|| 0u64,
|_start, chunk| chunk.iter().sum::<u64>(),
|acc, partial| *acc += partial,
|_| {},
|worker_results| {
for partial in worker_results {
result += partial;
}
},
);
assert_eq!(result, 500_500);
}
#[test]
fn parallel_work_queue_empty() {
let pool = ThreadPool::new(2);
let items: Arc<[u64]> = Arc::from(Vec::<u64>::new().into_boxed_slice());
let mut result = Vec::<u64>::new();
parallel_work_queue(
&pool,
2,
&items,
64,
Vec::<u64>::new,
|_start, chunk| chunk.to_vec(),
|acc, mut partial| acc.append(&mut partial),
|_| {},
|worker_results| {
for partial in worker_results {
result.extend(partial);
}
},
);
assert!(result.is_empty());
}
#[test]
fn parallel_work_queue_single_thread() {
let pool = ThreadPool::new(1);
let items: Arc<[i32]> = (0..100i32).collect::<Vec<_>>().into();
let mut result = 0i32;
parallel_work_queue(
&pool,
1,
&items,
10,
|| 0i32,
|_start, chunk| chunk.iter().sum::<i32>(),
|acc, partial| *acc += partial,
|_| {},
|worker_results| {
for partial in worker_results {
result += partial;
}
},
);
assert_eq!(result, (0..100).sum::<i32>());
}
#[test]
fn pool_drop_joins_threads() {
let flag = Arc::new(AtomicUsize::new(0));
{
let pool = ThreadPool::new(2);
let f = Arc::clone(&flag);
pool.spawn(move || {
std::thread::sleep(std::time::Duration::from_millis(30));
f.store(1, Ordering::SeqCst);
});
} // pool dropped here – should join
assert_eq!(flag.load(Ordering::SeqCst), 1);
}
#[test]
fn parallel_work_queue_many_workers_few_chunks() {
// More workers than chunks — extra workers should gracefully no-op.
let pool = ThreadPool::new(8);
let items: Arc<[u64]> = (1..=10u64).collect::<Vec<_>>().into();
let mut result = 0u64;
parallel_work_queue(
&pool,
8,
&items,
5,
|| 0u64,
|_start, chunk| chunk.iter().sum::<u64>(),
|acc, partial| *acc += partial,
|_| {},
|worker_results| {
for partial in worker_results {
result += partial;
}
},
);
assert_eq!(result, 55);
}
#[test]
fn parallel_work_queue_single_thread_pool_no_deadlock() {
// With a 1-thread pool the coordinator must NOT be a pool job —
// it runs on a dedicated OS thread and submits all worker jobs to
// the pool. This ensures the single pool thread is always free to
// run those workers and no deadlock can occur.
let (tx, rx) = std::sync::mpsc::channel();
let pool = Arc::new(ThreadPool::new(1));
let items: Arc<[u64]> = (1..=100u64).collect::<Vec<_>>().into();
let pool_coord = Arc::clone(&pool);
// Coordinator is a dedicated thread, not a pool job.
std::thread::spawn(move || {
parallel_work_queue(
&pool_coord,
1,
&items,
10,
|| 0u64,
|_start, chunk| chunk.iter().sum::<u64>(),
|acc, partial| *acc += partial,
|_| {},
|worker_results| {
let _ = tx.send(worker_results.into_iter().sum::<u64>());
},
);
});
let result = rx
.recv_timeout(std::time::Duration::from_secs(5))
.expect("deadlock or timeout");
assert_eq!(result, 5050);
}
}

25
src/tmux.rs
View file

@ -294,12 +294,12 @@ pub fn run_with(opts: &SkimOptions) -> Option<SkimOutput> {
if line.is_empty() {
None
} else {
Some(MatchedItem {
item: Arc::new(SkimTmuxOutput { line: line.to_string() }),
rank: Rank::default(),
rank_builder: Arc::new(RankBuilder::default()),
matched_range: None,
})
Some(MatchedItem::new(
Arc::new(SkimTmuxOutput { line: line.to_string() }),
Rank::default(),
None,
&RankBuilder::default(),
))
}
} else {
None
@ -310,15 +310,16 @@ pub fn run_with(opts: &SkimOptions) -> Option<SkimOutput> {
debug!("Adding output line: {line}");
// --print-score is always enabled in the child, so every item is followed by its score.
let score: i32 = stdout.next().unwrap_or_default().parse().unwrap_or_default();
let item = MatchedItem {
item: Arc::new(SkimTmuxOutput { line: line.to_string() }),
rank: Rank {
let rank = Rank {
score,
..Default::default()
},
rank_builder: Arc::new(RankBuilder::default()),
matched_range: None,
};
let item = MatchedItem::new(
Arc::new(SkimTmuxOutput { line: line.to_string() }),
rank,
None,
&RankBuilder::default(),
);
output_lines.push(item);
}

90
src/tui/app.rs
View file

@ -21,6 +21,7 @@ use super::event::Action;
use super::header::Header;
use super::item_list::ItemList;
use super::{input, preview};
use crate::thread_pool::{self, ThreadPool};
use color_eyre::eyre::{Result, bail};
use crossterm::event::{KeyCode, KeyEvent, KeyModifiers, MouseEvent, MouseEventKind};
use input::Input;
@ -30,7 +31,6 @@ use ratatui::crossterm::event::KeyCode::Char;
use ratatui::layout::Rect;
use ratatui::prelude::Backend;
use ratatui::widgets::Widget;
use rayon::ThreadPool;
use std::sync::LazyLock;
static NUM_THREADS: LazyLock<usize> = LazyLock::new(|| {
@ -47,8 +47,10 @@ const HIDE_GRACE_MS: u128 = 500;
pub struct App {
/// Pool of items to be filtered
pub item_pool: Arc<ItemPool>,
/// Separate thread pool for use by skim
pub thread_pool: Arc<ThreadPool>,
/// Thread pool used by the matcher (⌊2N/3⌋ threads).
pub matcher_pool: Arc<ThreadPool>,
/// Thread pool used by the reader pipeline (⌈N/3⌉ threads).
pub reader_pool: Arc<ThreadPool>,
/// Whether the application should quit
pub should_quit: bool,
@ -195,17 +197,14 @@ impl Default for App {
let initial_header_height = header.height();
let layout_template = LayoutTemplate::from_options(&opts, initial_header_height);
let layout = layout_template.apply(Rect::default());
let (reader_threads, matcher_threads) = thread_pool::partition_threads(*NUM_THREADS);
Self {
input: Input::from_options(&opts, theme.clone()),
preview: Preview::from_options(&opts, theme.clone()),
header,
item_list: ItemList::from_options(&opts, theme.clone()),
thread_pool: Arc::new(
rayon::ThreadPoolBuilder::new()
.num_threads(*NUM_THREADS)
.build()
.unwrap(),
),
matcher_pool: Arc::new(ThreadPool::new(matcher_threads)),
reader_pool: Arc::new(ThreadPool::new(reader_threads)),
item_pool: Arc::default(),
theme,
should_quit: false,
@ -250,23 +249,20 @@ impl App {
///
/// # Panics
///
/// Panics if the Rayon thread pool cannot be built (system resource exhaustion).
/// Panics if the thread pool cannot be built (system resource exhaustion).
#[must_use]
pub fn from_options(options: SkimOptions, theme: Arc<crate::theme::ColorTheme>, cmd: String) -> Self {
let header = Header::from_options(&options, theme.clone());
let initial_header_height = header.height();
let layout_template = LayoutTemplate::from_options(&options, initial_header_height);
let layout = layout_template.apply(Rect::default());
let (reader_threads, matcher_threads) = thread_pool::partition_threads(*NUM_THREADS);
Self {
input: Input::from_options(&options, theme.clone()),
preview: Preview::from_options(&options, theme.clone()),
header,
thread_pool: Arc::new(
rayon::ThreadPoolBuilder::new()
.num_threads(*NUM_THREADS)
.build()
.unwrap(),
),
matcher_pool: Arc::new(ThreadPool::new(matcher_threads)),
reader_pool: Arc::new(ThreadPool::new(reader_threads)),
item_pool: Arc::new(ItemPool::from_options(&options)),
item_list: ItemList::from_options(&options, theme.clone()),
theme,
@ -694,12 +690,12 @@ impl App {
..Default::default()
};
self.item_pool.append(vec![item.clone()]);
self.item_list.append(&mut vec![MatchedItem {
self.item_list.append(&mut vec![MatchedItem::new(
item,
rank,
rank_builder: self.matcher.rank_builder.clone(),
matched_range: None,
}]);
None,
&self.matcher.rank_builder,
)]);
self.item_list.select_row(self.item_list.items.len() - 1);
self.restart_matcher_debounced();
return Ok(Self::on_selection_changed());
@ -1162,7 +1158,7 @@ impl App {
/// If `force` is false, the matcher will only be restarted if there are new items
/// to process or if the previous matcher has completed.
pub fn restart_matcher(&mut self, force: bool) {
use crate::tui::item_list::{MergeStrategy, ProcessedItems};
use crate::tui::item_list::MergeStrategy;
// Check if query meets minimum length requirement
if let Some(min_length) = self.options.min_query_length
&& !self.options.disabled
@ -1198,54 +1194,30 @@ impl App {
&self.input
};
let item_pool = self.item_pool.clone();
let thread_pool = &self.thread_pool;
let processed_items = self.item_list.processed_items.clone();
let thread_pool = &self.matcher_pool;
let no_sort = self.options.no_sort;
if force {
self.item_pool.reset();
}
let needs_render = self.needs_render.clone();
self.matcher_control = self.matcher.run(query, &item_pool, thread_pool, move |mut matches| {
debug!("Got {} results from matcher, sending to item list...", matches.len());
if !no_sort {
matches.sort();
}
if force {
// Full re-match: replace all results
*processed_items.lock() = Some(ProcessedItems {
items: matches,
merge: MergeStrategy::Replace,
});
} else {
// Incremental: merge new matches into any unconsumed processed items,
// and mark with merge strategy so the render loop merges with item_list.items
let merge_strategy = if no_sort {
let merge_strategy = if force {
MergeStrategy::Replace
} else if no_sort {
MergeStrategy::Append
} else {
MergeStrategy::SortedMerge
};
let mut guard = processed_items.lock();
if let Some(ref mut existing) = *guard {
if no_sort {
existing.items.extend(matches);
} else {
// Merge incoming matches into existing sorted list in-place.
MatchedItem::merge_into_sorted(&mut existing.items, matches);
}
} else {
*guard = Some(ProcessedItems {
items: matches,
merge: merge_strategy,
});
}
}
needs_render.store(true, Ordering::Relaxed);
});
self.matcher_control = self.matcher.run(
query,
&item_pool,
thread_pool,
self.item_list.processed_items.clone(),
merge_strategy,
no_sort,
self.needs_render.clone(),
);
}
}

7
src/util.rs
View file

@ -258,12 +258,7 @@ mod test {
use std::sync::Arc;
fn make_item(s: &'static str) -> MatchedItem {
MatchedItem {
item: Arc::new(s),
rank: Rank::default(),
rank_builder: Arc::new(RankBuilder::default()),
matched_range: None,
}
MatchedItem::new(Arc::new(s), Rank::default(), None, &RankBuilder::default())
}
#[test]