Skip to content

Commit 0bdb61b

Browse files
authored
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
1 parent a7c0500 commit 0bdb61b

3 files changed

Lines changed: 140 additions & 18 deletions

File tree

‎crates/host_env/src/multiprocessing.rs‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,13 @@ pub struct SemHandle {
2323
raw: *mut sem_t,
2424
}
2525

26+
// POSIX named semaphores are safe to post/wait from any thread that holds
27+
// a `sem_t *` to the same kernel object.
28+
#[cfg(unix)]
29+
unsafe impl Send for SemHandle {}
30+
#[cfg(unix)]
31+
unsafe impl Sync for SemHandle {}
32+
2633
#[cfg(unix)]
2734
#[derive(Copy, Clone, Debug, Eq, PartialEq)]
2835
pub enum SemError {
@@ -110,7 +117,9 @@ pub struct SemHandle {
110117
raw: HANDLE,
111118
}
112119

120+
#[cfg(windows)]
113121
unsafe impl Send for SemHandle {}
122+
#[cfg(windows)]
114123
unsafe impl Sync for SemHandle {}
115124

116125
#[cfg(unix)]

‎crates/stdlib/src/multiprocessing.rs‎

Lines changed: 44 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@ mod _multiprocessing {
1111
types::Constructor,
1212
};
1313
use core::sync::atomic::{AtomicI32, AtomicU32, Ordering};
14+
use rustpython_common::lock::PyMutex;
1415
use rustpython_host_env::multiprocessing as host_multiprocessing;
1516

1617
// These match the values in Lib/multiprocessing/synchronize.py
@@ -49,6 +50,8 @@ mod _multiprocessing {
4950
name: Option<String>,
5051
last_tid: AtomicU32,
5152
count: AtomicI32,
53+
/// Serializes owner bookkeeping. Dropped around blocking waits.
54+
crit: PyMutex<()>,
5255
}
5356

5457
type SemHandle = host_multiprocessing::SemHandle;
@@ -109,21 +112,26 @@ mod _multiprocessing {
109112
};
110113

111114
// Check whether we already own the lock
112-
if self.kind == RECURSIVE_MUTEX && ismine!(self) {
113-
self.count.fetch_add(1, Ordering::Release);
114-
return Ok(true);
115-
}
116-
117-
// Check whether we can acquire without blocking
118-
match host_multiprocessing::wait_for_single_object(self.handle.as_raw(), 0) {
119-
x if x == host_multiprocessing::wait_object_0() => {
120-
self.last_tid
121-
.store(host_multiprocessing::current_thread_id(), Ordering::Release);
115+
{
116+
let _crit = self.crit.lock();
117+
if self.kind == RECURSIVE_MUTEX && ismine!(self) {
122118
self.count.fetch_add(1, Ordering::Release);
123119
return Ok(true);
124120
}
125-
x if x == host_multiprocessing::wait_failed() => return Err(vm.new_last_os_error()),
126-
_ => {}
121+
122+
// Check whether we can acquire without blocking
123+
match host_multiprocessing::wait_for_single_object(self.handle.as_raw(), 0) {
124+
x if x == host_multiprocessing::wait_object_0() => {
125+
self.last_tid
126+
.store(host_multiprocessing::current_thread_id(), Ordering::Release);
127+
self.count.fetch_add(1, Ordering::Release);
128+
return Ok(true);
129+
}
130+
x if x == host_multiprocessing::wait_failed() => {
131+
return Err(vm.new_last_os_error());
132+
}
133+
_ => {}
134+
}
127135
}
128136

129137
// Poll with signal checking (CPython uses WaitForMultipleObjectsEx
@@ -148,6 +156,7 @@ mod _multiprocessing {
148156

149157
match res {
150158
x if x == host_multiprocessing::wait_object_0() => {
159+
let _crit = self.crit.lock();
151160
self.last_tid
152161
.store(host_multiprocessing::current_thread_id(), Ordering::Release);
153162
self.count.fetch_add(1, Ordering::Release);
@@ -173,6 +182,7 @@ mod _multiprocessing {
173182

174183
#[pymethod]
175184
fn release(&self, vm: &VirtualMachine) -> PyResult<()> {
185+
let _crit = self.crit.lock();
176186
if self.kind == RECURSIVE_MUTEX {
177187
if !ismine!(self) {
178188
return Err(vm.new_assertion_error(
@@ -229,12 +239,14 @@ mod _multiprocessing {
229239
name,
230240
last_tid: AtomicU32::new(0),
231241
count: AtomicI32::new(0),
242+
crit: PyMutex::new(()),
232243
};
233244
zelf.into_ref_with_type(vm, cls).map(Into::into)
234245
}
235246

236247
#[pymethod]
237248
fn _after_fork(&self) {
249+
let _crit = self.crit.lock();
238250
self.count.store(0, Ordering::Release);
239251
self.last_tid.store(0, Ordering::Release);
240252
}
@@ -246,6 +258,7 @@ mod _multiprocessing {
246258

247259
#[pymethod]
248260
fn _count(&self) -> i32 {
261+
let _crit = self.crit.lock();
249262
self.count.load(Ordering::Acquire)
250263
}
251264

@@ -306,6 +319,7 @@ mod _multiprocessing {
306319
name,
307320
last_tid: AtomicU32::new(0),
308321
count: AtomicI32::new(0),
322+
crit: PyMutex::new(()),
309323
})
310324
}
311325
}
@@ -354,6 +368,7 @@ mod _multiprocessing {
354368
types::Constructor,
355369
};
356370
use core::sync::atomic::{AtomicI32, AtomicU64, Ordering};
371+
use rustpython_common::lock::PyMutex;
357372
#[cfg(target_vendor = "apple")]
358373
use rustpython_host_env::multiprocessing::sem_t;
359374
use rustpython_host_env::multiprocessing::{
@@ -437,6 +452,8 @@ mod _multiprocessing {
437452
name: Option<String>,
438453
last_tid: AtomicU64, // unsigned long
439454
count: AtomicI32, // int
455+
/// Serializes owner bookkeeping. Dropped around blocking waits.
456+
crit: PyMutex<()>,
440457
}
441458

442459
type SemHandle = host_multiprocessing::SemHandle;
@@ -483,6 +500,7 @@ mod _multiprocessing {
483500
.or_else(|| args.args.get(1))
484501
.cloned();
485502

503+
let _crit = self.crit.lock();
486504
if self.kind == RECURSIVE_MUTEX && ismine!(self) {
487505
self.count.fetch_add(1, Ordering::Release);
488506
return Ok(true);
@@ -516,10 +534,7 @@ mod _multiprocessing {
516534
// if (res < 0 && errno == EAGAIN && blocking)
517535
if matches!(try_status, TryAcquireStatus::WouldBlock) && blocking {
518536
// Couldn't acquire immediately, need to block.
519-
//
520-
// Save errno inside the allow_threads closure, before
521-
// attach_thread() runs — matches CPython which saves
522-
// `err = errno` before Py_END_ALLOW_THREADS.
537+
drop(_crit);
523538

524539
#[cfg(not(target_vendor = "apple"))]
525540
{
@@ -573,6 +588,9 @@ mod _multiprocessing {
573588
}
574589
}
575590
}
591+
let _crit = self.crit.lock();
592+
self.mark_acquired();
593+
return Ok(true);
576594
} else if !matches!(try_status, TryAcquireStatus::Acquired) {
577595
// Non-blocking path failed, or blocking=false
578596
match try_status {
@@ -583,17 +601,21 @@ mod _multiprocessing {
583601
}
584602
}
585603

604+
self.mark_acquired();
605+
Ok(true)
606+
}
607+
608+
fn mark_acquired(&self) {
586609
self.count.fetch_add(1, Ordering::Release);
587610
self.last_tid
588611
.store(host_multiprocessing::current_thread_id(), Ordering::Release);
589-
590-
Ok(true)
591612
}
592613

593614
/// Release the semaphore/lock.
594615
// _multiprocessing_SemLock_release_impl
595616
#[pymethod]
596617
fn release(&self, vm: &VirtualMachine) -> PyResult<()> {
618+
let _crit = self.crit.lock();
597619
if self.kind == RECURSIVE_MUTEX {
598620
// if (!ISMINE(self))
599621
if !ismine!(self) {
@@ -698,6 +720,7 @@ mod _multiprocessing {
698720
name,
699721
last_tid: AtomicU64::new(0),
700722
count: AtomicI32::new(0),
723+
crit: PyMutex::new(()),
701724
};
702725
zelf.into_ref_with_type(vm, cls).map(Into::into)
703726
}
@@ -706,6 +729,7 @@ mod _multiprocessing {
706729
// _multiprocessing_SemLock__after_fork_impl
707730
#[pymethod]
708731
fn _after_fork(&self) {
732+
let _crit = self.crit.lock();
709733
self.count.store(0, Ordering::Release);
710734
// Also reset last_tid for safety
711735
self.last_tid.store(0, Ordering::Release);
@@ -722,6 +746,7 @@ mod _multiprocessing {
722746
// _multiprocessing_SemLock__count_impl
723747
#[pymethod]
724748
fn _count(&self) -> i32 {
749+
let _crit = self.crit.lock();
725750
self.count.load(Ordering::Acquire)
726751
}
727752

@@ -826,6 +851,7 @@ mod _multiprocessing {
826851
name,
827852
last_tid: AtomicU64::new(0),
828853
count: AtomicI32::new(0),
854+
crit: PyMutex::new(()),
829855
})
830856
}
831857
}
Lines changed: 87 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,87 @@
1+
"""Thread+process Condition notify must wake every waiter."""
2+
3+
import multiprocessing
4+
import threading
5+
6+
7+
def wait_on_condition(cond, started, woken):
8+
cond.acquire()
9+
started.release()
10+
cond.wait()
11+
woken.release()
12+
cond.release()
13+
14+
15+
def join_workers(workers):
16+
for worker in workers:
17+
worker.join(timeout=5)
18+
assert not worker.is_alive()
19+
20+
21+
def wait_sem(sem, msg):
22+
assert sem.acquire(timeout=5), msg
23+
24+
25+
def notify_all_six(ctx):
26+
cond = ctx.Condition()
27+
started = ctx.Semaphore(0)
28+
woken = ctx.Semaphore(0)
29+
30+
workers = []
31+
for _ in range(3):
32+
proc = ctx.Process(target=wait_on_condition, args=(cond, started, woken))
33+
proc.daemon = True
34+
proc.start()
35+
workers.append(proc)
36+
37+
thread = threading.Thread(target=wait_on_condition, args=(cond, started, woken))
38+
thread.daemon = True
39+
thread.start()
40+
workers.append(thread)
41+
42+
for i in range(6):
43+
wait_sem(started, f"waiter {i} did not enter Condition.wait()")
44+
45+
cond.acquire()
46+
cond.notify_all()
47+
cond.release()
48+
49+
for i in range(6):
50+
wait_sem(woken, f"waiter {i} did not wake")
51+
join_workers(workers)
52+
53+
54+
def notify_one_then_the_other(ctx):
55+
cond = ctx.Condition()
56+
started = ctx.Semaphore(0)
57+
woken = ctx.Semaphore(0)
58+
59+
proc = ctx.Process(target=wait_on_condition, args=(cond, started, woken))
60+
proc.daemon = True
61+
proc.start()
62+
thread = threading.Thread(target=wait_on_condition, args=(cond, started, woken))
63+
thread.daemon = True
64+
thread.start()
65+
66+
wait_sem(started, "process did not enter Condition.wait()")
67+
wait_sem(started, "thread did not enter Condition.wait()")
68+
69+
cond.acquire()
70+
cond.notify()
71+
cond.release()
72+
wait_sem(woken, "first waiter did not wake")
73+
74+
cond.acquire()
75+
cond.notify()
76+
cond.release()
77+
wait_sem(woken, "second waiter did not wake")
78+
join_workers([proc, thread])
79+
80+
81+
if __name__ == "__main__":
82+
multiprocessing.freeze_support()
83+
ctx = multiprocessing.get_context("spawn")
84+
# Enough repeats to surface a lost-wakeup race in CI snippets.
85+
for _ in range(20):
86+
notify_all_six(ctx)
87+
notify_one_then_the_other(ctx)

0 commit comments

Comments
 (0)