From 29d18eb489fb76f698c5296d9694ab94dd2531aa Mon Sep 17 00:00:00 2001
From: Petr Viktorin <encukou@gmail.com>
Date: Wed, 13 Nov 2024 10:25:10 +0100
Subject: [PATCH] gh-71936: Fix race condition in multiprocessing.Pool
 (GH-124973)

* gh-71936: Fix race condition in multiprocessing.Pool

Proxes of shared objects register a Finalizer in BaseProxy._incref(), and it
will call BaseProxy._decref() when it is GCed. This may cause a race condition
with Pool(maxtasksperchild=None) on Windows.

A connection would be closed and raised TypeError when a GC occurs between
_ConnectionBase._check_writable() and _ConnectionBase._send_bytes() in
_ConnectionBase.send() in the second or later task, and a new object
is allocated that shares the id() of a previously deleted one.

Instead of using the id() of the token (or the proxy), use a unique,
non-reusable number.

(cherry picked from commit ba088c8f9cf7163b0f28c507cb1343befe21997e)

Co-authored-by: Petr Viktorin <encukou@gmail.com>
Co-Authored-By: Akinori Hattori <hattya@gmail.com>
---
 Lib/multiprocessing/managers.py               | 33 +++++++++++--------
 Misc/ACKS                                     |  1 +
 ...2-10-15-10-18-20.gh-issue-71936.MzJjc_.rst |  1 +
 3 files changed, 22 insertions(+), 13 deletions(-)
 create mode 100644 Misc/NEWS.d/next/Library/2022-10-15-10-18-20.gh-issue-71936.MzJjc_.rst

diff --git a/Lib/multiprocessing/managers.py b/Lib/multiprocessing/managers.py
index 76b915de74d94e..ef791c2751688a 100644
--- a/Lib/multiprocessing/managers.py
+++ b/Lib/multiprocessing/managers.py
@@ -758,22 +758,29 @@ class BaseProxy(object):
     _address_to_local = {}
     _mutex = util.ForkAwareThreadLock()
 
+    # Each instance gets a `_serial` number. Unlike `id(...)`, this number
+    # is never reused.
+    _next_serial = 1
+
     def __init__(self, token, serializer, manager=None,
                  authkey=None, exposed=None, incref=True, manager_owned=False):
         with BaseProxy._mutex:
-            tls_idset = BaseProxy._address_to_local.get(token.address, None)
-            if tls_idset is None:
-                tls_idset = util.ForkAwareLocal(), ProcessLocalSet()
-                BaseProxy._address_to_local[token.address] = tls_idset
+            tls_serials = BaseProxy._address_to_local.get(token.address, None)
+            if tls_serials is None:
+                tls_serials = util.ForkAwareLocal(), ProcessLocalSet()
+                BaseProxy._address_to_local[token.address] = tls_serials
+
+            self._serial = BaseProxy._next_serial
+            BaseProxy._next_serial += 1
 
         # self._tls is used to record the connection used by this
         # thread to communicate with the manager at token.address
-        self._tls = tls_idset[0]
+        self._tls = tls_serials[0]
 
-        # self._idset is used to record the identities of all shared
-        # objects for which the current process owns references and
+        # self._all_serials is a set used to record the identities of all
+        # shared objects for which the current process owns references and
         # which are in the manager at token.address
-        self._idset = tls_idset[1]
+        self._all_serials = tls_serials[1]
 
         self._token = token
         self._id = self._token.id
@@ -856,20 +863,20 @@ def _incref(self):
         dispatch(conn, None, 'incref', (self._id,))
         util.debug('INCREF %r', self._token.id)
 
-        self._idset.add(self._id)
+        self._all_serials.add(self._serial)
 
         state = self._manager and self._manager._state
 
         self._close = util.Finalize(
             self, BaseProxy._decref,
-            args=(self._token, self._authkey, state,
-                  self._tls, self._idset, self._Client),
+            args=(self._token, self._serial, self._authkey, state,
+                  self._tls, self._all_serials, self._Client),
             exitpriority=10
             )
 
     @staticmethod
-    def _decref(token, authkey, state, tls, idset, _Client):
-        idset.discard(token.id)
+    def _decref(token, serial, authkey, state, tls, idset, _Client):
+        idset.discard(serial)
 
         # check whether manager is still alive
         if state is None or state.value == State.STARTED:
diff --git a/Misc/ACKS b/Misc/ACKS
index 05cb6874759c31..d3bf8450fcbb4f 100644
--- a/Misc/ACKS
+++ b/Misc/ACKS
@@ -730,6 +730,7 @@ Larry Hastings
 Tim Hatch
 Zac Hatfield-Dodds
 Shane Hathaway
+Akinori Hattori
 Michael Haubenwallner
 Janko Hauser
 Flavian Hautbois
diff --git a/Misc/NEWS.d/next/Library/2022-10-15-10-18-20.gh-issue-71936.MzJjc_.rst b/Misc/NEWS.d/next/Library/2022-10-15-10-18-20.gh-issue-71936.MzJjc_.rst
new file mode 100644
index 00000000000000..a0959cc086fa9e
--- /dev/null
+++ b/Misc/NEWS.d/next/Library/2022-10-15-10-18-20.gh-issue-71936.MzJjc_.rst
@@ -0,0 +1 @@
+Fix a race condition in :class:`multiprocessing.pool.Pool`.