Use pthread_sigmask to avoid the race condition

pitrou · Sep 4, 2018 · Jun 20, 2018 · Jun 24, 2018 · Jun 24, 2018 · Jul 1, 2018
commit f5e5cf904e640997dbb3bf5d7eaef3d7e850e790
diff --git a/Lib/multiprocessing/semaphore_tracker.py b/Lib/multiprocessing/semaphore_tracker.py
@@ -10,6 +10,7 @@
 # the next reboot.  Without this semaphore tracker process, "killall
 # python" would probably leave unlinked semaphores.
 #
+
 import os
 import signal
 import sys
@@ -21,6 +22,8 @@
 from . import util
 
 __all__ = ['ensure_running', 'register', 'unregister']
+_HAVE_SIGMASK = hasattr(signal, 'pthread_sigmask')
+_IGNORED_SIGNALS = (signal.SIGINT, signal.SIGTERM)
 
 
 class SemaphoreTracker(object):
@@ -59,15 +62,19 @@ def ensure_running(self):
                 fds_to_pass.append(sys.stderr.fileno())
             except Exception:
                 pass
-            cmd = 'from multiprocessing.semaphore_tracker import main;main({})'
+            cmd = 'from multiprocessing.semaphore_tracker import main;main(%d)'
             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.format(r)]
+                args += ['-c', cmd % r]
+                if _HAVE_SIGMASK:
+                    signal.pthread_sigmask(signal.SIG_BLOCK, _IGNORED_SIGNALS)
                 pid = util.spawnv_passfds(exe, args, fds_to_pass)
+                if _HAVE_SIGMASK:
+                    signal.pthread_sigmask(signal.SIG_UNBLOCK, _IGNORED_SIGNALS)
             except:
                 os.close(w)
                 raise
@@ -103,12 +110,13 @@ def _send(self, cmd, name):
 unregister = _semaphore_tracker.unregister
 getfd = _semaphore_tracker.getfd
 
-
 def main(fd):
     '''Run semaphore tracker.'''
     # protect the process from ^C and "killall python" etc
     signal.signal(signal.SIGINT, signal.SIG_IGN)
     signal.signal(signal.SIGTERM, signal.SIG_IGN)
+    if _HAVE_SIGMASK:
+        signal.pthread_sigmask(signal.SIG_UNBLOCK, _IGNORED_SIGNALS)
 
     for f in (sys.stdin, sys.stdout):
         try:
@@ -127,9 +135,6 @@ def main(fd):
                         cache.add(name)
                     elif cmd == b'UNREGISTER':
                         cache.remove(name)
-                    elif cmd == b'PING':
-                        with open(sys.stderr.fileno(), "wb") as output:
-                            output.write(b"PONG\n")
                     else:
                         raise RuntimeError('unrecognized command %r' % cmd)
                 except Exception:

diff --git a/Lib/test/_test_multiprocessing.py b/Lib/test/_test_multiprocessing.py
@@ -4477,38 +4477,12 @@ def check_semaphore_tracker_death(self, signum, should_die):
         if pid:
             os.kill(pid, signal.SIGKILL)
             time.sleep(0.5)  # give it time to die
-        testfn = test.support.TESTFN
-        self.addCleanup(test.support.unlink, testfn)
-        tracker_stderr_r = open(testfn, "wb")
-        old_stderr = os.dup(sys.stderr.fileno())
-        os.dup2(tracker_stderr_r.fileno(), sys.stderr.fileno())
-        try:
-            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", "")
-            deadline = time.monotonic() + 5
-            with open(testfn, "rb") as tracker_stderr:
-                while True:
-                    if time.monotonic() >= deadline:
-                        raise TimeoutError("Reading data "
-                                           "from tracker stderr took too long")
-                    data = tracker_stderr.readline()
-                    if not data:
-                        continue
-                    if b"PONG" not in data:
-                        raise ValueError("Invalid data in stderr: {}!".format(data))
-                    break
-        finally:
-            tracker_stderr_r.close()
-            os.dup2(sys.stderr.fileno(), old_stderr)
-            os.close(old_stderr)
+        with warnings.catch_warnings(record=True) as all_warn:
+            _semaphore_tracker.ensure_running()
+        pid = _semaphore_tracker._pid
 
         os.kill(pid, signum)
-        time.sleep(1.0)  # give it time to die
+        time.sleep(0.5)  # give it time to die
 
         ctx = multiprocessing.get_context("spawn")
         with warnings.catch_warnings(record=True) as all_warn:
@@ -4534,6 +4508,10 @@ def test_semaphore_tracker_sigint(self):
         # Catchable signal (ignored by semaphore tracker)
         self.check_semaphore_tracker_death(signal.SIGINT, False)
 
+    def test_semaphore_tracker_sigterm(self):
+        # Catchable signal (ignored by semaphore tracker)
+        self.check_semaphore_tracker_death(signal.SIGTERM, False)
+
     def test_semaphore_tracker_sigkill(self):
         # Uncatchable signal.
         self.check_semaphore_tracker_death(signal.SIGKILL, True)