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

Show file tree
Hide file tree
Changes from 4 commits
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
35 changes: 17 additions & 18 deletions src/processor.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -89,7 +89,7 @@ impl FileEventQueue {
pub(crate) trait EventProcessor {
fn get_events(&mut self) -> Vec<RawEvent>;
fn get_errors(&mut self) -> Vec<NotifyError>;
fn add_event(&mut self, event: NotifyEvent);
fn add_event(&mut self, event: NotifyEvent, time: Instant);
fn add_error(&mut self, error: NotifyError);
}

Expand Down Expand Up @@ -117,8 +117,7 @@ impl<T: FileIdCache> CrossPlatformEventProcessor<T> {
}
}

fn handle_rename_from(&mut self, event: NotifyEvent) {
let time = Instant::now();
fn handle_rename_from(&mut self, event: NotifyEvent, time: Instant) {
let path = &event.paths[0];

// store event
Expand All @@ -130,7 +129,7 @@ impl<T: FileIdCache> CrossPlatformEventProcessor<T> {
self.push_event(event, time);
}

fn handle_rename_to(&mut self, event: NotifyEvent) {
fn handle_rename_to(&mut self, event: NotifyEvent, time: Instant) {
self.file_cache.add_path(&event.paths[0]);

let trackers_match = self
Expand All @@ -155,11 +154,11 @@ impl<T: FileIdCache> CrossPlatformEventProcessor<T> {
// connect rename
let (mut rename_event, _) = self.rename_event.take().unwrap(); // unwrap is safe because `rename_event` must be set at this point
let path = rename_event.paths.remove(0);
let time = rename_event.time;
self.push_rename_event(path, event, time);
let rename_time = rename_event.time;
self.push_rename_event(path, event, rename_time);
} else {
// move in
self.push_event(event, Instant::now());
self.push_event(event, time);
}

self.rename_event = None;
Expand Down Expand Up @@ -351,12 +350,12 @@ impl<T: FileIdCache> EventProcessor for CrossPlatformEventProcessor<T> {
}

/// Add new event to debouncer cache
fn add_event(&mut self, event: NotifyEvent) {
fn add_event(&mut self, event: NotifyEvent, time: Instant) {
// log::trace!("raw event: {event:?}");

if event.need_rescan() {
self.file_cache.rescan();
self.rescan_event = Some(event.into());
self.rescan_event = Some(RawEvent::new(event, time));
return;
}

Expand All @@ -366,22 +365,22 @@ impl<T: FileIdCache> EventProcessor for CrossPlatformEventProcessor<T> {
EventKind::Create(_) => {
self.file_cache.add_path(path);

self.push_event(event, Instant::now());
self.push_event(event, time);
}
EventKind::Modify(ModifyKind::Name(rename_mode)) => {
match rename_mode {
RenameMode::Any => {
if event.paths[0].exists() {
self.handle_rename_to(event);
self.handle_rename_to(event, time);
} else {
self.handle_rename_from(event);
self.handle_rename_from(event, time);
}
}
RenameMode::To => {
self.handle_rename_to(event);
self.handle_rename_to(event, time);
}
RenameMode::From => {
self.handle_rename_from(event);
self.handle_rename_from(event, time);
}
RenameMode::Both => {
// ignore and handle `To` and `From` events instead
Expand All @@ -392,7 +391,7 @@ impl<T: FileIdCache> EventProcessor for CrossPlatformEventProcessor<T> {
}
}
EventKind::Remove(_) => {
self.push_remove_event(event, Instant::now());
self.push_remove_event(event, time);
}
EventKind::Other => {
// ignore meta events
Expand All @@ -402,7 +401,7 @@ impl<T: FileIdCache> EventProcessor for CrossPlatformEventProcessor<T> {
self.file_cache.add_path(path);
}

self.push_event(event, Instant::now());
self.push_event(event, time);
}
}
}
Expand Down Expand Up @@ -458,8 +457,8 @@ impl EventProcessor for BatchProcessor {
std::mem::take(&mut self.errors)
}

fn add_event(&mut self, event: NotifyEvent) {
self.events.push(RawEvent::new(event, Instant::now()));
fn add_event(&mut self, event: NotifyEvent, time: Instant) {
self.events.push(RawEvent::new(event, time));
}

fn add_error(&mut self, error: NotifyError) {
Expand Down
98 changes: 56 additions & 42 deletions src/watcher.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,6 @@
use std::io::ErrorKind as IOErrorKind;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use std::time::{Duration, Instant};

use crate::events::EventType;
use crate::events::access::from_access_kind;
Expand All @@ -16,7 +15,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 @@ -25,13 +24,18 @@ use crate::processor::{BatchProcessor, EventProcessor, RawEvent};

pyo3::create_exception!(_inotify_toolkit_lib, WatcherError, PyException);

type TimestampedEvent = (Instant, Result<Event, notify::Error>);
type EventReceiver = mpsc::Receiver<TimestampedEvent>;

#[derive(Debug)]
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: Option<EventReceiver>,
rx_return: Option<oneshot::Receiver<EventReceiver>>,
tx: broadcast::Sender<Vec<EventType>>,
stop_tx: Option<oneshot::Sender<()>>,
drain_handle: Option<tokio::task::JoinHandle<()>>,
Expand All @@ -44,35 +48,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::channel(event_buffer_size);

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.try_send((Instant::now(), e)) {
eprintln!("event channel full or closed, dropping event: {e}");
Comment on lines +128 to +129

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 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.

Suggested change
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);
}
}

Copilot uses AI. Check for mistakes.
}
},
config,
Expand All @@ -81,8 +70,10 @@ impl Watcher {
Ok(Self {
debug,
event_buffer_size,
buffering_duration,
inner,
processor,
event_rx: Some(event_rx),
rx_return: None,
tx,
stop_tx: None,
drain_handle: None,
Expand Down Expand Up @@ -149,47 +140,64 @@ impl Watcher {
let _ = tx.send(());
}

self.recover_event_rx();

let (new_tx, _rx) = broadcast::channel::<Vec<EventType>>(self.event_buffer_size);
self.tx = new_tx;
}

fn recover_event_rx(&mut self) {
if let Some(handle) = self.drain_handle.take() {
handle.abort();
}

let (new_tx, _rx) = broadcast::channel::<Vec<EventType>>(self.event_buffer_size);
self.tx = new_tx;
if let Some(mut rx_return) = self.rx_return.take() {
if let Ok(rx) = rx_return.try_recv() {
self.event_rx = Some(rx);

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.

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.

Suggested change
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;
}

Copilot uses AI. Check for mistakes.
}
}
}

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(());
}

if let Some(handle) = self.drain_handle.take() {
handle.abort();
}
self.recover_event_rx();

let mut event_rx = match self.event_rx.take() {
Some(rx) => rx,
None => return,
};

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.

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.

Copilot uses AI. Check for mistakes.

let (stop_tx, mut stop_rx) = oneshot::channel();
self.stop_tx = Some(stop_tx);

let proc = Arc::clone(&self.processor);
let (rx_return_tx, rx_return_rx) = oneshot::channel();
self.rx_return = Some(rx_return_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;
}
};
(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),
}
}
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,10 +215,16 @@ 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}");
}
}
}
}
}

let _ = rx_return_tx.send(event_rx);
}));
}

Expand Down