[Python-checkins] bpo-40221: Update multiprocessing to use _at_fork_reinit (GH-19511)

[Dong-hee Na]

https://github.com/python/cpython/commit/a5900ecf9f22e65bef489633692e9ea6941379c5
commit: a5900ecf9f22e65bef489633692e9ea6941379c5
branch: master
author: Dong-hee Na <donghee.na92 at gmail.com>
committer: GitHub <noreply at github.com>
date: 2020-04-15T01:35:36+09:00
summary:

bpo-40221: Update multiprocessing to use _at_fork_reinit (GH-19511)

files:
M Lib/multiprocessing/queues.py
M Lib/multiprocessing/resource_sharer.py

diff --git a/Lib/multiprocessing/queues.py b/Lib/multiprocessing/queues.py
index 835070118387e..c0a284d10c807 100644
--- a/Lib/multiprocessing/queues.py
+++ b/Lib/multiprocessing/queues.py
@@ -49,8 +49,7 @@ def __init__(self, maxsize=0, *, ctx):
         self._sem = ctx.BoundedSemaphore(maxsize)
         # For use by concurrent.futures
         self._ignore_epipe = False
-
-        self._after_fork()
+        self._reset()
 
         if sys.platform != 'win32':
             register_after_fork(self, Queue._after_fork)
@@ -63,11 +62,17 @@ def __getstate__(self):
     def __setstate__(self, state):
         (self._ignore_epipe, self._maxsize, self._reader, self._writer,
          self._rlock, self._wlock, self._sem, self._opid) = state
-        self._after_fork()
+        self._reset()
 
     def _after_fork(self):
         debug('Queue._after_fork()')
-        self._notempty = threading.Condition(threading.Lock())
+        self._reset(after_fork=True)
+
+    def _reset(self, after_fork=False):
+        if after_fork:
+            self._notempty._at_fork_reinit()
+        else:
+            self._notempty = threading.Condition(threading.Lock())
         self._buffer = collections.deque()
         self._thread = None
         self._jointhread = None
diff --git a/Lib/multiprocessing/resource_sharer.py b/Lib/multiprocessing/resource_sharer.py
index 8d5c9900f69fe..66076509a1202 100644
--- a/Lib/multiprocessing/resource_sharer.py
+++ b/Lib/multiprocessing/resource_sharer.py
@@ -63,7 +63,6 @@ class _ResourceSharer(object):
     def __init__(self):
         self._key = 0
         self._cache = {}
-        self._old_locks = []
         self._lock = threading.Lock()
         self._listener = None
         self._address = None
@@ -113,10 +112,7 @@ def _afterfork(self):
         for key, (send, close) in self._cache.items():
             close()
         self._cache.clear()
-        # If self._lock was locked at the time of the fork, it may be broken
-        # -- see issue 6721.  Replace it without letting it be gc'ed.
-        self._old_locks.append(self._lock)
-        self._lock = threading.Lock()
+        self._lock._at_fork_reinit()
         if self._listener is not None:
             self._listener.close()
         self._listener = None