From c6bd169a76f18f33b2d327d534c32bcc85d59242 Mon Sep 17 00:00:00 2001 From: Alex Pasmantier Date: Fri, 24 Jul 2026 22:06:13 +0200 Subject: [PATCH 1/2] perf(channels): read source output in chunks instead of line by line The reader loop paid per-line costs (an async read_until, a buffer clone and an Instant::now) which dominated ingestion time on large sources. Read the pipe in large chunks instead and ship complete entries to the blocking tasks, which now also split the lines. Loading 3.5M lines end-to-end goes from ~0.76s to ~0.23s. --- Cargo.lock | 1 + Cargo.toml | 1 + television/channels/channel.rs | 201 +++++++++++++++------------------ 3 files changed, 95 insertions(+), 108 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index df6282a2..f61c6cc8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2595,6 +2595,7 @@ dependencies = [ "frizbee", "human-panic", "lazy-regex", + "memchr", "parking_lot", "phantom-test", "ratatui", diff --git a/Cargo.toml b/Cargo.toml index 5ed56e3f..2d38b6be 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -50,6 +50,7 @@ which = "8.0" clap_complete = "4.5" rayon = "1.11" smallvec = "1.15" +memchr = "2" fast-strip-ansi = "0.13" # the parser `fast-strip-ansi` is built on, driven directly to collect the # styling of a line alongside its stripped text (see utils/ansi.rs) diff --git a/television/channels/channel.rs b/television/channels/channel.rs index 74e0102f..11a40423 100644 --- a/television/channels/channel.rs +++ b/television/channels/channel.rs @@ -20,7 +20,7 @@ use std::sync::atomic::AtomicBool; use std::time::Duration; use tokio::process::Command as TokioCommand; use tokio::{ - io::{AsyncBufReadExt, BufReader}, + io::{AsyncBufReadExt, AsyncRead, AsyncReadExt, BufReader}, time::Instant, }; use tracing::debug; @@ -255,19 +255,86 @@ impl Channel

{ } } -const DEFAULT_LINE_BUFFER_SIZE: usize = 256; -// Batch size for pushing candidates to the injector -// 10k * 500 bytes (pessimistic avg line size) = ~5 MB -const BATCH_SIZE: usize = 10_000; -// Automatically flush batch after this interval +// Read the source's output in chunks of at least this size: reading in bulk +// instead of line by line keeps the per-line overhead (syscalls, timestamps, +// allocations) off the reader loop +const READ_CHUNK_SIZE: usize = 64 * 1024; +// Flush accumulated bytes to a processing task after this size +// (~100k entries at 40 bytes per line) +const FLUSH_SIZE: usize = 4 * 1024 * 1024; +// Automatically flush after this interval so first results reach the +// screen quickly on slow sources const UPDATE_INTERVAL: Duration = Duration::from_millis(200); // Maximum number of concurrent flush tasks to prevent unbounded memory growth -// 4 * 10_000 * average line size = ~20 MB +// 4 * ~2x FLUSH_SIZE (raw bytes + processed entries) = ~32 MB const MAX_CONCURRENT_FLUSHES: usize = 4; const DEFAULT_DELIMITER: u8 = b'\n'; +/// Reads `reader` in large chunks and ships complete entries to blocking +/// tasks that split, process and push them to the injector in batches. +/// +/// Returns whether the reader produced any output. +async fn stream_entries( + mut reader: R, + delimiter: u8, + processor: &P, + injector: &Injector, +) -> bool +where + R: AsyncRead + Unpin, + P: EntryProcessor, +{ + let mut acc: Vec = Vec::new(); + let mut flush_handles = tokio::task::JoinSet::new(); + let mut produced_output = false; + let mut last_flush = Instant::now(); + + loop { + acc.reserve(READ_CHUNK_SIZE); + let n = reader.read_buf(&mut acc).await.unwrap_or(0); + if n == 0 { + break; + } + + if acc.len() >= FLUSH_SIZE || last_flush.elapsed() >= UPDATE_INTERVAL { + // Only complete entries are flushed: bytes after the last + // delimiter stay in the accumulator + if let Some(pos) = memchr::memrchr(delimiter, &acc) { + let rest = acc.split_off(pos + 1); + let chunk = std::mem::replace(&mut acc, rest); + + if flush_handles.len() >= MAX_CONCURRENT_FLUSHES { + // Wait for any task to complete + let _ = flush_handles.join_next().await; + } + let inj = injector.clone(); + let mut proc = processor.clone(); + flush_handles.spawn_blocking(move || { + flush_chunk(&chunk, &inj, &mut proc, delimiter); + }); + produced_output = true; + last_flush = Instant::now(); + } + } + } + + // Flush whatever is left (the last entry may not be delimited) + if !acc.is_empty() { + let inj = injector.clone(); + let mut proc = processor.clone(); + flush_handles.spawn_blocking(move || { + flush_chunk(&acc, &inj, &mut proc, delimiter); + }); + produced_output = true; + } + + // Wait for all remaining flush tasks to complete + while flush_handles.join_next().await.is_some() {} + + produced_output +} + /// Collects entries before pushing them to the injector. -#[allow(clippy::unused_async)] pub async fn load_candidates( command: CommandSpec, entry_delimiter: Option, @@ -291,63 +358,16 @@ pub async fn load_candidates( .expect("failed to execute process"); // FIXME: handle error if let Some(out) = child.stdout.take() { - let mut produced_output = false; - let mut reader = BufReader::new(out); - let mut buf = Vec::with_capacity(DEFAULT_LINE_BUFFER_SIZE); - let mut batch = Vec::with_capacity(BATCH_SIZE); - let mut flush_handles = tokio::task::JoinSet::new(); - let delimiter = entry_delimiter .as_ref() .map(|d| *d as u8) .unwrap_or(DEFAULT_DELIMITER); - let mut last_flush = Instant::now(); - while { - buf.clear(); - let n = reader.read_until(delimiter, &mut buf).await.unwrap_or(0); - n > 0 - } { - batch.push(buf.clone()); - - // Flush batch when it reaches the target size - if batch.len() >= BATCH_SIZE - || last_flush.elapsed() >= UPDATE_INTERVAL - { - if flush_handles.len() >= MAX_CONCURRENT_FLUSHES { - // Wait for any task to complete - let _ = flush_handles.join_next().await; - } - - let batch_to_flush = std::mem::replace( - &mut batch, - Vec::with_capacity(BATCH_SIZE), - ); - let inj = injector.clone(); - let mut proc = processor.clone(); - flush_handles.spawn_blocking(move || { - flush_batch(batch_to_flush, &inj, &mut proc, delimiter); - }); - produced_output = true; - last_flush = Instant::now(); - } - } + let produced_output = + stream_entries(out, delimiter, &processor, &injector).await; debug!("Finished reading command output."); - // Flush any remaining entries in the batch - if !batch.is_empty() { - let inj = injector.clone(); - let mut proc = processor.clone(); - flush_handles.spawn_blocking(move || { - flush_batch(batch, &inj, &mut proc, delimiter); - }); - produced_output = true; - } - - // Wait for all remaining flush tasks to complete - while flush_handles.join_next().await.is_some() {} - // if the command didn't produce any output, check stderr and display that instead if !produced_output { let tv_message = @@ -380,74 +400,39 @@ pub async fn load_stdin_candidates( ) { debug!("Loading candidates from stdin"); let stdin = tokio::io::stdin(); - let mut reader = BufReader::new(stdin); - let mut buf = Vec::with_capacity(DEFAULT_LINE_BUFFER_SIZE); - let mut batch = Vec::with_capacity(BATCH_SIZE); - let mut flush_handles = tokio::task::JoinSet::new(); let delimiter = entry_delimiter .as_ref() .map(|d| *d as u8) .unwrap_or(DEFAULT_DELIMITER); - let mut last_flush = Instant::now(); - while { - buf.clear(); - let n = reader.read_until(delimiter, &mut buf).await.unwrap_or(0); - n > 0 - } { - batch.push(buf.clone()); - - if batch.len() >= BATCH_SIZE || last_flush.elapsed() >= UPDATE_INTERVAL - { - if flush_handles.len() >= MAX_CONCURRENT_FLUSHES { - let _ = flush_handles.join_next().await; - } - - let batch_to_flush = - std::mem::replace(&mut batch, Vec::with_capacity(BATCH_SIZE)); - let inj = injector.clone(); - let mut proc = processor.clone(); - flush_handles.spawn_blocking(move || { - flush_batch(batch_to_flush, &inj, &mut proc, delimiter); - }); - last_flush = Instant::now(); - } - } + stream_entries(stdin, delimiter, &processor, &injector).await; debug!("Finished reading stdin."); - - if !batch.is_empty() { - let inj = injector.clone(); - let mut proc = processor.clone(); - flush_handles.spawn_blocking(move || { - flush_batch(batch, &inj, &mut proc, delimiter); - }); - } - - while flush_handles.join_next().await.is_some() {} } -/// Flushes a batch of entries to the injector. +/// Splits a chunk of complete entries on `delimiter`, filters +/// empty/whitespace-only lines and runs the processor up front so the whole +/// chunk is pushed under a single injector call. /// This is called from a blocking task spawned in the threadpool. -fn flush_batch( - batch: Vec>, +fn flush_chunk( + chunk: &[u8], injector: &Injector, processor: &mut P, delimiter: u8, ) { - // decode utf8, filter empty/whitespace-only lines and run the processor - // up front so the whole batch is pushed under a single injector call - let mut entries = Vec::with_capacity(batch.len()); - for mut bytes in batch { - if bytes.is_empty() || bytes.iter().all(u8::is_ascii_whitespace) { + let mut entries = Vec::new(); + let mut start = 0; + for end in memchr::memchr_iter(delimiter, chunk) + .chain(std::iter::once(chunk.len())) + { + let line = &chunk[start..end]; + start = end + 1; + if line.is_empty() || line.iter().all(u8::is_ascii_whitespace) { continue; } - if bytes.last() == Some(&delimiter) { - bytes.pop(); - } - if let Ok(line) = String::from_utf8(bytes) { - entries.push(processor.process(line)); + if let Ok(line) = std::str::from_utf8(line) { + entries.push(processor.process(line.to_string())); } } injector.push_batch(entries); From af56d41b78d573b17522c528603b658a08f3b4f2 Mon Sep 17 00:00:00 2001 From: Alex Pasmantier Date: Fri, 24 Jul 2026 22:23:09 +0200 Subject: [PATCH 2/2] perf(previewer): share the preview text behind an Arc Generating a preview deep-copied the parsed Text twice when caching (once for the preview, once for the cache) and every cache hit copied it again; arriving previews were also compared structurally against the current one. Share the text behind an Arc instead: the cache stores and returns Arc clones, and a pointer check now short-circuits the content comparison. Bouncing between two cached ~200k-line previews drops from ~130ms of CPU per selection change to ~10ms. --- television/previewer/cache.rs | 21 +++++++-------- television/previewer/mod.rs | 48 ++++++++++++++--------------------- television/previewer/state.rs | 14 ++++++---- television/screen/preview.rs | 5 +++- 4 files changed, 43 insertions(+), 45 deletions(-) diff --git a/television/previewer/cache.rs b/television/previewer/cache.rs index a517a18c..f1d9153c 100644 --- a/television/previewer/cache.rs +++ b/television/previewer/cache.rs @@ -1,4 +1,5 @@ use rustc_hash::FxHashMap; +use std::sync::Arc; use crate::utils::cache::RingSet; use ratatui::text::Text; @@ -15,7 +16,7 @@ const DEFAULT_CACHE_SIZE: usize = 50; /// The cache is implemented as an LRU cache with a fixed size. #[derive(Debug)] pub struct Cache { - entries: FxHashMap>, + entries: FxHashMap>>, ring_set: RingSet, } @@ -28,17 +29,17 @@ impl Cache { } } - pub fn get(&self, key: &str) -> Option> { + pub fn get(&self, key: &str) -> Option>> { self.entries.get(key).cloned() } /// Insert a new preview into the cache. /// If the cache is full, the oldest entry will be removed. /// If the key is already in the cache, the preview will be updated. - pub fn insert(&mut self, key: &str, text: &Text<'static>) { + pub fn insert(&mut self, key: &str, text: &Arc>) { debug!("Inserting preview into cache for key: {:?}", key); let key = key.to_string(); - self.entries.insert(key.clone(), text.clone()); + self.entries.insert(key.clone(), Arc::clone(text)); if let Some(oldest_key) = self.ring_set.push(key) { debug!("Cache full, removing oldest entry: {:?}", oldest_key); self.entries.remove(&oldest_key); @@ -74,21 +75,21 @@ mod tests { fn test_preview_cache_ops() { let mut cache = Cache::new(2); let entry = "test"; - let preview = Text::raw("preview"); + let preview = Arc::new(Text::raw("preview")); cache.insert(entry, &preview); assert_eq!(cache.get(entry).unwrap(), preview); assert_eq!(cache.size(), 1); // override cache content for the same key - let other_preview = Text::raw("some content"); + let other_preview = Arc::new(Text::raw("some content")); cache.insert(entry, &other_preview); assert_eq!(cache.get(entry).unwrap(), other_preview); assert_eq!(cache.size(), 1); // insert new entries to trigger eviction let new_entry = "new_test"; - let new_preview = Text::raw("new preview"); + let new_preview = Arc::new(Text::raw("new preview")); cache.insert(new_entry, &new_preview); // the two previews should still be available assert_eq!(cache.size(), 2); @@ -96,15 +97,15 @@ mod tests { assert_eq!(cache.get(entry).unwrap(), other_preview); // this one should trigger eviction let another_entry = "another_test"; - cache.insert(another_entry, &Text::raw("another preview")); + cache.insert(another_entry, &Arc::new(Text::raw("another preview"))); assert_eq!(cache.size(), 2); assert!(cache.get(entry).is_none()); assert!(cache.get(new_entry).is_some()); assert!(cache.get(another_entry).is_some()); - assert_eq!(cache.get(new_entry).unwrap(), Text::raw("new preview")); + assert_eq!(*cache.get(new_entry).unwrap(), Text::raw("new preview")); assert_eq!( - cache.get(another_entry).unwrap(), + *cache.get(another_entry).unwrap(), Text::raw("another preview") ); } diff --git a/television/previewer/mod.rs b/television/previewer/mod.rs index 473f9f0c..b498a5c9 100644 --- a/television/previewer/mod.rs +++ b/television/previewer/mod.rs @@ -119,7 +119,9 @@ pub struct Preview { pub title: String, // NOTE: this does couple the previewer with ratatui but allows // to only parse ansi text once and reuse it in the UI. - pub content: Text<'static>, + // Shared behind an `Arc` so that the cache, the previewer and the + // render context never deep-copy the parsed text. + pub content: Arc>, pub target_line: Option, pub total_lines: u16, pub footer: Option, @@ -135,7 +137,7 @@ impl Default for Preview { entry_raw: EMPTY_STRING.to_string(), formatted_command: EMPTY_STRING.to_string(), title: DEFAULT_PREVIEW_TITLE.to_string(), - content: Text::from(EMPTY_STRING), + content: Arc::new(Text::from(EMPTY_STRING)), target_line: None, total_lines: 1, footer: None, @@ -151,7 +153,7 @@ impl Preview { entry_raw: String, formatted_command: String, title: &str, - displayable_content: Text<'static>, + displayable_content: Arc>, line_number: Option, total_lines: u16, footer: Option, @@ -330,7 +332,7 @@ fn sanitize_text(text: &mut Text<'static>) { fn build_preview_from_text( formatted_command: &str, entry: &Entry, - text: Text<'static>, + text: Arc>, title_template: Option<&Template>, footer_template: Option<&Template>, offset_expr: Option<&Template>, @@ -434,32 +436,20 @@ pub async fn try_preview( sanitize_text(&mut text); - let preview = if let Some(cache) = &cache { - let preview = build_preview_from_text( - &formatted_command, - &entry, - text.clone(), - title_template.as_ref(), - footer_template.as_ref(), - offset_expr.as_ref(), - cycle_index, - preview_count, - )?; + let text = Arc::new(text); + if let Some(cache) = &cache { cache.lock().insert(&formatted_command, &text); - preview - } else { - build_preview_from_text( - &formatted_command, - &entry, - text, - title_template.as_ref(), - footer_template.as_ref(), - offset_expr.as_ref(), - cycle_index, - preview_count, - )? - }; - // FIXME: ... and just send an Arc here as well + } + let preview = build_preview_from_text( + &formatted_command, + &entry, + text, + title_template.as_ref(), + footer_template.as_ref(), + offset_expr.as_ref(), + cycle_index, + preview_count, + )?; results_handle .send(preview) .with_context(|| "Failed to send preview result to main thread.") diff --git a/television/previewer/state.rs b/television/previewer/state.rs index 91935b60..1c74301e 100644 --- a/television/previewer/state.rs +++ b/television/previewer/state.rs @@ -1,11 +1,11 @@ use ratatui::text::Text; +use std::sync::Arc; use crate::previewer::Preview; #[derive(Debug, Clone, Default)] pub struct PreviewState { pub enabled: bool, - // FIXME: this should probably be an Arc pub preview: Preview, pub scroll: u16, } @@ -39,8 +39,13 @@ impl PreviewState { } pub fn update(&mut self, preview: Preview, scroll: u16) { + // cached previews for the same entry come back as the same `Arc`: + // the pointer check skips the deep content comparison + let content_changed = + !Arc::ptr_eq(&self.preview.content, &preview.content) + && self.preview.content != preview.content; if self.preview.entry_raw != preview.entry_raw - || self.preview.content != preview.content + || content_changed || self.preview.target_line != preview.target_line { self.preview = preview; @@ -51,7 +56,7 @@ impl PreviewState { // FIXME: does this really need to happen for every render? // What if we did it only when the preview content or scroll changes? pub fn for_render_context(&self, height: usize) -> Self { - // PERF: this allocates every time + // only the visible lines are copied for the render context let content_len = self.preview.content.lines.len(); let scroll = (self.scroll as usize).min(content_len); let num_lines = content_len.saturating_sub(scroll); @@ -67,12 +72,11 @@ impl PreviewState { PreviewState::new( self.enabled, - // PERF: this allocates every time Preview::new( self.preview.entry_raw.clone(), self.preview.formatted_command.clone(), &self.preview.title, - cropped_content, + Arc::new(cropped_content), adjusted_line_number, self.preview.total_lines, self.preview.footer.clone(), diff --git a/television/screen/preview.rs b/television/screen/preview.rs index 7e68b05f..740b6538 100644 --- a/television/screen/preview.rs +++ b/television/screen/preview.rs @@ -64,8 +64,11 @@ pub fn draw_preview_content_block( ); // render the preview content + // the render context's `Arc` is not shared, so this is a move, not a copy + let content = std::sync::Arc::try_unwrap(preview_state.preview.content) + .unwrap_or_else(|arc| (*arc).clone()); let rp = build_preview_paragraph( - preview_state.preview.content, + content, preview_state.preview.target_line, colorscheme.preview.highlight_bg, word_wrap,