bpo-33613, test_semaphore_tracker_sigint: fix race condition

Lib/multiprocessing/semaphore_tracker.py

@@ -60,14 +60,14 @@ def ensure_running(self):
                 fds_to_pass.append(sys.stderr.fileno())
             except Exception:
                 pass
-            cmd = 'from multiprocessing.semaphore_tracker import main;main(%d)'
+            cmd = 'from multiprocessing.semaphore_tracker import main;main({}, {})'
             r, w = os.pipe()
             try:
                 fds_to_pass.append(r)
                 # process will out live us, so no need to wait on pid
                 exe = spawn.get_executable()
                 args = [exe] + util._args_from_interpreter_flags()
-                args += ['-c', cmd % r]
+                args += ['-c', cmd.format(r, sys.stderr.fileno())]
                 pid = util.spawnv_passfds(exe, args, fds_to_pass)
             except:
                 os.close(w)
@@ -105,7 +105,7 @@ def _send(self, cmd, name):
 getfd = _semaphore_tracker.getfd
 
 
-def main(fd):
+def main(fd, fd_write):
     '''Run semaphore tracker.'''
     # protect the process from ^C and "killall python" etc
     signal.signal(signal.SIGINT, signal.SIG_IGN)
@@ -128,6 +128,8 @@ def main(fd):
                         cache.add(name)
                     elif cmd == b'UNREGISTER':
                         cache.remove(name)
+                    elif cmd == b'PING':
+                        os.write(fd_write, b"PONG\n")
                     else:
                         raise RuntimeError('unrecognized command %r' % cmd)
                 except Exception:

Lib/test/_test_multiprocessing.py

@@ -20,6 +20,7 @@
 import struct
 import operator
 import weakref
+import warnings
 import test.support
 import test.support.script_helper
 from test import support
@@ -4472,17 +4473,34 @@ def check_semaphore_tracker_death(self, signum, should_die):
         # bpo-31310: if the semaphore tracker process has died, it should
         # be restarted implicitly.
         from multiprocessing.semaphore_tracker import _semaphore_tracker
-        _semaphore_tracker.ensure_running()
         pid = _semaphore_tracker._pid
+        if pid:
+            os.kill(pid, signal.SIGKILL)
+            time.sleep(0.5)  # give it time to die
+        old_stderr = sys.stderr
+        r, w = os.pipe()
+        try:
+            sys.stderr = open(w, "bw")
+            with warnings.catch_warnings(record=True) as all_warn:
+                _semaphore_tracker.ensure_running()
+            pid = _semaphore_tracker._pid
+            # Wait until we receive the PONG from the child, indicating that
+            # the signal handlers have been registered. See bpo-33613 for more
+            # information.
+            _semaphore_tracker._send("PING", "")
+            with open(r, "rb") as pipe:
+                data = pipe.readline()
+                if b"PONG" not in data:
+                    raise ValueError("Invalid data in stderr!")
+        finally:
+            sys.stderr.close()
+            sys.stderr = old_stderr
+
         os.kill(pid, signum)
         time.sleep(1.0)  # give it time to die
 
         ctx = multiprocessing.get_context("spawn")
-        with contextlib.ExitStack() as stack:
-            if should_die:
-                stack.enter_context(self.assertWarnsRegex(
-                    UserWarning,
-                    "semaphore_tracker: process died"))
+        with warnings.catch_warnings(record=True) as all_warn:
             sem = ctx.Semaphore()
             sem.acquire()
             sem.release()
@@ -4492,6 +4510,14 @@ def check_semaphore_tracker_death(self, signum, should_die):
             del sem
             gc.collect()
             self.assertIsNone(wr())
+            if should_die:
+                self.assertEqual(len(all_warn), 1)
+                the_warn = all_warn[0]
+                issubclass(the_warn.category, UserWarning)
+                self.assertTrue("semaphore_tracker: process died"
+                                in str(the_warn.message))
+            else:
+                self.assertEqual(len(all_warn), 0)
 
     def test_semaphore_tracker_sigint(self):
         # Catchable signal (ignored by semaphore tracker)