DM-55482: Disable implicit threading in multiprocessing workers

doc/changes/DM-55482.bugfix.md

@@ -0,0 +1,2 @@
+Implicit threading is now disabled again inside each multiprocessing worker process.
+Runtime thread limits applied in the parent process do not propagate to spawned processes and inherited environment variables only cover libraries that have not yet created their thread pools.

python/lsst/pipe/base/mp_graph_executor.py

@@ -199,6 +199,12 @@ def _executeJob(
         fail_fast : `bool`
             If `True` then kill subprocess on RepeatableQuantumError.
         """
+        # This process was started with the spawn method, so runtime thread
+        # limits applied in the parent process do not carry over and implicit
+        # threading has to be disabled again. Inherited environment variables
+        # only cover libraries that have not yet created their thread pools.
+        disable_implicit_threading()
+
         # This terrible hack is a workaround for Python threading bug:
         # https://github.com/python/cpython/issues/102512. Should be removed
         # when fix for that bug is deployed. Inspired by

tests/test_mp_graph_executor.py

@@ -31,16 +31,18 @@
 import multiprocessing
 import multiprocessing.context
 import os
+import pickle
 import signal
 import sys
 import unittest
+import unittest.mock
 import warnings
 from typing import Literal
 
 import psutil
 
 from lsst.pipe.base.exec_fixup_data_id import ExecFixupDataId
-from lsst.pipe.base.mp_graph_executor import MPGraphExecutor, MPGraphExecutorError, MPTimeoutError
+from lsst.pipe.base.mp_graph_executor import MPGraphExecutor, MPGraphExecutorError, MPTimeoutError, _Job
 from lsst.pipe.base.quantum_reports import ExecutionStatus, Report
 from lsst.pipe.base.tests.mocks import (
     DynamicConnectionConfig,
@@ -421,6 +423,37 @@ def test_mpexec_num_fd(self) -> None:
         # quanta (20).
         self.assertLess(num_fds_1 - num_fds_0, 5)
 
+    def test_executejob_disables_implicit_threading(self) -> None:
+        """Check that the subprocess entry point disables implicit threading
+        itself, since runtime thread limits applied in the parent process do
+        not propagate to spawned processes.
+        """
+        helper = InMemoryRepo("base.yaml")
+        self.enterContext(helper)
+        helper.add_task(dimensions=["detector"])
+        qg = helper.make_quantum_graph()
+        qexec, _ = helper.make_single_quantum_executor()
+        qg.build_execution_quanta()
+        xgraph = qg.quantum_only_xgraph
+        quantum_id = next(iter(xgraph))
+        node = xgraph.nodes[quantum_id]
+        rcv_conn, snd_conn = multiprocessing.Pipe(False)
+        with unittest.mock.patch(
+            "lsst.pipe.base.mp_graph_executor.disable_implicit_threading"
+        ) as mock_disable:
+            _Job._executeJob(
+                pickle.dumps(qexec),
+                pickle.dumps(node["pipeline_node"]),
+                pickle.dumps(node["quantum"]),
+                quantum_id,
+                [],
+                snd_conn,
+                False,
+            )
+        mock_disable.assert_called_once_with()
+        # The job ran to completion and sent its report.
+        self.assertTrue(rcv_conn.poll())
+
 
 def setup_module(module):
     """Force spawn to be used if no method given explicitly.