Skip to content

🔨 Remove potential cross-threads contention by eliminating Mutex on Batch Processor #55

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
roma-glushko wants to merge 6 commits into
base: main
Choose a base branch
from
Open

🔨 Remove potential cross-threads contention by eliminating Mutex on Batch Processor #55

Changes from 1 commit
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
316a73a
Remove potential cross-threads contention by eliminating Mutex on Bat…
roma-glushko Mar 9, 2026
b52fac3
Use bounded channel & avoid mutex poisoning
roma-glushko Mar 9, 2026
1f4fc25
Make sure the event timestamp is not drifted by buffering time
roma-glushko Mar 9, 2026
674f272
Linting
roma-glushko Mar 9, 2026
15e247c
Made sure maturine is running on the new rust toolchain
roma-glushko Mar 9, 2026
bb84a81
Extract Event Queue Drain logic into a separate testable component
roma-glushko Mar 9, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
63 changes: 29 additions & 34 deletions src/watcher.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@ use notify::{
use pyo3::exceptions::{PyException, PyFileNotFoundError, PyOSError, PyPermissionError};
use pyo3::prelude::*;
use tokio::{
sync::{broadcast, oneshot},
sync::{broadcast, mpsc, oneshot},
time,
};
// use crate::file_cache::FileCache;
Expand All @@ -29,9 +29,10 @@ pyo3::create_exception!(_inotify_toolkit_lib, WatcherError, PyException);
pub(crate) struct Watcher {
debug: bool,
event_buffer_size: usize,
buffering_duration: Duration,
inner: RecommendedWatcher,
// file_cache: FileCache,
processor: Arc<Mutex<BatchProcessor>>, // TODO: use the EventProcessor trait instead
event_rx: Arc<Mutex<mpsc::UnboundedReceiver<Result<Event, notify::Error>>>>,
tx: broadcast::Sender<Vec<EventType>>,
stop_tx: Option<oneshot::Sender<()>>,
drain_handle: Option<tokio::task::JoinHandle<()>>,
Expand All @@ -44,35 +45,20 @@ impl Watcher {
debug: bool,
follow_symlinks: bool,
) -> Result<Self, notify::Error> {
// TODO: hide usage of file cache from Watcher
// let file_cache = FileCache::new();
// let file_cache_c = file_cache.clone();

let buffering_duration = Duration::from_millis(buffering_time_ms);
let processor = Arc::new(Mutex::new(BatchProcessor::new(buffering_duration)));
let processor_c = processor.clone();
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}");
}

Copilot AI Mar 9, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Copilot uses AI. Check for mistakes.
},
config,
Expand All @@ -81,8 +67,9 @@ impl Watcher {
Ok(Self {
debug,
event_buffer_size,
buffering_duration,
inner,
processor,
event_rx: Arc::new(Mutex::new(event_rx)),
tx,
stop_tx: None,
drain_handle: None,
Expand Down Expand Up @@ -157,7 +144,7 @@ impl Watcher {
self.tx = new_tx;
}

pub fn start_drain(&mut self, debounce_delay: Duration, event_filter: Option<EventFilter>) {
pub fn start_drain(&mut self, tick_duration: Duration, event_filter: Option<EventFilter>) {
if let Some(tx) = self.stop_tx.take() {
let _ = tx.send(());
}
Expand All @@ -169,27 +156,31 @@ impl Watcher {
let (stop_tx, mut stop_rx) = oneshot::channel();
self.stop_tx = Some(stop_tx);

let proc = Arc::clone(&self.processor);
let event_rx = Arc::clone(&self.event_rx);
let tx = self.tx.clone();
let debug = self.debug;
let buffering_duration = self.buffering_duration;

self.drain_handle = Some(pyo3_async_runtimes::tokio::get_runtime().spawn(async move {
let mut ticker = time::interval(debounce_delay);
let mut processor = BatchProcessor::new(buffering_duration);
let mut ticker = time::interval(tick_duration);
Comment on lines +227 to +229

Copilot AI Mar 9, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Copilot uses AI. Check for mistakes.

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),

Copilot AI Mar 9, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Copilot uses AI. Check for mistakes.
}
};
(p.get_events(), p.get_errors())
};
}
}

Copilot AI Mar 9, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Copilot uses AI. Check for mistakes.
Comment on lines 234 to +240

Copilot AI Mar 9, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Copilot uses AI. Check for mistakes.

let raw = processor.get_events();
let errs = processor.get_errors();

if debug && !raw.is_empty() { println!("processed: {:?}", raw); }
if !errs.is_empty() { eprintln!("errors: {:?}", errs); }
if raw.is_empty() { continue; }
Expand All @@ -207,7 +198,11 @@ impl Watcher {
}
}

if !batch.is_empty() { let _ = tx.send(batch); }
if !batch.is_empty() {
if let Err(e) = tx.send(batch) {
eprintln!("failed to broadcast events: {e}");
}
}
}
}
}
Expand Down