diff --git a/AGENTS.md b/AGENTS.md index ccfa0779..02ea01b2 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -39,7 +39,7 @@ ## Testing This application can be tested by : -- creating a new `tmux` session in the background (`tmux new-session -s -d`) +- creating a new `tmux` session in the background (`tmux new-session -s -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 -t ` and configuring the pane naming using `tmux set-window-option -t pane-base-index 0` - sending the command to run and input using `tmux send-keys -t ` - when ready, capturing the window using `tmux capture-pane -b -t .0` and then saving the capture to a file using `tmux save-buffer -b ` diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 247abfed..e668a346 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -78,8 +78,8 @@ stdin / command SkimOutput ``` -The **Reader** pulls raw text from stdin or a shell command and converts it into `Arc` 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` 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), 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]> │ - ├─ 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 │ │ └─ 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 @@ -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`. \ No newline at end of file +Set `options.cmd_collector = Rc::new(RefCell::new(my_source))` before calling `Skim::run_with`. diff --git a/benches/cli.rs b/benches/cli.rs index 982b427f..065e0c3b 100644 --- a/benches/cli.rs +++ b/benches/cli.rs @@ -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, + /// Run the final benchmark run under strace and write the trace to FILE. + /// Optionally specify the output file (default: auto-named + /// strace--.out). + #[arg( + short = 't', + long, + num_args = 0..=1, + default_missing_value = "", + value_name = "FILE" + )] + strace: Option, + /// 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, completed: bool, perf_file: Option, + strace_file: Option, /// 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, @@ -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 { @@ -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> = 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(()) } diff --git a/benches/gungraun.rs b/benches/gungraun.rs index b688f645..5d3fb6e7 100644 --- a/benches/gungraun.rs +++ b/benches/gungraun.rs @@ -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); diff --git a/src/bin/main.rs b/src/bin/main.rs index daa66543..9feca514 100644 --- a/src/bin/main.rs +++ b/src/bin/main.rs @@ -186,63 +186,71 @@ fn sk_main(mut opts: SkimOptions) -> Result { 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 diff --git a/src/engine/andor.rs b/src/engine/andor.rs index f5ed263b..c6d89fa7 100644 --- a/src/engine/andor.rs +++ b/src/engine/andor.rs @@ -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 { - // 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)?; diff --git a/src/engine/fuzzy.rs b/src/engine/fuzzy.rs index 20241327..943ea068 100644 --- a/src/engine/fuzzy.rs +++ b/src/engine/fuzzy.rs @@ -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, rank_builder: Arc, - 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::()) - } 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, + }) } } diff --git a/src/engine/normalized.rs b/src/engine/normalized.rs index 0e606e57..f718ebb2 100644 --- a/src/engine/normalized.rs +++ b/src/engine/normalized.rs @@ -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) diff --git a/src/engine/split.rs b/src/engine/split.rs index 63499843..aaca9361 100644 --- a/src/engine/split.rs +++ b/src/engine/split.rs @@ -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 diff --git a/src/fuzzy_matcher/arinae/algo.rs b/src/fuzzy_matcher/arinae/algo.rs index 2655df29..5fc363e4 100644 --- a/src/fuzzy_matcher/arinae/algo.rs +++ b/src/fuzzy_matcher/arinae/algo.rs @@ -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( /// 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( cho: &[C], pat: &[C], @@ -119,11 +120,11 @@ pub(super) fn full_dp>, indices_buf: &ThreadLocal>, use_last_match: bool, + banding: &BandingInfo, ) -> Option<(Score, MatchIndices)> { let n = pat.len(); let m = cho.len(); - let banding = compute_banding::(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( respect_case: bool, full_buf: &ThreadLocal>, use_last_match: bool, + banding: &BandingInfo, ) -> Option<(Score, usize, usize)> { let n = pat.len(); let m = cho.len(); - let banding = compute_banding::(pat, cho, respect_case)?; let j_start = banding.j_first; let col_off = j_start - 1; let mcols = m - col_off + 1; diff --git a/src/fuzzy_matcher/arinae/atom.rs b/src/fuzzy_matcher/arinae/atom.rs index ce2e9895..95dda5b0 100644 --- a/src/fuzzy_matcher/arinae/atom.rs +++ b/src/fuzzy_matcher/arinae/atom.rs @@ -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 + Copy { #[inline(always)] @@ -27,6 +27,17 @@ pub(super) trait Atom: PartialEq + Into + Copy { fn find_first_in(self, haystack: &[Self], respect_case: bool) -> Option { 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 { + 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 { + 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)] diff --git a/src/fuzzy_matcher/arinae/banding.rs b/src/fuzzy_matcher/arinae/banding.rs index d3ac405c..36ee9e67 100644 --- a/src/fuzzy_matcher/arinae/banding.rs +++ b/src/fuzzy_matcher/arinae/banding.rs @@ -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])>, diff --git a/src/fuzzy_matcher/arinae/helpers.rs b/src/fuzzy_matcher/arinae/helpers.rs index 3aee6e9b..abedf75b 100644 --- a/src/fuzzy_matcher/arinae/helpers.rs +++ b/src/fuzzy_matcher/arinae/helpers.rs @@ -30,7 +30,7 @@ pub(super) fn compute_last_match_cols( 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 diff --git a/src/fuzzy_matcher/arinae/mod.rs b/src/fuzzy_matcher/arinae/mod.rs index 6b4bb8ea..9f9443e5 100644 --- a/src/fuzzy_matcher/arinae/mod.rs +++ b/src/fuzzy_matcher/arinae/mod.rs @@ -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( &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::(cho, pat, bonuses, respect_case, &self.full_buf, &self.indices_buf, self.use_last_match), - (true, false) => full_dp::(cho, pat, bonuses, respect_case, &self.full_buf, &self.indices_buf, self.use_last_match), - (false, true) => full_dp::(cho, pat, bonuses, respect_case, &self.full_buf, &self.indices_buf, self.use_last_match), - (false, false) => full_dp::(cho, pat, bonuses, respect_case, &self.full_buf, &self.indices_buf, self.use_last_match), + (true, true) => full_dp::(cho, pat, bonuses, respect_case, &self.full_buf, &self.indices_buf, self.use_last_match, banding), + (true, false) => full_dp::(cho, pat, bonuses, respect_case, &self.full_buf, &self.indices_buf, self.use_last_match, banding), + (false, true) => full_dp::(cho, pat, bonuses, respect_case, &self.full_buf, &self.indices_buf, self.use_last_match, banding), + (false, false) => full_dp::(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::(pat, cho, respect_case)? + } else { + compute_banding::(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::(pat_buf, cho_buf, respect_case)? + } else { + compute_banding::(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::(pat, cho, respect_case)? + } else { + compute_banding::(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::(cho, pat, &bonus_buf, respect_case, &self.full_buf, self.use_last_match) + range_dp::( + cho, + pat, + &bonus_buf, + respect_case, + &self.full_buf, + self.use_last_match, + &banding, + ) } else { - range_dp::(cho, pat, &bonus_buf, respect_case, &self.full_buf, self.use_last_match) + range_dp::( + 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::(pat_buf, cho_buf, respect_case)? + } else { + compute_banding::(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::( @@ -249,6 +291,7 @@ impl ArinaeMatcher { respect_case, &self.full_buf, self.use_last_match, + &banding, ) } }; diff --git a/src/helper/item.rs b/src/helper/item.rs index 723f3d28..70c2f066 100644 --- a/src/helper/item.rs +++ b/src/helper/item.rs @@ -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 diff --git a/src/helper/item_reader.rs b/src/helper/item_reader.rs index cdbc657a..2591da25 100644 --- a/src/helper/item_reader.rs +++ b/src/helper/item_reader.rs @@ -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), - 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, - 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); - } - 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, + components_to_stop: &Arc, + ) -> (SkimItemReceiver, crate::prelude::Sender) { 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::(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>) { 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>)>, 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>)>, + 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, - input: CollectorInput, - ) -> (SkimItemReceiver, crate::prelude::Sender) { - 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> = error_text - .lines() - .map(|line| { - Arc::new(DefaultSkimItem::new( - line, - false, - &[], - &[], - &Regex::new(DELIMITER_STR).unwrap(), - )) as Arc - }) - .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, ) -> (SkimItemReceiver, crate::prelude::Sender) { - 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) { diff --git a/src/lib.rs b/src/lib.rs index d42d45f2..dd2e88f1 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -321,6 +321,9 @@ impl From 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(), } } diff --git a/src/reader.rs b/src/reader.rs index 9e327c8b..7c6ac9e1 100644 --- a/src/reader.rs +++ b/src/reader.rs @@ -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; } } diff --git a/src/skim.rs b/src/skim.rs index baf2a679..2213abda 100644 --- a/src/skim.rs +++ b/src/skim.rs @@ -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 { diff --git a/src/thread_pool.rs b/src/thread_pool.rs index 12675fbd..0c130de0 100644 --- a/src/thread_pool.rs +++ b/src/thread_pool.rs @@ -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>, { - { + 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 Slot { /// * `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( +pub fn parallel_work_queue( pool: &ThreadPool, num_workers: usize, - items: &Arc<[T]>, + items: &Arc, chunk_size: usize, identity: I, process_chunk: P, @@ -306,6 +315,7 @@ pub fn parallel_work_queue( 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( W: Fn(&mut R) + Send + Sync + 'static, G: FnOnce(Vec), { - 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( // 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( 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( // --------------------------------------------------------------------------- struct AtomicCounter { - state: Mutex, - 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>, } +// 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(); } } } diff --git a/src/tui/app.rs b/src/tui/app.rs index 6d0487c1..c1653142 100644 --- a/src/tui/app.rs +++ b/src/tui/app.rs @@ -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 = 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; diff --git a/src/tui/backend.rs b/src/tui/backend.rs index 0ca91c92..7c6f550d 100644 --- a/src/tui/backend.rs +++ b/src/tui/backend.rs @@ -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(), }) diff --git a/src/tui/item_renderer.rs b/src/tui/item_renderer.rs index 36a9ede5..f1b37a8c 100644 --- a/src/tui/item_renderer.rs +++ b/src/tui/item_renderer.rs @@ -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, }; diff --git a/src/tui/mod.rs b/src/tui/mod.rs index b3a4d084..fba58111 100644 --- a/src/tui/mod.rs +++ b/src/tui/mod.rs @@ -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 {