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: better performance on small datasets

Browse source
This commit is contained in:
Loric ANDRE 2026-04-07 10:34:12 +02:00
parent 40f0e29448
commit e227cafda2
24 changed files with 602 additions and 444 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
AGENTS.md
View file

@ -39,7 +39,7 @@
## Testing
This application can be tested by :
- creating a new `tmux` session in the background (`tmux new-session -s <session name> -d`)
- creating a new `tmux` session in the background (`tmux new-session -s <session name> -d`). Make sure to clear the `SKIM_DEFAULT_OPTIONS` env var.
- creating a new named tmux window in that session : `tmux new-window -d -P -F '#I' -n <window name> -t <session name>` and configuring the pane naming using `tmux set-window-option -t <window name> pane-base-index 0`
- sending the command to run and input using `tmux send-keys -t <window name> <keys>`
- when ready, capturing the window using `tmux capture-pane -b <window name> -t <window name>.0` and then saving the capture to a file using `tmux save-buffer -b <window name> <output file>`

62
ARCHITECTURE.md
View file

@ -78,8 +78,8 @@ stdin / command
SkimOutput
```
The **Reader** pulls raw text from stdin or a shell command and converts it into `Arc<dyn SkimItem>` batches, depositing them into the shared `ItemPool`.
The **Matcher** picks items up from the pool, evaluates every item against the current query string using a configured engine, and writes ranked `MatchedItem` results into `ProcessedItems`.
The **Reader** pulls raw text from stdin or a shell command and converts it into `Arc<dyn SkimItem>` batches, depositing them into the shared `ItemPool`.
The **Matcher** picks items up from the pool, evaluates every item against the current query string using a configured engine, and writes ranked `MatchedItem` results into `ProcessedItems`.
The **TUI** renders four composable widgets (Input, ItemList, Preview, Header), drives a `crossterm`-based event loop, and converts user keystrokes into typed `Action` values that are dispatched back to the `App` state machine.
---
@ -254,7 +254,7 @@ Each call to `tick()` runs a `tokio::select!` on four concurrent futures:
| Branch | Source | Action |
|---|---|---|
| `tui.next()` | crossterm keyboard/mouse/resize/paste events | Dispatch to `app.handle_event()` |
| `matcher_interval.tick()` | 10 ms periodic timer | `app.restart_matcher(false)` |
| `matcher_interval.tick()` | 10 ms periodic timer (adaptive: disabled once reader finishes and all items are matched) | `app.restart_matcher(false)` |
| `items_available.notified()` | `Notify` set by `ItemPool::append` | `app.restart_matcher(false)` |
| `listener.accept()` | IPC socket (when `--listen`) | Parse RON-encoded `Action`, push to event queue |
@ -347,6 +347,8 @@ The coordinate mapping is critical: match indices come back in terms of stripped
Without `--ansi`, any ANSI escape codes are passed through to `text()` and displayed as literal characters. If the raw input happens to contain escape sequences (but `--ansi` is not set), `escape_ansi()` is called to make them visible.
ANSI input uses the same parallel pipeline as plain input — there is no separate serial path. `DefaultSkimItem::new` handles the stripping inline inside the worker threads.
**Key files:** `src/helper/item.rs` (`DefaultSkimItem::new`, `strip_ansi`, `display`)
### Popup Mode (`--popup` / `--tmux`)
@ -388,29 +390,30 @@ The child `sk` process runs fully independently inside the popup. The parent rea
## Item Ingestion Pipeline
All inputs — plain stdin, `--ansi`, `--nth`/`--with-nth`, and shell commands — flow through a
single unified parallel pipeline (`parallel_bufread`). There is no serial fallback path.
```
Source (stdin bytes or child process stdout)
│
│ SkimItemReader::of_bufread() or CommandCollector::invoke()
│
├── Simple path (no ANSI, no --nth, no --with-nth):
│ raw_bufread()
│ ├─ Thread 1: I/O reader — reads 256 KB chunks, splits at line boundaries,
│ │ assigns monotonic sequence numbers, sends to MPMC channel
│ ├─ Thread N: workers — receive chunks, validate UTF-8, create DefaultSkimItem
│ │ (no metadata allocation, just Box<str>), send (seq, items) pairs
│ └─ Thread 1: reorder — collects (seq, items), emits in order through SkimItemReceiver
│
└── Complex path (ANSI | --nth | --with-nth):
read_lines_into_items() (single-threaded)
├─ reads line by line via read_until(line_ending)
├─ creates DefaultSkimItem::new(line, ansi, trans_fields, matching_fields, delimiter)
└─ buffers up to 1024 items before sending to SkimItemReceiver
└── parallel_bufread() (all inputs)
├─ Thread 1: I/O reader — reads 256 KB chunks, splits at line boundaries,
│ assigns monotonic sequence numbers, sends to MPMC channel
├─ Thread N: workers — receive chunks, validate UTF-8,
│ create DefaultSkimItem::new(line, ansi, trans_fields, matching_fields, delimiter)
│ (handles ANSI stripping, --nth / --with-nth transforms inline),
│ send (seq, items) pairs
├─ Thread 1: reorder — collects (seq, items), emits in order through SkimItemReceiver;
│ drops tx_pipeline_done on exit (signals killer thread)
└─ Thread 1: killer — waits on rx_interrupt OR rx_pipeline_done (whichever fires first);
kills child process if one exists, then exits
SkimItemReceiver channel
│
│ Reader::collect()
│ collect_items() spawns a thread that polls the channel every 1ms
│ collect_items() spawns a thread that blocks on recv_timeout (5ms) from the channel
│
└── ItemPool::append(items)
├─ respects --tac (reverse order)
@ -511,11 +514,11 @@ Matcher::run(query, item_pool, thread_pool, …)
├─ create matcher_engine from factory (synchronous)
├─ take items from pool synchronously (avoids race with restart)
│
└─ thread_pool.spawn(coordinator closure)
└─ std::thread::spawn(coordinator closure)
│
├─ shares items as Arc<[Arc<dyn SkimItem>]>
│
├─ parallel_work_queue(pool, num_workers, items, CHUNK_SIZE=512, …)
├─ parallel_work_queue(pool, num_workers, items, CHUNK_SIZE=4096, …)
│ │
│ ├─ Worker threads (num_cpus - 1):
│ │ ├─ atomically grab next chunk
@ -524,6 +527,8 @@ Matcher::run(query, item_pool, thread_pool, …)
│ │ └─ accumulate into worker-local Vec<MatchedItem>
│ │ └─ sort_unstable() on worker thread (parallel sort)
│ │
│ ├─ AtomicCounter barrier (lock-free AtomicUsize + thread::park/unpark)
│ │
│ └─ coordinator:
│ └─ merge_worker_results(worker_results, no_sort, …)
│ ├─ concatenate k sorted runs
@ -616,7 +621,7 @@ loop {
The main loop (`Skim::run()`) calls `tick()` in a loop, which `select!`s on the same channel plus the matcher interval and IPC listener.
Frame rate is capped at 30 fps (`FRAME_TIME_MS = 1000/30`). `App::handle_event(Heartbeat)` checks `needs_render` (an `AtomicBool` set by the matcher when new results arrive) and emits `Event::Render` only when the last render was more than `FRAME_TIME_MS` ago.
Frame rate is capped at 120 fps (`FRAME_TIME_MS = 1000/120`). `App::handle_event(Heartbeat)` checks `needs_render` (an `AtomicBool` set by the matcher when new results arrive) and emits `Event::Render` only when the last render was more than `FRAME_TIME_MS` ago.
### App State
@ -1072,10 +1077,12 @@ ThreadPool (N = num_cpus OS threads, persistent)
└─ Worker threads (N-1 slots per match run)
Reader threads (OS threads, per-invocation):
├─ collect_items thread: polls SkimItemReceiver, calls ItemPool::append
├─ I/O reader thread (parallel path only): reads large byte chunks
├─ Worker threads (parallel path only): parse lines, create DefaultSkimItem
└─ Reorder thread (parallel path only): sequence-ordered output
├─ collect_items thread: blocks on SkimItemReceiver (recv_timeout 5ms), calls ItemPool::append
├─ I/O reader thread: reads large byte chunks, splits lines, assigns sequence numbers
├─ Worker threads (N): parse lines, create DefaultSkimItem (ANSI strip + field transforms inline)
├─ Reorder thread: sequence-ordered output; drops tx_pipeline_done on EOF
└─ Killer thread (command inputs only): waits for rx_interrupt or rx_pipeline_done;
kills child process when either fires
Preview thread (OS thread, per preview spawn):
└─ reads PTY/child stdout → vt100::Parser or content Arc<RwLock>
@ -1124,8 +1131,9 @@ The global allocator is `mimalloc` (v3), chosen for its low-latency multi-thread
| `ItemPool::append` | `src/item.rs:467` | Add items, notify matcher |
| `ItemPool::take` | `src/item.rs:512` | Take un-matched items for matcher |
| `DefaultSkimItem::new` | `src/helper/item.rs:55` | ANSI strip, field transform, ranges |
| `SkimItemReader::raw_bufread` | `src/helper/item_reader.rs:300` | 3-stage parallel reader (simple path) |
| `SkimItemReader::read_lines_into_items` | `src/helper/item_reader.rs:217` | Single-threaded reader (complex path) |
| `SkimItemReader::parallel_bufread` | `src/helper/item_reader.rs:260` | Unified parallel pipeline (all inputs) |
| `spawn_io_reader` | `src/helper/item_reader.rs:352` | I/O reader thread: chunk reads + line splitting |
| `spawn_reorder_thread` | `src/helper/item_reader.rs:452` | Reorder thread: ordered output + pipeline-done signal |
| `Preview::spawn` | `src/tui/preview.rs:272` | Start PTY or plain child process |
| `Tui::new_with_height_and_backend` | `src/tui/backend.rs:68` | Terminal init + viewport sizing |
| `Tui::enter` | `src/tui/backend.rs:130` | Enable raw mode + start event pump |
@ -1218,4 +1226,4 @@ impl CommandCollector for MySource {
}
```
Set `options.cmd_collector = Rc::new(RefCell::new(my_source))` before calling `Skim::run_with`.
Set `options.cmd_collector = Rc::new(RefCell::new(my_source))` before calling `Skim::run_with`.

118
benches/cli.rs
View file

@ -6,20 +6,6 @@
//!
//! 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`.
//!
//! Invoke via the `bench-cli` cargo alias:
//!
//! ```text
//! cargo bench-cli # defaults
//! cargo bench-cli -- sk -n 500000 -q foo
//! cargo bench-cli -- ./old/sk ./new/sk -r 5 # compare two binaries
//! cargo bench-cli -- sk -r 5 # 5 runs, show average
//! cargo bench-cli -- sk -f input.txt -q search # use existing file
//! cargo bench-cli -- -g testdata.txt -n 2000000 # generate file and exit
//! cargo bench-cli -- sk -p # record perf (auto-named)
//! cargo bench-cli -- sk -p perf.data # record perf to perf.data
//! cargo bench-cli -- sk -j # JSON output
//! cargo bench-cli -- sk -r 3 -- --tiebreak=index # pass extra flags to sk
//! ```
use clap::Parser;
@ -145,6 +131,18 @@ struct Args {
)]
perf: Option<String>,
/// Run the final benchmark run under strace and write the trace to FILE.
/// Optionally specify the output file (default: auto-named
/// strace-<binary>-<timestamp>.out).
#[arg(
short = 't',
long,
num_args = 0..=1,
default_missing_value = "",
value_name = "FILE"
)]
strace: Option<String>,
/// Seconds the matched count must remain unchanged before a run is declared
/// complete (default: 5.0).
#[arg(short = 's', long, default_value_t = REQUIRED_STABLE_S, value_name = "SECS")]
@ -244,6 +242,7 @@ struct RunResult {
peak_cpu: Option<f64>,
completed: bool,
perf_file: Option<String>,
strace_file: Option<String>,
/// Time from launch until both the prompt+query and the `N/M` status counts
/// are visible — i.e. `max(prompt_appeared, status_appeared)`.
startup_s: Option<f64>,
@ -378,6 +377,7 @@ fn run_once(
run_index: u32,
session_suffix: &str,
perf_output: Option<&str>,
strace_output: Option<&str>,
tmux_server: &TmuxServer,
stable_secs: f64,
) -> Result<RunResult> {
@ -391,9 +391,13 @@ fn run_once(
Some(path) => format!("perf record -o {} -- ", path),
None => String::new(),
};
let strace_prefix = match strace_output {
Some(path) => format!("strace -C -ttt -o {} -- ", path),
None => String::new(),
};
let cmd_str = format!(
"cat {} | {}{} --prompt '{}' {}",
tmp_file, perf_prefix, binary_path, BENCH_PROMPT, extra_str
"cat {} | {}{}{} --prompt '{}' {}",
tmp_file, perf_prefix, strace_prefix, binary_path, BENCH_PROMPT, extra_str
);
// --- Phase 1: wait for the shell to be ready (any pane content appears) --
@ -559,6 +563,26 @@ fn run_once(
}
}
// Wait for strace to finish writing before killing the session
if strace_output.is_some() && pane_pid > 0 {
let strace_wait = Instant::now();
loop {
if strace_wait.elapsed().as_secs_f64() >= 15.0 {
eprintln!("Warning: strace did not exit within 15 s; trace data may be incomplete.");
break;
}
let still_running = Command::new("pgrep")
.args(["-P", &pane_pid.to_string(), "-f", "strace"])
.output()
.map(|o| o.status.success())
.unwrap_or(false);
if !still_running {
break;
}
thread::sleep(Duration::from_millis(100));
}
}
let monitor = monitor_cell.lock().unwrap().take();
let (peak_mem_kb, peak_cpu) = monitor.map(ResourceMonitor::join).unwrap_or((None, None));
@ -579,6 +603,7 @@ fn run_once(
peak_cpu,
completed,
perf_file: perf_output.map(str::to_owned),
strace_file: strace_output.map(str::to_owned),
startup_s: match (startup_prompt_s, startup_status_s) {
(Some(a), Some(b)) => Some(a.max(b)),
(Some(a), None) => Some(a),
@ -1070,6 +1095,23 @@ fn perf_path_for(binary: &str, explicit: &str) -> String {
format!("perf-{}-{}.data", base, ts)
}
fn strace_path_for(binary: &str, explicit: &str) -> String {
if !explicit.is_empty() {
return explicit.to_owned();
}
let ts = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_secs();
let base = Path::new(binary)
.file_name()
.and_then(|n| n.to_str())
.map(|s| s.replace(' ', "_"))
.filter(|s| !s.is_empty())
.unwrap_or_else(|| "sk".into());
format!("strace-{}-{}.out", base, ts)
}
// ---------------------------------------------------------------------------
// Main
// ---------------------------------------------------------------------------
@ -1137,6 +1179,8 @@ fn main() -> Result<()> {
let extra_args = &args.extra_args;
let record_perf = args.perf.is_some();
let perf_explicit = args.perf.as_deref().unwrap_or("");
let record_strace = args.strace.is_some();
let strace_explicit = args.strace.as_deref().unwrap_or("");
// ---- header ------------------------------------------------------------
eprintln!("=== Skim Ingestion + Matching Benchmark ===");
@ -1164,6 +1208,9 @@ fn main() -> Result<()> {
if record_perf {
eprintln!("Perf recording: enabled (final measured run only)");
}
if record_strace {
eprintln!("Strace recording: enabled (final measured run only)");
}
// ---- dedicated tmux server ---------------------------------------------
// Started once with a clean environment; all benchmark panes run inside
@ -1186,6 +1233,7 @@ fn main() -> Result<()> {
wu,
&format!("warmup_b{}", bi),
None,
None,
&tmux_server,
args.stable_secs,
)?;
@ -1211,6 +1259,19 @@ fn main() -> Result<()> {
vec![None; binaries.len()]
};
// Determine strace output paths (one per binary, recorded only on last run)
let strace_files: Vec<Option<String>> = if record_strace {
binaries
.iter()
.map(|binary| {
let explicit = if binaries.len() == 1 { strace_explicit } else { "" };
Some(strace_path_for(binary, explicit))
})
.collect()
} else {
vec![None; binaries.len()]
};
for run_num in 1..=runs {
for (bi, binary) in binaries.iter().enumerate() {
if runs > 1 || binaries.len() > 1 {
@ -1224,12 +1285,17 @@ fn main() -> Result<()> {
);
}
// Attach perf only on the final run for this binary
// Attach perf/strace only on the final run for this binary
let this_perf = if run_num == runs {
perf_files[bi].as_deref()
} else {
None
};
let this_strace = if run_num == runs {
strace_files[bi].as_deref()
} else {
None
};
let result = run_once(
binary,
@ -1239,6 +1305,7 @@ fn main() -> Result<()> {
run_num,
&format!("b{}", bi),
this_perf,
this_strace,
&tmux_server,
args.stable_secs,
)?;
@ -1260,6 +1327,9 @@ fn main() -> Result<()> {
if let Some(ref pf) = result.perf_file {
eprintln!("Perf data: {}", pf);
}
if let Some(ref sf) = result.strace_file {
eprintln!("Strace output: {}", sf);
}
}
all_results[bi].push(result);
@ -1305,5 +1375,19 @@ fn main() -> Result<()> {
}
}
}
// ---- strace summary ----------------------------------------------------
if record_strace {
eprintln!("\n=== Strace output ===");
for (binary, path) in binaries.iter().zip(&strace_files) {
if let Some(p) = path {
if Path::new(p).is_file() {
eprintln!(" [{}] strace output: {}", binary, p);
} else {
eprintln!(" [{}] strace output not found (strace may have failed)", binary);
}
}
}
}
Ok(())
}

5
benches/gungraun.rs
View file

@ -67,9 +67,6 @@ library_benchmark_group!(
);
#[cfg(not(any(target_arch = "x86_64", target_arch = "aarch64")))]
library_benchmark_group!(
name = benches,
benchmarks = [skim_v2, arinae, arinae_typos]
);
library_benchmark_group!(name = benches, benchmarks = [skim_v2, arinae, arinae_typos]);
main!(library_benchmark_groups = benches);

110
src/bin/main.rs
View file

@ -186,63 +186,71 @@ fn sk_main(mut opts: SkimOptions) -> Result<i32> {
return Ok(130);
}
// Output
if let Some(ref output_format) = bin_options.output_format {
print!(
"{}{}",
skim::printf(
output_format,
&bin_options.delimiter,
&bin_options.replstr,
&result.selected_items.iter(),
&result.current,
&result.query,
&result.cmd,
true
),
bin_options.output_ending
);
} else {
if bin_options.print_query {
print!("{}{}", result.query, bin_options.output_ending);
}
// Output — use a large BufWriter to batch all writes into a few syscalls
// instead of one syscall per item (Rust's default LineWriter flushes on \n).
{
let stdout = io::stdout();
let mut out = BufWriter::with_capacity(1 << 20, stdout.lock());
if bin_options.print_cmd {
print!("{}{}", result.cmd, bin_options.output_ending);
}
if let Some(ref output_format) = bin_options.output_format {
write!(
out,
"{}{}",
skim::printf(
output_format,
&bin_options.delimiter,
&bin_options.replstr,
&result.selected_items.iter(),
&result.current,
&result.query,
&result.cmd,
true
),
bin_options.output_ending
)?;
} else {
if bin_options.print_query {
write!(out, "{}{}", result.query, bin_options.output_ending)?;
}
if bin_options.print_header {
print!("{}{}", result.header, bin_options.output_ending);
}
if bin_options.print_cmd {
write!(out, "{}{}", result.cmd, bin_options.output_ending)?;
}
if bin_options.print_current {
if let Some(ref current) = result.current {
print!("{}{}", current.output(), bin_options.output_ending);
} else {
print!("{}", bin_options.output_ending);
}
}
if let Event::Action(Action::Accept(Some(accept_key))) = result.final_event {
print!("{}{}", accept_key, bin_options.output_ending);
}
for item in &result.selected_items {
if bin_options.strip_ansi {
print!(
"{}{}",
skim::helper::item::strip_ansi(&item.output()).0,
bin_options.output_ending
);
} else {
print!("{}{}", item.output(), bin_options.output_ending);
}
if bin_options.print_score {
print!("{}{}", item.rank.score, bin_options.output_ending);
if bin_options.print_header {
write!(out, "{}{}", result.header, bin_options.output_ending)?;
}
if bin_options.print_current {
if let Some(ref current) = result.current {
write!(out, "{}{}", current.output(), bin_options.output_ending)?;
} else {
write!(out, "{}", bin_options.output_ending)?;
}
}
if let Event::Action(Action::Accept(Some(accept_key))) = result.final_event {
write!(out, "{}{}", accept_key, bin_options.output_ending)?;
}
for item in &result.selected_items {
if bin_options.strip_ansi {
write!(
out,
"{}{}",
skim::helper::item::strip_ansi(&item.output()).0,
bin_options.output_ending
)?;
} else {
write!(out, "{}{}", item.output(), bin_options.output_ending)?;
}
if bin_options.print_score {
write!(out, "{}{}", item.rank.score, bin_options.output_ending)?;
}
}
}
out.flush()?;
}
std::io::stdout().flush()?;
//------------------------------------------------------------------------------
// write the history with latest item

9
src/engine/andor.rs
View file

@ -83,6 +83,9 @@ impl AndEngine {
MatchRange::ByteRange(..) => {
ranges.extend(item.range_char_indices(text));
}
MatchRange::CharRange(start, end) => {
ranges.extend(start..end);
}
MatchRange::Chars(vec) => {
ranges.extend(vec.iter().copied());
}
@ -103,7 +106,11 @@ impl AndEngine {
impl MatchEngine for AndEngine {
fn match_item(&self, item: &dyn SkimItem) -> Option<MatchResult> {
// mock
// Fast path: single sub-engine — skip merge entirely.
if self.engines.len() == 1 {
return self.engines[0].match_item(item);
}
let mut results = vec![];
for engine in &self.engines {
let result = engine.match_item(item)?;

142
src/engine/fuzzy.rs
View file

@ -2,7 +2,6 @@ use std::cmp::min;
use std::fmt::{Display, Error, Formatter};
use std::sync::Arc;
use crate::fuzzy_matcher::MatchIndices;
use crate::fuzzy_matcher::arinae::ArinaeMatcher;
#[cfg(any(target_arch = "x86_64", target_arch = "aarch64"))]
use crate::fuzzy_matcher::frizbee::FrizbeeMatcher;
@ -48,10 +47,6 @@ pub struct FuzzyEngineBuilder {
/// - `Typos::Smart`: adaptive (`pattern_length` / 4)
/// - `Typos::Fixed(n)`: exactly n typos allowed
typos: Typos,
/// When true, use `fuzzy_match_range` instead of `fuzzy_indices` to avoid
/// per-character index computation (useful in filter mode where highlighting
/// is not needed).
filter_mode: bool,
/// When true, prefer the last (rightmost) occurrence on tied scores.
last_match: bool,
}
@ -82,8 +77,9 @@ impl FuzzyEngineBuilder {
self
}
pub fn filter_mode(mut self, filter_mode: bool) -> Self {
self.filter_mode = filter_mode;
/// No-op: `fuzzy_match_range` is now always used (`ByteRange`).
/// Kept for API backward compatibility.
pub fn filter_mode(self, _filter_mode: bool) -> Self {
self
}
@ -154,7 +150,6 @@ impl FuzzyEngineBuilder {
matcher,
query: self.query,
rank_builder: self.rank_builder,
filter_mode: self.filter_mode,
}
}
}
@ -164,7 +159,6 @@ pub struct FuzzyEngine {
query: String,
matcher: Box<dyn FuzzyMatcher>,
rank_builder: Arc<RankBuilder>,
filter_mode: bool,
}
impl FuzzyEngine {
@ -180,83 +174,67 @@ impl MatchEngine for FuzzyEngine {
let item_text = item.text();
let default_range = [(0, item_text.len())];
if self.filter_mode {
// Fast path: use fuzzy_match_range to avoid per-character index computation
let mut best: Option<(i64, usize, usize)> = None;
for &(start, end) in item.get_matching_ranges().unwrap_or(&default_range) {
let start = min(start, item_text.len());
let end = min(end, item_text.len());
// Always use fuzzy_match_range which returns (score, begin, end)
// without computing per-character match indices. This avoids an
// O(pattern_len) Vec allocation per matched item and skips the full
// traceback in the DP. The renderer handles ByteRange directly for
// both horizontal scrolling and match highlighting.
let mut best: Option<(i64, usize, usize)> = None;
for &(start, end) in item.get_matching_ranges().unwrap_or(&default_range) {
let start = min(start, item_text.len());
let end = min(end, item_text.len());
let result = if self.query.is_empty() {
Some((0i64, 0, 0))
} else if item_text[start..end].is_empty() {
None
} else {
self.matcher
.fuzzy_match_range(&item_text[start..end], &self.query)
.map(|(s, b, e)| {
let offset = if start != 0 {
item_text[..start].chars().count()
} else {
0
};
(s, b + offset, e + offset)
})
};
let result = if self.query.is_empty() {
Some((0i64, 0, 0))
} else if item_text[start..end].is_empty() {
None
} else {
self.matcher
.fuzzy_match_range(&item_text[start..end], &self.query)
.map(|(s, b, e)| {
let offset = if start != 0 {
item_text[..start].chars().count()
} else {
0
};
(s, b + offset, e + offset)
})
};
if result.is_some() {
best = result;
break;
}
if result.is_some() {
best = result;
break;
}
let (score, begin, end) = best?;
Some(MatchResult {
rank: self
.rank_builder
.build_rank(i32::try_from(score).unwrap_or(i32::MAX), begin, end, &item_text),
matched_range: MatchRange::ByteRange(begin, end),
})
} else {
let mut matched_result = None;
for &(start, end) in item.get_matching_ranges().unwrap_or(&default_range) {
let start = min(start, item_text.len());
let end = min(end, item_text.len());
let result = if self.query.is_empty() {
Some((0i64, MatchIndices::new()))
} else if item_text[start..end].is_empty() {
None
} else {
self.matcher.fuzzy_indices(&item_text[start..end], &self.query)
};
matched_result = result.map(|(s, vec)| {
if start != 0 {
let start_char = item_text[..start].chars().count();
(s, vec.iter().map(|x| x + start_char).collect::<MatchIndices>())
} else {
(s, vec)
}
});
if matched_result.is_some() {
break;
}
}
let (score, matched_indices) = matched_result?;
let begin = *matched_indices.first().unwrap_or(&0);
let end = *matched_indices.last().unwrap_or(&0);
let matched_range = MatchRange::Chars(matched_indices);
Some(MatchResult {
rank: self
.rank_builder
.build_rank(i32::try_from(score).unwrap_or(i32::MAX), begin, end, &item_text),
matched_range,
})
}
let (score, begin, end) = best?;
// `fuzzy_match_range` returns `end` as the inclusive index of the last
// matched character. Convert to exclusive upper bound for CharRange.
let end_excl = if self.query.is_empty() { end } else { end + 1 };
// When the span length equals the query length, the match is
// contiguous and CharRange is a perfect representation (no gaps).
// When the span is wider than the query, there are gaps between
// matched characters; fall back to fuzzy_indices for accurate
// per-character highlighting (only ~25 visible items need this).
let query_len = self.query.chars().count();
let matched_range = if end_excl - begin == query_len {
// Contiguous match — CharRange is exact.
MatchRange::CharRange(begin, end_excl)
} else {
// Non-contiguous match — compute exact indices for correct highlighting.
match self.matcher.fuzzy_indices(&item_text, &self.query) {
Some((_s, indices)) => MatchRange::Chars(indices),
None => MatchRange::CharRange(begin, end_excl),
}
};
Some(MatchResult {
rank: self
.rank_builder
.build_rank(i32::try_from(score).unwrap_or(i32::MAX), begin, end_excl, &item_text),
matched_range,
})
}
}

9
src/engine/normalized.rs
View file

@ -40,6 +40,15 @@ impl MatchEngine for NormalizedEngine {
// Map the matched range back to the original text
result.matched_range = match result.matched_range {
MatchRange::Chars(indices) => MatchRange::Chars(map_char_indices_to_original(&indices, &char_mapping)),
MatchRange::CharRange(start, end) => {
let orig_start = char_mapping.get(start).copied().unwrap_or(start);
let orig_end = if end > 0 {
char_mapping.get(end - 1).copied().map_or(end, |e| e + 1)
} else {
0
};
MatchRange::CharRange(orig_start, orig_end)
}
MatchRange::ByteRange(start, end) => {
let (orig_start, orig_end) = map_byte_range_to_original(start, end, &byte_mapping, &item_text);
MatchRange::ByteRange(orig_start, orig_end)

2
src/engine/split.rs
View file

@ -54,6 +54,7 @@ impl MatchEngine for SplitMatchEngine {
let mut combined_indices: MatchIndices = match before_result.matched_range {
MatchRange::Chars(indices) => indices,
MatchRange::CharRange(start, end) => (start..end).collect(),
MatchRange::ByteRange(start, end) => {
// Convert byte range to char indices for the before part
text_before
@ -70,6 +71,7 @@ impl MatchEngine for SplitMatchEngine {
let after_indices: MatchIndices = match after_result.matched_range {
MatchRange::Chars(indices) => indices.into_iter().map(|i| i + offset).collect(),
MatchRange::CharRange(start, end) => (start..end).map(|i| i + offset).collect(),
MatchRange::ByteRange(start, end) => {
// Convert byte range to char indices for the after part
text_after

7
src/fuzzy_matcher/arinae/algo.rs
View file

@ -6,7 +6,7 @@ use thread_local::ThreadLocal;
use crate::fuzzy_matcher::{IndexType, MatchIndices};
use super::banding::{compute_banding, typo_vband_row};
use super::banding::{BandingInfo, typo_vband_row};
use super::constants::{
CONSECUTIVE_BONUS, GAP_EXTEND, GAP_OPEN, MATCH_BONUS, MAX_PAT_LEN, MISMATCH_PENALTY, TYPO_PENALTY,
};
@ -111,6 +111,7 @@ fn compute_cell<const ALLOW_TYPOS: bool>(
/// and subsequent rows are dead (since UP/LEFT can only propagate
/// existing scores). We track this and allow early termination.
#[allow(clippy::too_many_lines)]
#[allow(clippy::too_many_arguments)]
pub(super) fn full_dp<const ALLOW_TYPOS: bool, const COMPUTE_INDICES: bool, C: Atom>(
cho: &[C],
pat: &[C],
@ -119,11 +120,11 @@ pub(super) fn full_dp<const ALLOW_TYPOS: bool, const COMPUTE_INDICES: bool, C: A
full_buf: &ThreadLocal<RefCell<SWMatrix>>,
indices_buf: &ThreadLocal<RefCell<MatchIndices>>,
use_last_match: bool,
banding: &BandingInfo,
) -> Option<(Score, MatchIndices)> {
let n = pat.len();
let m = cho.len();
let banding = compute_banding::<ALLOW_TYPOS, C>(pat, cho, respect_case)?;
let j_start = banding.j_first; // earliest match — skip columns before this
// Column offset: the matrix stores only columns from j_start onward.
@ -377,11 +378,11 @@ pub(super) fn range_dp<const ALLOW_TYPOS: bool, C: Atom>(
respect_case: bool,
full_buf: &ThreadLocal<RefCell<SWMatrix>>,
use_last_match: bool,
banding: &BandingInfo,
) -> Option<(Score, usize, usize)> {
let n = pat.len();
let m = cho.len();
let banding = compute_banding::<ALLOW_TYPOS, C>(pat, cho, respect_case)?;
let j_start = banding.j_first;
let col_off = j_start - 1;
let mcols = m - col_off + 1;

35
src/fuzzy_matcher/arinae/atom.rs
View file

@ -1,7 +1,7 @@
//! Byte/Char helpers
use super::Score;
use super::constants::SEPARATOR_TABLE;
use memchr::memchr;
use memchr::{memchr, memrchr};
pub(super) trait Atom: PartialEq + Into<char> + Copy {
#[inline(always)]
@ -27,6 +27,17 @@ pub(super) trait Atom: PartialEq + Into<char> + Copy {
fn find_first_in(self, haystack: &[Self], respect_case: bool) -> Option<usize> {
haystack.iter().position(|&c| self.eq(c, respect_case))
}
/// Return the index of the last occurrence of `self` in `haystack`,
/// or `None` if not found.
///
/// Implementations may override this with a SIMD-backed search (e.g.
/// `memrchr` for `u8` in case-sensitive mode).
#[inline(always)]
fn find_last_in(self, haystack: &[Self], respect_case: bool) -> Option<usize> {
haystack.iter().rposition(|&c| self.eq(c, respect_case))
}
/// Return the word-separator bonus for this character, or `0` if it is not
/// a separator. Uses a table lookup — a single bounds check replaces
/// several branches and the returned value encodes both *whether* the
@ -76,6 +87,28 @@ impl Atom for u8 {
}
}
}
/// Case-sensitive backward search uses SIMD-backed `memrchr`.
#[inline(always)]
fn find_last_in(self, haystack: &[Self], respect_case: bool) -> Option<usize> {
if respect_case {
memrchr(self, haystack)
} else {
let lo = self.to_ascii_lowercase();
let hi = self.to_ascii_uppercase();
if lo == hi {
memrchr(lo, haystack)
} else {
// Return the rightmost occurrence across both case variants.
let p_lo = memrchr(lo, haystack);
let p_hi = memrchr(hi, haystack);
match (p_lo, p_hi) {
(None, x) | (x, None) => x,
(Some(a), Some(b)) => Some(a.max(b)),
}
}
}
}
}
impl Atom for char {
#[inline(always)]

1
src/fuzzy_matcher/arinae/banding.rs
View file

@ -7,6 +7,7 @@ use super::constants::{MAX_PAT_LEN, TYPO_BAND_SLACK};
use super::helpers::{compute_last_match_cols, compute_row_col_bounds, find_first_char};
/// Precomputed banding information shared by both score-only and full DP.
#[derive(Clone)]
pub(super) struct BandingInfo {
/// Per-row column bounds (only present in exact mode).
pub(super) row_bounds: Option<([usize; MAX_PAT_LEN], [usize; MAX_PAT_LEN])>,

2
src/fuzzy_matcher/arinae/helpers.rs
View file

@ -30,7 +30,7 @@ pub(super) fn compute_last_match_cols<C: Atom>(
let mut last = [0usize; MAX_PAT_LEN];
let mut end = m; // search up to this choice index (exclusive)
for i in (0..n).rev() {
let found = cho[..end].iter().rposition(|&c| pat[i].eq(c, respect_case));
let found = pat[i].find_last_in(&cho[..end], respect_case);
match found {
Some(pos) => {
last[i] = pos + 1; // 1-indexed column

75
src/fuzzy_matcher/arinae/mod.rs
View file

@ -39,6 +39,7 @@ use thread_local::ThreadLocal;
use self::algo::{full_dp, range_dp};
use self::atom::Atom;
use self::banding::{BandingInfo, compute_banding};
use self::constants::{CAMEL_CASE_BONUS, START_OF_STRING_BONUS};
use self::prefilter::cheap_typo_prefilter;
@ -101,7 +102,7 @@ impl ArinaeMatcher {
}
/// Dispatch to `full_dp` with the appropriate const generics.
/// Assumes prefilters and bonuses have already been computed.
/// Assumes prefilters, banding, and bonuses have already been computed.
fn dispatch_dp<C: Atom>(
&self,
cho: &[C],
@ -109,13 +110,14 @@ impl ArinaeMatcher {
bonuses: &[Score],
respect_case: bool,
compute_indices: bool,
banding: &BandingInfo,
) -> Option<(ScoreType, MatchIndices)> {
#[rustfmt::skip]
let res = match (self.allow_typos, compute_indices) {
(true, true) => full_dp::<true , true , _>(cho, pat, bonuses, respect_case, &self.full_buf, &self.indices_buf, self.use_last_match),
(true, false) => full_dp::<true , false, _>(cho, pat, bonuses, respect_case, &self.full_buf, &self.indices_buf, self.use_last_match),
(false, true) => full_dp::<false, true , _>(cho, pat, bonuses, respect_case, &self.full_buf, &self.indices_buf, self.use_last_match),
(false, false) => full_dp::<false, false, _>(cho, pat, bonuses, respect_case, &self.full_buf, &self.indices_buf, self.use_last_match),
(true, true) => full_dp::<true , true , _>(cho, pat, bonuses, respect_case, &self.full_buf, &self.indices_buf, self.use_last_match, banding),
(true, false) => full_dp::<true , false, _>(cho, pat, bonuses, respect_case, &self.full_buf, &self.indices_buf, self.use_last_match, banding),
(false, true) => full_dp::<false, true , _>(cho, pat, bonuses, respect_case, &self.full_buf, &self.indices_buf, self.use_last_match, banding),
(false, false) => full_dp::<false, false, _>(cho, pat, bonuses, respect_case, &self.full_buf, &self.indices_buf, self.use_last_match, banding),
};
res.map(|(s, idx)| (ScoreType::from(s), idx))
}
@ -134,18 +136,24 @@ impl ArinaeMatcher {
let respect_case = self.respect_case(pat);
// Prefilter for typo mode.
// In exact mode (non-typo) we skip is_subsequence here: compute_banding
// calls compute_first_match_cols which already validates the subsequence
// and returns None if any pattern character is absent — no redundant scan.
if self.allow_typos && !cheap_typo_prefilter(pat, cho, respect_case) {
return None;
}
// Prepare bonuses
// Compute banding BEFORE bonuses: the banding check (subsequence scan) is
// a fast SIMD operation that rejects ~70% of items early. For those items
// we never allocate or fill the bonus buffer, saving an O(m) write pass.
let banding = if self.allow_typos {
compute_banding::<true, C>(pat, cho, respect_case)?
} else {
compute_banding::<false, C>(pat, cho, respect_case)?
};
// Only compute bonuses for items that survive the banding check.
let mut bonus_buf = self.bonus_buf.get_or(|| RefCell::new(Vec::new())).borrow_mut();
precompute_bonuses(cho, &mut bonus_buf);
self.dispatch_dp(cho, pat, &bonus_buf, respect_case, compute_indices)
self.dispatch_dp(cho, pat, &bonus_buf, respect_case, compute_indices, &banding)
}
fn run(&self, choice: &str, pattern: &str, compute_indices: bool) -> Option<(ScoreType, MatchIndices)> {
@ -175,16 +183,23 @@ impl ArinaeMatcher {
let respect_case = self.respect_case(pat_buf);
// Prefilter for typo mode only (see match_slices for rationale).
// Prefilter for typo mode only.
if self.allow_typos && !cheap_typo_prefilter(pat_buf, cho_buf, respect_case) {
return None;
}
// Compute banding before bonuses — rejects non-matches without allocating.
let banding = if self.allow_typos {
compute_banding::<true, char>(pat_buf, cho_buf, respect_case)?
} else {
compute_banding::<false, char>(pat_buf, cho_buf, respect_case)?
};
let mut bonus_buf = self.bonus_buf.get_or(|| RefCell::new(Vec::new())).borrow_mut();
precompute_bonuses(cho_buf, &mut bonus_buf);
// Call dispatch_dp directly to avoid double-borrowing bonus_buf.
self.dispatch_dp(cho_buf, pat_buf, &bonus_buf, respect_case, compute_indices)
self.dispatch_dp(cho_buf, pat_buf, &bonus_buf, respect_case, compute_indices, &banding)
}
/// Run the DP and return `(score, begin, end)` without collecting all indices.
@ -204,16 +219,37 @@ impl ArinaeMatcher {
let cho = choice.as_bytes();
let pat = pattern.as_bytes();
let respect_case = self.respect_case(pat);
// Exact mode: compute_banding validates the subsequence implicitly.
if self.allow_typos && !cheap_typo_prefilter(pat, cho, respect_case) {
return None;
}
// Compute banding before bonuses — rejects non-matches without allocating.
let banding = if self.allow_typos {
compute_banding::<true, u8>(pat, cho, respect_case)?
} else {
compute_banding::<false, u8>(pat, cho, respect_case)?
};
let mut bonus_buf = self.bonus_buf.get_or(|| RefCell::new(Vec::new())).borrow_mut();
precompute_bonuses(cho, &mut bonus_buf);
if self.allow_typos {
range_dp::<true, _>(cho, pat, &bonus_buf, respect_case, &self.full_buf, self.use_last_match)
range_dp::<true, _>(
cho,
pat,
&bonus_buf,
respect_case,
&self.full_buf,
self.use_last_match,
&banding,
)
} else {
range_dp::<false, _>(cho, pat, &bonus_buf, respect_case, &self.full_buf, self.use_last_match)
range_dp::<false, _>(
cho,
pat,
&bonus_buf,
respect_case,
&self.full_buf,
self.use_last_match,
&banding,
)
}
} else {
let mut bufs = self
@ -226,10 +262,15 @@ impl ArinaeMatcher {
cho_buf.clear();
cho_buf.extend(choice.chars());
let respect_case = self.respect_case(pat_buf);
// Exact mode: compute_banding validates the subsequence implicitly.
if self.allow_typos && !cheap_typo_prefilter(pat_buf, cho_buf, respect_case) {
return None;
}
// Compute banding before bonuses — rejects non-matches without allocating.
let banding = if self.allow_typos {
compute_banding::<true, char>(pat_buf, cho_buf, respect_case)?
} else {
compute_banding::<false, char>(pat_buf, cho_buf, respect_case)?
};
let mut bonus_buf = self.bonus_buf.get_or(|| RefCell::new(Vec::new())).borrow_mut();
precompute_bonuses(cho_buf, &mut bonus_buf);
if self.allow_typos {
@ -240,6 +281,7 @@ impl ArinaeMatcher {
respect_case,
&self.full_buf,
self.use_last_match,
&banding,
)
} else {
range_dp::<false, _>(
@ -249,6 +291,7 @@ impl ArinaeMatcher {
respect_case,
&self.full_buf,
self.use_last_match,
&banding,
)
}
};

4
src/helper/item.rs
View file

@ -86,7 +86,7 @@ impl DefaultSkimItem {
};
// Keep track of whether we have null bytes for special handling
let has_null_bytes = temp_text.contains('\0');
let has_null_bytes = memchr::memchr(b'\0', temp_text.as_bytes()).is_some();
// Preserve original text with null bytes for output if needed
if has_null_bytes && orig_text.is_none() {
@ -171,7 +171,7 @@ impl DefaultSkimItem {
}
fn contains_ansi_escape(s: &str) -> bool {
s.contains('\x1b')
memchr::memchr(b'\x1b', s.as_bytes()).is_some()
}
/// Getter for `stripped_text` stored in the metadata

297
src/helper/item_reader.rs
View file

@ -5,9 +5,8 @@ use std::error::Error;
use std::io::{BufRead, BufReader};
use std::process::{Child, Stdio};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::thread;
use std::time::{Duration, Instant};
use crate::thread_pool::ThreadPool;
@ -23,13 +22,6 @@ use crate::{SkimItem, SkimItemReceiver, SkimItemSender, SkimOptions};
const DELIMITER_STR: &str = r"[\t\n ]+";
const READ_BUFFER_SIZE: usize = 1024;
const ITEMS_BUFFER_SIZE: usize = 1024;
const SEND_TIMEOUT_MS: u64 = 100; // Send items if we haven't sent anything in 100ms
pub enum CollectorInput {
Pipe(Box<dyn BufRead + Send>),
Command(String),
}
/// Options for configuring how items are read and parsed
#[derive(Debug)]
@ -165,12 +157,6 @@ impl SkimItemReaderOption {
pub fn build(self) -> Self {
self
}
/// Returns true if no field transformations or ANSI parsing is needed
#[must_use]
pub fn is_simple(&self) -> bool {
!self.use_ansi_color && self.matching_fields.is_empty() && self.transform_fields.is_empty()
}
}
/// Reader for converting various input sources into streams of skim items
@ -233,90 +219,19 @@ impl SkimItemReader {
}
impl SkimItemReader {
/// Converts a `BufRead` source into a stream of skim items
/// Converts a `BufRead` source into a stream of skim items using the
/// parallel pipeline.
pub fn of_bufread(&self, source: impl BufRead + Send + 'static) -> SkimItemReceiver {
if self.option.is_simple() {
self.raw_bufread(source)
} else {
self.read_and_collect_from_command(Arc::new(AtomicUsize::new(0)), CollectorInput::Pipe(Box::new(source)))
.0
}
self.parallel_bufread(source, None, &Arc::new(AtomicUsize::new(0))).0
}
/// Helper function that contains the common logic for reading lines from a `BufRead` source
/// and converting them into `SkimItems`.
fn read_lines_into_items(
mut source: impl BufRead + Send + 'static,
tx_item: &SkimItemSender,
option: &Arc<SkimItemReaderOption>,
transform_fields: &[FieldRange],
matching_fields: &[FieldRange],
) {
let mut buffer = Vec::with_capacity(option.buf_size);
let mut items_to_send = Vec::with_capacity(ITEMS_BUFFER_SIZE);
let mut last_send_time = Instant::now();
let send_timeout = Duration::from_millis(SEND_TIMEOUT_MS);
loop {
buffer.clear();
// start reading
match source.read_until(option.line_ending, &mut buffer) {
Ok(0) => break,
Ok(_) => {
// Strip line endings
if buffer.ends_with(b"\r\n") {
buffer.pop();
buffer.pop();
} else if buffer.ends_with(&[option.line_ending]) {
buffer.pop();
}
let Ok(line) = std::str::from_utf8(&buffer) else {
continue;
};
trace!("got item {line}");
let raw_item = DefaultSkimItem::new(
line,
option.use_ansi_color,
transform_fields,
matching_fields,
&option.delimiter,
);
items_to_send.push(Arc::new(raw_item) as Arc<dyn SkimItem>);
}
Err(err) => {
trace!("Got {err:?} when reading, skipping");
} // String not UTF8 or other error, skip.
}
// Send batched items if buffer is full OR timeout has elapsed
let should_send = items_to_send.len() == ITEMS_BUFFER_SIZE
|| (!items_to_send.is_empty() && last_send_time.elapsed() >= send_timeout);
if should_send {
let batch = std::mem::replace(&mut items_to_send, Vec::with_capacity(ITEMS_BUFFER_SIZE));
match tx_item.send(batch) {
Ok(()) => {
last_send_time = Instant::now();
}
Err(e) => {
warn!("Failed to send items: {e:?}");
break;
}
}
}
}
// Send remaining items
if !items_to_send.is_empty() {
let _ = tx_item.send(items_to_send);
}
}
/// Parallel reader for the simple (no ANSI, no field transforms) case.
/// Core parallel reader pipeline.
///
/// All input — whether a plain pipe, a `--ansi`-decorated stream, or one
/// with `--nth`/`--with-nth` field transforms — goes through the same four
/// stages. Every per-line operation inside `DefaultSkimItem::new` is
/// stateless and purely functional, so chunks can be processed concurrently
/// without any coordination beyond sequence reordering.
///
/// Pipeline:
///
@ -326,14 +241,28 @@ impl SkimItemReader {
/// 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.
/// 3. **Pool jobs** — parse lines, validate UTF-8, apply ANSI stripping and
/// field transforms, 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 {
///
/// When `child` is `Some`, a **killer thread** is also spawned. It waits
/// on `rx_interrupt` and kills the child process on request (or when the
/// reader is dropped). This thread participates in `components_to_stop`
/// accounting so that [`ReaderControl::kill`] waits for it to finish.
///
/// Returns `(rx_item, tx_interrupt)`. The caller must send on `tx_interrupt`
/// to signal shutdown; the killer thread (if any) will then kill the child.
fn parallel_bufread(
&self,
source: impl BufRead + Send + 'static,
child: Option<Child>,
components_to_stop: &Arc<AtomicUsize>,
) -> (SkimItemReceiver, crate::prelude::Sender<i32>) {
let (tx_item, rx_item): (SkimItemSender, SkimItemReceiver) = kanal::bounded(1024 * 1024);
let option = self.option.clone();
let pool = Arc::clone(&self.thread_pool);
@ -361,10 +290,60 @@ impl SkimItemReader {
// so the reorder thread exits once the last pool job finishes.
});
// Stage 4: reorder thread.
Self::spawn_reorder_thread(rx_results, tx_item);
// A zero-capacity channel used as a completion signal: the reorder
// thread drops its sender when it exits, closing the channel. The
// killer thread waits on either this signal (natural EOF) or on the
// external interrupt (early termination request).
let (tx_pipeline_done, rx_pipeline_done) = kanal::bounded::<()>(0);
rx_item
// Stage 4: reorder thread.
Self::spawn_reorder_thread(rx_results, tx_item.clone(), tx_pipeline_done);
// Killer thread: exits when the pipeline drains naturally (child process
// reached EOF) OR when it receives an explicit interrupt signal.
//
// This thread participates in `components_to_stop` accounting so that
// [`ReaderControl::is_done`] correctly waits for cleanup to complete.
let (tx_interrupt, rx_interrupt) = crate::prelude::bounded::<i32>(8);
let components_to_stop_killer = components_to_stop.clone();
components_to_stop.fetch_add(1, Ordering::SeqCst);
thread::spawn(move || {
debug!("parallel reader: killer thread start");
// Wait for either a kill request or the pipeline finishing naturally.
// kanal doesn't have a multi-channel select, so we poll with a short
// timeout. Both channels are bounded so this never busy-spins in
// practice: the kill path is rare, and the done path fires quickly.
loop {
if rx_interrupt.try_recv().is_ok_and(|v| v.is_some()) {
// Explicit kill: terminate the child immediately.
if let Some(mut c) = child {
let _ = c.kill();
let _ = c.wait();
}
break;
}
// Channel closed = reorder thread exited = pipeline drained.
match rx_pipeline_done.recv_timeout(std::time::Duration::from_millis(1)) {
Ok(()) => break,
Err(kanal::ReceiveErrorTimeout::Closed | kanal::ReceiveErrorTimeout::SendClosed) => {
// Natural EOF: child already exited; just reap if present.
if let Some(mut c) = child {
let _ = c.wait();
}
break;
}
Err(kanal::ReceiveErrorTimeout::Timeout) => {
// Neither signal yet — loop.
}
}
}
components_to_stop_killer.fetch_sub(1, Ordering::SeqCst);
debug!("parallel reader: killer thread stop");
});
(rx_item, tx_interrupt)
}
/// Stage 1 of the parallel reader: reads large byte chunks from `source`,
@ -432,8 +411,6 @@ impl SkimItemReader {
}
/// 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;
@ -455,8 +432,6 @@ impl SkimItemReader {
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,
@ -469,9 +444,15 @@ impl SkimItemReader {
(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) {
/// Stage 4: receives item batches from workers and emits them through the
/// downstream channel in the original sequence order. Drops
/// `tx_pipeline_done` on exit to signal the killer thread that the
/// pipeline has drained naturally.
fn spawn_reorder_thread(
rx_results: kanal::Receiver<(usize, Vec<Arc<dyn SkimItem>>)>,
tx_item: SkimItemSender,
tx_pipeline_done: kanal::Sender<()>,
) {
thread::spawn(move || {
debug!("parallel reader: reorder thread start");
let mut expected = 0usize;
@ -493,96 +474,12 @@ impl SkimItemReader {
return;
}
}
// Dropping tx_pipeline_done closes the channel, waking the killer
// thread so it can decrement components_to_stop.
drop(tx_pipeline_done);
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(
&self,
components_to_stop: Arc<AtomicUsize>,
input: CollectorInput,
) -> (SkimItemReceiver, crate::prelude::Sender<i32>) {
let send_error = self.option.show_error;
let (command, source) = match input {
CollectorInput::Pipe(pipe) => (None, pipe),
CollectorInput::Command(cmd) => get_command_output(&cmd, send_error).expect("command not found"),
};
let (tx_interrupt, rx_interrupt) = crate::prelude::bounded(8);
let (tx_item, rx_item): (SkimItemSender, SkimItemReceiver) = crate::prelude::bounded(1024 * 1024);
let started = Arc::new(AtomicBool::new(false));
let started_clone = started.clone();
let components_to_stop_clone = components_to_stop.clone();
let tx_item_clone = tx_item.clone();
// listening to close signal and kill command if needed
thread::spawn(move || {
debug!("collector: command killer start");
components_to_stop_clone.fetch_add(1, Ordering::SeqCst);
started_clone.store(true, Ordering::SeqCst); // notify parent that it is started
let _ = rx_interrupt.recv();
if let Some(mut child) = command {
// clean up resources
let _ = child.kill();
let _ = child.wait();
if send_error {
let has_error = child
.try_wait()
.map(|os| os.is_none_or(|s| !s.success()))
.unwrap_or(false);
if has_error {
trace!("collector: sending error");
let output = child.wait_with_output().expect("could not retrieve error message");
let error_text = String::from_utf8_lossy(&output.stderr).to_string();
let error_items: Vec<Arc<dyn SkimItem>> = error_text
.lines()
.map(|line| {
Arc::new(DefaultSkimItem::new(
line,
false,
&[],
&[],
&Regex::new(DELIMITER_STR).unwrap(),
)) as Arc<dyn SkimItem>
})
.collect();
let _ = tx_item_clone.send(error_items);
}
}
}
components_to_stop_clone.fetch_sub(1, Ordering::SeqCst);
debug!("collector: command killer stop");
});
while !started.load(Ordering::SeqCst) {
// busy waiting for the thread to start. (components_to_stop is added)
}
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();
// 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);
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 killer thread will exit
components_to_stop.fetch_sub(1, Ordering::SeqCst);
debug!("collector: command collector stop");
});
(rx_item, tx_interrupt)
}
}
impl CommandCollector for SkimItemReader {
@ -591,7 +488,9 @@ impl CommandCollector for SkimItemReader {
cmd: &str,
components_to_stop: Arc<AtomicUsize>,
) -> (SkimItemReceiver, crate::prelude::Sender<i32>) {
self.read_and_collect_from_command(components_to_stop, CollectorInput::Command(cmd.to_string()))
let send_error = self.option.show_error;
let (child, source) = get_command_output(cmd, send_error).expect("command not found");
self.parallel_bufread(source, child, &components_to_stop)
}
fn set_thread_pool(&mut self, pool: Arc<ThreadPool>) {

4
src/lib.rs
View file

@ -321,6 +321,9 @@ impl From<usize> for Typos {
pub enum MatchRange {
/// Range of bytes (start, end)
ByteRange(usize, usize),
/// Range of character indices (start, end) — used by fuzzy matchers that
/// operate on `char` arrays rather than raw bytes.
CharRange(usize, usize),
/// Individual character indices that matched
Chars(MatchIndices),
}
@ -367,6 +370,7 @@ impl MatchResult {
let last = first + text[start..end].chars().count();
(first..last).collect()
}
&MatchRange::CharRange(start, end) => (start..end).collect(),
MatchRange::Chars(vec) => vec.clone(),
}
}

11
src/reader.rs
View file

@ -190,15 +190,16 @@ where
debug!("interrupt: {msg}");
break;
}
match rx_item.try_recv() {
Ok(Some(items)) => {
match rx_item.recv_timeout(std::time::Duration::from_millis(1)) {
Ok(items) => {
trace!("collect_item: got {} items", items.len());
callback(items);
}
Ok(None) => {
std::thread::sleep(std::time::Duration::from_millis(1));
Err(kanal::ReceiveErrorTimeout::Timeout) => {
// No items within the timeout — loop back to check the
// interrupt channel before blocking again.
}
Err(_) => {
Err(kanal::ReceiveErrorTimeout::Closed | kanal::ReceiveErrorTimeout::SendClosed) => {
break;
}
}

50
src/skim.rs
View file

@ -10,6 +10,7 @@ use crossterm::event::{KeyCode, KeyEvent, KeyModifiers};
use tokio::{runtime::Handle, select, task::block_in_place};
use crate::reader::{Reader, ReaderControl};
use crate::tui::TICK_RATE;
use crate::tui::{App, Event, Size, Tui, event::Action};
use crate::{SkimItem, SkimItemReceiver, SkimOptions, SkimOutput};
@ -223,6 +224,12 @@ where
{
self.reader_done = true;
self.app.restart_matcher(false);
// If the matcher already consumed everything, stop the periodic
// interval immediately rather than waiting for the next tick.
if self.app.matcher_control.stopped() && self.app.item_pool.num_not_taken() == 0 {
trace!("matcher interval stopped in check_reader: all items consumed");
self.matcher_interval = None;
}
true
} else {
false
@ -484,6 +491,22 @@ where
self.app.should_quit
}
/// If `needs_render` has been set (e.g. by the matcher thread), immediately
/// send a `Render` event to the TUI so the screen updates without waiting
/// for the next heartbeat tick. Respects the 30 FPS frame-rate cap.
fn try_flush_render(&mut self) {
use std::sync::atomic::Ordering;
if self.app.needs_render.load(Ordering::Relaxed)
&& self.app.last_render_timer.elapsed().as_millis() > 1000 / u128::from(TICK_RATE)
{
self.app.needs_render.store(false, Ordering::Relaxed);
self.app.last_render_timer = std::time::Instant::now();
if let Some(tui) = self.tui.as_ref() {
let _ = tui.event_tx.try_send(Event::Render);
}
}
}
/// Process a single event loop iteration.
///
/// This awaits the next event from the TUI, matcher, or IPC listener,
@ -537,12 +560,37 @@ where
None => std::future::pending::<()>().await,
}
} => {
self.app.restart_matcher(false);
// Check for a pending debounced restart (e.g. the user typed
// while a match-all was running). Checking here (every 10ms)
// rather than only on the heartbeat (83ms) eliminates most of
// the latency between query change and matcher restart.
if self.app.pending_matcher_restart {
self.app.restart_matcher(true);
} else {
self.app.restart_matcher(false);
}
// Check if the matcher (or reader) has set the render flag and
// flush a render event immediately instead of waiting for the
// next heartbeat tick. This can shave up to ~80ms of latency
// when the heartbeat runs at 12 Hz.
self.try_flush_render();
// Once the reader has finished and the matcher has consumed all
// items, stop the periodic interval — it can only produce empty
// ticks from this point. The `items_available` Notify branch
// still handles the (rare) case of late-arriving items.
if self.reader_done
&& self.app.matcher_control.stopped()
&& self.app.item_pool.num_not_taken() == 0
{
trace!("matcher interval stopped: reader done, matcher idle, no pending items");
self.matcher_interval = None;
}
}
// Wake immediately when new items arrive in the pool so the matcher
// can pick them up without waiting for the next periodic interval.
() = items_available.notified() => {
self.app.restart_matcher(false);
self.try_flush_render();
}
Ok(stream) = async {
match &self.listener {

84
src/thread_pool.rs
View file

@ -124,9 +124,10 @@ impl ThreadPool {
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.
/// Submits multiple closures in a single lock acquisition, then wakes
/// exactly as many workers as there are new jobs (capped at the pool size).
/// This avoids unnecessary wakes when fewer jobs than workers are
/// submitted.
///
/// # Panics
///
@ -135,13 +136,21 @@ impl ThreadPool {
where
I: IntoIterator<Item = Box<dyn FnOnce() + Send + 'static>>,
{
{
let count = {
let mut queue = self.shared.queue.lock().unwrap();
let before = queue.jobs.len();
for job in jobs {
queue.jobs.push_back(job);
}
} // lock dropped before notify
self.shared.job_available.notify_all();
queue.jobs.len() - before
}; // lock dropped before notify
if count >= self.num_threads {
self.shared.job_available.notify_all();
} else {
for _ in 0..count {
self.shared.job_available.notify_one();
}
}
}
}
@ -295,10 +304,10 @@ impl<R> Slot<R> {
/// * `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>(
pub fn parallel_work_queue<S, T, R, P, M, I, W, G>(
pool: &ThreadPool,
num_workers: usize,
items: &Arc<[T]>,
items: &Arc<S>,
chunk_size: usize,
identity: I,
process_chunk: P,
@ -306,6 +315,7 @@ pub fn parallel_work_queue<T, R, P, M, I, W, G>(
prepare: W,
merge: G,
) where
S: AsRef<[T]> + Send + Sync + ?Sized + 'static,
T: Send + Sync + 'static,
R: Send + 'static,
P: Fn(usize, &[T]) -> R + Send + Sync + 'static,
@ -314,7 +324,8 @@ pub fn parallel_work_queue<T, R, P, M, I, W, G>(
W: Fn(&mut R) + Send + Sync + 'static,
G: FnOnce(Vec<R>),
{
let total = items.len();
let items_slice: &[T] = AsRef::<[T]>::as_ref(&**items);
let total = items_slice.len();
if total == 0 {
merge(Vec::new());
return;
@ -332,6 +343,8 @@ pub fn parallel_work_queue<T, R, P, M, I, W, G>(
// Barrier: we wait until all workers have finished.
let remaining = Arc::new(AtomicCounter::new(num_workers));
// Register the coordinator thread so workers can unpark it.
remaining.set_waiter();
let process_chunk = Arc::new(process_chunk);
let reduce = Arc::new(reduce);
@ -366,7 +379,8 @@ pub fn parallel_work_queue<T, R, P, M, I, W, G>(
let start = chunk_idx * chunk_size;
let end = total.min(start + chunk_size);
let partial = w_process_chunk(start, &w_items[start..end]);
let slice: &[T] = AsRef::<[T]>::as_ref(&*w_items);
let partial = w_process_chunk(start, &slice[start..end]);
w_reduce(&mut local_acc, partial);
}
@ -421,41 +435,59 @@ pub fn parallel_work_queue<T, R, P, M, I, W, G>(
// ---------------------------------------------------------------------------
struct AtomicCounter {
state: Mutex<usize>,
done: Condvar,
count: AtomicUsize,
/// The thread that called `wait_for_zero`. Workers unpark it when the
/// count reaches zero. Set once by `wait_for_zero` before any worker can
/// finish, so plain `Relaxed` loads inside `dec_and_notify` are fine
/// (the `fetch_sub` with `AcqRel` provides the necessary ordering).
waiter: UnsafeCell<Option<thread::Thread>>,
}
// SAFETY: `waiter` is written exactly once (by the coordinator in
// `set_waiter`, before any worker can observe it via `dec_and_notify`)
// and read by workers only after that write is visible (guaranteed by the
// `AcqRel` ordering on the atomic counter operations).
unsafe impl Send for AtomicCounter {}
unsafe impl Sync for AtomicCounter {}
impl AtomicCounter {
fn new(n: usize) -> Self {
Self {
state: Mutex::new(n),
done: Condvar::new(),
count: AtomicUsize::new(n),
waiter: UnsafeCell::new(None),
}
}
/// Decrements the counter by one and notifies waiters if it reaches zero.
/// Register the current thread as the waiter.
///
/// Must be called exactly once, before any worker calls `dec_and_notify`.
fn set_waiter(&self) {
// SAFETY: called once by the coordinator before workers start.
unsafe { *self.waiter.get() = Some(thread::current()) };
}
/// Decrements the counter by one and unparks the waiter 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();
let prev = self.count.fetch_sub(1, Ordering::AcqRel);
debug_assert!(prev > 0, "AtomicCounter decremented below zero — double-decrement bug?");
if prev == 1 {
// We just decremented from 1 → 0.
// SAFETY: waiter was set before workers were dispatched.
if let Some(t) = unsafe { &*self.waiter.get() } {
t.unpark();
}
}
}
/// 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();
while self.count.load(Ordering::Acquire) > 0 {
thread::park();
}
}
}

7
src/tui/app.rs
View file

@ -6,12 +6,12 @@ use std::sync::atomic::{AtomicBool, Ordering};
use crate::item::{ItemPool, MatchedItem};
use crate::matcher::{Matcher, MatcherControl};
use crate::prelude::ExactOrFuzzyEngineFactory;
use crate::tui::SkimRender;
use crate::tui::input::StatusInfo;
use crate::tui::layout::{AppLayout, LayoutTemplate};
use crate::tui::options::TuiLayout;
use crate::tui::statusline::InfoDisplay;
use crate::tui::widget::SkimWidget;
use crate::tui::{SkimRender, TICK_RATE};
use crate::{ItemPreview, PreviewContext, SkimItem, SkimOptions};
use crate::{Rank, util};
@ -39,7 +39,6 @@ static NUM_THREADS: LazyLock<usize> = LazyLock::new(|| {
.map_or_else(|| 0, std::num::NonZero::get)
});
const FRAME_TIME_MS: u128 = 1000 / 30;
const MATCHER_DEBOUNCE_MS: u128 = 200;
const HIDE_GRACE_MS: u128 = 500;
@ -545,7 +544,7 @@ impl App {
self.restart_matcher(true);
}
if self.needs_render.load(Ordering::Relaxed)
&& self.last_render_timer.elapsed().as_millis() > FRAME_TIME_MS
&& self.last_render_timer.elapsed().as_millis() > 1000 / u128::from(TICK_RATE)
{
debug!("Triggering render");
self.needs_render.store(false, Ordering::Relaxed);
@ -1255,7 +1254,7 @@ impl App {
/// Restart matcher with debouncing to avoid excessive restarts during rapid typing
fn restart_matcher_debounced(&mut self) {
const DEBOUNCE_MS: u64 = 50;
const DEBOUNCE_MS: u64 = 10;
if self.options.disabled {
return;

5
src/tui/backend.rs
View file

@ -17,9 +17,8 @@ use tokio::task::JoinHandle;
use tokio_util::sync::CancellationToken;
use super::util::cursor_pos_from_tty;
use super::{Event, Size};
use super::{Event, Size, TICK_RATE};
const TICK_RATE: f64 = 12.;
static PANIC_HOOK_SET: Once = Once::new();
/// Terminal user interface handler for skim
@ -104,7 +103,7 @@ where
task: None,
event_rx: event_channel.1,
event_tx: event_channel.0,
tick_rate: TICK_RATE,
tick_rate: f64::from(TICK_RATE),
cancellation_token: CancellationToken::default(),
is_fullscreen: lines.is_none(),
})

2
src/tui/item_renderer.rs
View file

@ -98,6 +98,7 @@ impl<'a> ItemRenderer<'a> {
let diff = item_text[*start..*end].chars().count();
(msc, msc + diff)
}
Some(MatchRange::CharRange(start, end)) => (*start, *end),
None => (0, 0),
};
@ -182,6 +183,7 @@ impl<'a> ItemRenderer<'a> {
let matches = match &item.matched_range {
Some(MatchRange::ByteRange(start, end)) => crate::Matches::ByteRange(*start, *end),
Some(MatchRange::CharRange(start, end)) => crate::Matches::CharRange(*start, *end),
Some(MatchRange::Chars(chars)) => crate::Matches::CharIndices(chars.clone()),
None => crate::Matches::None,
};

3
src/tui/mod.rs
View file

@ -36,6 +36,9 @@ pub mod statusline;
/// Widget rendering utilities
pub mod widget;
/// Number of heartbeats per second
pub const TICK_RATE: u32 = 120;
/// Represents a size value, either as a percentage or fixed value
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Size {