Coverage for tests/test_mp_graph_executor.py: 99%
281 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-22 17:17 +0000
1# This file is part of pipe_base.
2#
3# Developed for the LSST Data Management System.
4# This product includes software developed by the LSST Project
5# (https://www.lsst.org).
6# See the COPYRIGHT file at the top-level directory of this distribution
7# for details of code ownership.
8#
9# This software is dual licensed under the GNU General Public License and also
10# under a 3-clause BSD license. Recipients may choose which of these licenses
11# to use; please see the files gpl-3.0.txt and/or bsd_license.txt,
12# respectively. If you choose the GPL option then the following text applies
13# (but note that there is still no warranty even if you opt for BSD instead):
14#
15# This program is free software: you can redistribute it and/or modify
16# it under the terms of the GNU General Public License as published by
17# the Free Software Foundation, either version 3 of the License, or
18# (at your option) any later version.
19#
20# This program is distributed in the hope that it will be useful,
21# but WITHOUT ANY WARRANTY; without even the implied warranty of
22# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
23# GNU General Public License for more details.
24#
25# You should have received a copy of the GNU General Public License
26# along with this program. If not, see <https://www.gnu.org/licenses/>.
28from __future__ import annotations
30import logging
31import multiprocessing
32import multiprocessing.context
33import os
34import pickle
35import signal
36import sys
37import unittest
38import unittest.mock
39import warnings
40from typing import Literal
42import psutil
44from lsst.pipe.base.exec_fixup_data_id import ExecFixupDataId
45from lsst.pipe.base.mp_graph_executor import MPGraphExecutor, MPGraphExecutorError, MPTimeoutError, _Job
46from lsst.pipe.base.quantum_reports import ExecutionStatus, Report
47from lsst.pipe.base.tests.mocks import (
48 DynamicConnectionConfig,
49 DynamicTestPipelineTask,
50 DynamicTestPipelineTaskConfig,
51 InMemoryRepo,
52)
54logging.basicConfig(level=logging.DEBUG)
56_LOG = logging.getLogger(__name__)
58TESTDIR = os.path.abspath(os.path.dirname(__file__))
61class NoMultiprocessingTask(DynamicTestPipelineTask):
62 """A test pipeline task that declares that it cannot be used in
63 multiprocessing.
64 """
66 canMultiprocess = False
69def _count_status(report: Report, status: ExecutionStatus) -> int:
70 """Count number of quanta with a given status."""
71 return len([qrep for qrep in report.quantaReports if qrep.status is status])
74class MPGraphExecutorTestCase(unittest.TestCase):
75 """A test case for MPGraphExecutor class."""
77 def test_mpexec_nomp(self) -> None:
78 """Make simple graph and execute."""
79 helper = InMemoryRepo("base.yaml")
80 self.enterContext(helper)
81 helper.add_task(dimensions=["detector"])
82 qgraph = helper.make_quantum_graph()
83 qexec, butler = helper.make_single_quantum_executor()
84 # run in single-process mode
85 mpexec = MPGraphExecutor(num_proc=1, timeout=100, quantum_executor=qexec)
86 mpexec.execute(qgraph) # type: ignore[arg-type]
87 self.assertCountEqual(
88 [ref.dataId["detector"] for ref in butler.get_datasets("dataset_auto1")], [1, 2, 3, 4]
89 )
90 report = mpexec.getReport()
91 assert report is not None
92 self.assertEqual(report.status, ExecutionStatus.SUCCESS)
93 self.assertIsNone(report.exitCode)
94 self.assertIsNone(report.exceptionInfo)
95 self.assertEqual(len(report.quantaReports), 4)
96 self.assertTrue(all(qrep.status == ExecutionStatus.SUCCESS for qrep in report.quantaReports))
97 self.assertTrue(all(qrep.exitCode is None for qrep in report.quantaReports))
98 self.assertTrue(all(qrep.exceptionInfo is None for qrep in report.quantaReports))
99 self.assertTrue(all(qrep.taskLabel == "task_auto1" for qrep in report.quantaReports))
101 def test_mpexec_mp(self) -> None:
102 """Make simple graph and execute."""
103 helper = InMemoryRepo("base.yaml")
104 self.enterContext(helper)
105 helper.add_task(dimensions=["detector"])
106 qg = helper.make_quantum_graph()
107 qexec, butler = helper.make_single_quantum_executor()
109 methods: list[Literal["spawn", "forkserver"]] = ["spawn"]
110 if sys.platform == "linux": 110 ↛ 113line 110 didn't jump to line 113 because the condition on line 110 was always true
111 methods.append("forkserver")
113 for method in methods:
114 with self.subTest(startMethod=method):
115 # Run in multi-process mode, the order of results is not
116 # defined.
117 mpexec = MPGraphExecutor(num_proc=3, timeout=100, quantum_executor=qexec, start_method=method)
118 mpexec.execute(qg) # type: ignore[arg-type]
119 report = mpexec.getReport()
120 assert report is not None
121 self.assertEqual(report.status, ExecutionStatus.SUCCESS)
122 self.assertIsNone(report.exitCode)
123 self.assertIsNone(report.exceptionInfo)
124 self.assertEqual(len(report.quantaReports), 4)
125 self.assertTrue(all(qrep.status == ExecutionStatus.SUCCESS for qrep in report.quantaReports))
126 self.assertTrue(all(qrep.exitCode == 0 for qrep in report.quantaReports))
127 self.assertTrue(all(qrep.exceptionInfo is None for qrep in report.quantaReports))
128 self.assertTrue(all(qrep.taskLabel == "task_auto1" for qrep in report.quantaReports))
130 def test_mpexec_nompsupport(self) -> None:
131 """Try to run MP for task that has no MP support which should fail."""
132 helper = InMemoryRepo("base.yaml")
133 self.enterContext(helper)
134 helper.add_task(task_class=NoMultiprocessingTask, dimensions=["detector"])
135 qg = helper.make_quantum_graph()
136 qexec, butler = helper.make_single_quantum_executor()
137 mpexec = MPGraphExecutor(num_proc=3, timeout=100, quantum_executor=qexec)
138 with self.assertRaisesRegex(
139 MPGraphExecutorError, "Task 'task_auto1' does not support multiprocessing"
140 ):
141 mpexec.execute(qg) # type: ignore[arg-type]
143 def test_mpexec_fixup(self) -> None:
144 """Make simple graph and execute, add dependencies by executing fixup
145 code.
146 """
147 helper = InMemoryRepo("base.yaml")
148 self.enterContext(helper)
149 helper.add_task(dimensions=["detector"])
150 qg = helper.make_quantum_graph()
151 for reverse in (False, True):
152 qexec, butler = helper.make_single_quantum_executor()
153 fixup = ExecFixupDataId("task_auto1", "detector", reverse=reverse)
154 mpexec = MPGraphExecutor(
155 num_proc=1, timeout=100, quantum_executor=qexec, execution_graph_fixup=fixup
156 )
157 mpexec.execute(qg) # type: ignore[arg-type]
158 expected = [1, 2, 3, 4]
159 if reverse:
160 expected = list(reversed(expected))
161 self.assertEqual(
162 [ref.dataId["detector"] for ref in butler.get_datasets("dataset_auto1")], expected
163 )
165 def test_mpexec_fixup_old_qg(self) -> None:
166 """Test using an old QuantumGraph object to initialize the executor,
167 with an ordering fixup.
168 """
169 helper = InMemoryRepo("base.yaml")
170 self.enterContext(helper)
171 helper.add_task(dimensions=["detector"])
172 qgraph = helper.make_quantum_graph_builder().build(attach_datastore_records=False)
173 for reverse in (False, True):
174 qexec, butler = helper.make_single_quantum_executor()
175 fixup = ExecFixupDataId("task_auto1", "detector", reverse=reverse)
176 mpexec = MPGraphExecutor(
177 num_proc=1, timeout=100, quantum_executor=qexec, execution_graph_fixup=fixup
178 )
179 mpexec.execute(qgraph) # type: ignore[arg-type]
180 expected = [1, 2, 3, 4]
181 if reverse:
182 expected = list(reversed(expected))
183 self.assertEqual(
184 [ref.dataId["detector"] for ref in butler.get_datasets("dataset_auto1")], expected
185 )
187 def test_mpexec_timeout(self) -> None:
188 """Fail due to timeout."""
189 helper = InMemoryRepo("base.yaml")
190 self.enterContext(helper)
191 helper.add_task(label="a")
192 helper.add_task(
193 label="b",
194 inputs={"input_connection": DynamicConnectionConfig(dataset_type_name="dataset_auto0")},
195 )
196 helper.add_task(
197 label="c",
198 inputs={"input_connection": DynamicConnectionConfig(dataset_type_name="dataset_auto0")},
199 config=DynamicTestPipelineTaskConfig(sleep=100.0),
200 )
201 qg = helper.make_quantum_graph()
203 # with failFast we'll get immediate MPTimeoutError
204 qexec, _ = helper.make_single_quantum_executor()
205 mpexec = MPGraphExecutor(num_proc=3, timeout=1, quantum_executor=qexec, fail_fast=True)
206 with self.assertRaises(MPTimeoutError):
207 mpexec.execute(qg) # type: ignore[arg-type]
208 report = mpexec.getReport()
209 assert report is not None and report.exceptionInfo is not None
210 self.assertEqual(report.status, ExecutionStatus.TIMEOUT)
211 self.assertEqual(report.exceptionInfo.className, "lsst.pipe.base.mp_graph_executor.MPTimeoutError")
212 self.assertGreater(len(report.quantaReports), 0)
213 self.assertEqual(_count_status(report, ExecutionStatus.TIMEOUT), 1)
214 self.assertTrue(any(qrep.exitCode is not None and qrep.exitCode < 0 for qrep in report.quantaReports))
215 self.assertTrue(all(qrep.exceptionInfo is None for qrep in report.quantaReports))
217 # with failFast=False exception happens after last task finishes
218 qexec, _ = helper.make_single_quantum_executor()
219 mpexec = MPGraphExecutor(num_proc=3, timeout=3, quantum_executor=qexec, fail_fast=False)
220 with self.assertRaises(MPTimeoutError):
221 mpexec.execute(qg) # type: ignore[arg-type]
222 assert report is not None and report.exceptionInfo is not None
223 self.assertEqual(report.status, ExecutionStatus.TIMEOUT)
224 self.assertEqual(report.exceptionInfo.className, "lsst.pipe.base.mp_graph_executor.MPTimeoutError")
225 # We expect two tasks ('a' and 'b') to finish successfully and one task
226 # ('c') to timeout, which should get us all three reports.
227 # Unfortunately on busy CPU there is no guarantee that tasks finish on
228 # time, so expect more timeouts and issue a warning.
229 if len(report.quantaReports) != 3: 229 ↛ 233line 229 didn't jump to line 233 because the condition on line 229 was always true
230 warnings.warn(
231 f"Possibly timed out tasks, expected three reports, received {len(report.quantaReports)})."
232 )
233 report = mpexec.getReport()
234 self.assertGreater(_count_status(report, ExecutionStatus.TIMEOUT), 0)
235 self.assertTrue(any(qrep.exitCode is not None and qrep.exitCode < 0 for qrep in report.quantaReports))
236 self.assertTrue(all(qrep.exceptionInfo is None for qrep in report.quantaReports))
238 def test_mpexec_failure(self) -> None:
239 """Failure in one task should not stop other tasks."""
240 helper = InMemoryRepo("base.yaml")
241 self.enterContext(helper)
242 helper.add_task(
243 config=DynamicTestPipelineTaskConfig(fail_condition="detector=2"),
244 dimensions=["detector"],
245 )
246 qg = helper.make_quantum_graph()
247 qexec, _ = helper.make_single_quantum_executor()
248 mpexec = MPGraphExecutor(num_proc=3, timeout=100, quantum_executor=qexec)
249 with self.assertRaisesRegex(MPGraphExecutorError, "One or more tasks failed"):
250 mpexec.execute(qg) # type: ignore[arg-type]
251 report = mpexec.getReport()
252 assert report is not None and report.exceptionInfo is not None
253 self.assertEqual(report.status, ExecutionStatus.FAILURE)
254 self.assertEqual(
255 report.exceptionInfo.className, "lsst.pipe.base.mp_graph_executor.MPGraphExecutorError"
256 )
257 self.assertGreater(len(report.quantaReports), 0)
258 self.assertEqual(_count_status(report, ExecutionStatus.FAILURE), 1)
259 self.assertEqual(_count_status(report, ExecutionStatus.SUCCESS), 3)
260 self.assertTrue(any(qrep.exitCode is not None and qrep.exitCode > 0 for qrep in report.quantaReports))
261 self.assertTrue(any(qrep.exceptionInfo is not None for qrep in report.quantaReports))
263 def test_mpexec_failure_dep(self) -> None:
264 """Failure in one task should skip dependents."""
265 helper = InMemoryRepo("base.yaml")
266 self.enterContext(helper)
267 helper.add_task(
268 "a", config=DynamicTestPipelineTaskConfig(fail_condition="detector=2"), dimensions=["detector"]
269 )
270 helper.add_task("b", dimensions=["detector"]) # depends on 'a', for the same detector.
271 qg = helper.make_quantum_graph()
272 qexec, _ = helper.make_single_quantum_executor()
273 mpexec = MPGraphExecutor(num_proc=3, timeout=100, quantum_executor=qexec)
274 with self.assertRaisesRegex(MPGraphExecutorError, "One or more tasks failed"):
275 mpexec.execute(qg) # type: ignore[arg-type]
276 report = mpexec.getReport()
277 assert report is not None and report.exceptionInfo is not None
278 self.assertEqual(report.status, ExecutionStatus.FAILURE)
279 self.assertEqual(
280 report.exceptionInfo.className, "lsst.pipe.base.mp_graph_executor.MPGraphExecutorError"
281 )
282 # Dependencies of failed tasks do not appear in quantaReports
283 self.assertGreater(len(report.quantaReports), 0)
284 self.assertEqual(_count_status(report, ExecutionStatus.FAILURE), 1)
285 self.assertEqual(_count_status(report, ExecutionStatus.SUCCESS), 6)
286 self.assertEqual(_count_status(report, ExecutionStatus.SKIPPED), 1)
287 self.assertTrue(any(qrep.exitCode is not None and qrep.exitCode > 0 for qrep in report.quantaReports))
288 self.assertTrue(any(qrep.exceptionInfo is not None for qrep in report.quantaReports))
290 def test_mpexec_failure_dep_nomp(self) -> None:
291 """Failure in one task should skip dependents, in-process version."""
292 helper = InMemoryRepo("base.yaml")
293 self.enterContext(helper)
294 helper.add_task(
295 "a", config=DynamicTestPipelineTaskConfig(fail_condition="detector=2"), dimensions=["detector"]
296 )
297 helper.add_task("b", dimensions=["detector"]) # depends on 'a', for the same detector.
298 qg = helper.make_quantum_graph()
299 qexec, butler = helper.make_single_quantum_executor()
300 mpexec = MPGraphExecutor(num_proc=1, timeout=100, quantum_executor=qexec)
301 with self.assertRaisesRegex(MPGraphExecutorError, "One or more tasks failed"):
302 mpexec.execute(qg) # type: ignore[arg-type]
303 self.assertCountEqual(
304 [ref.dataId["detector"] for ref in butler.get_datasets("dataset_auto1")], [1, 3, 4]
305 )
306 self.assertCountEqual(
307 [ref.dataId["detector"] for ref in butler.get_datasets("dataset_auto2")], [1, 3, 4]
308 )
309 report = mpexec.getReport()
310 assert report is not None and report.exceptionInfo is not None
311 self.assertEqual(report.status, ExecutionStatus.FAILURE)
312 self.assertEqual(
313 report.exceptionInfo.className, "lsst.pipe.base.mp_graph_executor.MPGraphExecutorError"
314 )
315 # Dependencies of failed tasks do not appear in quantaReports
316 self.assertGreater(len(report.quantaReports), 0)
317 self.assertEqual(_count_status(report, ExecutionStatus.FAILURE), 1)
318 self.assertEqual(_count_status(report, ExecutionStatus.SUCCESS), 6)
319 self.assertEqual(_count_status(report, ExecutionStatus.SKIPPED), 1)
320 self.assertTrue(all(qrep.exitCode is None for qrep in report.quantaReports))
321 self.assertTrue(any(qrep.exceptionInfo is not None for qrep in report.quantaReports))
323 def test_mpexec_failure_failfast(self) -> None:
324 """Fast fail stops quickly.
326 Timing delay of task 'b' should be sufficient to process
327 failure and raise exception before task 'c'.
328 """
329 helper = InMemoryRepo("base.yaml")
330 self.enterContext(helper)
331 helper.add_task(
332 "a", config=DynamicTestPipelineTaskConfig(fail_condition="detector=2"), dimensions=["detector"]
333 )
334 helper.add_task("b", config=DynamicTestPipelineTaskConfig(sleep=100.0), dimensions=["detector"])
335 helper.add_task("c", dimensions=["detector"]) # depends on 'b', for the same detector.
336 qg = helper.make_quantum_graph()
337 qexec, _ = helper.make_single_quantum_executor()
338 mpexec = MPGraphExecutor(num_proc=3, timeout=100, quantum_executor=qexec, fail_fast=True)
339 with self.assertRaisesRegex(MPGraphExecutorError, "failed, exit code=1"):
340 mpexec.execute(qg) # type: ignore[arg-type]
341 report = mpexec.getReport()
342 assert report is not None and report.exceptionInfo is not None
343 self.assertEqual(report.status, ExecutionStatus.FAILURE)
344 self.assertEqual(
345 report.exceptionInfo.className, "lsst.pipe.base.mp_graph_executor.MPGraphExecutorError"
346 )
347 # Dependencies of failed tasks do not appear in quantaReports
348 self.assertGreater(len(report.quantaReports), 0)
349 self.assertEqual(_count_status(report, ExecutionStatus.FAILURE), 1)
350 self.assertTrue(any(qrep.exitCode is not None and qrep.exitCode > 0 for qrep in report.quantaReports))
351 self.assertTrue(any(qrep.exceptionInfo is not None for qrep in report.quantaReports))
353 def test_mpexec_crash(self) -> None:
354 """Check task crash due to signal."""
355 helper = InMemoryRepo("base.yaml")
356 self.enterContext(helper)
357 helper.add_task(
358 config=DynamicTestPipelineTaskConfig(fail_condition="detector=2", fail_signal=signal.SIGILL),
359 dimensions=["detector"],
360 )
361 qg = helper.make_quantum_graph()
362 qexec, _ = helper.make_single_quantum_executor()
363 mpexec = MPGraphExecutor(num_proc=3, timeout=100, quantum_executor=qexec)
364 with self.assertRaisesRegex(MPGraphExecutorError, "One or more tasks failed"):
365 mpexec.execute(qg) # type: ignore[arg-type]
366 report = mpexec.getReport()
367 assert report is not None and report.exceptionInfo is not None
368 self.assertEqual(report.status, ExecutionStatus.FAILURE)
369 self.assertEqual(
370 report.exceptionInfo.className, "lsst.pipe.base.mp_graph_executor.MPGraphExecutorError"
371 )
372 # Dependencies of failed tasks do not appear in quantaReports
373 self.assertGreater(len(report.quantaReports), 0)
374 self.assertEqual(_count_status(report, ExecutionStatus.FAILURE), 1)
375 self.assertEqual(_count_status(report, ExecutionStatus.SUCCESS), 3)
376 self.assertTrue(any(qrep.exitCode == -signal.SIGILL for qrep in report.quantaReports))
377 self.assertTrue(all(qrep.exceptionInfo is None for qrep in report.quantaReports))
379 def test_mpexec_crash_failfast(self) -> None:
380 """Check task crash due to signal with --fail-fast."""
381 helper = InMemoryRepo("base.yaml")
382 self.enterContext(helper)
383 helper.add_task(
384 "a",
385 config=DynamicTestPipelineTaskConfig(fail_condition="detector=2", fail_signal=signal.SIGILL),
386 dimensions=["detector"],
387 )
388 helper.add_task("b", config=DynamicTestPipelineTaskConfig(sleep=100.0), dimensions=["detector"])
389 helper.add_task("c", dimensions=["detector"]) # depends on 'b', for the same detector.
390 qg = helper.make_quantum_graph()
391 qexec, _ = helper.make_single_quantum_executor()
392 mpexec = MPGraphExecutor(num_proc=3, timeout=100, quantum_executor=qexec, fail_fast=True)
393 with self.assertRaisesRegex(MPGraphExecutorError, "failed, killed by signal 4 .Illegal instruction"):
394 mpexec.execute(qg) # type: ignore[arg-type]
395 report = mpexec.getReport()
396 assert report is not None and report.exceptionInfo is not None
397 self.assertEqual(report.status, ExecutionStatus.FAILURE)
398 self.assertEqual(
399 report.exceptionInfo.className, "lsst.pipe.base.mp_graph_executor.MPGraphExecutorError"
400 )
401 self.assertEqual(_count_status(report, ExecutionStatus.FAILURE), 1)
402 self.assertTrue(any(qrep.exitCode == -signal.SIGILL for qrep in report.quantaReports))
403 self.assertTrue(all(qrep.exceptionInfo is None for qrep in report.quantaReports))
405 def test_mpexec_num_fd(self) -> None:
406 """Check that number of open files stays reasonable."""
407 helper = InMemoryRepo("base.yaml")
408 self.enterContext(helper)
409 helper.add_task("a", task_class=NoMultiprocessingTask, dimensions=["detector", "visit"])
410 helper.add_task("b", task_class=NoMultiprocessingTask, dimensions=["detector", "visit"])
411 qg = helper.make_quantum_graph()
412 qexec, _ = helper.make_single_quantum_executor()
413 this_proc = psutil.Process()
414 num_fds_0 = this_proc.num_fds()
416 # run in multi-process mode, the order of results is not defined
417 mpexec = MPGraphExecutor(num_proc=3, timeout=100, quantum_executor=qexec)
418 mpexec.execute(qg) # type: ignore[arg-type]
420 num_fds_1 = this_proc.num_fds()
421 # They should be the same but allow small growth just in case.
422 # Without DM-26728 fix the difference would be equal to number of
423 # quanta (20).
424 self.assertLess(num_fds_1 - num_fds_0, 5)
426 def test_executejob_disables_implicit_threading(self) -> None:
427 """Check that the subprocess entry point disables implicit threading
428 itself, since runtime thread limits applied in the parent process do
429 not propagate to spawned processes.
430 """
431 helper = InMemoryRepo("base.yaml")
432 self.enterContext(helper)
433 helper.add_task(dimensions=["detector"])
434 qg = helper.make_quantum_graph()
435 qexec, _ = helper.make_single_quantum_executor()
436 qg.build_execution_quanta()
437 xgraph = qg.quantum_only_xgraph
438 quantum_id = next(iter(xgraph))
439 node = xgraph.nodes[quantum_id]
440 rcv_conn, snd_conn = multiprocessing.Pipe(False)
441 with unittest.mock.patch(
442 "lsst.pipe.base.mp_graph_executor.disable_implicit_threading"
443 ) as mock_disable:
444 _Job._executeJob(
445 pickle.dumps(qexec),
446 pickle.dumps(node["pipeline_node"]),
447 pickle.dumps(node["quantum"]),
448 quantum_id,
449 [],
450 snd_conn,
451 False,
452 )
453 mock_disable.assert_called_once_with()
454 # The job ran to completion and sent its report.
455 self.assertTrue(rcv_conn.poll())
458def setup_module(module):
459 """Force spawn to be used if no method given explicitly.
461 This can be removed when Python 3.14 changes the default.
463 Parameters
464 ----------
465 module : `~types.ModuleType`
466 Module to set up.
467 """
468 multiprocessing.set_start_method("spawn", force=True)
471if __name__ == "__main__":
472 # Do not need to force start mode when running standalone.
473 multiprocessing.set_start_method("spawn")
474 unittest.main()