🔨 Remove potential cross-threads contention by eliminating Mutex on Batch Processor#55
🔨 Remove potential cross-threads contention by eliminating Mutex on Batch Processor#55roma-glushko wants to merge 6 commits into
Conversation
There was a problem hiding this comment.
Pull request overview
This PR refactors the Rust Watcher event ingestion path to avoid cross-thread contention by removing the shared Arc<Mutex<BatchProcessor>> and routing raw notify events through a Tokio mpsc channel to the drain task.
Changes:
- Replace direct, mutex-protected
BatchProcessoraccess in the notify callback withmpsc::unbounded_channel()sending. - Move
BatchProcessorownership into the drain task and drain the channel on each tick usingtry_recv. - Add
buffering_durationtoWatcherstate to construct the drain-sideBatchProcessor.
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| let (event_tx, event_rx) = mpsc::unbounded_channel(); | ||
|
|
||
| let (tx, _rx) = broadcast::channel::<Vec<EventType>>(event_buffer_size); | ||
|
|
||
| let config = notify::Config::default().with_follow_symlinks(follow_symlinks); | ||
|
|
||
| let inner = RecommendedWatcher::new( | ||
| move |e: Result<Event, notify::Error>| { | ||
| let mut event_processor = match processor_c.lock() { | ||
| Ok(guard) => guard, | ||
| Err(e) => { | ||
| eprintln!("notifykit: event processor lock poisoned, dropping event: {e}"); | ||
| return; | ||
| } | ||
| }; | ||
|
|
||
| if debug { | ||
| println!("raw event: {:?}", e); | ||
| } | ||
|
|
||
| match e { | ||
| Ok(e) => event_processor.add_event(e), | ||
| Err(e) => event_processor.add_error(e), | ||
| if let Err(e) = event_tx.send(e) { | ||
| eprintln!("failed to send event: {e}"); | ||
| } |
There was a problem hiding this comment.
Using an unbounded mpsc channel here can lead to unbounded memory growth if filesystem events arrive faster than the drain loop can process them (e.g., slow consumer, large tick_duration, or drain temporarily stopped). Consider switching to a bounded channel (mpsc::channel) sized for expected bursts and using try_send / drop-with-metrics when full, or otherwise introducing backpressure/limits so the process can’t OOM under sustained event load.
| let mut processor = BatchProcessor::new(buffering_duration); | ||
| let mut ticker = time::interval(tick_duration); | ||
|
|
||
| loop { | ||
| tokio::select! { | ||
| _ = &mut stop_rx => break, | ||
| _ = ticker.tick() => { | ||
| let (raw, errs) = { | ||
| let mut p = match proc.lock() { | ||
| Ok(guard) => guard, | ||
| Err(e) => { | ||
| eprintln!("notifykit: event processor lock poisoned, skipping drain tick: {e}"); | ||
| continue; | ||
| if let Ok(mut rx) = event_rx.lock() { | ||
| while let Ok(result) = rx.try_recv() { | ||
| match result { | ||
| Ok(event) => processor.add_event(event), | ||
| Err(error) => processor.add_error(error), |
There was a problem hiding this comment.
Events are now timestamped when they’re drained (processor.add_event() uses Instant::now()) rather than when notify delivers them. That changes the effective debounce/buffering window by up to tick_duration (and can merge/split batches differently compared to the previous design). If the original timing semantics matter, consider sending an Instant alongside each event (captured in the notify callback) or restructuring the drain task to recv events continuously and only flush on ticker ticks.
| if let Ok(mut rx) = event_rx.lock() { | ||
| while let Ok(result) = rx.try_recv() { | ||
| match result { | ||
| Ok(event) => processor.add_event(event), | ||
| Err(error) => processor.add_error(error), | ||
| } | ||
| }; | ||
| (p.get_events(), p.get_errors()) | ||
| }; | ||
| } | ||
| } |
There was a problem hiding this comment.
The async drain task uses a std::sync::Mutex around the UnboundedReceiver. Even if contention is expected to be low, this is a blocking mutex inside the Tokio runtime and it can be poisoned (reintroducing the ‘poisoned mutex’ failure mode the PR description is trying to avoid). Consider replacing this with tokio::sync::Mutex (no poisoning) or refactoring so the receiver is moved into the drain task (no Arc/Mutex needed) and polled via recv/try_recv within the task.
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 2 out of 2 changed files in this pull request and generated 2 comments.
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| if let Some(mut rx_return) = self.rx_return.take() { | ||
| if let Ok(rx) = rx_return.try_recv() { | ||
| self.event_rx = Some(rx); |
There was a problem hiding this comment.
recover_event_rx() drops rx_return if the drain task hasn’t returned the receiver yet (e.g. try_recv() returns Empty). Because self.rx_return is take()n unconditionally, you lose the only handle that could ever recover event_rx, and subsequent start_drain() calls will permanently have event_rx = None (and/or the mpsc channel becomes closed after abort). Consider (a) not take()ing the receiver unless it is ready (put it back on Empty), and (b) avoiding handle.abort() here or ensuring the receiver is returned even on abort (e.g. via a Drop guard), otherwise restart/stop can leave the watcher unable to drain events.
| if let Some(mut rx_return) = self.rx_return.take() { | |
| if let Ok(rx) = rx_return.try_recv() { | |
| self.event_rx = Some(rx); | |
| if let Some(rx_return) = &mut self.rx_return { | |
| match rx_return.try_recv() { | |
| Ok(rx) => { | |
| // Successfully recovered the event receiver; we no longer need rx_return. | |
| self.event_rx = Some(rx); | |
| self.rx_return = None; | |
| } | |
| Err(tokio::sync::oneshot::error::TryRecvError::Empty) => { | |
| // The drain task has not yet returned the receiver; keep rx_return so | |
| // that a future call to recover_event_rx can try again. | |
| } | |
| Err(tokio::sync::oneshot::error::TryRecvError::Closed) => { | |
| // The oneshot will never yield a receiver; drop it to avoid retrying. | |
| self.rx_return = None; | |
| } |
| let mut event_rx = match self.event_rx.take() { | ||
| Some(rx) => rx, | ||
| None => return, | ||
| }; |
There was a problem hiding this comment.
start_drain() silently returns when self.event_rx is None. With the current “move receiver into task and return it later” design, this can happen after a stop/restart attempt or if recover_event_rx() couldn’t immediately reclaim the receiver. Instead of returning, consider blocking until the receiver is available (or recreating the channel/sender pair) so callers don’t end up with a watcher that never emits events and only logs dropped events.
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 3 out of 3 changed files in this pull request and generated 4 comments.
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| _ = ticker.tick() => { | ||
| let (raw, errs) = { | ||
| let mut p = match proc.lock() { | ||
| Ok(guard) => guard, | ||
| Err(e) => { | ||
| eprintln!("notifykit: event processor lock poisoned, skipping drain tick: {e}"); | ||
| continue; | ||
| } | ||
| }; | ||
| (p.get_events(), p.get_errors()) | ||
| }; | ||
| while let Ok((time, result)) = event_rx.try_recv() { | ||
| match result { | ||
| Ok(event) => processor.add_event(event, time), | ||
| Err(error) => processor.add_error(error), | ||
| } | ||
| } |
There was a problem hiding this comment.
The drain task only consumes from event_rx on each ticker.tick() via try_recv(). If tick_duration is large or events arrive in bursts, the bounded channel can fill up and try_send will start dropping events even though the drain task is idle between ticks. Consider also awaiting event_rx.recv() (and buffering timestamps) so the receiver keeps up continuously while still emitting batches on tick boundaries.
| let handle = runtime.spawn(async move { | ||
| let mut processor = BatchProcessor::new(buffering_duration); | ||
| let mut ticker = time::interval(tick_duration); |
There was a problem hiding this comment.
BatchProcessor is now created inside the drain task. When stop_and_recover stops a running drain, any events already moved from event_rx into processor but not yet emitted will be dropped when the task exits. Previously buffered events survived drain restarts because the processor lived outside the task. If drain restarts are expected (e.g., multiple events() calls), consider persisting the processor across drain lifecycles or flushing pending buffered events back into the receiver/channel before exit.
| if let Err(e) = event_tx.try_send((Instant::now(), e)) { | ||
| eprintln!("event channel full or closed, dropping event: {e}"); |
There was a problem hiding this comment.
The log message says "dropping event" but {e} here is a TrySendError whose Display typically only reports the reason (full/closed), not the event being dropped. Consider logging the dropped payload (or using {:?} and matching Full/Closed) so debug output is actionable.
| if let Err(e) = event_tx.try_send((Instant::now(), e)) { | |
| eprintln!("event channel full or closed, dropping event: {e}"); | |
| if let Err(err) = event_tx.try_send((Instant::now(), e)) { | |
| match err { | |
| tokio::sync::mpsc::error::TrySendError::Full((_ts, payload)) => { | |
| eprintln!("event channel full, dropping event: {:?}", payload); | |
| } | |
| tokio::sync::mpsc::error::TrySendError::Closed((_ts, payload)) => { | |
| eprintln!("event channel closed, dropping event: {:?}", payload); | |
| } | |
| } |
| fn recover_event_rx(&mut self, runtime: &tokio::runtime::Runtime) { | ||
| if let Some(handle) = self.drain_handle.take() { | ||
| // Wait for the task to finish so it can return the receiver. | ||
| // The caller must send the stop signal before calling this. | ||
| let _ = runtime.block_on(handle); | ||
| } | ||
|
|
||
| if let Some(mut rx_return) = self.rx_return.take() { | ||
| match rx_return.try_recv() { | ||
| Ok(rx) => self.event_rx = Some(rx), | ||
| Err(oneshot::error::TryRecvError::Empty) => { | ||
| // Task hasn't returned the receiver yet; keep the handle for later | ||
| self.rx_return = Some(rx_return); | ||
| } | ||
| Err(oneshot::error::TryRecvError::Closed) => { | ||
| // Task dropped the sender (e.g. panic); receiver is lost |
There was a problem hiding this comment.
recover_event_rx blocks the current thread with runtime.block_on(handle). In libraries this can be risky because calling Runtime::block_on from within an active Tokio runtime context will panic / can deadlock depending on where stop()/start_drain() is invoked. Consider making recovery non-blocking (e.g., abort + detach, or perform the join from a dedicated blocking thread) so stop()/start_drain() are safe to call regardless of caller context.
| fn recover_event_rx(&mut self, runtime: &tokio::runtime::Runtime) { | |
| if let Some(handle) = self.drain_handle.take() { | |
| // Wait for the task to finish so it can return the receiver. | |
| // The caller must send the stop signal before calling this. | |
| let _ = runtime.block_on(handle); | |
| } | |
| if let Some(mut rx_return) = self.rx_return.take() { | |
| match rx_return.try_recv() { | |
| Ok(rx) => self.event_rx = Some(rx), | |
| Err(oneshot::error::TryRecvError::Empty) => { | |
| // Task hasn't returned the receiver yet; keep the handle for later | |
| self.rx_return = Some(rx_return); | |
| } | |
| Err(oneshot::error::TryRecvError::Closed) => { | |
| // Task dropped the sender (e.g. panic); receiver is lost | |
| fn recover_event_rx(&mut self, _runtime: &tokio::runtime::Runtime) { | |
| if let Some(_handle) = self.drain_handle.take() { | |
| // Do not block on the task here; let it finish in the background. | |
| // The task is responsible for sending back the receiver via `rx_return`. | |
| } | |
| if let Some(mut rx_return) = self.rx_return.take() { | |
| match rx_return.try_recv() { | |
| Ok(rx) => self.event_rx = Some(rx), | |
| Err(oneshot::error::TryRecvError::Empty) => { | |
| // Task hasn't returned the receiver yet; keep the receiver channel for later. | |
| self.rx_return = Some(rx_return); | |
| } | |
| Err(oneshot::error::TryRecvError::Closed) => { | |
| // Task dropped the sender (e.g. panic); receiver is lost. |
c-h-russell-walker
left a comment
c-h-russell-walker
left a comment
There was a problem hiding this comment.
Very nice - PR looks good to me 👍
Remove a potential cross-thread contention during event queuing by replacing
Arc<Mutex<BatchProcessor>>with boundedmpsc::channel.The mutex was contended across threads because:
If the mutex got poisoned (e.g. a panic in either side), events would be dropped.
mpsc::channelshould help to submit events in an non-blocking way. If we can't queue events we will be dropping them.