//! Search thread that handles searching a terminal for a string match. //! This is expected to run on a dedicated thread to try to prevent too much //! overhead to other terminal read/write operations. //! //! The current architecture of search does acquire global locks for accessing //! terminal data, so there's still added contention, but we do our best to //! minimize this by trading off memory usage (copying data to minimize lock //! time). //! //! Deviation: Zig drives this from a libxev loop (async wakeup/stop handles, //! a refresh timer). Here the loop is built from `crossbeam-channel`: a //! bounded mailbox, a control channel for wakeup/stop, and a deadline //! based refresh timer. The terminal and its mutex are combined into an //! `Arc>`. use std::sync::{Arc, Mutex, MutexGuard, PoisonError}; use std::time::{Duration, Instant}; use crossbeam_channel::{Receiver, Sender, TrySendError}; use super::screen::{ScreenSearch, Select, TickError}; use super::viewport::ViewportSearch; use crate::highlight::{Flattened, Untracked}; use crate::page_list::{Chunk, Direction as ListDirection}; use crate::point::{Coordinate, Point}; use crate::screen_set::Key as ScreenKey; use crate::terminal::Terminal; // TODO: Some stuff that could be improved: // - pause the refresh timer when the terminal isn't focused // - we probably want to know our progress through the search // for viewport matches so we can show n/total UI. // - notifications should be coalesced to avoid spamming a massive // amount of events if the terminal is changing rapidly. /// The interval at which we refresh the terminal state to check if /// there are any changes that require us to re-search. This should be /// balanced to be fast enough to be responsive but not so fast that /// we hold the terminal lock too often. pub const REFRESH_INTERVAL: Duration = Duration::from_millis(24); // 40 FPS /// Capacity of the mailbox (Zig: `BlockingQueue(Message, 64)`). pub const MAILBOX_CAPACITY: usize = 64; /// Callback for events from the search thread. pub type EventCallback = Arc; /// The options used to initialize the thread. #[derive(Clone)] pub struct Options { /// The terminal data to search. The mutex must be held while /// reading/writing the terminal. pub terminal: Arc>, /// The callback for events from the search thread. This can be `None` /// if you don't want to receive events, which could be useful for a /// one-time search (although, odd, you should use our search /// structures directly then). pub event_cb: Option, } /// The messages that can be sent to the thread. #[derive(Debug, Clone, PartialEq, Eq)] pub enum Message { /// Change the search term. If no prior search term is given this /// will start a search. If an existing search term is given this will /// stop the prior search and start a new one. ChangeNeedle(Vec), /// Select a search result. Select(Select), } /// Events that can be emitted from the search thread. The caller /// chooses to handle these as they see fit. #[derive(Debug, Clone, PartialEq, Eq)] pub enum Event { /// Search is quitting. The search thread is exiting. Quit, /// Search is complete for the given needle on all screens. Complete, /// Total matches on the current active screen have changed. TotalMatches(usize), /// Selected match changed. SelectedMatch(Option), /// Matches in the viewport have changed. ViewportMatches(Vec), } #[derive(Debug, Clone, PartialEq, Eq)] pub struct SelectedMatch { pub idx: usize, pub highlight: Flattened, } #[derive(Debug)] enum Ctl { Wakeup, Stop, } /// A cloneable handle used to talk to the thread from other threads /// (Zig: `thread.mailbox`, `thread.wakeup`, `thread.stop`). #[derive(Clone)] pub struct Handle { mailbox: Sender, ctl: Sender, } impl Handle { /// Push a message to the mailbox, blocking while it is full (Zig: /// `mailbox.push(msg, .forever)`), and wake the thread. Returns false /// if the thread is gone. pub fn send(&self, msg: Message) -> bool { if self.mailbox.send(msg).is_err() { return false; } self.wakeup(); true } /// Push a message without blocking. Returns false if the mailbox is /// full or the thread is gone. pub fn try_send(&self, msg: Message) -> bool { match self.mailbox.try_send(msg) { Ok(()) => { self.wakeup(); true } Err(TrySendError::Full(_) | TrySendError::Disconnected(_)) => false, } } /// Wake up the thread so it drains its mailbox. pub fn wakeup(&self) { let _ = self.ctl.try_send(Ctl::Wakeup); } /// Stop the thread on the next loop iteration. pub fn stop(&self) { let _ = self.ctl.send(Ctl::Stop); } } fn lock(m: &Mutex) -> MutexGuard<'_, Terminal> { m.lock().unwrap_or_else(PoisonError::into_inner) } /// The search thread state. pub struct Thread { opts: Options, mailbox_tx: Sender, mailbox_rx: Receiver, ctl_tx: Sender, ctl_rx: Receiver, /// Search state. Starts as `None` and is populated when a search is /// started (a needle is given). search: Option, stopped: bool, next_refresh: Instant, } impl Thread { /// Initialize the thread. This does not START the thread. This only sets /// up all the internal state necessary prior to starting the thread. It /// is up to the caller to run `thread_main` on a dedicated thread (or /// use `spawn`). pub fn new(opts: Options) -> Self { let (mailbox_tx, mailbox_rx) = crossbeam_channel::bounded(MAILBOX_CAPACITY); let (ctl_tx, ctl_rx) = crossbeam_channel::unbounded(); Self { opts, mailbox_tx, mailbox_rx, ctl_tx, ctl_rx, search: None, stopped: false, next_refresh: Instant::now() + REFRESH_INTERVAL, } } /// A handle to message/stop this thread. pub fn handle(&self) -> Handle { Handle { mailbox: self.mailbox_tx.clone(), ctl: self.ctl_tx.clone(), } } /// Create the thread state and run it on a new OS thread named /// "search". Returns the handle and the join handle. pub fn spawn(opts: Options) -> std::io::Result<(Handle, std::thread::JoinHandle<()>)> { let thread = Thread::new(opts); let handle = thread.handle(); let join = std::thread::Builder::new() .name("search".into()) .spawn(move || thread.thread_main())?; Ok((handle, join)) } /// The main entrypoint for the thread. pub fn thread_main(mut self) { log::debug!("starting search thread"); // Drain the mailbox immediately (Zig sends an initial wakeup). self.drain_mailbox(); self.next_refresh = Instant::now() + REFRESH_INTERVAL; // Unlike some of our other threads, we interleave search work // with our loop so that we can try to make forward search progress // while also listening for messages. loop { // If our loop is canceled then we drain our messages and quit. if self.stopped { while let Ok(message) = self.mailbox_rx.try_recv() { log::debug!("mailbox message ignored during shutdown={message:?}"); } break; } let Some(s) = self.search.as_mut() else { // If we're not actively searching, we can block the loop // until it does some work. self.run_once(); continue; }; // If we have an active search, we always send any pending // notifications. Even if the search is complete, there may be // notifications to send. if let Some(cb) = &self.opts.event_cb { s.notify(&self.opts.terminal, cb); } if s.is_complete() { // If our search is complete, there's no more work to do, we // can block until we have an event. self.run_once(); continue; } // Tick the search. This will trigger any event callbacks, lock // for data loading, etc. match s.tick() { // We're complete now when we were not before. Notify! Tick::Complete => {} // Forward progress was made. Tick::Progress => {} // All searches are blocked. Let's grab the lock and feed data. Tick::Blocked => { let mut t = lock(&self.opts.terminal); s.feed(&mut t); } } // We have an active search, so we only want to process messages // we have but otherwise return immediately so we can continue the // search. self.run_no_wait(); } log::debug!("starting search thread shutdown"); if let Some(mut s) = self.search.take() { let mut t = lock(&self.opts.terminal); s.deinit(&mut t); } // Send the quit message if let Some(cb) = &self.opts.event_cb { cb(Event::Quit); } log::debug!("search thread exited"); } /// Block until something happens (message, control event, or the /// refresh timer firing) and process it. fn run_once(&mut self) { let timeout = self.next_refresh.saturating_duration_since(Instant::now()); crossbeam_channel::select! { recv(self.mailbox_rx) -> msg => { if let Ok(msg) = msg { self.handle_message(msg); } } recv(self.ctl_rx) -> ctl => { match ctl { Ok(Ctl::Stop) | Err(_) => self.stopped = true, Ok(Ctl::Wakeup) => {} } } default(timeout) => {} } self.run_no_wait(); } /// Process anything already pending without blocking. fn run_no_wait(&mut self) { while let Ok(ctl) = self.ctl_rx.try_recv() { match ctl { Ctl::Stop => self.stopped = true, Ctl::Wakeup => {} } } if self.stopped { return; } self.drain_mailbox(); if Instant::now() >= self.next_refresh { self.refresh(); } } /// Drain the mailbox. fn drain_mailbox(&mut self) { while let Ok(message) = self.mailbox_rx.try_recv() { self.handle_message(message); } } fn handle_message(&mut self, message: Message) { log::debug!("mailbox message={message:?}"); match message { Message::ChangeNeedle(v) => self.change_needle(&v), Message::Select(v) => self.select(v), } } /// The refresh timer callback. fn refresh(&mut self) { // Run our feed if we have a search active. if let Some(s) = self.search.as_mut() { let mut t = lock(&self.opts.terminal); s.feed(&mut t); } self.next_refresh = Instant::now() + REFRESH_INTERVAL; } fn select(&mut self, sel: Select) { let Some(s) = self.search.as_mut() else { return; }; let key = s.last_screen.key; let Some(screen_search) = s.screens[key_index(key)].as_mut() else { return; }; let mut t = lock(&self.opts.terminal); let Some(screen) = t.screens.get_mut(key) else { return; }; // Make the selection. Ignore the result because we don't // care if the selection didn't change. if let Err(e) = screen_search.select(screen, sel) { log::warn!("error selecting search match err={e:?}"); return; } // Grab our match if we have one. If we don't have a selection // then we do nothing. let Some(flattened) = screen_search.selected_match() else { return; }; // No matter what we reset our selected match cache. This will // trigger a callback which will trigger the renderer to wake up // so it can be notified the screen scrolled. s.last_screen.selected = None; // Grab the viewport. Viewports and selections are usually small // so this check isn't very expensive, despite appearing O(N^2), // both Ns are usually equal to 1. let overlaps = { let it = screen.pages.page_iterator( ListDirection::RightDown, Point::Viewport(Coordinate { x: 0, y: 0 }), None, ); let mut overlaps = false; 'outer: for chunk in it { for hl_chunk in &flattened.chunks { if chunk.overlaps(Chunk { node: hl_chunk.node, start: hl_chunk.start, end: hl_chunk.end, }) { overlaps = true; break 'outer; } } } overlaps }; if overlaps { return; } screen.scroll(crate::screen::Scroll::Pin(flattened.start_pin())); } /// Change the search term to the given value. fn change_needle(&mut self, needle: &[u8]) { log::debug!( "changing search needle to '{}'", String::from_utf8_lossy(needle) ); // Stop the previous search if let Some(s) = self.search.as_mut() { // If our search is unchanged, do nothing. if s.viewport.needle().eq_ignore_ascii_case(needle) { return; } { let mut t = lock(&self.opts.terminal); s.deinit(&mut t); } self.search = None; // When the search changes then we need to emit that it stopped. if let Some(cb) = &self.opts.event_cb { cb(Event::TotalMatches(0)); cb(Event::SelectedMatch(None)); cb(Event::ViewportMatches(Vec::new())); } } // No needle means stop the search. if needle.is_empty() { return; } // Setup our search state. let mut search = Search::new(needle); // We need to grab the terminal lock and do an initial feed. { let mut t = lock(&self.opts.terminal); search.feed(&mut t); } self.search = Some(search); } } fn key_index(key: ScreenKey) -> usize { match key { ScreenKey::Primary => 0, ScreenKey::Alternate => 1, } } const KEYS: [ScreenKey; 2] = [ScreenKey::Primary, ScreenKey::Alternate]; /// Search state. struct Search { /// Active viewport search for the active screen. viewport: ViewportSearch, /// The searchers for all the screens (indexed by screen key order). screens: [Option; 2], /// All state related to screen switches, collected so that when /// we switch screens it makes everything related stale, too. last_screen: ScreenState, /// True if we sent the complete notification yet. last_complete: bool, /// The last viewport matches we found. stale_viewport_matches: bool, } struct ScreenState { /// Last active screen key key: ScreenKey, /// Last notified total matches count total: Option, /// Last notified selected match index selected: Option, } struct ScreenStateSelected { idx: usize, highlight: Untracked, } impl ScreenState { fn new(key: ScreenKey) -> Self { Self { key, total: None, selected: None, } } } #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum Tick { /// All searches are complete. Complete, /// Progress was made on at least one screen. Progress, /// All incomplete searches are blocked on feed. Blocked, } impl Search { fn new(needle: &[u8]) -> Self { let mut vp = ViewportSearch::new(needle); // We use dirty tracking for active area changes. Start with it // dirty so the first change is re-searched. vp.active_dirty = Some(true); Self { viewport: vp, screens: [None, None], last_screen: ScreenState::new(ScreenKey::Primary), last_complete: false, stale_viewport_matches: true, } } /// Release tracked pins. Requires the terminal (and therefore its lock). fn deinit(&mut self, t: &mut Terminal) { for key in KEYS { if let Some(mut ss) = self.screens[key_index(key)].take() && let Some(screen) = t.screens.get_mut(key) && ss.is_for_screen(screen) { ss.deinit(screen); } } } /// Returns true if all searches on all screens are complete. fn is_complete(&self) -> bool { self.screens.iter().flatten().all(|s| s.state.is_complete()) } /// Tick the search forward as much as possible without acquiring /// the big lock. Returns the overall tick progress. fn tick(&mut self) -> Tick { let mut result = Tick::Complete; for (i, ss) in self.screens.iter_mut().enumerate() { let Some(ss) = ss else { continue }; match ss.tick() { Ok(()) => result = Tick::Progress, // Ignore... nothing we can do. Err(TickError::Page(e)) => { log::warn!("error ticking screen search key={:?} err={e:?}", KEYS[i]); } // Ignore, good for us. State remains whatever it is. Err(TickError::SearchComplete) => {} // Ignore, too, progressed Err(TickError::FeedRequired) => { // If we think we're complete, we're not because we're // blocked now (nothing made progress). If we made some // progress, we remain in progress since blocked means // no progress at all. if result == Tick::Complete { result = Tick::Blocked; } } } } result } /// Update any state that requires the terminal lock, such as /// feeding additional data to the searches or updating the active screen. fn feed(&mut self, t: &mut Terminal) { // Update our active screen if t.screens.active_key != self.last_screen.key { // The default values will force resets of a bunch of other // state too to force recalculations and notifications. self.last_screen = ScreenState::new(t.screens.active_key); } // Reconcile our screens with the terminal screens. Remove // searchers for screens that no longer exist and add searchers // for screens that do exist but we don't have yet. for key in KEYS { let i = key_index(key); let remove = match (&self.screens[i], t.screens.get(key)) { (None, _) => false, // If the screen doesn't exist at all, remove it. (Some(_), None) => true, // If the screen pointer changed, remove it, the screen // was totally reinitialized. (Some(ss), Some(actual)) => !ss.is_for_screen(actual), }; if remove { // The old screen (and its tracked pins) is gone, so there // is nothing to untrack. self.screens[i] = None; } } for key in KEYS { let i = key_index(key); if self.screens[i].is_some() { continue; } let Some(screen) = t.screens.get_mut(key) else { continue; }; match ScreenSearch::new(screen, self.viewport.needle()) { Ok(ss) => self.screens[i] = Some(ss), Err(e) => { // OOM is probably going to sink the entire ship but // we can just ignore it and wait on the next // reconciliation to try again. log::warn!("error initializing screen search for key={key:?} err={e:?}"); } } } // See the `search_viewport_dirty` flag on the terminal to know // what exactly this is for. But, if this is set, we know the renderer // found the viewport/active area dirty, so we should mark it as // dirty in our viewport searcher so it forces a re-search. if t.flags.search_viewport_dirty { t.flags.search_viewport_dirty = false; // Mark our viewport dirty so it researches the active self.viewport.active_dirty = Some(true); // Reload our active area for our active screen let key = t.screens.active_key; if let Some(ss) = self.screens[key_index(key)].as_mut() && let Err(e) = ss.reload_active(t.screens.active_mut()) { log::warn!("error reloading active area for screen key={key:?} err={e:?}"); } } // Check our viewport for changes. if self.viewport.update(&t.screens.active().pages) { self.stale_viewport_matches = true; } // Feed data for key in KEYS { let i = key_index(key); let Some(ss) = self.screens[i].as_mut() else { continue; }; if ss.state.needs_feed() && let Some(screen) = t.screens.get_mut(key) && let Err(e) = ss.feed(screen) { log::warn!("error feeding screen search key={key:?} err={e:?}"); } } } /// Notify about any changes to the search state. /// /// This doesn't require any locking as it only reads internal state /// (the terminal mutex is only passed through for API symmetry and is /// not locked). fn notify(&mut self, _terminal: &Arc>, cb: &EventCallback) { let Some(screen_search) = self.screens[key_index(self.last_screen.key)].as_ref() else { return; }; // Check our total match data let total = screen_search.matches_len(); if Some(total) != self.last_screen.total { log::debug!("notifying total matches={total}"); self.last_screen.total = Some(total); cb(Event::TotalMatches(total)); } // Check our viewport matches. If they're stale, we do the // viewport search now. We do this as part of notify and not // tick because the viewport search is very fast and doesn't // require ticked progress or feeds. if self.stale_viewport_matches { // We always make stale as false. Even if we fail below // we require a re-feed to re-search the viewport. The feed // process will make it stale again. self.stale_viewport_matches = false; let mut results: Vec = Vec::new(); while let Some(hl) = self.viewport.next() { results.push(hl); } log::debug!("notifying viewport matches len={}", results.len()); cb(Event::ViewportMatches(results)); } // Check our last selected match data. let flattened = screen_search .selected .as_ref() .and_then(|m| screen_search.selected_match().map(|f| (m.idx, f))); if let Some((idx, flattened)) = flattened { let untracked = flattened.untracked(); if let Some(prev) = &self.last_screen.selected && prev.idx == idx && prev.highlight.eql(untracked) { // Same selection, don't update it. } else { // New selection, notify! self.last_screen.selected = Some(ScreenStateSelected { idx, highlight: untracked, }); log::debug!("notifying selection updated idx={idx}"); cb(Event::SelectedMatch(Some(SelectedMatch { idx, highlight: flattened, }))); } } else if self.last_screen.selected.is_some() { log::debug!("notifying selection cleared"); self.last_screen.selected = None; cb(Event::SelectedMatch(None)); } // Send our complete notification if we just completed. if !self.last_complete && self.is_complete() { log::debug!("notifying search complete"); self.last_complete = true; cb(Event::Complete); } } } #[cfg(test)] mod tests { use super::super::test_util::*; use super::*; use crate::point::Tag; use std::sync::mpsc; #[derive(Default)] struct TestUserData { total: usize, selected: Option, viewport: Vec, } #[test] fn search_thread_basic() { let mut t = terminal(20, 2); feed(&mut t, b"Hello, world"); let terminal = Arc::new(Mutex::new(t)); let ud = Arc::new(Mutex::new(TestUserData::default())); let (done_tx, done_rx) = mpsc::channel::<()>(); let done_tx = Mutex::new(done_tx); let ud_cb = Arc::clone(&ud); let cb: EventCallback = Arc::new(move |event| { let mut ud = ud_cb.lock().unwrap(); match event { Event::Quit => {} Event::Complete => { let _ = done_tx.lock().unwrap().send(()); } Event::TotalMatches(v) => ud.total = v, Event::SelectedMatch(v) => ud.selected = v, Event::ViewportMatches(v) => ud.viewport = v, } }); let (handle, join) = Thread::spawn(Options { terminal: Arc::clone(&terminal), event_cb: Some(cb), }) .unwrap(); // Start our search assert!(handle.send(Message::ChangeNeedle(b"world".to_vec()))); // Wait for completion done_rx.recv_timeout(Duration::from_secs(5)).unwrap(); // Stop the thread handle.stop(); join.join().unwrap(); // 1 total matches let ud = ud.lock().unwrap(); assert_eq!(1, ud.total); assert_eq!(1, ud.viewport.len()); let sel = ud.viewport[0].untracked(); let t = terminal.lock().unwrap(); assert_eq!( t.screen().pages.point_from_pin(Tag::Screen, sel.start), Some(screen_pt(7, 0)) ); assert_eq!( t.screen().pages.point_from_pin(Tag::Screen, sel.end), Some(screen_pt(11, 0)) ); } #[test] fn search_thread_select_and_state() { // Extra coverage: selection round trip through the mailbox. let mut t = terminal(20, 2); feed(&mut t, b"foo\r\nbar foo\r\nfoo"); let terminal = Arc::new(Mutex::new(t)); let ud = Arc::new(Mutex::new(TestUserData::default())); let (done_tx, done_rx) = mpsc::channel::<()>(); let done_tx = Mutex::new(done_tx); let ud_cb = Arc::clone(&ud); let cb: EventCallback = Arc::new(move |event| { let mut ud = ud_cb.lock().unwrap(); match event { Event::Complete => { let _ = done_tx.lock().unwrap().send(()); } Event::TotalMatches(v) => ud.total = v, Event::SelectedMatch(v) => { ud.selected = v; let _ = done_tx.lock().unwrap().send(()); } _ => {} } }); let (handle, join) = Thread::spawn(Options { terminal: Arc::clone(&terminal), event_cb: Some(cb), }) .unwrap(); handle.send(Message::ChangeNeedle(b"FOO".to_vec())); done_rx.recv_timeout(Duration::from_secs(5)).unwrap(); handle.send(Message::Select(Select::Next)); done_rx.recv_timeout(Duration::from_secs(5)).unwrap(); // Clearing the needle stops the search and resets state. handle.send(Message::ChangeNeedle(Vec::new())); done_rx.recv_timeout(Duration::from_secs(5)).unwrap(); handle.stop(); join.join().unwrap(); let ud = ud.lock().unwrap(); assert_eq!(0, ud.total); assert!(ud.selected.is_none()); } }