//! The location to where write-related messages are sent //! (`termio/mailbox.zig`). //! //! Zig has a single `spsc` variant: a fixed-capacity blocking queue plus an //! `xev.Async` wakeup. Here the queue is a bounded `crossbeam-channel` (so a //! full queue can hand the rejected message back to the sender, which the //! `BlockingQueue` cannot) and the wakeup is an eventfd ([`Wakeup`]). use std::sync::{Mutex, MutexGuard, PoisonError}; use crossbeam_channel::{Receiver, Sender, TrySendError, bounded}; use crate::{message::Message, wakeup::Wakeup}; /// Capacity of the queue. Hardcoded to a value that empirically has made /// sense for Ghostty usage (`BlockingQueue(termio.Message, 64)`). pub const CAPACITY: usize = 64; /// The termio mailbox. Cloning is cheap and shares the same queue; all /// clones are producers, but only the IO thread should call [`Mailbox::pop`] /// / [`Mailbox::drain`] (single consumer by convention, like Zig). #[derive(Clone, Debug)] pub struct Mailbox { tx: Sender, rx: Receiver, wakeup: Wakeup, } impl Mailbox { /// Init the SPSC mailbox (`initSPSC`). pub fn new() -> std::io::Result { let (tx, rx) = bounded(CAPACITY); Ok(Mailbox { tx, rx, wakeup: Wakeup::new()?, }) } /// The wakeup the IO thread waits on. pub fn wakeup(&self) -> &Wakeup { &self.wakeup } /// Sends the given message without notifying there are messages. /// /// If the queue is full this wakes up the IO thread so it can make room /// and then blocks until the message is queued. Do NOT call this while /// holding the terminal lock if the queue may be full (the IO thread may /// need the lock to make progress); use [`Mailbox::send_locked`] or /// [`Mailbox::try_send`] in that case. pub fn send(&self, msg: Message) { match self.tx.try_send(msg) { Ok(()) => {} Err(TrySendError::Full(msg)) => { if let Err(err) = self.wakeup.notify() { log::warn!("failed to wake up writer, data will be dropped err={err}"); return; } // Disconnected can't happen: we own a receiver too. let _ = self.tx.send(msg); } Err(TrySendError::Disconnected(_)) => {} } } /// Like [`Mailbox::send`] for callers that hold a lock (the Zig /// `mutex` argument): if the queue is full the guard is released while /// blocking and re-acquired afterwards. pub fn send_locked<'a, T>( &self, msg: Message, mutex: &'a Mutex, guard: MutexGuard<'a, T>, ) -> MutexGuard<'a, T> { match self.tx.try_send(msg) { Ok(()) | Err(TrySendError::Disconnected(_)) => guard, Err(TrySendError::Full(msg)) => { if let Err(err) = self.wakeup.notify() { log::warn!("failed to wake up writer, data will be dropped err={err}"); return guard; } // Unlock the renderer state so the writer thread can acquire // it, then queue our message before continuing. drop(guard); let _ = self.tx.send(msg); mutex.lock().unwrap_or_else(PoisonError::into_inner) } } } /// Queue without ever blocking. Returns the message back if the queue is /// full. pub fn try_send(&self, msg: Message) -> Result<(), Message> { match self.tx.try_send(msg) { Ok(()) => Ok(()), Err(TrySendError::Full(m) | TrySendError::Disconnected(m)) => Err(m), } } /// Notify that there are new messages. pub fn notify(&self) { if let Err(err) = self.wakeup.notify() { log::warn!("failed to notify writer, data will be dropped err={err}"); } } /// Pop one message without blocking (IO thread). pub fn pop(&self) -> Option { self.rx.try_recv().ok() } /// Pop everything currently queued (IO thread). pub fn drain(&self) -> impl Iterator + '_ { std::iter::from_fn(|| self.pop()) } /// Number of queued messages. pub fn len(&self) -> usize { self.rx.len() } pub fn is_empty(&self) -> bool { self.rx.is_empty() } } #[cfg(test)] mod tests { use std::sync::Arc; use super::*; #[test] fn send_pop_order() { let mb = Mailbox::new().unwrap(); mb.send(Message::Focused(true)); mb.send(Message::LinefeedMode(false)); assert_eq!(mb.len(), 2); assert!(matches!(mb.pop(), Some(Message::Focused(true)))); assert!(matches!(mb.pop(), Some(Message::LinefeedMode(false)))); assert!(mb.pop().is_none()); } #[test] fn try_send_returns_message_when_full() { let mb = Mailbox::new().unwrap(); for _ in 0..CAPACITY { mb.try_send(Message::Crash).unwrap(); } let back = mb.try_send(Message::Focused(true)).unwrap_err(); assert!(matches!(back, Message::Focused(true))); } /// Block until the wakeup is signaled. #[cfg(unix)] fn wait_readable(w: &Wakeup) { let mut fds = [rustix::event::PollFd::from_borrowed_fd( w.as_fd(), rustix::event::PollFlags::IN, )]; rustix::event::poll(&mut fds, None).unwrap(); } /// Block until the wakeup is signaled (this consumes the auto-reset /// event; the callers' `drain` after it is then a harmless no-op). #[cfg(windows)] #[allow(unsafe_code)] fn wait_readable(w: &Wakeup) { use std::os::windows::io::AsRawHandle; // SAFETY: valid event handle owned by the wakeup. unsafe { windows_sys::Win32::System::Threading::WaitForSingleObject( w.as_handle().as_raw_handle().cast(), windows_sys::Win32::System::Threading::INFINITE, ) }; } #[test] fn blocking_send_wakes_consumer_and_completes() { let mb = Mailbox::new().unwrap(); for _ in 0..CAPACITY { mb.send(Message::Crash); } let consumer = mb.clone(); let t = std::thread::spawn(move || { // Wait for the wakeup the full-queue path posts, then drain. wait_readable(consumer.wakeup()); consumer.wakeup().drain(); consumer.drain().count() }); mb.send(Message::Focused(false)); // blocks until the consumer drains let n = t.join().unwrap(); assert!(n >= CAPACITY); } #[test] fn send_locked_releases_lock_when_full() { let mb = Mailbox::new().unwrap(); for _ in 0..CAPACITY { mb.send(Message::Crash); } let m = Arc::new(Mutex::new(0u32)); let consumer = mb.clone(); let m2 = Arc::clone(&m); // The consumer needs the lock to make progress (like the IO thread). let t = std::thread::spawn(move || { wait_readable(consumer.wakeup()); let mut g = m2.lock().unwrap(); *g += 1; consumer.drain().count() }); let g = m.lock().unwrap(); let g = mb.send_locked(Message::Focused(true), &m, g); assert!(*g <= 1); drop(g); t.join().unwrap(); } }