From 81471f1fb842261dd6de57fd76e85eeed0d76654 Mon Sep 17 00:00:00 2001 From: Michael Ramos Date: Tue, 27 Oct 2026 18:38:32 -0400 Subject: [PATCH] fix: hand loaded files to the screen without sleep-polling The input worker published each file through a two-slot channel and slept 5 ms whenever it was full, while the screen took the two files and waited for the next wake. On a busy machine, loading the Linux v6.0..v7.0 patch spent most of its time asleep: about 30 s to reach all files against a 5 s parse. Files now go through a queue bounded by 1,024 files or 8 MiB of index bytes. When the queue is full, the parser blocks on a condition variable until the screen drains it. The screen takes everything queued on each wake. Prepared sources stay limited to two, because each holds both complete sides. Closing the worker wakes a blocked parser. --- src/engine.rs | 35 +++-- src/input/worker.rs | 324 +++++++++++++++++++++++++++++++++++++++--- tests/input_worker.rs | 18 ++- 3 files changed, 333 insertions(+), 44 deletions(-) diff --git a/src/engine.rs b/src/engine.rs index 057fc4f..621660e 100644 --- a/src/engine.rs +++ b/src/engine.rs @@ -4,6 +4,7 @@ mod reading; mod selection; use std::{ + collections::VecDeque, fs::File, path::PathBuf, sync::{ @@ -312,18 +313,24 @@ impl Engine { self.app.as_mut().context("The review has no viewer state") } - /// Clear the host wake token before calling this. Each drain is bounded so input can run. + /// Clear the host wake token before calling this. Each drain is bounded so input can run: + /// the input worker queues a bounded number of files, and this takes all of them at once. pub fn poll(&mut self) -> Result<()> { if self.input.is_none() { return Ok(()); } if (self.loading || self.full_sources) && self.error.is_none() { - for _ in 0..64 { - let Some(input) = &self.input else { - break; - }; - match input.try_reply() { - Ok(Ok(input::Message::File(file))) => { + let replies = match self.input.as_ref().map(input::Worker::try_replies) { + Some(Ok(replies)) => replies, + Some(Err(mpsc::TryRecvError::Disconnected)) => { + self.fail_input("The input worker stopped before completing its work.".into()); + VecDeque::new() + } + Some(Err(mpsc::TryRecvError::Empty)) | None => VecDeque::new(), + }; + for reply in replies { + match reply { + Ok(input::Message::File(file)) => { let app = self.app()?; let exposes_more = app .visible @@ -337,7 +344,7 @@ impl Engine { self.dirty |= exposes_more; self.progress = true; } - Ok(Ok(input::Message::Complete { files })) => { + Ok(input::Message::Complete { files }) => { crate::delivery_event!("indexed", 0, 0); ensure!( self.app()?.doc.files.len() == files, @@ -345,9 +352,8 @@ impl Engine { ); self.loading = false; self.dirty = true; - break; } - Ok(Ok(input::Message::Source { file, state })) => { + Ok(input::Message::Source { file, state }) => { self.source_failed |= matches!(state, SourceState::Unavailable(_)); self.app .as_mut() @@ -356,17 +362,10 @@ impl Engine { self.request = true; self.dirty = true; } - Ok(Err(error)) => { + Err(error) => { self.fail_input(format!("{error:#}")); break; } - Err(mpsc::TryRecvError::Disconnected) => { - self.fail_input( - "The input worker stopped before completing its work.".into(), - ); - break; - } - Err(mpsc::TryRecvError::Empty) => break, } } } diff --git a/src/input/worker.rs b/src/input/worker.rs index 7a9e04a..ce02936 100644 --- a/src/input/worker.rs +++ b/src/input/worker.rs @@ -1,8 +1,10 @@ use super::*; use std::{ + collections::VecDeque, sync::{ + Condvar, Mutex, MutexGuard, atomic::Ordering, - mpsc::{self, Receiver, SyncSender}, + mpsc::{self, SyncSender}, }, thread::{self, JoinHandle}, time::Duration, @@ -14,11 +16,194 @@ pub enum Message { Source { file: usize, state: SourceState }, } +/// Published files waiting for the screen are bounded by count and index bytes. A full +/// queue pauses ingestion until the screen drains it; it never delays keyboard input. +/// The count also bounds the screen's per-turn drain. +const QUEUED_FILES: usize = 1024; +const QUEUED_FILE_BYTES: usize = 8 << 20; +/// Prepared sources retain both complete sides, so only two may wait. +const QUEUED_SOURCES: usize = 2; + +#[derive(Default)] +struct Queue { + replies: VecDeque>, + files: usize, + file_bytes: usize, + sources: usize, + /// The worker was closed; a waiting producer must stop. + closed: bool, + /// The producer exited; nothing more will arrive. + finished: bool, +} + +#[derive(Clone, Copy)] +enum Cost { + File(usize), + Source, + Final, +} + +impl Cost { + fn of(reply: &Result) -> Self { + match reply { + Ok(Message::File(file)) => Self::File( + size_of::() + + file.source_index_bytes() + + file.projection_bytes() + + file.metadata_bytes(), + ), + Ok(Message::Source { .. }) => Self::Source, + Ok(Message::Complete { .. }) | Err(_) => Self::Final, + } + } +} + +impl Queue { + fn admits(&self, cost: Cost) -> bool { + match cost { + // An oversized file still enters an empty queue; it is accounted for in peak memory. + Cost::File(bytes) => { + self.files == 0 + || (self.files < QUEUED_FILES + && self.file_bytes.saturating_add(bytes) <= QUEUED_FILE_BYTES) + } + Cost::Source => self.sources < QUEUED_SOURCES, + // At most one completion and one error are ever published. + Cost::Final => true, + } + } + + fn add(&mut self, cost: Cost) { + match cost { + Cost::File(bytes) => { + self.files += 1; + self.file_bytes = self.file_bytes.saturating_add(bytes); + } + Cost::Source => self.sources += 1, + Cost::Final => {} + } + } + + fn remove(&mut self, cost: Cost) { + match cost { + Cost::File(bytes) => { + self.files = self.files.saturating_sub(1); + self.file_bytes = self.file_bytes.saturating_sub(bytes); + } + Cost::Source => self.sources = self.sources.saturating_sub(1), + Cost::Final => {} + } + } +} + +/// Single-producer handoff from the input thread to the screen. The producer blocks on a +/// condition variable when the queue is full; the screen takes everything queued at once. +#[derive(Default)] +struct Mailbox { + queue: Mutex, + space: Condvar, +} + +impl Mailbox { + fn lock(&self) -> Result> { + self.queue + .lock() + .map_err(|_| anyhow::anyhow!("The input queue lock failed")) + } + + fn deliver(&self, reply: Result, stop: &Cancellation) -> Result<()> { + let cost = Cost::of(&reply); + let mut queue = self.lock()?; + loop { + if queue.closed { + return Err(Cancelled.into()); + } + stop.check()?; + if queue.admits(cost) { + break; + } + queue = self + .space + .wait(queue) + .map_err(|_| anyhow::anyhow!("The input queue lock failed"))?; + } + queue.add(cost); + queue.replies.push_back(reply); + Ok(()) + } + + fn finish(&self) { + if let Ok(mut queue) = self.queue.lock() { + queue.finished = true; + } + } + + fn close(&self) { + let replies = self.queue.lock().map(|mut queue| { + queue.closed = true; + queue.files = 0; + queue.file_bytes = 0; + queue.sources = 0; + std::mem::take(&mut queue.replies) + }); + self.space.notify_all(); + // Release queued files and sources outside the lock. + drop(replies); + } + + fn try_reply(&self) -> std::result::Result, mpsc::TryRecvError> { + let mut queue = self + .queue + .lock() + .map_err(|_| mpsc::TryRecvError::Disconnected)?; + match queue.replies.pop_front() { + Some(reply) => { + queue.remove(Cost::of(&reply)); + drop(queue); + self.space.notify_one(); + Ok(reply) + } + None if queue.finished => Err(mpsc::TryRecvError::Disconnected), + None => Err(mpsc::TryRecvError::Empty), + } + } + + fn try_replies(&self) -> std::result::Result>, mpsc::TryRecvError> { + let mut queue = self + .queue + .lock() + .map_err(|_| mpsc::TryRecvError::Disconnected)?; + if queue.replies.is_empty() { + return Err(if queue.finished { + mpsc::TryRecvError::Disconnected + } else { + mpsc::TryRecvError::Empty + }); + } + queue.files = 0; + queue.file_bytes = 0; + queue.sources = 0; + let replies = std::mem::take(&mut queue.replies); + drop(queue); + self.space.notify_one(); + Ok(replies) + } +} + +/// Marks the producer finished before the exit notification, so the screen observes disconnection. +struct Finish(Arc); + +impl Drop for Finish { + fn drop(&mut self) { + self.0.finish(); + } +} + pub struct Worker { cancel: Cancellation, destination: Arc, wake: SyncSender<()>, - replies: Receiver>, + replies: Arc, join: Option>>, } @@ -40,25 +225,17 @@ impl Worker { let cancel = Cancellation::default(); let destination = Arc::new(AtomicUsize::new(usize::MAX)); let (wake, requests) = mpsc::sync_channel(1); - let (send, replies) = mpsc::sync_channel(2); + let replies = Arc::new(Mailbox::default()); + let mailbox = replies.clone(); let stop = cancel.clone(); let wanted = destination.clone(); let join = thread::spawn(move || { let _notify_exit = crate::wake::NotifyOnExit(notify.clone()); - let send = send; - let deliver = |mut value| -> Result<()> { - loop { - stop.check()?; - match send.try_send(value) { - Ok(()) => { - let _ = notify.try_send(()); - return Ok(()); - } - Err(mpsc::TrySendError::Disconnected(_)) => return Err(Cancelled.into()), - Err(mpsc::TrySendError::Full(v)) => value = v, - } - thread::sleep(Duration::from_millis(5)); - } + let _finish = Finish(mailbox.clone()); + let deliver = |value| -> Result<()> { + mailbox.deliver(value, &stop)?; + let _ = notify.try_send(()); + Ok(()) }; let result = (|| -> Result<()> { let input = match input { @@ -120,13 +297,23 @@ impl Worker { let _ = self.wake.try_send(()); } } + /// Take the oldest reply. `Disconnected` means the worker exited and nothing remains. pub fn try_reply(&self) -> std::result::Result, mpsc::TryRecvError> { - self.replies.try_recv() + self.replies.try_reply() + } + + /// Take every queued reply in order, releasing a producer that waits for space. + /// `Disconnected` means the worker exited and nothing remains. + pub fn try_replies( + &self, + ) -> std::result::Result>, mpsc::TryRecvError> { + self.replies.try_replies() } /// Cancel and join. Input errors arrive through replies; this reports cleanup failure or panic. pub fn close(&mut self) -> Result<()> { self.cancel.stop(); + self.replies.close(); let _ = self.wake.try_send(()); let result = self.join.take().map_or(Ok(()), |join| { join.join() @@ -136,7 +323,7 @@ impl Worker { result => result, }) }); - while self.replies.try_recv().is_ok() {} + self.replies.close(); result } } @@ -146,3 +333,102 @@ impl Drop for Worker { let _ = self.close(); } } + +#[cfg(test)] +mod tests { + use super::*; + + fn source(file: usize) -> Result { + Ok(Message::Source { + file, + state: SourceState::Pending, + }) + } + + #[test] + fn queued_files_are_bounded_by_count_and_bytes_but_one_always_fits() { + let mut queue = Queue::default(); + assert!(queue.admits(Cost::File(QUEUED_FILE_BYTES * 2))); + for _ in 0..QUEUED_FILES { + assert!(queue.admits(Cost::File(1))); + queue.add(Cost::File(1)); + } + assert!(!queue.admits(Cost::File(1))); + assert!(queue.admits(Cost::Final)); + queue.remove(Cost::File(1)); + assert!(queue.admits(Cost::File(1))); + + let mut queue = Queue::default(); + queue.add(Cost::File(QUEUED_FILE_BYTES - 10)); + assert!(queue.admits(Cost::File(10))); + assert!(!queue.admits(Cost::File(11))); + assert!(queue.admits(Cost::Source)); + } + + #[test] + fn a_full_queue_blocks_the_producer_until_the_screen_drains_everything() { + let mailbox = Arc::new(Mailbox::default()); + let stop = Cancellation::default(); + for file in 0..QUEUED_SOURCES { + mailbox.deliver(source(file), &stop).unwrap(); + } + let (done, delivered) = mpsc::sync_channel(1); + let producer = { + let mailbox = mailbox.clone(); + let stop = stop.clone(); + thread::spawn(move || { + let result = mailbox.deliver(source(QUEUED_SOURCES), &stop); + done.send(()).unwrap(); + result + }) + }; + assert_eq!( + delivered.recv_timeout(Duration::from_millis(50)), + Err(mpsc::RecvTimeoutError::Timeout) + ); + let replies = mailbox.try_replies().unwrap(); + assert_eq!(replies.len(), QUEUED_SOURCES); + delivered.recv_timeout(Duration::from_secs(10)).unwrap(); + producer.join().unwrap().unwrap(); + let replies = mailbox.try_replies().unwrap(); + assert!(matches!( + replies.front(), + Some(Ok(Message::Source { file, .. })) if *file == QUEUED_SOURCES + )); + assert!(matches!( + mailbox.try_replies(), + Err(mpsc::TryRecvError::Empty) + )); + mailbox.finish(); + assert!(matches!( + mailbox.try_reply(), + Err(mpsc::TryRecvError::Disconnected) + )); + } + + #[test] + fn closing_releases_a_blocked_producer_and_drops_queued_replies() { + let mailbox = Arc::new(Mailbox::default()); + let stop = Cancellation::default(); + for file in 0..QUEUED_SOURCES { + mailbox.deliver(source(file), &stop).unwrap(); + } + let producer = { + let mailbox = mailbox.clone(); + let stop = stop.clone(); + thread::spawn(move || mailbox.deliver(source(QUEUED_SOURCES), &stop)) + }; + thread::sleep(Duration::from_millis(20)); + stop.stop(); + mailbox.close(); + let error = producer.join().unwrap().unwrap_err(); + assert!(error.is::()); + assert!(mailbox.lock().unwrap().replies.is_empty()); + assert!( + mailbox + .deliver(source(0), &Cancellation::default()) + .unwrap_err() + .is::() + ); + } +} diff --git a/tests/input_worker.rs b/tests/input_worker.rs index b73e619..7b87e78 100644 --- a/tests/input_worker.rs +++ b/tests/input_worker.rs @@ -52,17 +52,17 @@ fn start(repo: &std::path::Path) -> (Worker, mpsc::Receiver<()>) { } #[test] -fn closing_a_full_patch_queue_releases_the_producer_but_keeps_published_source_valid() { +fn closing_queued_patch_files_releases_the_producer_but_keeps_published_source_valid() { let repo = readiness::repository(false); let (worker, notified) = start(repo.path()); - for _ in 0..2 { + // All twelve files and the completion are queued without the screen draining any. + for _ in 0..13 { notified.recv_timeout(Duration::from_secs(10)).unwrap(); } let Message::File(file) = worker.try_reply().unwrap().unwrap() else { panic!("expected the first file"); }; assert_eq!(file.id, 0); - notified.recv_timeout(Duration::from_secs(10)).unwrap(); assert_eq!( notified.recv_timeout(Duration::from_millis(20)), Err(mpsc::RecvTimeoutError::Timeout) @@ -152,7 +152,7 @@ fn closing_patch_readers_releases_blocked_input_and_queues_before_a_new_review() unistd::pipe, }; use std::{fs::File, io::Write, os::fd::AsRawFd}; - for mode in ["idle", "partial", "full-queue"] { + for mode in ["idle", "partial", "queued"] { let (read_end, write_end) = pipe().unwrap(); let observed = File::from(read_end); let original = fcntl(observed.as_raw_fd(), FcntlArg::F_GETFL).unwrap(); @@ -166,17 +166,21 @@ fn closing_patch_readers_releases_blocked_input_and_queues_before_a_new_review() assert!(Instant::now() < deadline, "patch reader did not start"); thread::sleep(Duration::from_millis(1)); } - let retained = if mode == "full-queue" { + let retained = if mode == "queued" { for n in 0..8 { writeln!(producer, "diff --git a/{n}.txt b/{n}.txt\n--- a/{n}.txt\n+++ b/{n}.txt\n@@ -1 +1 @@\n-old\n+new").unwrap(); } - for _ in 0..2 { + // The eighth file stays pending until its successor's header or EOF frames it. + for _ in 0..7 { notified.recv_timeout(Duration::from_secs(5)).unwrap(); } let Message::File(file) = worker.try_reply().unwrap().unwrap() else { panic!("expected file"); }; - notified.recv_timeout(Duration::from_secs(5)).unwrap(); + assert!(matches!( + notified.recv_timeout(Duration::from_millis(20)), + Err(mpsc::RecvTimeoutError::Timeout) + )); Some(file) } else { if mode == "partial" {