From 0bdb61bfb6aff0123c85be49ea29a84a45472a24 Mon Sep 17 00:00:00 2001 From: "Jeong, YunWon" <69878+youknowone@users.noreply.github.com> Date: Sat, 19 Sep 2026 15:19:57 +0900 Subject: [PATCH] Serialize SemLock bookkeeping for Condition notify (#8739) * Serialize SemLock owner bookkeeping around waits Hold a mutex around SemLock acquire/release/_count so the per-object owner count cannot tear while threads and processes share a Condition. Drop the mutex only for the blocking wait. Mark Unix SemHandle Send/Sync. Add a spawn Condition.notify_all snippet. Assisted-by: Grok:4.6 * Repeat the Condition.notify_all snippet 20 times Give the snippets job more chances to see a lost wakeup. Assisted-by: Grok:4.6 * Cover one-by-one Condition.notify in the snippet Wake a process waiter and a thread waiter with two notify() calls, not only notify_all(), and keep the 20-repeat loop. Assisted-by: Grok:4.6 * address review: bound snippet waits with timeouts Assisted-by: Grok:4.6 --- crates/host_env/src/multiprocessing.rs | 9 ++ crates/stdlib/src/multiprocessing.rs | 62 +++++++++---- .../stdlib_multiprocessing_condition.py | 87 +++++++++++++++++++ 3 files changed, 140 insertions(+), 18 deletions(-) create mode 100644 extra_tests/snippets/stdlib_multiprocessing_condition.py diff --git a/crates/host_env/src/multiprocessing.rs b/crates/host_env/src/multiprocessing.rs index 067a86307..37c0798a1 100644 --- a/crates/host_env/src/multiprocessing.rs +++ b/crates/host_env/src/multiprocessing.rs @@ -23,6 +23,13 @@ pub struct SemHandle { raw: *mut sem_t, } +// POSIX named semaphores are safe to post/wait from any thread that holds +// a `sem_t *` to the same kernel object. +#[cfg(unix)] +unsafe impl Send for SemHandle {} +#[cfg(unix)] +unsafe impl Sync for SemHandle {} + #[cfg(unix)] #[derive(Copy, Clone, Debug, Eq, PartialEq)] pub enum SemError { @@ -110,7 +117,9 @@ pub struct SemHandle { raw: HANDLE, } +#[cfg(windows)] unsafe impl Send for SemHandle {} +#[cfg(windows)] unsafe impl Sync for SemHandle {} #[cfg(unix)] diff --git a/crates/stdlib/src/multiprocessing.rs b/crates/stdlib/src/multiprocessing.rs index c79af002b..22d546d49 100644 --- a/crates/stdlib/src/multiprocessing.rs +++ b/crates/stdlib/src/multiprocessing.rs @@ -11,6 +11,7 @@ mod _multiprocessing { types::Constructor, }; use core::sync::atomic::{AtomicI32, AtomicU32, Ordering}; + use rustpython_common::lock::PyMutex; use rustpython_host_env::multiprocessing as host_multiprocessing; // These match the values in Lib/multiprocessing/synchronize.py @@ -49,6 +50,8 @@ mod _multiprocessing { name: Option, last_tid: AtomicU32, count: AtomicI32, + /// Serializes owner bookkeeping. Dropped around blocking waits. + crit: PyMutex<()>, } type SemHandle = host_multiprocessing::SemHandle; @@ -109,21 +112,26 @@ mod _multiprocessing { }; // Check whether we already own the lock - if self.kind == RECURSIVE_MUTEX && ismine!(self) { - self.count.fetch_add(1, Ordering::Release); - return Ok(true); - } - - // Check whether we can acquire without blocking - match host_multiprocessing::wait_for_single_object(self.handle.as_raw(), 0) { - x if x == host_multiprocessing::wait_object_0() => { - self.last_tid - .store(host_multiprocessing::current_thread_id(), Ordering::Release); + { + let _crit = self.crit.lock(); + if self.kind == RECURSIVE_MUTEX && ismine!(self) { self.count.fetch_add(1, Ordering::Release); return Ok(true); } - x if x == host_multiprocessing::wait_failed() => return Err(vm.new_last_os_error()), - _ => {} + + // Check whether we can acquire without blocking + match host_multiprocessing::wait_for_single_object(self.handle.as_raw(), 0) { + x if x == host_multiprocessing::wait_object_0() => { + self.last_tid + .store(host_multiprocessing::current_thread_id(), Ordering::Release); + self.count.fetch_add(1, Ordering::Release); + return Ok(true); + } + x if x == host_multiprocessing::wait_failed() => { + return Err(vm.new_last_os_error()); + } + _ => {} + } } // Poll with signal checking (CPython uses WaitForMultipleObjectsEx @@ -148,6 +156,7 @@ mod _multiprocessing { match res { x if x == host_multiprocessing::wait_object_0() => { + let _crit = self.crit.lock(); self.last_tid .store(host_multiprocessing::current_thread_id(), Ordering::Release); self.count.fetch_add(1, Ordering::Release); @@ -173,6 +182,7 @@ mod _multiprocessing { #[pymethod] fn release(&self, vm: &VirtualMachine) -> PyResult<()> { + let _crit = self.crit.lock(); if self.kind == RECURSIVE_MUTEX { if !ismine!(self) { return Err(vm.new_assertion_error( @@ -229,12 +239,14 @@ mod _multiprocessing { name, last_tid: AtomicU32::new(0), count: AtomicI32::new(0), + crit: PyMutex::new(()), }; zelf.into_ref_with_type(vm, cls).map(Into::into) } #[pymethod] fn _after_fork(&self) { + let _crit = self.crit.lock(); self.count.store(0, Ordering::Release); self.last_tid.store(0, Ordering::Release); } @@ -246,6 +258,7 @@ mod _multiprocessing { #[pymethod] fn _count(&self) -> i32 { + let _crit = self.crit.lock(); self.count.load(Ordering::Acquire) } @@ -306,6 +319,7 @@ mod _multiprocessing { name, last_tid: AtomicU32::new(0), count: AtomicI32::new(0), + crit: PyMutex::new(()), }) } } @@ -354,6 +368,7 @@ mod _multiprocessing { types::Constructor, }; use core::sync::atomic::{AtomicI32, AtomicU64, Ordering}; + use rustpython_common::lock::PyMutex; #[cfg(target_vendor = "apple")] use rustpython_host_env::multiprocessing::sem_t; use rustpython_host_env::multiprocessing::{ @@ -437,6 +452,8 @@ mod _multiprocessing { name: Option, last_tid: AtomicU64, // unsigned long count: AtomicI32, // int + /// Serializes owner bookkeeping. Dropped around blocking waits. + crit: PyMutex<()>, } type SemHandle = host_multiprocessing::SemHandle; @@ -483,6 +500,7 @@ mod _multiprocessing { .or_else(|| args.args.get(1)) .cloned(); + let _crit = self.crit.lock(); if self.kind == RECURSIVE_MUTEX && ismine!(self) { self.count.fetch_add(1, Ordering::Release); return Ok(true); @@ -516,10 +534,7 @@ mod _multiprocessing { // if (res < 0 && errno == EAGAIN && blocking) if matches!(try_status, TryAcquireStatus::WouldBlock) && blocking { // Couldn't acquire immediately, need to block. - // - // Save errno inside the allow_threads closure, before - // attach_thread() runs — matches CPython which saves - // `err = errno` before Py_END_ALLOW_THREADS. + drop(_crit); #[cfg(not(target_vendor = "apple"))] { @@ -573,6 +588,9 @@ mod _multiprocessing { } } } + let _crit = self.crit.lock(); + self.mark_acquired(); + return Ok(true); } else if !matches!(try_status, TryAcquireStatus::Acquired) { // Non-blocking path failed, or blocking=false match try_status { @@ -583,17 +601,21 @@ mod _multiprocessing { } } + self.mark_acquired(); + Ok(true) + } + + fn mark_acquired(&self) { self.count.fetch_add(1, Ordering::Release); self.last_tid .store(host_multiprocessing::current_thread_id(), Ordering::Release); - - Ok(true) } /// Release the semaphore/lock. // _multiprocessing_SemLock_release_impl #[pymethod] fn release(&self, vm: &VirtualMachine) -> PyResult<()> { + let _crit = self.crit.lock(); if self.kind == RECURSIVE_MUTEX { // if (!ISMINE(self)) if !ismine!(self) { @@ -698,6 +720,7 @@ mod _multiprocessing { name, last_tid: AtomicU64::new(0), count: AtomicI32::new(0), + crit: PyMutex::new(()), }; zelf.into_ref_with_type(vm, cls).map(Into::into) } @@ -706,6 +729,7 @@ mod _multiprocessing { // _multiprocessing_SemLock__after_fork_impl #[pymethod] fn _after_fork(&self) { + let _crit = self.crit.lock(); self.count.store(0, Ordering::Release); // Also reset last_tid for safety self.last_tid.store(0, Ordering::Release); @@ -722,6 +746,7 @@ mod _multiprocessing { // _multiprocessing_SemLock__count_impl #[pymethod] fn _count(&self) -> i32 { + let _crit = self.crit.lock(); self.count.load(Ordering::Acquire) } @@ -826,6 +851,7 @@ mod _multiprocessing { name, last_tid: AtomicU64::new(0), count: AtomicI32::new(0), + crit: PyMutex::new(()), }) } } diff --git a/extra_tests/snippets/stdlib_multiprocessing_condition.py b/extra_tests/snippets/stdlib_multiprocessing_condition.py new file mode 100644 index 000000000..2b4bc1394 --- /dev/null +++ b/extra_tests/snippets/stdlib_multiprocessing_condition.py @@ -0,0 +1,87 @@ +"""Thread+process Condition notify must wake every waiter.""" + +import multiprocessing +import threading + + +def wait_on_condition(cond, started, woken): + cond.acquire() + started.release() + cond.wait() + woken.release() + cond.release() + + +def join_workers(workers): + for worker in workers: + worker.join(timeout=5) + assert not worker.is_alive() + + +def wait_sem(sem, msg): + assert sem.acquire(timeout=5), msg + + +def notify_all_six(ctx): + cond = ctx.Condition() + started = ctx.Semaphore(0) + woken = ctx.Semaphore(0) + + workers = [] + for _ in range(3): + proc = ctx.Process(target=wait_on_condition, args=(cond, started, woken)) + proc.daemon = True + proc.start() + workers.append(proc) + + thread = threading.Thread(target=wait_on_condition, args=(cond, started, woken)) + thread.daemon = True + thread.start() + workers.append(thread) + + for i in range(6): + wait_sem(started, f"waiter {i} did not enter Condition.wait()") + + cond.acquire() + cond.notify_all() + cond.release() + + for i in range(6): + wait_sem(woken, f"waiter {i} did not wake") + join_workers(workers) + + +def notify_one_then_the_other(ctx): + cond = ctx.Condition() + started = ctx.Semaphore(0) + woken = ctx.Semaphore(0) + + proc = ctx.Process(target=wait_on_condition, args=(cond, started, woken)) + proc.daemon = True + proc.start() + thread = threading.Thread(target=wait_on_condition, args=(cond, started, woken)) + thread.daemon = True + thread.start() + + wait_sem(started, "process did not enter Condition.wait()") + wait_sem(started, "thread did not enter Condition.wait()") + + cond.acquire() + cond.notify() + cond.release() + wait_sem(woken, "first waiter did not wake") + + cond.acquire() + cond.notify() + cond.release() + wait_sem(woken, "second waiter did not wake") + join_workers([proc, thread]) + + +if __name__ == "__main__": + multiprocessing.freeze_support() + ctx = multiprocessing.get_context("spawn") + # Enough repeats to surface a lost-wakeup race in CI snippets. + for _ in range(20): + notify_all_six(ctx) + notify_one_then_the_other(ctx)