10000 [3.6] bpo-31310: multiprocessing's semaphore tracker should be launched again if crashed (GH-3247) by pitrou · Pull Request #4254 · python/cpython · GitHub
[go: up one dir, main page]

Skip to content

[3.6] bpo-31310: multiprocessing's semaphore tracker should be launched again if crashed (GH-3247) #4254

8000
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Nov 3, 2017
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 17 additions & 3 deletions Lib/multiprocessing/semaphore_tracker.py
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@ class SemaphoreTracker(object):
def __init__(self):
self._lock = threading.Lock()
self._fd = None
self._pid = None

def getfd(self):
self.ensure_running()
Expand All @@ -40,8 +41,20 @@ def ensure_running(self):
This can be run from any process. Usually a child process will use
the semaphore created by its parent.'''
with self._lock:
if self._fd is not None:
return
if self._pid is not None:
# semaphore tracker was launched before, is it still running?
pid, status = os.waitpid(self._pid, os.WNOHANG)
if not pid:
# => still alive
return
# => dead, launch it again
os.close(self._fd)
self._fd = None
self._pid = None

warnings.warn('semaphore_tracker: process died unexpectedly, '
'relaunching. Some semaphores might leak.')

fds_to_pass = []
try:
fds_to_pass.append(sys.stderr.fileno())
Expand All @@ -55,12 +68,13 @@ def ensure_running(self):
exe = spawn.get_executable()
args = [exe] + util._args_from_interpreter_flags()
args += ['-c', cmd % r]
util.spawnv_passfds(exe, args, fds_to_pass)
pid = util.spawnv_passfds(exe, args, fds_to_pass)
except:
os.close(w)
raise
else:
self._fd = w
self._pid = pid
finally:
os.close(r)

Expand Down
43 changes: 39 additions & 4 deletions Lib/test/_test_multiprocessing.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@

import unittest
import queue as pyqueue
import contextlib
import time
import io
import itertools
Expand Down Expand Up @@ -4125,14 +4126,14 @@ def test_preload_resources(self):
self.fail("failed spawning forkserver or grandchild")


#
# Check that killing process does not leak named semaphores
#

@unittest.skipIf(sys.platform == "win32",
"test semantics don't make sense on Windows")
class TestSemaphoreTracker(unittest.TestCase):

def test_semaphore_tracker(self):
#
# Check that killing process does not leak named semaphores
#
import subprocess
cmd = '''if 1:
import multiprocessing as mp, time, os
Expand Down Expand Up @@ -4166,6 +4167,40 @@ def test_semaphore_tracker(self):
self.assertRegex(err, expected)
self.assertRegex(err, r'semaphore_tracker: %r: \[Errno' % name1)

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
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"))
sem = ctx.Semaphore()
sem.acquire()
sem.release()
wr = weakref.ref(sem)
# ensure `sem` gets collected, which triggers communication with
# the semaphore tracker
del sem
gc.collect()
self.assertIsNone(wr())

def test_semaphore_tracker_sigint(self):
# Catchable signal (ignored by semaphore tracker)
self.check_semaphore_tracker_death(signal.SIGINT, False)

def test_semaphore_tracker_sigkill(self):
# Uncatchable signal.
self.check_semaphore_tracker_death(signal.SIGKILL, True)


class TestSimpleQueue(unittest.TestCase):

@classmethod
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
multiprocessing's semaphore tracker should be launched again if crashed.
0