diff --git a/crates/komorebi-transport/src/subscribers.rs b/crates/komorebi-transport/src/subscribers.rs index 6d131c5b..8ee004c1 100644 --- a/crates/komorebi-transport/src/subscribers.rs +++ b/crates/komorebi-transport/src/subscribers.rs @@ -43,7 +43,23 @@ pub fn notify_sockets( match connect(path) { Ok(mut stream) => { tracing::debug!(target: LOG_TARGET, "pushed notification to subscriber: {socket}"); - stream.write_all(notification.as_bytes())?; + match stream.write_all(notification.as_bytes()) { + Ok(()) => {} + // a subscriber that does not read (the stream has a write timeout) + // is skipped, so that it cannot hold up the one that is pushing + Err(error) + if matches!( + error.kind(), + std::io::ErrorKind::TimedOut | std::io::ErrorKind::WouldBlock + ) => + { + tracing::warn!( + target: LOG_TARGET, + "subscriber {socket} did not take the notification in time: {error}" + ); + } + Err(error) => return Err(error), + } } Err(_) => { stale_sockets.push(socket.clone()); @@ -132,4 +148,41 @@ mod tests { assert!(live.accept().is_ok()); let _ = std::fs::remove_dir_all(&dir); } + + /// A stream whose write times out, as one with a write timeout does when nobody reads it + struct Stuck; + + impl Write for Stuck { + fn write(&mut self, _: &[u8]) -> std::io::Result { + Err(std::io::ErrorKind::TimedOut.into()) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + + #[test] + fn a_subscriber_that_times_out_is_skipped_and_stays_subscribed() { + let dir = dir("timeout"); + let mut sockets = HashMap::from([ + ("a.sock".to_string(), dir.join("a.sock")), + ("b.sock".to_string(), dir.join("b.sock")), + ]); + let pushed = std::cell::Cell::new(0); + + let result = notify_sockets("{}", &mut sockets, &HashMap::new(), false, &dir, |_| { + pushed.set(pushed.get() + 1); + Ok(Stuck) + }); + + assert!(result.is_ok()); + assert_eq!( + pushed.get(), + 2, + "the subscribers after the stuck one are tried" + ); + assert_eq!(sockets.len(), 2, "a slow subscriber is not removed"); + let _ = std::fs::remove_dir_all(&dir); + } } diff --git a/komorebi/src/lib.rs b/komorebi/src/lib.rs index aa95cae9..3c140675 100644 --- a/komorebi/src/lib.rs +++ b/komorebi/src/lib.rs @@ -14,13 +14,13 @@ pub mod focus_manager; pub mod live_resize; pub use komorebi_collections::Lockable; pub use komorebi_collections::lockable_sequence; +pub mod lock_holder; pub mod monitor; pub mod monitor_reconciliator; pub mod pan; pub mod process_command; pub mod process_event; pub mod process_movement; -pub mod lock_holder; pub mod reaper; pub mod scrolling; pub mod set_window_position; @@ -389,7 +389,13 @@ pub fn notify_subscribers( &options, state_has_been_modified || is_override_event, &DATA_DIR, - |path| UnixStream::connect(path), + |path| { + let stream = UnixStream::connect(path)?; + // a subscriber that does not read must not hold up the writer, which has the + // window manager locked + stream.set_write_timeout(Some(std::time::Duration::from_secs(1)))?; + Ok(stream) + }, )?; let mut stale_pipes = vec![];