From 6d949356f4475988374f6b6be34ce6094bcf115b Mon Sep 17 00:00:00 2001
From: Tim Jenness <tjenness@lsst.org>
Date: Wed, 15 Jul 2026 15:30:47 -0700
Subject: [PATCH] Disable implicit threading in multiprocessing workers

Worker processes are started with the spawn method, so runtime thread
limits applied in the parent process do not carry over and inherited
environment variables only cover libraries that have not yet created
their thread pools. Call disable_implicit_threading again at the top
of the subprocess entry point.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 doc/changes/DM-55482.bugfix.md             |  2 ++
 python/lsst/pipe/base/mp_graph_executor.py |  6 ++++
 tests/test_mp_graph_executor.py            | 35 +++++++++++++++++++++-
 3 files changed, 42 insertions(+), 1 deletion(-)
 create mode 100644 doc/changes/DM-55482.bugfix.md

diff --git a/doc/changes/DM-55482.bugfix.md b/doc/changes/DM-55482.bugfix.md
new file mode 100644
index 000000000..f758f7d45
--- /dev/null
+++ b/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.
diff --git a/python/lsst/pipe/base/mp_graph_executor.py b/python/lsst/pipe/base/mp_graph_executor.py
index c883b89df..ca90f3dd9 100644
--- a/python/lsst/pipe/base/mp_graph_executor.py
+++ b/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
diff --git a/tests/test_mp_graph_executor.py b/tests/test_mp_graph_executor.py
index 95ebbea22..4ebd0f16f 100644
--- a/tests/test_mp_graph_executor.py
+++ b/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.