PYTHON-3257 Fix "connection pool paused" errors in child after fork (#944)

Author: ShaneHarvey

pymongo/pool.py

@@ -1420,7 +1420,7 @@ def _get_socket(self):
         # See test.test_client:TestClient.test_fork for an example of
         # what could go wrong otherwise
         if self.pid != os.getpid():
-            self.reset()
+            self.reset_without_pause()
 
         if self.closed:
             if self.enabled_for_cmap:
@@ -1526,7 +1526,7 @@ def return_socket(self, sock_info):
         if self.enabled_for_cmap:
             listeners.publish_connection_checked_in(self.address, sock_info.id)
         if self.pid != os.getpid():
-            self.reset()
+            self.reset_without_pause()
         else:
             if self.closed:
                 sock_info.close_socket(ConnectionClosedReason.POOL_CLOSED)

pymongo/topology.py

@@ -169,20 +169,24 @@ def open(self):
           forking.
 
         """
+        pid = os.getpid()
         if self._pid is None:
-            self._pid = os.getpid()
-        else:
-            if os.getpid() != self._pid:
-                warnings.warn(
-                    "MongoClient opened before fork. Create MongoClient only "
-                    "after forking. See PyMongo's documentation for details: "
-                    "https://pymongo.readthedocs.io/en/stable/faq.html#"
-                    "is-pymongo-fork-safe"
-                )
-                with self._lock:
-                    # Reset the session pool to avoid duplicate sessions in
-                    # the child process.
-                    self._session_pool.reset()
+            self._pid = pid
+        elif pid != self._pid:
+            self._pid = pid
+            warnings.warn(
+                "MongoClient opened before fork. Create MongoClient only "
+                "after forking. See PyMongo's documentation for details: "
+                "https://pymongo.readthedocs.io/en/stable/faq.html#"
+                "is-pymongo-fork-safe"
+            )
+            with self._lock:
+                # Close servers and clear the pools.
+                for server in self._servers.values():
+                    server.close()
+                # Reset the session pool to avoid duplicate sessions in
+                # the child process.
+                self._session_pool.reset()
 
         with self._lock:
             self._ensure_opened()