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.