gh-126434: Use multiprocessing.Value for multiprocessing.Event to avoid deadlock when there is reentrant usage of set from is_set, e.g. when handling system signals

Lib/multiprocessing/synchronize.py

@@ -328,41 +328,70 @@ def wait_for(self, predicate, timeout=None):
 class Event(object):
 
     def __init__(self, *, ctx):
-        self._cond = ctx.Condition(ctx.Lock())
-        self._flag = ctx.Semaphore(0)
+        self._flag = ctx.Value('i', 0)
+        # Allocate a ctypes.c_ulonglong to hold the set_id:
+        # Represents the C unsigned long long datatype.
+        # The constructor accepts an optional integer initializer; no overflow checking is done.
+        # From https://docs.python.org/3/library/ctypes.html#ctypes.c_ulonglong
+        # See multiprocessing/sharedctypes.py for typecode to ctypes definitions
+        self._set_id = ctx.Value('Q', 0)
 
     def is_set(self):
-        with self._cond:
-            if self._flag.acquire(False):
-                self._flag.release()
-                return True
-            return False
+        # From https://docs.python.org/3/library/multiprocessing.html#multiprocessing.Value :
+        # If lock is True (the default) then a new recursive lock object is created to synchronize access to the value.
+        with self._flag:
+            return self._flag.value == 1
 
     def set(self):
-        with self._cond:
-            self._flag.acquire(False)
-            self._flag.release()
-            self._cond.notify_all()
+        # If `set` is called from a signal handler, this is fine as the lock is recursive (i.e. it won't deadlock).
+        # If the thread interrupted by the signal handler is wait()-ing and the signal handler calls set(),
+        # this is fine as wait() spins on the value. Thus, after the signal handler is done, the thread will
+        # return from wait()
+        # Fixes https://github.com/python/cpython/issues/85772
+        with self._flag:
+            with self._set_id:
+                if self._flag.value == 0:
+                    # There is a theoretical chance of race here. It requires the following conditions:
+                    # The interrupted thread must be wait()ing.
+                    # Then set must be called reentrant for the maximum value of c_ulonglong times,
+                    # and all interruptions must happen exactly after `if self._flag.value == 0:`.
+                    # The _set_id value will then wrap around. Then clear() must be called
+                    # before the original wait() code continues. The wait() code will then continue
+                    # to (incorrectly) wait. I think this case is safe to ignore. The stack
+                    # will grow too large before there is any chance of this actually happening.
+
+                    self._flag.value = 1
+                    self._set_id.value += 1
+                    # There is no race here by reentrant set when reaching the maximum value for `self._set_id.value`.
+                    # ctypes.c_ulonglong will overflow without any exception:
+                    # https://docs.python.org/3/library/ctypes.html#ctypes.c_ulonglong
+                    # > no overflow checking is done.
+                    # This means that we do not need to check if some maximum value is reached:
+                    # C will wrap around the value for us.
 
     def clear(self):
-        with self._cond:
-            self._flag.acquire(False)
+        with self._flag:
+            self._flag.value = 0
 
     def wait(self, timeout=None):
-        with self._cond:
-            if self._flag.acquire(False):
-                self._flag.release()
-            else:
-                self._cond.wait(timeout)
-
-            if self._flag.acquire(False):
-                self._flag.release()
+        start_time = time.monotonic()
+        set_id = self._set_id.value
+        while True:
+            if self._flag.value == 1:
                 return True
-            return False
+            elif set_id != self._set_id.value:
+                return True # flag is unset, but set_id changed, so there must have been a `set` followed by a `clear`
+                            # during `time.sleep()`. Fixes https://github.com/python/cpython/issues/95826
+            elif timeout is not None and (time.monotonic() - start_time) > timeout:
+                return False
+            else:
+                # Fixes https://github.com/python/cpython/issues/85772 by spinning and sleeping.
+                time.sleep(0.010) # sleep 10 milliseconds
 
     def __repr__(self) -> str:
         set_status = 'set' if self.is_set() else 'unset'
         return f"<{type(self).__qualname__} at {id(self):#x} {set_status}>"
+
 #
 # Barrier
 #

Lib/test/multiprocessingdata/is_set_set.py

@@ -0,0 +1,29 @@
+import multiprocessing
+import os
+import signal
+import concurrent.futures
+import time
+
+
+def send_sigint(pid):
+    time.sleep(1)
+    os.kill(pid, signal.SIGINT)
+
+
+def run_signal_handler_set_is_set_test():
+    shutdown_event = multiprocessing.Event()
+
+    def sigterm_handler(_signo, _stack_frame):
+        shutdown_event.set()
+
+    signal.signal(signal.SIGINT, sigterm_handler)
+
+    with concurrent.futures.ProcessPoolExecutor() as executor:
+        f = executor.submit(send_sigint, os.getpid())
+        while not shutdown_event.is_set():
+            pass
+        f.result()
+
+
+if __name__ == '__main__':
+    run_signal_handler_set_is_set_test()

Lib/test/multiprocessingdata/set_clear_race.py

@@ -0,0 +1,36 @@
+import multiprocessing
+import sys
+
+
+# Reproduction code copied and modified from https://github.com/python/cpython/issues/95826
+# Fixes the issue above
+
+class SimpleRepro:
+    def __init__(self):
+        self.heartbeat_event = multiprocessing.Event()
+        self.shutdown_event = multiprocessing.Event()
+        self.child_proc = multiprocessing.Process(target=self.child_process, daemon=True)
+        self.child_proc.start()
+
+    def child_process(self):
+        while True:
+            if self.shutdown_event.is_set():
+                return
+            self.heartbeat_event.set()
+            self.heartbeat_event.clear()
+
+    def test_heartbeat(self):
+        exit_code = 0
+        for i in range(2000):
+            success = self.heartbeat_event.wait(100)
+            if not success:
+                exit_code = 1
+                break
+        self.shutdown_event.set()
+        self.child_proc.join()
+        sys.exit(exit_code)
+
+
+if __name__ == '__main__':
+    foo = SimpleRepro()
+    foo.test_heartbeat()

Lib/test/multiprocessingdata/wait_set_no_deadlock.py

@@ -0,0 +1,30 @@
+import multiprocessing
+import signal
+import concurrent.futures
+import time
+import os
+
+
+# Shows that https://github.com/python/cpython/issues/85772 is fixed
+
+def send_sigint(pid):
+    time.sleep(1) # Make sure shutdown_event.wait() is called
+    os.kill(pid, signal.SIGINT)
+
+
+def run_signal_handler_wait_set_test():
+    shutdown_event = multiprocessing.Event()
+
+    def sigterm_handler(_signo, _stack_frame):
+        shutdown_event.set()
+
+    signal.signal(signal.SIGINT, sigterm_handler)
+
+    with concurrent.futures.ProcessPoolExecutor() as executor:
+        f = executor.submit(send_sigint, os.getpid())
+        shutdown_event.wait()
+        f.result()
+
+
+if __name__ == '__main__':
+    run_signal_handler_wait_set_test()

Lib/test/test_multiprocessing_event/__init__.py

@@ -0,0 +1,5 @@
+import os.path
+from test import support
+
+def load_tests(*args):
+    return support.load_package_tests(os.path.dirname(__file__), *args)

Lib/test/test_multiprocessing_event/test_event.py

@@ -0,0 +1,73 @@
+import unittest
+from test import support
+import sys
+import signal
+import os
+
+
+try:
+    import multiprocessing
+    from concurrent.futures import ProcessPoolExecutor
+    _have_multiprocessing = True
+except (NotImplementedError, ModuleNotFoundError):
+    _have_multiprocessing = False
+
+
+@unittest.skipUnless(_have_multiprocessing,
+                     "requires multiprocessing")
+@unittest.skipUnless(hasattr(signal, 'signal'),
+                     "Requires signal.signal")
+@unittest.skipUnless(hasattr(signal, 'SIGINT'),
+                     "Requires signal.SIGINT")
+@unittest.skipUnless(hasattr(os, 'kill'),
+                     "Requires os.kill")
+@unittest.skipUnless(hasattr(os, 'getppid'),
+                     "Requires os.getppid")
+@support.requires_subprocess()
+class TestEventSignalHandling(unittest.TestCase):
+    def test_no_race_for_is_set_set(self):
+        import subprocess
+        script = support.findfile("is_set_set.py", subdir="multiprocessingdata")
+        for x in range(10):
+            try:
+                assert 0 == subprocess.call([sys.executable, script], timeout=60)
+            except subprocess.TimeoutExpired:
+                assert False, 'subprocess.Timeoutexpired for is_set_set.py'
+
+    def test_no_race_set_clear(self):
+        import subprocess
+        script = support.findfile("set_clear_race.py", subdir="multiprocessingdata")
+        assert 0 == subprocess.call([sys.executable, script])
+
+    def test_wait_set_no_deadlock(self):
+        import subprocess
+        script = support.findfile("wait_set_no_deadlock.py", subdir="multiprocessingdata")
+        assert 0 == subprocess.call([sys.executable, script])
+
+    def test_wait_timeout(self):
+        event = multiprocessing.Event()
+        # https://docs.python.org/3/library/multiprocessing.html#multiprocessing.Event
+        # multiprocessing.Event: A clone of threading.Event.
+
+        # threading.Event: https://docs.python.org/3/library/threading.html#threading.Event
+
+        # threading.Event.wait(): https://docs.python.org/3/library/threading.html#threading.Event.wait
+        # Block as long as the internal flag is false and the timeout, if given, has not expired.
+        # The return value represents the reason that this blocking method returned;
+        # True if returning because the internal flag is set to true, or
+        # False if a timeout is given and the internal flag did not become true within the given wait time.
+
+        # When the timeout argument is present and not None, it should be a floating-point number
+        # specifying a timeout for the operation in seconds, or fractions thereof.
+
+        # wait() supports both integer and float:
+        flag_set = event.wait(1)
+        assert flag_set == False
+
+        flag_set = event.wait(0.1)
+        assert flag_set == False
+
+
+if __name__ == '__main__':
+    unittest.main()
+

Makefile.pre.in

@@ -2449,6 +2449,7 @@ TESTSUBDIRS=	idlelib/idle_test \
 		test/leakers \
 		test/libregrtest \
 		test/mathdata \
+		test/multiprocessingdata \
 		test/regrtestdata \
 		test/regrtestdata/import_from_tests \
 		test/regrtestdata/import_from_tests/test_regrtest_b \
@@ -2519,6 +2520,7 @@ TESTSUBDIRS=	idlelib/idle_test \
 		test/test_interpreters \
 		test/test_json \
 		test/test_module \
+		test/test_multiprocessing_event \
 		test/test_multiprocessing_fork \
 		test/test_multiprocessing_forkserver \
 		test/test_multiprocessing_spawn \

Misc/NEWS.d/next/Library/2024-11-05-11-11-29.gh-issue-126434.Hw3wdI.rst

@@ -0,0 +1 @@
+All of :mod:`multiprocessing` ``Event`` is now reentrant and thread safe, and can thus be used from signal handlers.