PYTHON-3406 Reinstate warning and docs that PyMongo is not fork safe (#1050)

Log child process C-level stacks when fork tests deadlock.
Encode hostname to bytes to avoid getaddrinfo importlib deadlock.

Author: ShaneHarvey

doc/changelog.rst

@@ -13,8 +13,11 @@ PyMongo 4.3 brings a number of improvements including:
   :class:`bson.codec_options.DatetimeConversion`, and
   :class:`bson.codec_options.CodecOptions`'s ``datetime_conversion``
   parameter for more details (`PYTHON-1824`_).
-- Added support for using a :class:`~pymongo.mongo_client.MongoClient` after
-  an :py:func:`os.fork` (`PYTHON-2484`_).
+- PyMongo now resets its locks and other shared state in the child process
+  after a :py:func:`os.fork` to reduce the frequency of deadlocks. Note that
+  deadlocks are still possible because libraries that PyMongo depends like
+  OpenSSL cannot be made fork() safe in multithreaded applications.
+  (`PYTHON-2484`_). For more info see :ref:`pymongo-fork-safe`.
 
 Bug fixes
 .........

doc/faq.rst

@@ -14,21 +14,15 @@ for threaded applications.
 Is PyMongo fork-safe?
 ---------------------
 
-Starting in PyMongo 4.3, forking on a compatible Python interpreter while
-using a client will result in all locks held by :class:`~bson.objectid
-.ObjectId` and :class:`~pymongo.mongo_client.MongoClient` being released in
-the child, as well as state shared between child and parent processes being
-reset.
-
-If greenlet has been imported (usually with a library like gevent or
-Eventlet), care must be taken when using instances of :class:`~pymongo
-.mongo_client.MongoClient` with ``fork()``. Specifically, instances of
-MongoClient must not be copied from a parent process to a child process.
-Instead, the parent process and each child process must create their own
-instances of MongoClient. Instances of MongoClient copied from the parent
-process have a high probability of deadlock in the child process due to the
-inherent incompatibilities between ``fork()``, threads, and locks described
-:ref:`below<pymongo-fork-safe-details>`.
+PyMongo is not fork-safe. Care must be taken when using instances of
+:class:`~pymongo.mongo_client.MongoClient` with ``fork()``. Specifically,
+instances of MongoClient must not be copied from a parent process to
+a child process. Instead, the parent process and each child process must
+create their own instances of MongoClient. Instances of MongoClient copied from
+the parent process have a high probability of deadlock in the child process due
+to the inherent incompatibilities between ``fork()``, threads, and locks
+described :ref:`below <pymongo-fork-safe-details>`. PyMongo will attempt to
+issue a warning if there is a chance of this deadlock occurring.
 
 .. _pymongo-fork-safe-details:
 
@@ -44,10 +38,33 @@ created by ``fork()`` only has one thread, so any locks that were taken out by
 other threads in the parent will never be released in the child. The next time
 the child process attempts to acquire one of these locks, deadlock occurs.
 
+Starting in version 4.3, PyMongo utilizes :py:func:`os.register_at_fork` to
+reset its locks and other shared state in the child process after a
+:py:func:`os.fork` to reduce the frequency of deadlocks. However deadlocks
+are still possible because libraries that PyMongo depends on, like `OpenSSL`_
+and `getaddrinfo(3)`_ (on some platforms), are not fork() safe in a
+multithreaded application. Linux also imposes the restriction that:
+
+    After a `fork()`_ in a multithreaded program, the child can
+    safely call only async-signal-safe functions (see
+    `signal-safety(7)`_) until such time as it calls `execve(2)`_.
+
+PyMongo relies on functions that are *not* `async-signal-safe`_ and hence the
+child process can experience deadlocks or crashes when attempting to call
+a non `async-signal-safe`_ function. For examples of deadlocks or crashes
+that could occur see `PYTHON-3406`_.
+
 For a long but interesting read about the problems of Python locks in
 multithreaded contexts with ``fork()``, see http://bugs.python.org/issue6721.
 
 .. _not fork-safe: http://bugs.python.org/issue6721
+.. _OpenSSL: https://github.com/openssl/openssl/issues/19066
+.. _fork(): https://man7.org/linux/man-pages/man2/fork.2.html
+.. _signal-safety(7): https://man7.org/linux/man-pages/man7/signal-safety.7.html
+.. _async-signal-safe: https://man7.org/linux/man-pages/man7/signal-safety.7.html
+.. _execve(2): https://man7.org/linux/man-pages/man2/execve.2.html
+.. _getaddrinfo(3): https://man7.org/linux/man-pages/man3/gai_strerror.3.html
+.. _PYTHON-3406: https://jira.mongodb.org/browse/PYTHON-3406
 
 .. _connection-pooling:
 

pymongo/mongo_client.py

@@ -82,7 +82,7 @@
     ServerSelectionTimeoutError,
     WaitQueueTimeoutError,
 )
-from pymongo.lock import _create_lock, _release_locks
+from pymongo.lock import _HAS_REGISTER_AT_FORK, _create_lock, _release_locks
 from pymongo.pool import ConnectionClosedReason
 from pymongo.read_preferences import ReadPreference, _ServerMode
 from pymongo.server_selectors import writable_server_selector
@@ -831,9 +831,10 @@ def __init__(
             self._encrypter = _Encrypter(self, self.__options.auto_encryption_opts)
         self._timeout = self.__options.timeout
 
-        # Add this client to the list of weakly referenced items.
-        # This will be used later if we fork.
-        MongoClient._clients[self._topology._topology_id] = self
+        if _HAS_REGISTER_AT_FORK:
+            # Add this client to the list of weakly referenced items.
+            # This will be used later if we fork.
+            MongoClient._clients[self._topology._topology_id] = self
 
     def _init_background(self):
         self._topology = Topology(self._topology_settings)
@@ -2177,7 +2178,7 @@ def _after_fork_child():
         client._after_fork()
 
 
-if hasattr(os, "register_at_fork"):
+if _HAS_REGISTER_AT_FORK:
     # This will run in the same thread as the fork was called.
     # If we fork in a critical region on the same thread, it should break.
     # This is fine since we would never call fork directly from a critical region.

pymongo/pool.py

@@ -979,9 +979,11 @@ def _create_connection(address, options):
     This is a modified version of create_connection from CPython >= 2.7.
     """
     host, port = address
+    # Avoid the getaddrinfo importlib deadlock on fork() described in PYTHON-3406.
+    host = host.encode("idna")
 
     # Check if dealing with a unix domain socket
-    if host.endswith(".sock"):
+    if host.endswith(b".sock"):
         if not hasattr(socket, "AF_UNIX"):
             raise ConnectionFailure("UNIX-sockets are not supported on this system")
         sock = socket.socket(socket.AF_UNIX)
@@ -998,7 +1000,7 @@ def _create_connection(address, options):
     # is 'localhost' (::1 is fine). Avoids slow connect issues
     # like PYTHON-356.
     family = socket.AF_INET
-    if socket.has_ipv6 and host != "localhost":
+    if socket.has_ipv6 and host != b"localhost":
         family = socket.AF_UNSPEC
 
     err = None

pymongo/topology.py

@@ -36,7 +36,7 @@
     WriteError,
 )
 from pymongo.hello import Hello
-from pymongo.lock import _HAS_REGISTER_AT_FORK, _create_lock
+from pymongo.lock import _create_lock
 from pymongo.monitor import SrvMonitor
 from pymongo.pool import PoolOptions
 from pymongo.server import Server
@@ -174,13 +174,12 @@ def open(self):
             self._pid = pid
         elif pid != self._pid:
             self._pid = pid
-            if not _HAS_REGISTER_AT_FORK:
-                warnings.warn(
-                    "MongoClient opened before fork. May not be entirely fork-safe, "
-                    "proceed with caution. See PyMongo's documentation for details: "
-                    "https://pymongo.readthedocs.io/en/stable/faq.html#"
-                    "is-pymongo-fork-safe"
-                )
+            warnings.warn(
+                "MongoClient opened before fork. May not be entirely fork-safe, "
+                "proceed with caution. 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():

test/__init__.py

@@ -21,8 +21,10 @@
 import os
 import signal
 import socket
+import subprocess
 import sys
 import threading
+import time
 import traceback
 import unittest
 import warnings
@@ -1011,8 +1013,28 @@ def fork(
             with self.fork(target=lambda: print('in child')) as proc:
                 self.assertTrue(proc.pid)  # Child process was started
         """
+
+        def _print_threads(*args: object) -> None:
+            if _print_threads.called:
+                return
+            _print_threads.called = True
+            print_thread_tracebacks()
+
+        _print_threads.called = False
+
+        def _target() -> None:
+            signal.signal(signal.SIGUSR1, _print_threads)
+            try:
+                target()
+            except Exception as exc:
+                sys.stderr.write(f"Child process failed with: {exc}\n")
+                _print_threads()
+                # Sleep for a while to let the parent attach via GDB.
+                time.sleep(2 * timeout)
+                raise
+
         ctx = multiprocessing.get_context("fork")
-        proc = ctx.Process(target=target)
+        proc = ctx.Process(target=_target)
         proc.start()
         try:
             yield proc  # type: ignore
@@ -1021,15 +1043,47 @@ def fork(
             pid = proc.pid
             assert pid
             if proc.exitcode is None:
-                # If it failed, SIGINT to get traceback and wait 10s.
-                os.kill(pid, signal.SIGINT)
-                proc.join(10)
-                proc.kill()
-                proc.join(1)
+                # gdb to get C-level tracebacks
+                print_thread_stacks(pid)
+                # If it failed, SIGUSR1 to get thread tracebacks.
+                os.kill(pid, signal.SIGUSR1)
+                proc.join(5)
+                if proc.exitcode is None:
+                    # SIGINT to get main thread traceback in case SIGUSR1 didn't work.
+                    os.kill(pid, signal.SIGINT)
+                    proc.join(5)
+                if proc.exitcode is None:
+                    # SIGKILL in case SIGINT didn't work.
+                    proc.kill()
+                    proc.join(1)
                 self.fail(f"child timed out after {timeout}s (see traceback in logs): deadlock?")
             self.assertEqual(proc.exitcode, 0)
 
 
+def print_thread_tracebacks() -> None:
+    """Print all Python thread tracebacks."""
+    for thread_id, frame in sys._current_frames().items():
+        sys.stderr.write(f"\n--- Traceback for thread {thread_id} ---\n")
+        traceback.print_stack(frame, file=sys.stderr)
+
+
+def print_thread_stacks(pid: int) -> None:
+    """Print all C-level thread stacks for a given process id."""
+    if sys.platform == "darwin":
+        cmd = ["lldb", "--attach-pid", f"{pid}", "--batch", "--one-line", '"thread backtrace all"']
+    else:
+        cmd = ["gdb", f"--pid={pid}", "--batch", '--eval-command="thread apply all bt"']
+
+    try:
+        res = subprocess.run(
+            cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, encoding="utf-8"
+        )
+    except Exception as exc:
+        sys.stderr.write(f"Could not print C-level thread stacks because {cmd[0]} failed: {exc}")
+    else:
+        sys.stderr.write(res.stdout)
+
+
 class IntegrationTest(PyMongoTestCase):
     """Base class for TestCases that need a connection to MongoDB to pass."""
 

test/test_fork.py

@@ -12,23 +12,19 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-"""Test that pymongo is fork safe."""
+"""Test that pymongo resets its own locks after a fork."""
 
 import os
 import sys
 import unittest
 from multiprocessing import Pipe
 
-from bson.objectid import ObjectId
-
 sys.path[0:0] = [""]
 
 from test import IntegrationTest
-from test.utils import (
-    ExceptionCatchingThread,
-    is_greenthread_patched,
-    rs_or_single_client,
-)
+from test.utils import is_greenthread_patched
+
+from bson.objectid import ObjectId
 
 
 @unittest.skipIf(
@@ -91,55 +87,6 @@ def target():
             passed, msg = parent_conn.recv()
             self.assertTrue(passed, msg)
 
-    def test_many_threaded(self):
-        # Fork randomly while doing operations.
-        clients = []
-        for _ in range(10):
-            c = rs_or_single_client()
-            clients.append(c)
-            self.addCleanup(c.close)
-
-        class ForkThread(ExceptionCatchingThread):
-            def __init__(self, runner, clients):
-                self.runner = runner
-                self.clients = clients
-                self.fork = False
-
-                super().__init__(target=self.fork_behavior)
-
-            def fork_behavior(self) -> None:
-                def action(client):
-                    client.admin.command("ping")
-                    return 0
-
-                for i in range(200):
-                    # Pick a random client.
-                    rc = self.clients[i % len(self.clients)]
-                    if i % 50 == 0 and self.fork:
-                        # Fork
-                        def target():
-                            for c_ in self.clients:
-                                action(c_)
-                                c_.close()
-
-                        with self.runner.fork(target=target) as proc:
-                            self.runner.assertTrue(proc.pid)
-                    action(rc)
-
-        threads = [ForkThread(self, clients) for _ in range(10)]
-        threads[-1].fork = True
-        for t in threads:
-            t.start()
-
-        for t in threads:
-            t.join()
-
-        for t in threads:
-            self.assertIsNone(t.exc)
-
-        for c in clients:
-            c.close()
-
 
 if __name__ == "__main__":
     unittest.main()

test/unified_format.py

@@ -1591,7 +1591,8 @@ def run_scenario(self, spec, uri=None):
                     if i < attempts - 1:
                         print(
                             f"Retrying after attempt {i+1} of {self.id()} failed with:\n"
-                            f"{traceback.format_exc()}"
+                            f"{traceback.format_exc()}",
+                            file=sys.stderr,
                         )
                         self.setUp()
                         continue