@@ -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 }
0 commit comments