mirror of
https://github.com/lotabout/skim.git
synced 2026-09-10 07:16:23 -04:00
chore: nitpicks after multithreading review (#1172)
* chore: nitpicks after multithreading review * chore: remove call sites from ARCHITECTURE.md * chore: minor reliability corrections
This commit is contained in:
parent
0392eeb70d
commit
17c517fd02
|
|
@ -34,7 +34,6 @@
|
||||||
- Changing the threading model or synchronization primitives
|
- Changing the threading model or synchronization primitives
|
||||||
- Adding or removing public API surface (`SkimItem`, `SkimOptions`, `SkimOutput`, etc.)
|
- Adding or removing public API surface (`SkimItem`, `SkimOptions`, `SkimOutput`, etc.)
|
||||||
- Changing the event/action system or key binding infrastructure
|
- Changing the event/action system or key binding infrastructure
|
||||||
- Keep call-site line numbers in the cross-reference table up to date when the referenced functions move.
|
|
||||||
|
|
||||||
|
|
||||||
## Testing
|
## Testing
|
||||||
|
|
|
||||||
111
ARCHITECTURE.md
111
ARCHITECTURE.md
|
|
@ -428,7 +428,8 @@ Source (stdin bytes or child process stdout)
|
||||||
└── parallel_bufread() (all inputs)
|
└── parallel_bufread() (all inputs)
|
||||||
├─ Thread 1: I/O reader — reads 256 KB chunks, splits at line boundaries,
|
├─ Thread 1: I/O reader — reads 256 KB chunks, splits at line boundaries,
|
||||||
│ assigns monotonic sequence numbers, sends to MPMC channel
|
│ assigns monotonic sequence numbers, sends to MPMC channel
|
||||||
├─ Thread N: workers — receive chunks, validate UTF-8,
|
├─ Thread 1: bounded dispatcher — uses in-flight tokens to limit queued/running jobs
|
||||||
|
├─ Pool workers: receive chunk jobs, validate UTF-8,
|
||||||
│ create DefaultSkimItem::new(line, ansi, trans_fields, matching_fields, delimiter)
|
│ create DefaultSkimItem::new(line, ansi, trans_fields, matching_fields, delimiter)
|
||||||
│ .hidden_fields(hidden_fields, delimiter)
|
│ .hidden_fields(hidden_fields, delimiter)
|
||||||
│ (handles ANSI stripping, --nth / --with-nth / --hide-nth inline),
|
│ (handles ANSI stripping, --nth / --with-nth / --hide-nth inline),
|
||||||
|
|
@ -491,7 +492,7 @@ Engines are composable through the factory pattern. Starting from `Matcher::crea
|
||||||
options
|
options
|
||||||
│
|
│
|
||||||
├── if regex mode:
|
├── if regex mode:
|
||||||
│ RegexEngineFactory
|
│ RegexEngineFactory (configured with the same RankBuilder / --tiebreak criteria)
|
||||||
│ └─ if normalize: NormalizedEngineFactory(RegexEngineFactory)
|
│ └─ if normalize: NormalizedEngineFactory(RegexEngineFactory)
|
||||||
│
|
│
|
||||||
└── else (fuzzy/exact mode):
|
└── else (fuzzy/exact mode):
|
||||||
|
|
@ -581,12 +582,12 @@ Matcher::run(query, item_pool, thread_pool, …)
|
||||||
│ └─ merge_worker_results(worker_results, no_sort, …)
|
│ └─ merge_worker_results(worker_results, no_sort, …)
|
||||||
│ ├─ concatenate k sorted runs
|
│ ├─ concatenate k sorted runs
|
||||||
│ ├─ sort() (stable; driftsort detects k runs → O(n log k))
|
│ ├─ sort() (stable; driftsort detects k runs → O(n log k))
|
||||||
│ └─ write into SpinLock<Option<ProcessedItems>>
|
│ └─ validate query generation and write into Mutex<Option<ProcessedItems>>
|
||||||
│
|
│
|
||||||
└─ stopped.store(true)
|
└─ stopped.store(true)
|
||||||
```
|
```
|
||||||
|
|
||||||
Interruption is cooperative: each chunk checks `interrupt.load(Relaxed)` before processing. `MatcherControl::kill()` sets `interrupt = true`; `MatcherControl::drop()` also calls `kill()`.
|
Interruption is cooperative: each chunk checks `interrupt.load(Relaxed)` before processing. `MatcherControl::kill()` sets `interrupt = true`; `MatcherControl::drop()` also calls `kill()`. Each forced restart increments a shared query generation. Published `ProcessedItems` carry that generation; both publication and consumption reject stale generations, so a cancelled matcher cannot replace newer results.
|
||||||
|
|
||||||
### Ranking & Sorting
|
### Ranking & Sorting
|
||||||
|
|
||||||
|
|
@ -828,13 +829,14 @@ The `StatusInfo` struct rendered inside the input line shows:
|
||||||
`ItemList` (`src/tui/item_list.rs`) maintains:
|
`ItemList` (`src/tui/item_list.rs`) maintains:
|
||||||
|
|
||||||
- `items: Vec<MatchedItem>` — the currently displayed matched items
|
- `items: Vec<MatchedItem>` — the currently displayed matched items
|
||||||
- `processed_items: Arc<SpinLock<Option<ProcessedItems>>>` — shared with matcher
|
- `processed_items: Arc<Mutex<Option<ProcessedItems>>>` — shared with matcher
|
||||||
|
- `matcher_generation: Arc<AtomicUsize>` — identifies the active query generation
|
||||||
- `selection: Vec<usize>` — indices of multi-selected items
|
- `selection: Vec<usize>` — indices of multi-selected items
|
||||||
- `current: usize` — focused item index (0 = bottom in default layout)
|
- `current: usize` — focused item index (0 = bottom in default layout)
|
||||||
- `offset: usize` — scroll offset (number of items scrolled)
|
- `offset: usize` — scroll offset (number of items scrolled)
|
||||||
- `manual_hscroll: i16` — user-driven horizontal scroll
|
- `manual_hscroll: i16` — user-driven horizontal scroll
|
||||||
|
|
||||||
On each render, `ItemList::render()` checks `processed_items` and swaps them in atomically via the `SpinLock`. Depending on `MergeStrategy`:
|
On each render, `ItemList::render()` checks `processed_items`, rejects results from an old query generation, and swaps current results in through the mutex. Depending on `MergeStrategy`:
|
||||||
|
|
||||||
- `Replace`: replaces `items` entirely.
|
- `Replace`: replaces `items` entirely.
|
||||||
- `SortedMerge`: performs an O(n+m) merge preserving order.
|
- `SortedMerge`: performs an O(n+m) merge preserving order.
|
||||||
|
|
@ -868,7 +870,7 @@ Pre-selection is applied when items first appear: `DefaultSkimSelector::should_s
|
||||||
|
|
||||||
`Preview` (`src/tui/preview.rs`) renders a side/top/bottom pane showing expanded information about the focused item. Its stored content is one of three variants:
|
`Preview` (`src/tui/preview.rs`) renders a side/top/bottom pane showing expanded information about the focused item. Its stored content is one of three variants:
|
||||||
|
|
||||||
**Plain text mode** (no `pty`): spawns `sh -c <cmd>` on Unix or `cmd /c <cmd>` on Windows. On Windows, `Command::raw_arg` is used so `cmd.exe` receives shell metacharacters exactly as written. The child captures stdout (capped at `PREVIEW_MAX_BYTES`), parses it with `ansi_to_tui::IntoText`, stores as `PreviewContent::Text`, and sends `Event::PreviewReady`.
|
**Plain text mode** (no `pty`): spawns `sh -c <cmd>` on Unix or `cmd /c <cmd>` on Windows. On Windows, `Command::raw_arg` is used so `cmd.exe` receives shell metacharacters exactly as written. The worker drains stdout and stderr concurrently, but retains at most `PREVIEW_MAX_BYTES` from each stream. Cancellation terminates the child process group (the process tree on Windows), so selection changes do not leave old preview commands running. Successful stdout or failed stderr is parsed with `ansi_to_tui::IntoText`, stored as `PreviewContent::Text`, and followed by `Event::PreviewReady`.
|
||||||
|
|
||||||
**PTY mode** (`--preview-window pty`): creates a real pseudo-terminal pair via `portable_pty`. The child process sees a properly sized terminal (via `ROWS`/`COLUMNS` env and PTY dimensions). Output is parsed by a `vt100::Parser` with a scrollback buffer, stored as `PreviewContent::Terminal(Arc<RwLock<vt100::Parser>>)`. This enables interactive preview programs (e.g. `bat`, `delta`).
|
**PTY mode** (`--preview-window pty`): creates a real pseudo-terminal pair via `portable_pty`. The child process sees a properly sized terminal (via `ROWS`/`COLUMNS` env and PTY dimensions). Output is parsed by a `vt100::Parser` with a scrollback buffer, stored as `PreviewContent::Terminal(Arc<RwLock<vt100::Parser>>)`. This enables interactive preview programs (e.g. `bat`, `delta`).
|
||||||
|
|
||||||
|
|
@ -893,12 +895,14 @@ else if pty mode:
|
||||||
→ Event::PreviewReady when EOF
|
→ Event::PreviewReady when EOF
|
||||||
|
|
||||||
else:
|
else:
|
||||||
sh -c <cmd>
|
start shell in a dedicated process group with piped stdout + stderr
|
||||||
thread: wait for output → content.write() = PreviewContent::Text(…)
|
thread: drain both streams with bounded retention; poll child status
|
||||||
|
→ cancellation kills the process group
|
||||||
|
→ content.write() = PreviewContent::Text(…)
|
||||||
→ Event::PreviewReady
|
→ Event::PreviewReady
|
||||||
```
|
```
|
||||||
|
|
||||||
Scroll state: `scroll_y`, `scroll_x` (in lines/columns). `page_up/down`, `scroll_up/down/left/right` modify these. `PreviewPosition` supports fixed, percentage, and negative offsets. When `PreviewReady` fires, an optional offset expression (from `--preview-window +expr`) is evaluated to auto-scroll to the matched line.
|
Scroll state: `scroll_y`, `scroll_x` (in lines/columns) and `total_lines` use `usize`; conversion to ratatui's `u16` coordinates saturates at render time. `page_up/down`, `scroll_up/down/left/right` modify these. `PreviewPosition` supports fixed, percentage, and negative offsets. When `PreviewReady` fires, an optional offset expression (from `--preview-window +expr`) is evaluated to auto-scroll to the matched line.
|
||||||
|
|
||||||
### Header Widget
|
### Header Widget
|
||||||
|
|
||||||
|
|
@ -1275,21 +1279,23 @@ Main thread (Tokio runtime)
|
||||||
│
|
│
|
||||||
└─ Tokio task: Tui event pump (crossterm EventStream + tick timer)
|
└─ Tokio task: Tui event pump (crossterm EventStream + tick timer)
|
||||||
|
|
||||||
ThreadPool (N = num_cpus OS threads, persistent)
|
Matcher ThreadPool (persistent)
|
||||||
├─ Matcher coordinator (1 slot per match run)
|
└─ Worker threads process atomic match chunks; one separate coordinator thread waits for completion
|
||||||
└─ Worker threads (N-1 slots per match run)
|
|
||||||
|
|
||||||
Reader threads (OS threads, per-invocation):
|
Reader ThreadPool (persistent)
|
||||||
├─ collect_items thread: blocks on SkimItemReceiver (recv_timeout 5ms), calls ItemPool::append
|
└─ Short chunk jobs parse items; an in-flight token limit bounds the work queue
|
||||||
|
|
||||||
|
Reader threads (OS threads, per invocation):
|
||||||
|
├─ collect_items thread: blocks on SkimItemReceiver (recv_timeout 1ms), calls ItemPool::append
|
||||||
├─ I/O reader thread: reads large byte chunks, splits lines, assigns sequence numbers
|
├─ 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)
|
├─ Bounded dispatcher thread: submits chunk jobs only while an in-flight token is available
|
||||||
├─ Reorder thread: sequence-ordered output; drops tx_pipeline_done on EOF
|
├─ Reorder thread: sequence-ordered output; drops tx_pipeline_done on EOF
|
||||||
└─ Killer thread (command inputs only): waits for rx_interrupt or rx_pipeline_done;
|
└─ Killer thread: waits for rx_interrupt or rx_pipeline_done; kills a command child if present
|
||||||
kills child process when either fires
|
|
||||||
|
|
||||||
Preview thread (OS thread, per preview spawn):
|
Preview threads (OS threads, per preview spawn):
|
||||||
└─ reads PTY/child stdout or decodes image path → PreviewContent Arc<RwLock>
|
├─ PTY reader, image decoder, or plain-child monitor
|
||||||
→ sends Event::PreviewReady
|
└─ Plain mode also has bounded stdout and stderr drain threads
|
||||||
|
→ writes PreviewContent Arc<RwLock> and sends Event::PreviewReady
|
||||||
|
|
||||||
IPC handler task (Tokio, per connection):
|
IPC handler task (Tokio, per connection):
|
||||||
└─ reads RON actions → sends Event::Action to TUI channel
|
└─ reads RON actions → sends Event::Action to TUI channel
|
||||||
|
|
@ -1300,9 +1306,9 @@ Popup stdin relay thread (OS thread, only in --popup/--tmux mode):
|
||||||
|
|
||||||
**Synchronization primitives used:**
|
**Synchronization primitives used:**
|
||||||
|
|
||||||
- `Arc<SpinLock<Option<ProcessedItems>>>` — matcher-to-ItemList result handoff
|
- `Arc<Mutex<Option<ProcessedItems>>>` — matcher-to-ItemList result handoff without CPU-spinning under merge contention
|
||||||
- `Arc<AtomicBool>` — `needs_render` (matcher → event loop), `stopped` / `interrupt` (MatcherControl)
|
- `Arc<AtomicBool>` — `needs_render` (matcher → event loop), `stopped` / `interrupt` (MatcherControl)
|
||||||
- `Arc<AtomicUsize>` — `processed` / `matched` counters, reader `components_to_stop`
|
- `Arc<AtomicUsize>` — `processed` / `matched` counters, matcher query generation, reader `components_to_stop`
|
||||||
- `Arc<tokio::sync::Notify>` — `items_available` (ItemPool → Skim::tick wakeup)
|
- `Arc<tokio::sync::Notify>` — `items_available` (ItemPool → Skim::tick wakeup)
|
||||||
- `Arc<std::sync::RwLock<PreviewContent>>` — preview thread → Preview widget
|
- `Arc<std::sync::RwLock<PreviewContent>>` — preview thread → Preview widget
|
||||||
- `kanal::Sender/Receiver<Vec<Arc<dyn SkimItem>>>` — item batches through pipeline
|
- `kanal::Sender/Receiver<Vec<Arc<dyn SkimItem>>>` — item batches through pipeline
|
||||||
|
|
@ -1311,65 +1317,6 @@ The global allocator is `mimalloc` (v3), chosen for its low-latency multi-thread
|
||||||
|
|
||||||
---
|
---
|
||||||
|
|
||||||
## Important Call Sites (Cross-Reference)
|
|
||||||
|
|
||||||
| Call site | File | What it does |
|
|
||||||
| --- | --- | --- |
|
|
||||||
| `Skim::run_with` | `src/skim.rs:72` | Top-level library entry point |
|
|
||||||
| `Skim::run_items` | `src/skim.rs:114` | Convenience wrapper for iterator inputs |
|
|
||||||
| `Skim::init_tui` | `src/skim.rs:138` | Initialize default crossterm TUI backend |
|
|
||||||
| `Skim::init` | `src/skim.rs:159` | Constructs all subsystems from options |
|
|
||||||
| `Skim::start` | `src/skim.rs:203` | Starts reader + initial matcher pass |
|
|
||||||
| `Skim::handle_reload` | `src/skim.rs:235` | Kills reader, clears pool, restarts |
|
|
||||||
| `Skim::init_tui_with` | `src/skim.rs:307` | Install a caller-provided TUI backend |
|
|
||||||
| `Skim::enter` | `src/skim.rs:394` | Enter terminal, resolve image picker, start listener/event pump |
|
|
||||||
| `Skim::should_enter` | `src/skim.rs:438` | Filter/select-1/exit-0/sync gate |
|
|
||||||
| `Skim::output` | `src/skim.rs:555` | Collect & return SkimOutput |
|
|
||||||
| `Skim::tick` | `src/skim.rs:636` | Single async event loop iteration |
|
|
||||||
| `App::from_options` | `src/tui/app.rs:289` | Build all widgets from options |
|
|
||||||
| `App::run_preview` | `src/tui/app.rs:503` | Expand cmd, debounce, call Preview::spawn |
|
|
||||||
| `App::handle_event` | `src/tui/app.rs:628` | Dispatch all Event variants |
|
|
||||||
| `App::handle_action` | `src/tui/app.rs:833` | Apply action follow-up bindings |
|
|
||||||
| `App::dispatch_conditional` | `src/tui/app.rs:852` | Dispatch the selected conditional subaction chain without follow-up bindings |
|
|
||||||
| `App::dispatch_action` | `src/tui/app.rs:875` | Dispatch one Action variant without follow-up bindings |
|
|
||||||
| `Tui::run_execute` | `src/tui/backend.rs:363` | Suspend reader, run `execute` child with its own tty stdin, restart reader |
|
|
||||||
| `App::restart_matcher` | `src/tui/app.rs:1352` | Kill old match pass, start new one |
|
|
||||||
| `App::expand_cmd` | `src/tui/app.rs:1428` | Substitute `{}`, `{q}`, `{n}` etc. |
|
|
||||||
| `App::handle_mouse` | `src/tui/app.rs:1496` | Handle mouse behavior and emit `double-click` |
|
|
||||||
| `Widget::render (App)` | `src/tui/app.rs:151` | Root render; calls all sub-widgets |
|
|
||||||
| `Matcher::run` | `src/matcher.rs:~260` | Parallel match dispatch |
|
|
||||||
| `merge_worker_results` | `src/matcher.rs:28` | Merge k sorted runs → ProcessedItems |
|
|
||||||
| `ItemPool::append` | `src/item.rs:469` | Add items, notify matcher |
|
|
||||||
| `ItemPool::take` | `src/item.rs:502` | Take un-matched items for matcher |
|
|
||||||
| `DefaultSkimItem::new` | `src/helper/item.rs:64` | ANSI strip, field transform, matching ranges (hidden ranges set later via `hidden_fields` builder) |
|
|
||||||
| `SkimItemReader::parallel_bufread` | `src/helper/item_reader.rs:287` | Unified parallel pipeline (all inputs) |
|
|
||||||
| `spawn_io_reader` | `src/helper/item_reader.rs:378` | I/O reader thread: chunk reads + line splitting |
|
|
||||||
| `spawn_reorder_thread` | `src/helper/item_reader.rs:483` | Reorder thread: ordered output + pipeline-done signal |
|
|
||||||
| `Preview::spawn` | `src/tui/preview.rs:319` | Start image, PTY, or plain preview worker |
|
|
||||||
| `Tui::new_with_height_and_backend` | `src/tui/backend.rs:81` | Terminal init + viewport sizing |
|
|
||||||
| `Tui::enter` | `src/tui/backend.rs:134` | Enable raw mode + terminal setup |
|
|
||||||
| `Tui::start` | `src/tui/backend.rs:235` | Spawn crossterm EventStream task (fresh cancellation token each call) |
|
|
||||||
| `Tui::stop_and_join` | `src/tui/backend.rs:221` | Cancel event pump and block until `EventStream` is dropped (before `execute`) |
|
|
||||||
| `Tui::force_full_redraw` | `src/tui/backend.rs:202` | Reset ratatui diff buffers for a full repaint with no cursor query (after `execute`) |
|
|
||||||
| `Tui::min_height` | `src/tui/backend.rs:400` | Grow an inline viewport and scroll the terminal when needed |
|
|
||||||
| `popup::run_with` | `src/popup/mod.rs:86` | Delegate to multiplexer popup + parse output |
|
|
||||||
| `popup::check_env` | `src/popup/mod.rs:72` | Guard: multiplexer present and not already in popup |
|
|
||||||
| `check_and_run_popup` | `src/bin/main.rs:131` | Check popup conditions, dispatch to popup::run_with |
|
|
||||||
| `sk_main` | `src/bin/main.rs:144` | CLI orchestration + output printing |
|
|
||||||
| `SkimEvent` | `src/binds.rs:25` | Bindable synthetic events, including `double-click`, routed through reserved `KeyEvent`s |
|
|
||||||
| `parse_key` | `src/binds.rs:226` | `"ctrl-a"` → `KeyEvent` |
|
|
||||||
| `parse_action_binds` | `src/binds.rs:335` | `"reload:first"`, `"act-up:suppress+down"` → action follow-up map |
|
|
||||||
| `parse_action_chain` | `src/binds.rs:383` | `"down+select"` → `Vec<Action>` |
|
|
||||||
| `Action::name` | `src/tui/actions.rs:254` | `Action` → canonical bind name (generated from the `define_action_catalog!` list that also defines the `Action` enum and `parse_action`) |
|
|
||||||
| `Matcher::create_engine_factory_with_builder` | `src/matcher.rs:189` | Build engine factory chain from options |
|
|
||||||
| `ExactOrFuzzyEngineFactory::create_engine_with_case` | `src/engine/factory.rs:93` | Parse query prefixes, build engine |
|
|
||||||
| `AndOrEngineFactory::parse_andor` | `src/engine/factory.rs:176` | Split query into AND/OR tree |
|
|
||||||
| `FuzzyEngine::match_item` | `src/engine/fuzzy.rs:175` | Fuzzy match a single item |
|
|
||||||
| `LayoutTemplate::from_options` | `src/tui/layout.rs:76` | Compute widget constraint tree |
|
|
||||||
| `LayoutTemplate::apply` | `src/tui/layout.rs:165` | Split Rect into AppLayout |
|
|
||||||
| `ItemRenderer::render_item` | `src/tui/item_renderer.rs:84` | Full per-item render pipeline |
|
|
||||||
| `ColorTheme::init_from_options` | `src/theme.rs:56` | Parse `--color` spec |
|
|
||||||
|
|
||||||
---
|
---
|
||||||
|
|
||||||
## Public Library API
|
## Public Library API
|
||||||
|
|
|
||||||
|
|
@ -88,7 +88,7 @@ kanal = "0.1.1"
|
||||||
log = "0.4.31"
|
log = "0.4.31"
|
||||||
memchr = "2.8.1"
|
memchr = "2.8.1"
|
||||||
mimalloc = { version = "0.1.48", features = ["v3"] }
|
mimalloc = { version = "0.1.48", features = ["v3"] }
|
||||||
nix = { version = "0.31.3", features = ["fs", "poll"] }
|
nix = { version = "0.31.3", features = ["fs", "poll", "signal"] }
|
||||||
portable-pty = "0.9.0"
|
portable-pty = "0.9.0"
|
||||||
ratatui = "0.30.0"
|
ratatui = "0.30.0"
|
||||||
ratatui-image = { version = "11.0.4", features = ["crossterm"], default-features = false, optional = true }
|
ratatui-image = { version = "11.0.4", features = ["crossterm"], default-features = false, optional = true }
|
||||||
|
|
|
||||||
|
|
@ -265,16 +265,13 @@ impl SkimItemReader {
|
||||||
/// 1. **I/O thread** (dedicated) — reads large byte chunks (~256 KB) from
|
/// 1. **I/O thread** (dedicated) — reads large byte chunks (~256 KB) from
|
||||||
/// `source`, splitting on line boundaries, and sends them tagged with
|
/// `source`, splitting on line boundaries, and sends them tagged with
|
||||||
/// monotonic sequence numbers into a bounded channel.
|
/// monotonic sequence numbers into a bounded channel.
|
||||||
/// 2. **Dispatcher thread** (dedicated, lightweight) — drains that channel
|
/// 2. **Bounded dispatcher** — submits chunk jobs to the pool while a token
|
||||||
/// and submits one pool job per chunk. The bounded channel provides
|
/// limit bounds queued and running work. It stops draining the input
|
||||||
/// natural back-pressure on the I/O thread when the pool is busy.
|
/// channel when that limit is reached, which applies back-pressure to I/O.
|
||||||
/// 3. **Pool jobs** — parse lines, validate UTF-8, apply ANSI stripping and
|
/// 3. **Pool jobs** — parse lines, validate UTF-8, apply ANSI stripping and
|
||||||
/// field transforms, and create `DefaultSkimItem` + `Arc` per line.
|
/// field transforms, and create `DefaultSkimItem` + `Arc` per line.
|
||||||
/// Because these jobs share the same pool as the matcher, reader and
|
/// 4. **Reorder thread** (dedicated) — collects `(seq, items)` from workers
|
||||||
/// matcher compete for the same thread budget rather than over-subscribing
|
/// and emits them in sequence order so downstream index assignment
|
||||||
/// 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.
|
/// and `--tac` behaviour are correct.
|
||||||
///
|
///
|
||||||
/// When `child` is `Some`, a **killer thread** is also spawned. It waits
|
/// When `child` is `Some`, a **killer thread** is also spawned. It waits
|
||||||
|
|
@ -303,18 +300,27 @@ impl SkimItemReader {
|
||||||
// Stage 1: I/O thread.
|
// Stage 1: I/O thread.
|
||||||
Self::spawn_io_reader(source, tx_chunks, line_ending);
|
Self::spawn_io_reader(source, tx_chunks, line_ending);
|
||||||
|
|
||||||
// Stage 2: dispatcher thread — bridges the bounded channel to the pool.
|
// Stage 2: dispatch at most a fixed number of queued or running jobs.
|
||||||
|
// A worker returns its token only after it sends the parsed result.
|
||||||
|
let max_in_flight = num_threads * 4;
|
||||||
|
let (tx_permits, rx_permits) = std::sync::mpsc::sync_channel(max_in_flight);
|
||||||
|
for _ in 0..max_in_flight {
|
||||||
|
tx_permits.send(()).expect("permit receiver is alive");
|
||||||
|
}
|
||||||
thread::spawn(move || {
|
thread::spawn(move || {
|
||||||
while let Ok((seq, chunk)) = rx_chunks.recv() {
|
while let Ok((seq, chunk)) = rx_chunks.recv() {
|
||||||
|
if rx_permits.recv().is_err() {
|
||||||
|
break;
|
||||||
|
}
|
||||||
let tx = tx_results.clone();
|
let tx = tx_results.clone();
|
||||||
|
let return_permit = tx_permits.clone();
|
||||||
let opt = option.clone();
|
let opt = option.clone();
|
||||||
pool.spawn(move || {
|
pool.spawn(move || {
|
||||||
let result = Self::process_chunk(seq, &chunk, &opt);
|
let result = Self::process_chunk(seq, &chunk, &opt);
|
||||||
let _ = tx.send(result);
|
let _ = tx.send(result);
|
||||||
|
let _ = return_permit.send(());
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
// rx_chunks closed → all chunks dispatched; tx_results dropped here
|
|
||||||
// so the reorder thread exits once the last pool job finishes.
|
|
||||||
});
|
});
|
||||||
|
|
||||||
// A zero-capacity channel used as a completion signal: the reorder
|
// A zero-capacity channel used as a completion signal: the reorder
|
||||||
|
|
|
||||||
169
src/matcher.rs
169
src/matcher.rs
|
|
@ -2,14 +2,13 @@
|
||||||
use crate::thread_pool::{self, ThreadPool};
|
use crate::thread_pool::{self, ThreadPool};
|
||||||
use crate::tui::item_list::{MergeStrategy, ProcessedItems};
|
use crate::tui::item_list::{MergeStrategy, ProcessedItems};
|
||||||
use std::rc::Rc;
|
use std::rc::Rc;
|
||||||
use std::sync::Arc;
|
|
||||||
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
|
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
|
||||||
|
use std::sync::{Arc, Mutex};
|
||||||
|
|
||||||
use crate::engine::normalized::NormalizedEngineFactory;
|
use crate::engine::normalized::NormalizedEngineFactory;
|
||||||
use crate::engine::split::SplitMatchEngineFactory;
|
use crate::engine::split::SplitMatchEngineFactory;
|
||||||
use crate::item::{ItemPool, MatchedItem, RankBuilder};
|
use crate::item::{ItemPool, MatchedItem, RankBuilder};
|
||||||
use crate::prelude::{AndOrEngineFactory, ExactOrFuzzyEngineFactory, RegexEngineFactory};
|
use crate::prelude::{AndOrEngineFactory, ExactOrFuzzyEngineFactory, RegexEngineFactory};
|
||||||
use crate::spinlock::SpinLock;
|
|
||||||
use crate::{CaseMatching, MatchEngineFactory, SkimItem, SkimOptions};
|
use crate::{CaseMatching, MatchEngineFactory, SkimItem, SkimOptions};
|
||||||
|
|
||||||
/// Merges per-worker match results and writes them into `processed_items`.
|
/// Merges per-worker match results and writes them into `processed_items`.
|
||||||
|
|
@ -37,10 +36,16 @@ fn input_index(tac: bool, start: usize, batch_len: usize, batch_index: usize) ->
|
||||||
fn merge_worker_results(
|
fn merge_worker_results(
|
||||||
worker_results: Vec<Vec<MatchedItem>>,
|
worker_results: Vec<Vec<MatchedItem>>,
|
||||||
no_sort: bool,
|
no_sort: bool,
|
||||||
processed_items: &SpinLock<Option<ProcessedItems>>,
|
processed_items: &Mutex<Option<ProcessedItems>>,
|
||||||
merge_strategy: MergeStrategy,
|
merge_strategy: MergeStrategy,
|
||||||
|
generation: usize,
|
||||||
|
current_generation: &AtomicUsize,
|
||||||
needs_render: &AtomicBool,
|
needs_render: &AtomicBool,
|
||||||
) {
|
) {
|
||||||
|
if current_generation.load(Ordering::Acquire) != generation {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
let total_len: usize = worker_results.iter().map(Vec::len).sum();
|
let total_len: usize = worker_results.iter().map(Vec::len).sum();
|
||||||
let mut items = Vec::with_capacity(total_len);
|
let mut items = Vec::with_capacity(total_len);
|
||||||
for chunk in worker_results {
|
for chunk in worker_results {
|
||||||
|
|
@ -55,18 +60,32 @@ fn merge_worker_results(
|
||||||
|
|
||||||
trace!("matcher stop, total matched: {}", items.len());
|
trace!("matcher stop, total matched: {}", items.len());
|
||||||
|
|
||||||
// Single lock, single write into processed_items.
|
// Validate while holding the result lock so an old matcher cannot overwrite
|
||||||
let mut guard = processed_items.lock();
|
// results that belong to a newer query generation.
|
||||||
|
let mut guard = processed_items
|
||||||
|
.lock()
|
||||||
|
.unwrap_or_else(std::sync::PoisonError::into_inner);
|
||||||
|
if current_generation.load(Ordering::Acquire) != generation {
|
||||||
|
return;
|
||||||
|
}
|
||||||
if matches!(merge_strategy, MergeStrategy::Replace) {
|
if matches!(merge_strategy, MergeStrategy::Replace) {
|
||||||
*guard = Some(ProcessedItems {
|
*guard = Some(ProcessedItems {
|
||||||
items,
|
items,
|
||||||
merge: MergeStrategy::Replace,
|
merge: MergeStrategy::Replace,
|
||||||
|
generation,
|
||||||
});
|
});
|
||||||
drop(guard);
|
drop(guard);
|
||||||
needs_render.store(true, Ordering::Relaxed);
|
needs_render.store(true, Ordering::Relaxed);
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
match &mut *guard {
|
match &mut *guard {
|
||||||
|
Some(existing) if existing.generation != generation => {
|
||||||
|
*guard = Some(ProcessedItems {
|
||||||
|
items,
|
||||||
|
merge: MergeStrategy::Replace,
|
||||||
|
generation,
|
||||||
|
});
|
||||||
|
}
|
||||||
Some(existing) => {
|
Some(existing) => {
|
||||||
if no_sort {
|
if no_sort {
|
||||||
if matches!(merge_strategy, MergeStrategy::Prepend) {
|
if matches!(merge_strategy, MergeStrategy::Prepend) {
|
||||||
|
|
@ -84,6 +103,7 @@ fn merge_worker_results(
|
||||||
*guard = Some(ProcessedItems {
|
*guard = Some(ProcessedItems {
|
||||||
items,
|
items,
|
||||||
merge: merge_strategy,
|
merge: merge_strategy,
|
||||||
|
generation,
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
@ -200,13 +220,14 @@ impl Matcher {
|
||||||
#[must_use]
|
#[must_use]
|
||||||
pub fn create_engine_factory_with_builder(options: &SkimOptions) -> (Rc<dyn MatchEngineFactory>, Arc<RankBuilder>) {
|
pub fn create_engine_factory_with_builder(options: &SkimOptions) -> (Rc<dyn MatchEngineFactory>, Arc<RankBuilder>) {
|
||||||
if options.regex {
|
if options.regex {
|
||||||
let regex_factory = RegexEngineFactory::builder();
|
let rank_builder = Arc::new(RankBuilder::new(options.tiebreak.clone()).tac(options.tac));
|
||||||
|
let regex_factory = RegexEngineFactory::builder().rank_builder(rank_builder.clone());
|
||||||
let factory: Rc<dyn MatchEngineFactory> = if options.normalize {
|
let factory: Rc<dyn MatchEngineFactory> = if options.normalize {
|
||||||
Rc::new(NormalizedEngineFactory::new(regex_factory))
|
Rc::new(NormalizedEngineFactory::new(regex_factory))
|
||||||
} else {
|
} else {
|
||||||
Rc::new(regex_factory)
|
Rc::new(regex_factory)
|
||||||
};
|
};
|
||||||
(factory, Arc::new(RankBuilder::default().tac(options.tac)))
|
(factory, rank_builder)
|
||||||
} else {
|
} else {
|
||||||
let rank_builder = Arc::new(RankBuilder::new(options.tiebreak.clone()).tac(options.tac));
|
let rank_builder = Arc::new(RankBuilder::new(options.tiebreak.clone()).tac(options.tac));
|
||||||
log::debug!("Creating matcher for algo {:?}", options.algorithm);
|
log::debug!("Creating matcher for algo {:?}", options.algorithm);
|
||||||
|
|
@ -275,10 +296,12 @@ impl Matcher {
|
||||||
query: &str,
|
query: &str,
|
||||||
item_pool: &Arc<ItemPool>,
|
item_pool: &Arc<ItemPool>,
|
||||||
thread_pool: &Arc<ThreadPool>,
|
thread_pool: &Arc<ThreadPool>,
|
||||||
processed_items: Arc<SpinLock<Option<ProcessedItems>>>,
|
processed_items: Arc<Mutex<Option<ProcessedItems>>>,
|
||||||
merge_strategy: MergeStrategy,
|
merge_strategy: MergeStrategy,
|
||||||
no_sort: bool,
|
no_sort: bool,
|
||||||
tac: bool,
|
tac: bool,
|
||||||
|
generation: usize,
|
||||||
|
current_generation: Arc<AtomicUsize>,
|
||||||
needs_render: Arc<AtomicBool>,
|
needs_render: Arc<AtomicBool>,
|
||||||
) -> MatcherControl {
|
) -> MatcherControl {
|
||||||
let matcher_engine = self.engine_factory.create_engine_with_case(query, self.case_matching);
|
let matcher_engine = self.engine_factory.create_engine_with_case(query, self.case_matching);
|
||||||
|
|
@ -401,7 +424,15 @@ impl Matcher {
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
merge_worker_results(worker_results, no_sort, &processed_items, merge_strategy, &needs_render);
|
merge_worker_results(
|
||||||
|
worker_results,
|
||||||
|
no_sort,
|
||||||
|
&processed_items,
|
||||||
|
merge_strategy,
|
||||||
|
generation,
|
||||||
|
¤t_generation,
|
||||||
|
&needs_render,
|
||||||
|
);
|
||||||
},
|
},
|
||||||
);
|
);
|
||||||
stopped.store(true, Ordering::Relaxed);
|
stopped.store(true, Ordering::Relaxed);
|
||||||
|
|
@ -469,31 +500,123 @@ mod tests {
|
||||||
assert!(engine.match_item(&"foobar".to_string()).is_some());
|
assert!(engine.match_item(&"foobar".to_string()).is_some());
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn regex_factory_uses_configured_tiebreak() {
|
||||||
|
let options = SkimOptionsBuilder::default()
|
||||||
|
.regex(true)
|
||||||
|
.tiebreak(vec![crate::RankCriteria::Length])
|
||||||
|
.build()
|
||||||
|
.unwrap();
|
||||||
|
let (factory, rank_builder) = Matcher::create_engine_factory_with_builder(&options);
|
||||||
|
let engine = factory.create_engine("a");
|
||||||
|
let mut matches: Vec<_> = ["aaaa", "a", "aaa"]
|
||||||
|
.into_iter()
|
||||||
|
.enumerate()
|
||||||
|
.map(|(index, text)| {
|
||||||
|
let item: Arc<dyn SkimItem> = Arc::new(text.to_string());
|
||||||
|
let mut result = engine.match_item(item.as_ref()).unwrap();
|
||||||
|
result.rank.index = i32::try_from(index).unwrap();
|
||||||
|
MatchedItem::new(item, result.rank, Some(result.matched_range), &rank_builder)
|
||||||
|
})
|
||||||
|
.collect();
|
||||||
|
matches.sort();
|
||||||
|
|
||||||
|
let output: Vec<_> = matches.iter().map(|item| item.text().into_owned()).collect();
|
||||||
|
assert_eq!(output, ["a", "aaa", "aaaa"]);
|
||||||
|
}
|
||||||
|
|
||||||
|
fn merge_test_results(
|
||||||
|
worker_results: Vec<Vec<MatchedItem>>,
|
||||||
|
no_sort: bool,
|
||||||
|
processed_items: &Mutex<Option<ProcessedItems>>,
|
||||||
|
merge_strategy: MergeStrategy,
|
||||||
|
needs_render: &AtomicBool,
|
||||||
|
) {
|
||||||
|
let generation = AtomicUsize::new(0);
|
||||||
|
merge_worker_results(
|
||||||
|
worker_results,
|
||||||
|
no_sort,
|
||||||
|
processed_items,
|
||||||
|
merge_strategy,
|
||||||
|
0,
|
||||||
|
&generation,
|
||||||
|
needs_render,
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn merge_worker_results_replace_sorts() {
|
fn merge_worker_results_replace_sorts() {
|
||||||
let processed = SpinLock::new(None);
|
let processed = Mutex::new(None);
|
||||||
let needs_render = AtomicBool::new(false);
|
let needs_render = AtomicBool::new(false);
|
||||||
let workers = vec![vec![matched("b", 1)], vec![matched("a", 0)]];
|
let workers = vec![vec![matched("b", 1)], vec![matched("a", 0)]];
|
||||||
merge_worker_results(workers, false, &processed, MergeStrategy::Replace, &needs_render);
|
merge_test_results(workers, false, &processed, MergeStrategy::Replace, &needs_render);
|
||||||
|
|
||||||
assert!(needs_render.load(Ordering::Relaxed));
|
assert!(needs_render.load(Ordering::Relaxed));
|
||||||
let guard = processed.lock();
|
let guard = processed.lock().unwrap();
|
||||||
let items = &guard.as_ref().unwrap().items;
|
let items = &guard.as_ref().unwrap().items;
|
||||||
assert_eq!(items.len(), 2);
|
assert_eq!(items.len(), 2);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn stale_generation_cannot_publish_results() {
|
||||||
|
let processed = Mutex::new(None);
|
||||||
|
let needs_render = AtomicBool::new(false);
|
||||||
|
let generation = AtomicUsize::new(2);
|
||||||
|
|
||||||
|
merge_worker_results(
|
||||||
|
vec![vec![matched("stale", 0)]],
|
||||||
|
false,
|
||||||
|
&processed,
|
||||||
|
MergeStrategy::Replace,
|
||||||
|
1,
|
||||||
|
&generation,
|
||||||
|
&needs_render,
|
||||||
|
);
|
||||||
|
|
||||||
|
assert!(processed.lock().unwrap().is_none());
|
||||||
|
assert!(!needs_render.load(Ordering::Relaxed));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn current_results_replace_pending_stale_generation() {
|
||||||
|
let processed = Mutex::new(Some(ProcessedItems {
|
||||||
|
items: vec![matched("stale", 0)],
|
||||||
|
merge: MergeStrategy::SortedMerge,
|
||||||
|
generation: 0,
|
||||||
|
}));
|
||||||
|
let needs_render = AtomicBool::new(false);
|
||||||
|
let generation = AtomicUsize::new(1);
|
||||||
|
|
||||||
|
merge_worker_results(
|
||||||
|
vec![vec![matched("current", 1)]],
|
||||||
|
false,
|
||||||
|
&processed,
|
||||||
|
MergeStrategy::SortedMerge,
|
||||||
|
1,
|
||||||
|
&generation,
|
||||||
|
&needs_render,
|
||||||
|
);
|
||||||
|
|
||||||
|
let guard = processed.lock().unwrap();
|
||||||
|
let result = guard.as_ref().unwrap();
|
||||||
|
assert_eq!(result.generation, 1);
|
||||||
|
assert_eq!(result.items.len(), 1);
|
||||||
|
assert_eq!(result.items[0].text(), "current");
|
||||||
|
assert!(matches!(result.merge, MergeStrategy::Replace));
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn merge_worker_results_no_sort_preserves_chunk_order() {
|
fn merge_worker_results_no_sort_preserves_chunk_order() {
|
||||||
let processed = SpinLock::new(None);
|
let processed = Mutex::new(None);
|
||||||
let needs_render = AtomicBool::new(false);
|
let needs_render = AtomicBool::new(false);
|
||||||
let workers = vec![
|
let workers = vec![
|
||||||
vec![matched("a", 0), matched("b", 1)],
|
vec![matched("a", 0), matched("b", 1)],
|
||||||
vec![matched("c", 2), matched("d", 3)],
|
vec![matched("c", 2), matched("d", 3)],
|
||||||
vec![matched("e", 4), matched("f", 5)],
|
vec![matched("e", 4), matched("f", 5)],
|
||||||
];
|
];
|
||||||
merge_worker_results(workers, true, &processed, MergeStrategy::Replace, &needs_render);
|
merge_test_results(workers, true, &processed, MergeStrategy::Replace, &needs_render);
|
||||||
|
|
||||||
let guard = processed.lock();
|
let guard = processed.lock().unwrap();
|
||||||
let items = &guard.as_ref().unwrap().items;
|
let items = &guard.as_ref().unwrap().items;
|
||||||
let indexes: Vec<i32> = items.iter().map(|item| item.rank.index).collect();
|
let indexes: Vec<i32> = items.iter().map(|item| item.rank.index).collect();
|
||||||
assert_eq!(indexes, vec![0, 1, 2, 3, 4, 5]);
|
assert_eq!(indexes, vec![0, 1, 2, 3, 4, 5]);
|
||||||
|
|
@ -501,11 +624,11 @@ mod tests {
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn merge_worker_results_append_no_sort_extends_existing() {
|
fn merge_worker_results_append_no_sort_extends_existing() {
|
||||||
let processed = SpinLock::new(None);
|
let processed = Mutex::new(None);
|
||||||
let needs_render = AtomicBool::new(false);
|
let needs_render = AtomicBool::new(false);
|
||||||
|
|
||||||
// First append establishes the existing list.
|
// First append establishes the existing list.
|
||||||
merge_worker_results(
|
merge_test_results(
|
||||||
vec![vec![matched("a", 0)]],
|
vec![vec![matched("a", 0)]],
|
||||||
true,
|
true,
|
||||||
&processed,
|
&processed,
|
||||||
|
|
@ -513,7 +636,7 @@ mod tests {
|
||||||
&needs_render,
|
&needs_render,
|
||||||
);
|
);
|
||||||
// Second append with no_sort extends the existing list in place.
|
// Second append with no_sort extends the existing list in place.
|
||||||
merge_worker_results(
|
merge_test_results(
|
||||||
vec![vec![matched("b", 1)]],
|
vec![vec![matched("b", 1)]],
|
||||||
true,
|
true,
|
||||||
&processed,
|
&processed,
|
||||||
|
|
@ -521,7 +644,7 @@ mod tests {
|
||||||
&needs_render,
|
&needs_render,
|
||||||
);
|
);
|
||||||
|
|
||||||
let guard = processed.lock();
|
let guard = processed.lock().unwrap();
|
||||||
assert_eq!(guard.as_ref().unwrap().items.len(), 2);
|
assert_eq!(guard.as_ref().unwrap().items.len(), 2);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -538,17 +661,17 @@ mod tests {
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn merge_worker_results_prepend_no_sort_places_new_batch_first() {
|
fn merge_worker_results_prepend_no_sort_places_new_batch_first() {
|
||||||
let processed = SpinLock::new(None);
|
let processed = Mutex::new(None);
|
||||||
let needs_render = AtomicBool::new(false);
|
let needs_render = AtomicBool::new(false);
|
||||||
|
|
||||||
merge_worker_results(
|
merge_test_results(
|
||||||
vec![vec![matched("c", 2), matched("b", 1), matched("a", 0)]],
|
vec![vec![matched("c", 2), matched("b", 1), matched("a", 0)]],
|
||||||
true,
|
true,
|
||||||
&processed,
|
&processed,
|
||||||
MergeStrategy::Prepend,
|
MergeStrategy::Prepend,
|
||||||
&needs_render,
|
&needs_render,
|
||||||
);
|
);
|
||||||
merge_worker_results(
|
merge_test_results(
|
||||||
vec![vec![matched("e", 4), matched("d", 3)]],
|
vec![vec![matched("e", 4), matched("d", 3)]],
|
||||||
true,
|
true,
|
||||||
&processed,
|
&processed,
|
||||||
|
|
@ -556,7 +679,7 @@ mod tests {
|
||||||
&needs_render,
|
&needs_render,
|
||||||
);
|
);
|
||||||
|
|
||||||
let guard = processed.lock();
|
let guard = processed.lock().unwrap();
|
||||||
let indexes: Vec<i32> = guard
|
let indexes: Vec<i32> = guard
|
||||||
.as_ref()
|
.as_ref()
|
||||||
.unwrap()
|
.unwrap()
|
||||||
|
|
|
||||||
|
|
@ -10,7 +10,9 @@ use crate::{SkimItem, SkimItemReceiver};
|
||||||
use std::cell::RefCell;
|
use std::cell::RefCell;
|
||||||
use std::rc::Rc;
|
use std::rc::Rc;
|
||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
|
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||||
|
use std::thread::JoinHandle;
|
||||||
|
use std::time::Duration;
|
||||||
|
|
||||||
/// Trait for collecting items from command output
|
/// Trait for collecting items from command output
|
||||||
pub trait CommandCollector {
|
pub trait CommandCollector {
|
||||||
|
|
@ -39,6 +41,7 @@ pub struct ReaderControl {
|
||||||
tx_interrupt: Sender<i32>,
|
tx_interrupt: Sender<i32>,
|
||||||
tx_interrupt_cmd: Option<Sender<i32>>,
|
tx_interrupt_cmd: Option<Sender<i32>>,
|
||||||
components_to_stop: Arc<AtomicUsize>,
|
components_to_stop: Arc<AtomicUsize>,
|
||||||
|
collector_handle: Option<JoinHandle<()>>,
|
||||||
items: Arc<SpinLock<Vec<Arc<dyn SkimItem>>>>,
|
items: Arc<SpinLock<Vec<Arc<dyn SkimItem>>>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -52,7 +55,14 @@ impl ReaderControl {
|
||||||
|
|
||||||
let _ = self.tx_interrupt_cmd.clone().map(|tx| tx.send(1));
|
let _ = self.tx_interrupt_cmd.clone().map(|tx| tx.send(1));
|
||||||
let _ = self.tx_interrupt.send(1);
|
let _ = self.tx_interrupt.send(1);
|
||||||
while self.components_to_stop.load(Ordering::SeqCst) != 0 {}
|
if let Some(handle) = self.collector_handle.take() {
|
||||||
|
let _ = handle.join();
|
||||||
|
}
|
||||||
|
// Command collectors can own additional components outside the reader's
|
||||||
|
// join handle. Wait without consuming a CPU while they process the signal.
|
||||||
|
while self.components_to_stop.load(Ordering::Acquire) != 0 {
|
||||||
|
std::thread::sleep(Duration::from_millis(1));
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Takes all items collected so far
|
/// Takes all items collected so far
|
||||||
|
|
@ -123,12 +133,14 @@ impl Reader {
|
||||||
);
|
);
|
||||||
|
|
||||||
let components_to_stop_clone = components_to_stop.clone();
|
let components_to_stop_clone = components_to_stop.clone();
|
||||||
let tx_interrupt = collect_items(components_to_stop_clone, rx_item, move |items| _ = app_tx.send(items));
|
let (tx_interrupt, collector_handle) =
|
||||||
|
collect_items(components_to_stop_clone, rx_item, move |items| _ = app_tx.send(items));
|
||||||
|
|
||||||
ReaderControl {
|
ReaderControl {
|
||||||
tx_interrupt,
|
tx_interrupt,
|
||||||
tx_interrupt_cmd,
|
tx_interrupt_cmd,
|
||||||
components_to_stop,
|
components_to_stop,
|
||||||
|
collector_handle: Some(collector_handle),
|
||||||
items,
|
items,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
@ -149,7 +161,7 @@ impl Reader {
|
||||||
);
|
);
|
||||||
|
|
||||||
let components_to_stop_clone = components_to_stop.clone();
|
let components_to_stop_clone = components_to_stop.clone();
|
||||||
let tx_interrupt = collect_items(components_to_stop_clone, rx_item, move |items| {
|
let (tx_interrupt, collector_handle) = collect_items(components_to_stop_clone, rx_item, move |items| {
|
||||||
item_pool.append(items);
|
item_pool.append(items);
|
||||||
});
|
});
|
||||||
debug!("collect: started ({components_to_stop:?} components)");
|
debug!("collect: started ({components_to_stop:?} components)");
|
||||||
|
|
@ -158,6 +170,7 @@ impl Reader {
|
||||||
tx_interrupt,
|
tx_interrupt,
|
||||||
tx_interrupt_cmd,
|
tx_interrupt_cmd,
|
||||||
components_to_stop,
|
components_to_stop,
|
||||||
|
collector_handle: Some(collector_handle),
|
||||||
items,
|
items,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
@ -172,18 +185,19 @@ impl Default for Reader {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
fn collect_items<F>(components_to_stop: Arc<AtomicUsize>, rx_item: SkimItemReceiver, callback: F) -> Sender<i32>
|
fn collect_items<F>(
|
||||||
|
components_to_stop: Arc<AtomicUsize>,
|
||||||
|
rx_item: SkimItemReceiver,
|
||||||
|
callback: F,
|
||||||
|
) -> (Sender<i32>, JoinHandle<()>)
|
||||||
where
|
where
|
||||||
F: Fn(Vec<Arc<dyn SkimItem>>) + Send + 'static,
|
F: Fn(Vec<Arc<dyn SkimItem>>) + Send + 'static,
|
||||||
{
|
{
|
||||||
let (tx_interrupt, rx_interrupt) = crate::prelude::bounded(8);
|
let (tx_interrupt, rx_interrupt) = crate::prelude::bounded(8);
|
||||||
|
|
||||||
let started = Arc::new(AtomicBool::new(false));
|
components_to_stop.fetch_add(1, Ordering::AcqRel);
|
||||||
let started_clone = started.clone();
|
let handle = std::thread::spawn(move || {
|
||||||
std::thread::spawn(move || {
|
|
||||||
debug!("collect_item start");
|
debug!("collect_item start");
|
||||||
components_to_stop.fetch_add(1, Ordering::SeqCst);
|
|
||||||
started_clone.store(true, Ordering::SeqCst); // notify parent that it is started
|
|
||||||
|
|
||||||
loop {
|
loop {
|
||||||
if let Ok(Some(msg)) = rx_interrupt.try_recv() {
|
if let Ok(Some(msg)) = rx_interrupt.try_recv() {
|
||||||
|
|
@ -205,15 +219,11 @@ where
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
components_to_stop.fetch_sub(1, Ordering::SeqCst);
|
components_to_stop.fetch_sub(1, Ordering::AcqRel);
|
||||||
debug!("collect_item stop");
|
debug!("collect_item stop");
|
||||||
});
|
});
|
||||||
|
|
||||||
while !started.load(Ordering::SeqCst) {
|
(tx_interrupt, handle)
|
||||||
// busy waiting for the thread to start. (components_to_stop is added)
|
|
||||||
}
|
|
||||||
|
|
||||||
tx_interrupt
|
|
||||||
}
|
}
|
||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
|
|
|
||||||
10
src/skim.rs
10
src/skim.rs
|
|
@ -470,6 +470,7 @@ where
|
||||||
.item_list
|
.item_list
|
||||||
.processed_items
|
.processed_items
|
||||||
.lock()
|
.lock()
|
||||||
|
.unwrap_or_else(std::sync::PoisonError::into_inner)
|
||||||
.take()
|
.take()
|
||||||
.unwrap_or_default()
|
.unwrap_or_default()
|
||||||
.items
|
.items
|
||||||
|
|
@ -521,7 +522,14 @@ where
|
||||||
app.matcher_control.get_num_matched()
|
app.matcher_control.get_num_matched()
|
||||||
);
|
);
|
||||||
if app.matcher_control.get_num_matched() == min_items_before_enter - 1 {
|
if app.matcher_control.get_num_matched() == min_items_before_enter - 1 {
|
||||||
app.item_list.items = app.item_list.processed_items.lock().take().unwrap_or_default().items;
|
app.item_list.items = app
|
||||||
|
.item_list
|
||||||
|
.processed_items
|
||||||
|
.lock()
|
||||||
|
.unwrap_or_else(std::sync::PoisonError::into_inner)
|
||||||
|
.take()
|
||||||
|
.unwrap_or_default()
|
||||||
|
.items;
|
||||||
debug!("early exit, result: {:?}", app.results());
|
debug!("early exit, result: {:?}", app.results());
|
||||||
return false;
|
return false;
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -26,9 +26,9 @@ pub struct SpinLockGuard<'a, T: ?Sized + 'a> {
|
||||||
}
|
}
|
||||||
|
|
||||||
impl<'a, T: ?Sized + 'a> SpinLockGuard<'a, T> {
|
impl<'a, T: ?Sized + 'a> SpinLockGuard<'a, T> {
|
||||||
/// Creates a new guard for the given lock
|
/// Creates a guard after its lock has been acquired.
|
||||||
pub fn new(pool: &'a SpinLock<T>) -> SpinLockGuard<'a, T> {
|
fn new(lock: &'a SpinLock<T>) -> SpinLockGuard<'a, T> {
|
||||||
Self { __lock: pool }
|
Self { __lock: lock }
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -582,7 +582,7 @@ impl App {
|
||||||
u16::try_from(u32::from(self.preview.rows) * u32::from(p) / 100).unwrap_or(u16::MAX)
|
u16::try_from(u32::from(self.preview.rows) * u32::from(p) / 100).unwrap_or(u16::MAX)
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
self.preview.scroll_y = v_scroll;
|
self.preview.scroll_y = usize::from(v_scroll);
|
||||||
self.preview.scroll_down(v_offset);
|
self.preview.scroll_down(v_offset);
|
||||||
|
|
||||||
let h_scroll = match preview_position.h_scroll {
|
let h_scroll = match preview_position.h_scroll {
|
||||||
|
|
@ -599,7 +599,7 @@ impl App {
|
||||||
u16::try_from(u32::from(self.preview.cols) * u32::from(p) / 100).unwrap_or(u16::MAX)
|
u16::try_from(u32::from(self.preview.cols) * u32::from(p) / 100).unwrap_or(u16::MAX)
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
self.preview.scroll_x = h_scroll.saturating_add(h_offset);
|
self.preview.scroll_x = usize::from(h_scroll).saturating_add(usize::from(h_offset));
|
||||||
}
|
}
|
||||||
ItemPreview::TextWithPos(t, preview_position) | ItemPreview::AnsiWithPos(t, preview_position) => self
|
ItemPreview::TextWithPos(t, preview_position) | ItemPreview::AnsiWithPos(t, preview_position) => self
|
||||||
.preview
|
.preview
|
||||||
|
|
@ -1381,6 +1381,12 @@ impl App {
|
||||||
if self.query_below_min_length() {
|
if self.query_below_min_length() {
|
||||||
// Query is too short, clear items and don't run matcher
|
// Query is too short, clear items and don't run matcher
|
||||||
self.matcher_control.kill();
|
self.matcher_control.kill();
|
||||||
|
self.item_list.matcher_generation.fetch_add(1, Ordering::AcqRel);
|
||||||
|
self.item_list
|
||||||
|
.processed_items
|
||||||
|
.lock()
|
||||||
|
.unwrap_or_else(std::sync::PoisonError::into_inner)
|
||||||
|
.take();
|
||||||
self.item_list.items.clear();
|
self.item_list.items.clear();
|
||||||
self.item_list.current = 0;
|
self.item_list.current = 0;
|
||||||
self.item_list.offset = 0;
|
self.item_list.offset = 0;
|
||||||
|
|
@ -1413,6 +1419,12 @@ impl App {
|
||||||
self.item_pool.reset();
|
self.item_pool.reset();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
let generation = if force {
|
||||||
|
self.item_list.matcher_generation.fetch_add(1, Ordering::AcqRel) + 1
|
||||||
|
} else {
|
||||||
|
self.item_list.matcher_generation.load(Ordering::Acquire)
|
||||||
|
};
|
||||||
|
|
||||||
let merge_strategy = if force {
|
let merge_strategy = if force {
|
||||||
MergeStrategy::Replace
|
MergeStrategy::Replace
|
||||||
} else if no_sort && self.options.tac {
|
} else if no_sort && self.options.tac {
|
||||||
|
|
@ -1431,6 +1443,8 @@ impl App {
|
||||||
merge_strategy,
|
merge_strategy,
|
||||||
no_sort,
|
no_sort,
|
||||||
self.options.tac,
|
self.options.tac,
|
||||||
|
generation,
|
||||||
|
self.item_list.matcher_generation.clone(),
|
||||||
self.needs_render.clone(),
|
self.needs_render.clone(),
|
||||||
);
|
);
|
||||||
// A new search is in flight; arm the `result`/`zero`/`one` events to
|
// A new search is in flight; arm the `result`/`zero`/`one` events to
|
||||||
|
|
|
||||||
|
|
@ -1,5 +1,6 @@
|
||||||
use std::rc::Rc;
|
use std::rc::Rc;
|
||||||
use std::sync::Arc;
|
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||||
|
use std::sync::{Arc, Mutex};
|
||||||
|
|
||||||
use indexmap::IndexSet;
|
use indexmap::IndexSet;
|
||||||
use ratatui::widgets::{
|
use ratatui::widgets::{
|
||||||
|
|
@ -10,7 +11,6 @@ use regex::Regex;
|
||||||
|
|
||||||
use crate::item::MatchedItem;
|
use crate::item::MatchedItem;
|
||||||
use crate::options::feature_flag;
|
use crate::options::feature_flag;
|
||||||
use crate::spinlock::SpinLock;
|
|
||||||
use crate::theme::ColorTheme;
|
use crate::theme::ColorTheme;
|
||||||
use crate::tui::BorderType;
|
use crate::tui::BorderType;
|
||||||
use crate::tui::item_renderer::ItemRenderer;
|
use crate::tui::item_renderer::ItemRenderer;
|
||||||
|
|
@ -36,6 +36,7 @@ pub(crate) enum MergeStrategy {
|
||||||
pub(crate) struct ProcessedItems {
|
pub(crate) struct ProcessedItems {
|
||||||
pub(crate) items: Vec<MatchedItem>,
|
pub(crate) items: Vec<MatchedItem>,
|
||||||
pub(crate) merge: MergeStrategy,
|
pub(crate) merge: MergeStrategy,
|
||||||
|
pub(crate) generation: usize,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Default for ProcessedItems {
|
impl Default for ProcessedItems {
|
||||||
|
|
@ -43,6 +44,7 @@ impl Default for ProcessedItems {
|
||||||
Self {
|
Self {
|
||||||
items: Vec::new(),
|
items: Vec::new(),
|
||||||
merge: MergeStrategy::Replace,
|
merge: MergeStrategy::Replace,
|
||||||
|
generation: 0,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
@ -52,7 +54,8 @@ impl Default for ProcessedItems {
|
||||||
pub struct ItemList {
|
pub struct ItemList {
|
||||||
pub(crate) items: Vec<MatchedItem>,
|
pub(crate) items: Vec<MatchedItem>,
|
||||||
pub(crate) selection: IndexSet<MatchedItem>,
|
pub(crate) selection: IndexSet<MatchedItem>,
|
||||||
pub(crate) processed_items: Arc<SpinLock<Option<ProcessedItems>>>,
|
pub(crate) processed_items: Arc<Mutex<Option<ProcessedItems>>>,
|
||||||
|
pub(crate) matcher_generation: Arc<AtomicUsize>,
|
||||||
pub(crate) direction: ListDirection,
|
pub(crate) direction: ListDirection,
|
||||||
pub(crate) offset: usize,
|
pub(crate) offset: usize,
|
||||||
/// How many leading sub-lines of items[offset] have been scrolled off the top.
|
/// How many leading sub-lines of items[offset] have been scrolled off the top.
|
||||||
|
|
@ -420,7 +423,8 @@ impl SkimWidget for ItemList {
|
||||||
(None, 0)
|
(None, 0)
|
||||||
};
|
};
|
||||||
|
|
||||||
let processed_items = Arc::new(SpinLock::new(None));
|
let processed_items = Arc::new(Mutex::new(None));
|
||||||
|
let matcher_generation = Arc::new(AtomicUsize::new(0));
|
||||||
|
|
||||||
let interactive = options.interactive;
|
let interactive = options.interactive;
|
||||||
let no_clear_if_empty = options.no_clear_if_empty;
|
let no_clear_if_empty = options.no_clear_if_empty;
|
||||||
|
|
@ -429,6 +433,7 @@ impl SkimWidget for ItemList {
|
||||||
// Spawn background processing thread with the appropriate configuration
|
// Spawn background processing thread with the appropriate configuration
|
||||||
Self {
|
Self {
|
||||||
processed_items,
|
processed_items,
|
||||||
|
matcher_generation,
|
||||||
reserved: 0, // header_lines are now displayed in the Header widget, not ItemList
|
reserved: 0, // header_lines are now displayed in the Header widget, not ItemList
|
||||||
direction: match options.layout {
|
direction: match options.layout {
|
||||||
TuiLayout::Default => ratatui::widgets::ListDirection::BottomToTop,
|
TuiLayout::Default => ratatui::widgets::ListDirection::BottomToTop,
|
||||||
|
|
@ -511,7 +516,13 @@ impl SkimWidget for ItemList {
|
||||||
// Check for pre-processed items from background thread (non-blocking).
|
// Check for pre-processed items from background thread (non-blocking).
|
||||||
// Bind the result separately so the lock guard is dropped before a merge
|
// Bind the result separately so the lock guard is dropped before a merge
|
||||||
// mutates the item list.
|
// mutates the item list.
|
||||||
let processed = this.processed_items.lock().take();
|
let processed = this
|
||||||
|
.processed_items
|
||||||
|
.lock()
|
||||||
|
.unwrap_or_else(std::sync::PoisonError::into_inner)
|
||||||
|
.take();
|
||||||
|
let current_generation = this.matcher_generation.load(Ordering::Acquire);
|
||||||
|
let processed = processed.filter(|result| result.generation == current_generation);
|
||||||
let items_updated = if let Some(processed) = processed {
|
let items_updated = if let Some(processed) = processed {
|
||||||
debug!("Render: Got {} processed items", processed.items.len());
|
debug!("Render: Got {} processed items", processed.items.len());
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -240,7 +240,11 @@ fn render_list(il: &mut ItemList, w: u16, h: u16) {
|
||||||
}
|
}
|
||||||
|
|
||||||
fn set_processed(il: &ItemList, items: Vec<MatchedItem>, merge: MergeStrategy) {
|
fn set_processed(il: &ItemList, items: Vec<MatchedItem>, merge: MergeStrategy) {
|
||||||
*il.processed_items.lock() = Some(ProcessedItems { items, merge });
|
*il.processed_items.lock().unwrap() = Some(ProcessedItems {
|
||||||
|
items,
|
||||||
|
merge,
|
||||||
|
generation: il.matcher_generation.load(std::sync::atomic::Ordering::Acquire),
|
||||||
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
|
|
@ -253,6 +257,21 @@ fn render_applies_replace_strategy() {
|
||||||
assert_eq!(il.items[0].text(), "new");
|
assert_eq!(il.items[0].text(), "new");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn render_discards_results_from_stale_generation() {
|
||||||
|
let mut il = list(2);
|
||||||
|
*il.processed_items.lock().unwrap() = Some(ProcessedItems {
|
||||||
|
items: vec![matched("stale", 0)],
|
||||||
|
merge: MergeStrategy::Replace,
|
||||||
|
generation: 0,
|
||||||
|
});
|
||||||
|
il.matcher_generation.store(1, std::sync::atomic::Ordering::Release);
|
||||||
|
|
||||||
|
render_list(&mut il, 20, 5);
|
||||||
|
assert_eq!(il.items.len(), 2);
|
||||||
|
assert!(il.items.iter().all(|item| item.text() != "stale"));
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn render_applies_append_strategy() {
|
fn render_applies_append_strategy() {
|
||||||
let mut il = list(2);
|
let mut il = list(2);
|
||||||
|
|
|
||||||
|
|
@ -15,9 +15,10 @@ use tui_term::widget::PseudoTerminal;
|
||||||
|
|
||||||
use std::env;
|
use std::env;
|
||||||
use std::io::Read;
|
use std::io::Read;
|
||||||
use std::sync::{Arc, RwLock, mpsc};
|
use std::process::{Child, Stdio};
|
||||||
|
use std::sync::{Arc, Mutex, RwLock, mpsc};
|
||||||
use std::thread::JoinHandle;
|
use std::thread::JoinHandle;
|
||||||
use std::time::Instant;
|
use std::time::{Duration, Instant};
|
||||||
|
|
||||||
use super::statusline::spinner_char;
|
use super::statusline::spinner_char;
|
||||||
use super::util::{find_csi_end, find_osc_end, handle_csi_query, handle_osc_query};
|
use super::util::{find_csi_end, find_osc_end, handle_csi_query, handle_osc_query};
|
||||||
|
|
@ -31,6 +32,50 @@ use crate::{SkimItem, SkimOptions};
|
||||||
pub type PreviewCallbackFn = dyn Fn(Vec<Arc<dyn SkimItem>>) -> Vec<String> + Send + Sync + 'static;
|
pub type PreviewCallbackFn = dyn Fn(Vec<Arc<dyn SkimItem>>) -> Vec<String> + Send + Sync + 'static;
|
||||||
const PREVIEW_MAX_BYTES: usize = 1024 * 1024;
|
const PREVIEW_MAX_BYTES: usize = 1024 * 1024;
|
||||||
const VT_SCROLLBACK: usize = 100_000;
|
const VT_SCROLLBACK: usize = 100_000;
|
||||||
|
type PlainChild = Arc<Mutex<Option<Child>>>;
|
||||||
|
|
||||||
|
fn read_bounded(mut reader: impl Read) -> Vec<u8> {
|
||||||
|
let mut output = Vec::with_capacity(PREVIEW_MAX_BYTES);
|
||||||
|
let mut buffer = [0; 8192];
|
||||||
|
loop {
|
||||||
|
match reader.read(&mut buffer) {
|
||||||
|
Ok(0) | Err(_) => break,
|
||||||
|
Ok(read) => {
|
||||||
|
let retained = PREVIEW_MAX_BYTES.saturating_sub(output.len()).min(read);
|
||||||
|
output.extend_from_slice(&buffer[..retained]);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
output
|
||||||
|
}
|
||||||
|
|
||||||
|
fn terminate_plain_child(child: &PlainChild) {
|
||||||
|
let Ok(mut guard) = child.lock() else {
|
||||||
|
return;
|
||||||
|
};
|
||||||
|
let Some(child) = guard.as_mut() else {
|
||||||
|
return;
|
||||||
|
};
|
||||||
|
|
||||||
|
#[cfg(unix)]
|
||||||
|
if let Ok(process_group) = i32::try_from(child.id()) {
|
||||||
|
use nix::sys::signal::{Signal, killpg};
|
||||||
|
use nix::unistd::Pid;
|
||||||
|
|
||||||
|
let _ = killpg(Pid::from_raw(process_group), Signal::SIGKILL);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(windows)]
|
||||||
|
{
|
||||||
|
let _ = std::process::Command::new("taskkill")
|
||||||
|
.args(["/PID", &child.id().to_string(), "/T", "/F"])
|
||||||
|
.stdout(Stdio::null())
|
||||||
|
.stderr(Stdio::null())
|
||||||
|
.status();
|
||||||
|
}
|
||||||
|
|
||||||
|
let _ = child.kill();
|
||||||
|
}
|
||||||
|
|
||||||
/// Preview content options
|
/// Preview content options
|
||||||
pub(crate) enum PreviewContent {
|
pub(crate) enum PreviewContent {
|
||||||
|
|
@ -81,11 +126,12 @@ pub struct Preview {
|
||||||
pub cmd: String,
|
pub cmd: String,
|
||||||
pub rows: u16,
|
pub rows: u16,
|
||||||
pub cols: u16,
|
pub cols: u16,
|
||||||
pub scroll_y: u16,
|
pub scroll_y: usize,
|
||||||
pub scroll_x: u16,
|
pub scroll_x: usize,
|
||||||
pub thread_handle: Option<JoinHandle<()>>,
|
pub thread_handle: Option<JoinHandle<()>>,
|
||||||
/// Channel to signal thread interruption
|
/// Channel to signal thread interruption
|
||||||
interrupt_tx: Option<mpsc::Sender<()>>,
|
interrupt_tx: Option<mpsc::Sender<()>>,
|
||||||
|
plain_child: Option<PlainChild>,
|
||||||
pub theme: Arc<ColorTheme>,
|
pub theme: Arc<ColorTheme>,
|
||||||
/// Border type
|
/// Border type
|
||||||
pub border: BorderType,
|
pub border: BorderType,
|
||||||
|
|
@ -97,7 +143,7 @@ pub struct Preview {
|
||||||
image: bool,
|
image: bool,
|
||||||
#[cfg(feature = "image")]
|
#[cfg(feature = "image")]
|
||||||
image_picker: Option<Picker>,
|
image_picker: Option<Picker>,
|
||||||
pub total_lines: u16,
|
pub total_lines: usize,
|
||||||
loading: bool,
|
loading: bool,
|
||||||
spinner_start: Instant,
|
spinner_start: Instant,
|
||||||
}
|
}
|
||||||
|
|
@ -150,18 +196,12 @@ impl Preview {
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Convert a Size value to an actual offset based on preview dimensions
|
/// Convert a Size value to an actual offset based on preview dimensions
|
||||||
fn size_to_offset(&self, size: super::Size, is_vertical: bool) -> u16 {
|
fn size_to_offset(&self, size: super::Size, is_vertical: bool) -> usize {
|
||||||
|
let dimension = if is_vertical { self.rows } else { self.cols };
|
||||||
match size {
|
match size {
|
||||||
super::Size::Fixed(n) => n,
|
super::Size::Fixed(n) => usize::from(n),
|
||||||
super::Size::Percent(p) => {
|
super::Size::Percent(p) => usize::from(dimension) * usize::from(p) / 100,
|
||||||
let dimension = if is_vertical { self.rows } else { self.cols };
|
super::Size::Neg(n) => usize::from(dimension.saturating_sub(n)),
|
||||||
// Result is at most dimension (a u16), so truncation cannot occur.
|
|
||||||
u16::try_from(u32::from(dimension) * u32::from(p) / 100).unwrap_or(u16::MAX)
|
|
||||||
}
|
|
||||||
super::Size::Neg(n) => {
|
|
||||||
let dimension = if is_vertical { self.rows } else { self.cols };
|
|
||||||
dimension.saturating_sub(n)
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -226,7 +266,7 @@ impl Preview {
|
||||||
let Ok(mut content) = self.content.write() else {
|
let Ok(mut content) = self.content.write() else {
|
||||||
return Err(eyre::eyre!("Failed to acquire content for writing"));
|
return Err(eyre::eyre!("Failed to acquire content for writing"));
|
||||||
};
|
};
|
||||||
self.total_lines = text.lines.len().try_into().unwrap();
|
self.total_lines = text.lines.len();
|
||||||
*content = PreviewContent::Text(text);
|
*content = PreviewContent::Text(text);
|
||||||
self.scroll_y = 0;
|
self.scroll_y = 0;
|
||||||
self.scroll_x = 0;
|
self.scroll_x = 0;
|
||||||
|
|
@ -256,7 +296,7 @@ impl Preview {
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn scroll_up(&mut self, lines: u16) {
|
pub fn scroll_up(&mut self, lines: u16) {
|
||||||
self.scroll_y = self.scroll_y.saturating_sub(lines);
|
self.scroll_y = self.scroll_y.saturating_sub(usize::from(lines));
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn scroll_down(&mut self, lines: u16) {
|
pub fn scroll_down(&mut self, lines: u16) {
|
||||||
|
|
@ -265,26 +305,26 @@ impl Preview {
|
||||||
self.total_lines, self.rows
|
self.total_lines, self.rows
|
||||||
);
|
);
|
||||||
if self.total_lines > 0 {
|
if self.total_lines > 0 {
|
||||||
self.scroll_y = self
|
self.scroll_y = self.scroll_y.saturating_add(usize::from(lines)).min(
|
||||||
.scroll_y
|
self.total_lines
|
||||||
.saturating_add(lines)
|
.saturating_sub(usize::from(self.rows.saturating_sub(1))),
|
||||||
.min(self.total_lines.saturating_sub(self.rows.saturating_sub(1)));
|
);
|
||||||
} else {
|
} else {
|
||||||
// We might not have the actual total_lines value
|
// We might not have the actual total_lines value
|
||||||
self.scroll_y = self.scroll_y.saturating_add(lines);
|
self.scroll_y = self.scroll_y.saturating_add(usize::from(lines));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn scroll_left(&mut self, cols: u16) {
|
pub fn scroll_left(&mut self, cols: u16) {
|
||||||
self.scroll_x = self.scroll_x.saturating_sub(cols);
|
self.scroll_x = self.scroll_x.saturating_sub(usize::from(cols));
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn scroll_right(&mut self, cols: u16) {
|
pub fn scroll_right(&mut self, cols: u16) {
|
||||||
self.scroll_x = self.scroll_x.saturating_add(cols);
|
self.scroll_x = self.scroll_x.saturating_add(usize::from(cols));
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn set_offset(&mut self, offset: u16) {
|
pub fn set_offset(&mut self, offset: u16) {
|
||||||
self.scroll_y = offset.saturating_sub(1); // -1 because line numbers are 1-indexed
|
self.scroll_y = usize::from(offset.saturating_sub(1)); // -1 because line numbers are 1-indexed
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn page_up(&mut self) {
|
pub fn page_up(&mut self) {
|
||||||
|
|
@ -302,6 +342,11 @@ impl Preview {
|
||||||
let _ = tx.send(());
|
let _ = tx.send(());
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if let Some(child) = self.plain_child.take() {
|
||||||
|
trace!("killing plain preview child process group");
|
||||||
|
terminate_plain_child(&child);
|
||||||
|
}
|
||||||
|
|
||||||
if let Some(mut child) = self.pty_child.take() {
|
if let Some(mut child) = self.pty_child.take() {
|
||||||
trace!("killing pty child process");
|
trace!("killing pty child process");
|
||||||
match child.try_wait() {
|
match child.try_wait() {
|
||||||
|
|
@ -490,39 +535,86 @@ impl Preview {
|
||||||
shell_cmd
|
shell_cmd
|
||||||
.env("ROWS", self.rows.to_string())
|
.env("ROWS", self.rows.to_string())
|
||||||
.env("COLUMNS", self.cols.to_string())
|
.env("COLUMNS", self.cols.to_string())
|
||||||
.env("PAGER", "");
|
.env("PAGER", "")
|
||||||
|
.stdout(Stdio::piped())
|
||||||
|
.stderr(Stdio::piped());
|
||||||
if let Ok(cwd) = env::current_dir() {
|
if let Ok(cwd) = env::current_dir() {
|
||||||
shell_cmd.current_dir(cwd);
|
shell_cmd.current_dir(cwd);
|
||||||
}
|
}
|
||||||
|
#[cfg(unix)]
|
||||||
|
{
|
||||||
|
use std::os::unix::process::CommandExt as _;
|
||||||
|
shell_cmd.process_group(0);
|
||||||
|
}
|
||||||
|
#[cfg(windows)]
|
||||||
|
{
|
||||||
|
use std::os::windows::process::CommandExt as _;
|
||||||
|
const CREATE_NEW_PROCESS_GROUP: u32 = 0x0000_0200;
|
||||||
|
shell_cmd.creation_flags(CREATE_NEW_PROCESS_GROUP);
|
||||||
|
}
|
||||||
|
|
||||||
let (interrupt_tx, interrupt_rx) = mpsc::channel();
|
let (interrupt_tx, interrupt_rx) = mpsc::channel();
|
||||||
self.interrupt_tx = Some(interrupt_tx);
|
self.interrupt_tx = Some(interrupt_tx);
|
||||||
|
|
||||||
self.thread_handle = Some(std::thread::spawn(move || {
|
let mut child = match shell_cmd.spawn() {
|
||||||
if interrupt_rx.try_recv().is_ok() {
|
Ok(child) => child,
|
||||||
return;
|
Err(error) => {
|
||||||
}
|
log::info!("Shell cmd in error: {error:?}");
|
||||||
|
|
||||||
let try_out = shell_cmd.output();
|
|
||||||
if try_out.is_err() {
|
|
||||||
log::info!("Shell cmd in error: {try_out:?}");
|
|
||||||
let _ = event_tx_clone.blocking_send(Event::PreviewReady);
|
let _ = event_tx_clone.blocking_send(Event::PreviewReady);
|
||||||
return;
|
return Ok(());
|
||||||
}
|
}
|
||||||
|
};
|
||||||
|
let stdout = child.stdout.take().expect("stdout was configured as piped");
|
||||||
|
let stderr = child.stderr.take().expect("stderr was configured as piped");
|
||||||
|
let child = Arc::new(Mutex::new(Some(child)));
|
||||||
|
self.plain_child = Some(child.clone());
|
||||||
|
|
||||||
let mut out = try_out.unwrap();
|
self.thread_handle = Some(std::thread::spawn(move || {
|
||||||
|
let stdout_reader = std::thread::spawn(move || read_bounded(stdout));
|
||||||
|
let stderr_reader = std::thread::spawn(move || read_bounded(stderr));
|
||||||
|
|
||||||
if interrupt_rx.try_recv().is_ok() {
|
let status = loop {
|
||||||
return;
|
match interrupt_rx.recv_timeout(Duration::from_millis(10)) {
|
||||||
}
|
Ok(()) | Err(mpsc::RecvTimeoutError::Disconnected) => {
|
||||||
|
terminate_plain_child(&child);
|
||||||
if let Ok(mut c) = content.write() {
|
break None;
|
||||||
if out.status.success() {
|
}
|
||||||
out.stdout.resize(PREVIEW_MAX_BYTES.min(out.stdout.len()), 0);
|
Err(mpsc::RecvTimeoutError::Timeout) => {}
|
||||||
*c = PreviewContent::Text(out.stdout.into_text().unwrap_or_default());
|
|
||||||
} else {
|
|
||||||
*c = PreviewContent::Text(out.stderr.clone().into_text().unwrap_or_default());
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
let wait_result = match child.lock() {
|
||||||
|
Ok(mut guard) => guard.as_mut().map(Child::try_wait),
|
||||||
|
Err(_) => break None,
|
||||||
|
};
|
||||||
|
match wait_result {
|
||||||
|
Some(Ok(Some(status))) => break Some(status),
|
||||||
|
Some(Ok(None)) => {}
|
||||||
|
Some(Err(error)) => {
|
||||||
|
log::info!("Failed to wait for preview command: {error:?}");
|
||||||
|
break None;
|
||||||
|
}
|
||||||
|
None => break None,
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
// A shell can exit while a background descendant still owns the pipes.
|
||||||
|
// Terminate the whole process group before joining the drain threads.
|
||||||
|
terminate_plain_child(&child);
|
||||||
|
let stdout = stdout_reader.join().unwrap_or_default();
|
||||||
|
let stderr = stderr_reader.join().unwrap_or_default();
|
||||||
|
if let Ok(mut guard) = child.lock()
|
||||||
|
&& let Some(mut child) = guard.take()
|
||||||
|
&& status.is_none()
|
||||||
|
{
|
||||||
|
let _ = child.wait();
|
||||||
|
}
|
||||||
|
|
||||||
|
let Some(status) = status else {
|
||||||
|
return;
|
||||||
|
};
|
||||||
|
if let Ok(mut c) = content.write() {
|
||||||
|
let output = if status.success() { stdout } else { stderr };
|
||||||
|
*c = PreviewContent::Text(output.into_text().unwrap_or_default());
|
||||||
}
|
}
|
||||||
|
|
||||||
trace!("sending ready ping");
|
trace!("sending ready ping");
|
||||||
|
|
@ -538,12 +630,14 @@ impl Preview {
|
||||||
area: ratatui::layout::Rect,
|
area: ratatui::layout::Rect,
|
||||||
buf: &mut ratatui::prelude::Buffer,
|
buf: &mut ratatui::prelude::Buffer,
|
||||||
text: &Text,
|
text: &Text,
|
||||||
) -> u16 {
|
) -> usize {
|
||||||
// Calculate total lines in content
|
// Calculate total lines in content
|
||||||
let total_lines: u16 = text.lines.len().try_into().unwrap();
|
let total_lines = text.lines.len();
|
||||||
|
|
||||||
// Create paragraph with optional block
|
// Ratatui terminal coordinates are u16. Saturate previews that exceed that range.
|
||||||
let mut paragraph = Paragraph::new(text.clone()).scroll((self.scroll_y, self.scroll_x));
|
let scroll_y = u16::try_from(self.scroll_y).unwrap_or(u16::MAX);
|
||||||
|
let scroll_x = u16::try_from(self.scroll_x).unwrap_or(u16::MAX);
|
||||||
|
let mut paragraph = Paragraph::new(text.clone()).scroll((scroll_y, scroll_x));
|
||||||
|
|
||||||
// Enable wrapping if wrap is true
|
// Enable wrapping if wrap is true
|
||||||
if self.wrap {
|
if self.wrap {
|
||||||
|
|
@ -552,7 +646,7 @@ impl Preview {
|
||||||
|
|
||||||
// Add scroll position indicator at top-right if scrolled
|
// Add scroll position indicator at top-right if scrolled
|
||||||
if self.scroll_y > 0 && total_lines > 0 {
|
if self.scroll_y > 0 && total_lines > 0 {
|
||||||
let current_line = (self.scroll_y + 1) as usize; // +1 because scroll_y is 0-indexed but we want 1-indexed display
|
let current_line = self.scroll_y.saturating_add(1); // Display line numbers are 1-indexed.
|
||||||
let title = format!("{current_line}/{total_lines}");
|
let title = format!("{current_line}/{total_lines}");
|
||||||
|
|
||||||
outer = outer.title_top(Line::from(title).alignment(Alignment::Right).reversed());
|
outer = outer.title_top(Line::from(title).alignment(Alignment::Right).reversed());
|
||||||
|
|
@ -569,23 +663,21 @@ impl Preview {
|
||||||
area: ratatui::layout::Rect,
|
area: ratatui::layout::Rect,
|
||||||
buf: &mut ratatui::prelude::Buffer,
|
buf: &mut ratatui::prelude::Buffer,
|
||||||
parser: &std::sync::RwLock<tui_term::vt100::Parser>,
|
parser: &std::sync::RwLock<tui_term::vt100::Parser>,
|
||||||
) -> u16 {
|
) -> usize {
|
||||||
let mut total_lines = 0u16;
|
let mut total_lines = 0usize;
|
||||||
// For terminal content, manipulate scrollback to implement scrolling
|
// For terminal content, manipulate scrollback to implement scrolling
|
||||||
if let Ok(mut parser_guard) = parser.try_write() {
|
if let Ok(mut parser_guard) = parser.try_write() {
|
||||||
let scrollback_len = parser_guard.screen().scrollback();
|
let scrollback_len = parser_guard.screen().scrollback();
|
||||||
// Reset scrollback to its full size first
|
// Reset scrollback to its full size first
|
||||||
parser_guard.screen_mut().set_scrollback(VT_SCROLLBACK);
|
parser_guard.screen_mut().set_scrollback(VT_SCROLLBACK);
|
||||||
// If the scrollback is not empty, we seem to be off by one
|
// If the scrollback is not empty, we seem to be off by one
|
||||||
total_lines = (scrollback_len.saturating_sub(1) + parser_guard.screen().contents().lines().count())
|
total_lines = scrollback_len.saturating_sub(1) + parser_guard.screen().contents().lines().count();
|
||||||
.try_into()
|
|
||||||
.unwrap();
|
|
||||||
if self.scroll_y > 0 {
|
if self.scroll_y > 0 {
|
||||||
trace!("scrolling in vt buffer: {}/{}", self.scroll_y, total_lines);
|
trace!("scrolling in vt buffer: {}/{}", self.scroll_y, total_lines);
|
||||||
// Reduce scrollback by scroll_y to show earlier content
|
// Reduce scrollback by scroll_y to show earlier content
|
||||||
parser_guard
|
parser_guard
|
||||||
.screen_mut()
|
.screen_mut()
|
||||||
.set_scrollback(scrollback_len.saturating_sub(self.scroll_y.into()));
|
.set_scrollback(scrollback_len.saturating_sub(self.scroll_y));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -674,6 +766,7 @@ impl SkimWidget for Preview {
|
||||||
scroll_x: 0,
|
scroll_x: 0,
|
||||||
thread_handle: None,
|
thread_handle: None,
|
||||||
interrupt_tx: None,
|
interrupt_tx: None,
|
||||||
|
plain_child: None,
|
||||||
pty: None,
|
pty: None,
|
||||||
pty_child: None,
|
pty_child: None,
|
||||||
#[cfg(feature = "image")]
|
#[cfg(feature = "image")]
|
||||||
|
|
|
||||||
|
|
@ -5,7 +5,7 @@ use ratatui::layout::Size;
|
||||||
#[cfg(feature = "image")]
|
#[cfg(feature = "image")]
|
||||||
use ratatui_image::picker::Picker;
|
use ratatui_image::picker::Picker;
|
||||||
|
|
||||||
use super::Preview;
|
use super::{PREVIEW_MAX_BYTES, Preview, PreviewContent, read_bounded};
|
||||||
|
|
||||||
#[cfg(feature = "image")]
|
#[cfg(feature = "image")]
|
||||||
fn image(width: u32, height: u32) -> DynamicImage {
|
fn image(width: u32, height: u32) -> DynamicImage {
|
||||||
|
|
@ -60,6 +60,51 @@ fn content_loads_text_and_resets_scroll() {
|
||||||
assert!(!p.is_loading());
|
assert!(!p.is_loading());
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn large_text_content_does_not_overflow_line_count() {
|
||||||
|
let input = "x\n".repeat(70_000);
|
||||||
|
let mut preview = Preview::default();
|
||||||
|
preview.content(input.as_bytes()).unwrap();
|
||||||
|
assert_eq!(preview.total_lines, 70_000);
|
||||||
|
|
||||||
|
let content = preview.content.read().unwrap();
|
||||||
|
let PreviewContent::Text(text) = &*content else {
|
||||||
|
panic!("expected text preview");
|
||||||
|
};
|
||||||
|
let area = ratatui::layout::Rect::new(0, 0, 20, 5);
|
||||||
|
let mut buffer = ratatui::buffer::Buffer::empty(area);
|
||||||
|
assert_eq!(
|
||||||
|
preview.render_text(ratatui::widgets::Block::new(), area, &mut buffer, text),
|
||||||
|
70_000
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn bounded_reader_discards_output_after_limit() {
|
||||||
|
let input = vec![b'x'; PREVIEW_MAX_BYTES + 4096];
|
||||||
|
let output = read_bounded(std::io::Cursor::new(input));
|
||||||
|
assert_eq!(output.len(), PREVIEW_MAX_BYTES);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(unix)]
|
||||||
|
#[test]
|
||||||
|
fn plain_preview_can_be_cancelled() {
|
||||||
|
use ratatui::backend::TestBackend;
|
||||||
|
use std::time::{Duration, Instant};
|
||||||
|
|
||||||
|
let mut preview = Preview::default();
|
||||||
|
preview.pty = None;
|
||||||
|
let mut tui =
|
||||||
|
super::super::Tui::new_with_height_and_backend(TestBackend::new(20, 5), super::super::Size::Percent(100))
|
||||||
|
.unwrap();
|
||||||
|
preview.spawn(&mut tui, "sleep 30").unwrap();
|
||||||
|
|
||||||
|
let started = Instant::now();
|
||||||
|
preview.kill();
|
||||||
|
preview.thread_handle.take().unwrap().join().unwrap();
|
||||||
|
assert!(started.elapsed() < Duration::from_secs(2));
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn vertical_scroll_clamps_to_content() {
|
fn vertical_scroll_clamps_to_content() {
|
||||||
let mut p = Preview::default();
|
let mut p = Preview::default();
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue