Coverage for tests/test_mp_graph_executor.py: 14%
251 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-11-13 01:39 -0800
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 signal
35import sys
36import unittest
37import warnings
38from typing import Literal
40import psutil
42from lsst.pipe.base.exec_fixup_data_id import ExecFixupDataId
43from lsst.pipe.base.mp_graph_executor import MPGraphExecutor, MPGraphExecutorError, MPTimeoutError
44from lsst.pipe.base.quantum_reports import ExecutionStatus, Report
45from lsst.pipe.base.tests.mocks import (
46 DynamicConnectionConfig,
47 DynamicTestPipelineTask,
48 DynamicTestPipelineTaskConfig,
49 InMemoryRepo,
50)
52logging.basicConfig(level=logging.DEBUG)
54_LOG = logging.getLogger(__name__)
56TESTDIR = os.path.abspath(os.path.dirname(__file__))
59class NoMultiprocessingTask(DynamicTestPipelineTask):
60 """A test pipeline task that declares that it cannot be used in
61 multiprocessing.
62 """
64 canMultiprocess = False
67def _count_status(report: Report, status: ExecutionStatus) -> int:
68 """Count number of quanta with a given status."""
69 return len([qrep for qrep in report.quantaReports if qrep.status is status])
72class MPGraphExecutorTestCase(unittest.TestCase):
73 """A test case for MPGraphExecutor class."""
75 def test_mpexec_nomp(self) -> None:
76 """Make simple graph and execute."""
77 helper = InMemoryRepo("base.yaml")
78 helper.add_task(dimensions=["detector"])
79 qgraph = helper.make_quantum_graph()
80 qexec, butler = helper.make_single_quantum_executor()
81 # run in single-process mode
82 mpexec = MPGraphExecutor(num_proc=1, timeout=100, quantum_executor=qexec)
83 mpexec.execute(qgraph) # type: ignore[arg-type]
84 self.assertCountEqual(
85 [ref.dataId["detector"] for ref in butler.get_datasets("dataset_auto1")], [1, 2, 3, 4]
86 )
87 report = mpexec.getReport()
88 assert report is not None
89 self.assertEqual(report.status, ExecutionStatus.SUCCESS)
90 self.assertIsNone(report.exitCode)
91 self.assertIsNone(report.exceptionInfo)
92 self.assertEqual(len(report.quantaReports), 4)
93 self.assertTrue(all(qrep.status == ExecutionStatus.SUCCESS for qrep in report.quantaReports))
94 self.assertTrue(all(qrep.exitCode is None for qrep in report.quantaReports))
95 self.assertTrue(all(qrep.exceptionInfo is None for qrep in report.quantaReports))
96 self.assertTrue(all(qrep.taskLabel == "task_auto1" for qrep in report.quantaReports))
98 def test_mpexec_mp(self) -> None:
99 """Make simple graph and execute."""
100 helper = InMemoryRepo("base.yaml")
101 helper.add_task(dimensions=["detector"])
102 qg = helper.make_quantum_graph()
103 qexec, butler = helper.make_single_quantum_executor()
105 methods: list[Literal["spawn", "forkserver"]] = ["spawn"]
106 if sys.platform == "linux":
107 methods.append("forkserver")
109 for method in methods:
110 with self.subTest(startMethod=method):
111 # Run in multi-process mode, the order of results is not
112 # defined.
113 mpexec = MPGraphExecutor(num_proc=3, timeout=100, quantum_executor=qexec, start_method=method)
114 mpexec.execute(qg) # type: ignore[arg-type]
115 report = mpexec.getReport()
116 assert report is not None
117 self.assertEqual(report.status, ExecutionStatus.SUCCESS)
118 self.assertIsNone(report.exitCode)
119 self.assertIsNone(report.exceptionInfo)
120 self.assertEqual(len(report.quantaReports), 4)
121 self.assertTrue(all(qrep.status == ExecutionStatus.SUCCESS for qrep in report.quantaReports))
122 self.assertTrue(all(qrep.exitCode == 0 for qrep in report.quantaReports))
123 self.assertTrue(all(qrep.exceptionInfo is None for qrep in report.quantaReports))
124 self.assertTrue(all(qrep.taskLabel == "task_auto1" for qrep in report.quantaReports))
126 def test_mpexec_nompsupport(self) -> None:
127 """Try to run MP for task that has no MP support which should fail."""
128 helper = InMemoryRepo("base.yaml")
129 helper.add_task(task_class=NoMultiprocessingTask, dimensions=["detector"])
130 qg = helper.make_quantum_graph()
131 qexec, butler = helper.make_single_quantum_executor()
132 mpexec = MPGraphExecutor(num_proc=3, timeout=100, quantum_executor=qexec)
133 with self.assertRaisesRegex(
134 MPGraphExecutorError, "Task 'task_auto1' does not support multiprocessing"
135 ):
136 mpexec.execute(qg) # type: ignore[arg-type]
138 def test_mpexec_fixup(self) -> None:
139 """Make simple graph and execute, add dependencies by executing fixup
140 code.
141 """
142 helper = InMemoryRepo("base.yaml")
143 helper.add_task(dimensions=["detector"])
144 qg = helper.make_quantum_graph()
145 for reverse in (False, True):
146 qexec, butler = helper.make_single_quantum_executor()
147 fixup = ExecFixupDataId("task_auto1", "detector", reverse=reverse)
148 mpexec = MPGraphExecutor(
149 num_proc=1, timeout=100, quantum_executor=qexec, execution_graph_fixup=fixup
150 )
151 mpexec.execute(qg) # type: ignore[arg-type]
152 expected = [1, 2, 3, 4]
153 if reverse:
154 expected = list(reversed(expected))
155 self.assertEqual(
156 [ref.dataId["detector"] for ref in butler.get_datasets("dataset_auto1")], expected
157 )
159 def test_mpexec_fixup_old_qg(self) -> None:
160 """Test using an old QuantumGraph object to initialize the executor,
161 with an ordering fixup.
162 """
163 helper = InMemoryRepo("base.yaml")
164 helper.add_task(dimensions=["detector"])
165 qgraph = helper.make_quantum_graph_builder().build(attach_datastore_records=False)
166 for reverse in (False, True):
167 qexec, butler = helper.make_single_quantum_executor()
168 fixup = ExecFixupDataId("task_auto1", "detector", reverse=reverse)
169 mpexec = MPGraphExecutor(
170 num_proc=1, timeout=100, quantum_executor=qexec, execution_graph_fixup=fixup
171 )
172 mpexec.execute(qgraph) # type: ignore[arg-type]
173 expected = [1, 2, 3, 4]
174 if reverse:
175 expected = list(reversed(expected))
176 self.assertEqual(
177 [ref.dataId["detector"] for ref in butler.get_datasets("dataset_auto1")], expected
178 )
180 def test_mpexec_timeout(self) -> None:
181 """Fail due to timeout."""
182 helper = InMemoryRepo("base.yaml")
183 helper.add_task(label="a")
184 helper.add_task(
185 label="b",
186 inputs={"input_connection": DynamicConnectionConfig(dataset_type_name="dataset_auto0")},
187 )
188 helper.add_task(
189 label="c",
190 inputs={"input_connection": DynamicConnectionConfig(dataset_type_name="dataset_auto0")},
191 config=DynamicTestPipelineTaskConfig(sleep=100.0),
192 )
193 qg = helper.make_quantum_graph()
195 # with failFast we'll get immediate MPTimeoutError
196 qexec, _ = helper.make_single_quantum_executor()
197 mpexec = MPGraphExecutor(num_proc=3, timeout=1, quantum_executor=qexec, fail_fast=True)
198 with self.assertRaises(MPTimeoutError):
199 mpexec.execute(qg) # type: ignore[arg-type]
200 report = mpexec.getReport()
201 assert report is not None and report.exceptionInfo is not None
202 self.assertEqual(report.status, ExecutionStatus.TIMEOUT)
203 self.assertEqual(report.exceptionInfo.className, "lsst.pipe.base.mp_graph_executor.MPTimeoutError")
204 self.assertGreater(len(report.quantaReports), 0)
205 self.assertEqual(_count_status(report, ExecutionStatus.TIMEOUT), 1)
206 self.assertTrue(any(qrep.exitCode is not None and qrep.exitCode < 0 for qrep in report.quantaReports))
207 self.assertTrue(all(qrep.exceptionInfo is None for qrep in report.quantaReports))
209 # with failFast=False exception happens after last task finishes
210 qexec, _ = helper.make_single_quantum_executor()
211 mpexec = MPGraphExecutor(num_proc=3, timeout=3, quantum_executor=qexec, fail_fast=False)
212 with self.assertRaises(MPTimeoutError):
213 mpexec.execute(qg) # type: ignore[arg-type]
214 assert report is not None and report.exceptionInfo is not None
215 self.assertEqual(report.status, ExecutionStatus.TIMEOUT)
216 self.assertEqual(report.exceptionInfo.className, "lsst.pipe.base.mp_graph_executor.MPTimeoutError")
217 # We expect two tasks ('a' and 'b') to finish successfully and one task
218 # ('c') to timeout, which should get us all three reports.
219 # Unfortunately on busy CPU there is no guarantee that tasks finish on
220 # time, so expect more timeouts and issue a warning.
221 if len(report.quantaReports) != 3:
222 warnings.warn(
223 f"Possibly timed out tasks, expected three reports, received {len(report.quantaReports)})."
224 )
225 report = mpexec.getReport()
226 self.assertGreater(_count_status(report, ExecutionStatus.TIMEOUT), 0)
227 self.assertTrue(any(qrep.exitCode is not None and qrep.exitCode < 0 for qrep in report.quantaReports))
228 self.assertTrue(all(qrep.exceptionInfo is None for qrep in report.quantaReports))
230 def test_mpexec_failure(self) -> None:
231 """Failure in one task should not stop other tasks."""
232 helper = InMemoryRepo("base.yaml")
233 helper.add_task(
234 config=DynamicTestPipelineTaskConfig(fail_condition="detector=2"),
235 dimensions=["detector"],
236 )
237 qg = helper.make_quantum_graph()
238 qexec, _ = helper.make_single_quantum_executor()
239 mpexec = MPGraphExecutor(num_proc=3, timeout=100, quantum_executor=qexec)
240 with self.assertRaisesRegex(MPGraphExecutorError, "One or more tasks failed"):
241 mpexec.execute(qg) # type: ignore[arg-type]
242 report = mpexec.getReport()
243 assert report is not None and report.exceptionInfo is not None
244 self.assertEqual(report.status, ExecutionStatus.FAILURE)
245 self.assertEqual(
246 report.exceptionInfo.className, "lsst.pipe.base.mp_graph_executor.MPGraphExecutorError"
247 )
248 self.assertGreater(len(report.quantaReports), 0)
249 self.assertEqual(_count_status(report, ExecutionStatus.FAILURE), 1)
250 self.assertEqual(_count_status(report, ExecutionStatus.SUCCESS), 3)
251 self.assertTrue(any(qrep.exitCode is not None and qrep.exitCode > 0 for qrep in report.quantaReports))
252 self.assertTrue(any(qrep.exceptionInfo is not None for qrep in report.quantaReports))
254 def test_mpexec_failure_dep(self) -> None:
255 """Failure in one task should skip dependents."""
256 helper = InMemoryRepo("base.yaml")
257 helper.add_task(
258 "a", config=DynamicTestPipelineTaskConfig(fail_condition="detector=2"), dimensions=["detector"]
259 )
260 helper.add_task("b", dimensions=["detector"]) # depends on 'a', for the same detector.
261 qg = helper.make_quantum_graph()
262 qexec, _ = helper.make_single_quantum_executor()
263 mpexec = MPGraphExecutor(num_proc=3, timeout=100, quantum_executor=qexec)
264 with self.assertRaisesRegex(MPGraphExecutorError, "One or more tasks failed"):
265 mpexec.execute(qg) # type: ignore[arg-type]
266 report = mpexec.getReport()
267 assert report is not None and report.exceptionInfo is not None
268 self.assertEqual(report.status, ExecutionStatus.FAILURE)
269 self.assertEqual(
270 report.exceptionInfo.className, "lsst.pipe.base.mp_graph_executor.MPGraphExecutorError"
271 )
272 # Dependencies of failed tasks do not appear in quantaReports
273 self.assertGreater(len(report.quantaReports), 0)
274 self.assertEqual(_count_status(report, ExecutionStatus.FAILURE), 1)
275 self.assertEqual(_count_status(report, ExecutionStatus.SUCCESS), 6)
276 self.assertEqual(_count_status(report, ExecutionStatus.SKIPPED), 1)
277 self.assertTrue(any(qrep.exitCode is not None and qrep.exitCode > 0 for qrep in report.quantaReports))
278 self.assertTrue(any(qrep.exceptionInfo is not None for qrep in report.quantaReports))
280 def test_mpexec_failure_dep_nomp(self) -> None:
281 """Failure in one task should skip dependents, in-process version."""
282 helper = InMemoryRepo("base.yaml")
283 helper.add_task(
284 "a", config=DynamicTestPipelineTaskConfig(fail_condition="detector=2"), dimensions=["detector"]
285 )
286 helper.add_task("b", dimensions=["detector"]) # depends on 'a', for the same detector.
287 qg = helper.make_quantum_graph()
288 qexec, butler = helper.make_single_quantum_executor()
289 mpexec = MPGraphExecutor(num_proc=1, timeout=100, quantum_executor=qexec)
290 with self.assertRaisesRegex(MPGraphExecutorError, "One or more tasks failed"):
291 mpexec.execute(qg) # type: ignore[arg-type]
292 self.assertCountEqual(
293 [ref.dataId["detector"] for ref in butler.get_datasets("dataset_auto1")], [1, 3, 4]
294 )
295 self.assertCountEqual(
296 [ref.dataId["detector"] for ref in butler.get_datasets("dataset_auto2")], [1, 3, 4]
297 )
298 report = mpexec.getReport()
299 assert report is not None and report.exceptionInfo is not None
300 self.assertEqual(report.status, ExecutionStatus.FAILURE)
301 self.assertEqual(
302 report.exceptionInfo.className, "lsst.pipe.base.mp_graph_executor.MPGraphExecutorError"
303 )
304 # Dependencies of failed tasks do not appear in quantaReports
305 self.assertGreater(len(report.quantaReports), 0)
306 self.assertEqual(_count_status(report, ExecutionStatus.FAILURE), 1)
307 self.assertEqual(_count_status(report, ExecutionStatus.SUCCESS), 6)
308 self.assertEqual(_count_status(report, ExecutionStatus.SKIPPED), 1)
309 self.assertTrue(all(qrep.exitCode is None for qrep in report.quantaReports))
310 self.assertTrue(any(qrep.exceptionInfo is not None for qrep in report.quantaReports))
312 def test_mpexec_failure_failfast(self) -> None:
313 """Fast fail stops quickly.
315 Timing delay of task 'b' should be sufficient to process
316 failure and raise exception before task 'c'.
317 """
318 helper = InMemoryRepo("base.yaml")
319 helper.add_task(
320 "a", config=DynamicTestPipelineTaskConfig(fail_condition="detector=2"), dimensions=["detector"]
321 )
322 helper.add_task("b", config=DynamicTestPipelineTaskConfig(sleep=100.0), dimensions=["detector"])
323 helper.add_task("c", dimensions=["detector"]) # depends on 'b', for the same detector.
324 qg = helper.make_quantum_graph()
325 qexec, _ = helper.make_single_quantum_executor()
326 mpexec = MPGraphExecutor(num_proc=3, timeout=100, quantum_executor=qexec, fail_fast=True)
327 with self.assertRaisesRegex(MPGraphExecutorError, "failed, exit code=1"):
328 mpexec.execute(qg) # type: ignore[arg-type]
329 report = mpexec.getReport()
330 assert report is not None and report.exceptionInfo is not None
331 self.assertEqual(report.status, ExecutionStatus.FAILURE)
332 self.assertEqual(
333 report.exceptionInfo.className, "lsst.pipe.base.mp_graph_executor.MPGraphExecutorError"
334 )
335 # Dependencies of failed tasks do not appear in quantaReports
336 self.assertGreater(len(report.quantaReports), 0)
337 self.assertEqual(_count_status(report, ExecutionStatus.FAILURE), 1)
338 self.assertTrue(any(qrep.exitCode is not None and qrep.exitCode > 0 for qrep in report.quantaReports))
339 self.assertTrue(any(qrep.exceptionInfo is not None for qrep in report.quantaReports))
341 def test_mpexec_crash(self) -> None:
342 """Check task crash due to signal."""
343 helper = InMemoryRepo("base.yaml")
344 helper.add_task(
345 config=DynamicTestPipelineTaskConfig(fail_condition="detector=2", fail_signal=signal.SIGILL),
346 dimensions=["detector"],
347 )
348 qg = helper.make_quantum_graph()
349 qexec, _ = helper.make_single_quantum_executor()
350 mpexec = MPGraphExecutor(num_proc=3, timeout=100, quantum_executor=qexec)
351 with self.assertRaisesRegex(MPGraphExecutorError, "One or more tasks failed"):
352 mpexec.execute(qg) # type: ignore[arg-type]
353 report = mpexec.getReport()
354 assert report is not None and report.exceptionInfo is not None
355 self.assertEqual(report.status, ExecutionStatus.FAILURE)
356 self.assertEqual(
357 report.exceptionInfo.className, "lsst.pipe.base.mp_graph_executor.MPGraphExecutorError"
358 )
359 # Dependencies of failed tasks do not appear in quantaReports
360 self.assertGreater(len(report.quantaReports), 0)
361 self.assertEqual(_count_status(report, ExecutionStatus.FAILURE), 1)
362 self.assertEqual(_count_status(report, ExecutionStatus.SUCCESS), 3)
363 self.assertTrue(any(qrep.exitCode == -signal.SIGILL for qrep in report.quantaReports))
364 self.assertTrue(all(qrep.exceptionInfo is None for qrep in report.quantaReports))
366 def test_mpexec_crash_failfast(self) -> None:
367 """Check task crash due to signal with --fail-fast."""
368 helper = InMemoryRepo("base.yaml")
369 helper.add_task(
370 "a",
371 config=DynamicTestPipelineTaskConfig(fail_condition="detector=2", fail_signal=signal.SIGILL),
372 dimensions=["detector"],
373 )
374 helper.add_task("b", config=DynamicTestPipelineTaskConfig(sleep=100.0), dimensions=["detector"])
375 helper.add_task("c", dimensions=["detector"]) # depends on 'b', for the same detector.
376 qg = helper.make_quantum_graph()
377 qexec, _ = helper.make_single_quantum_executor()
378 mpexec = MPGraphExecutor(num_proc=3, timeout=100, quantum_executor=qexec, fail_fast=True)
379 with self.assertRaisesRegex(MPGraphExecutorError, "failed, killed by signal 4 .Illegal instruction"):
380 mpexec.execute(qg) # type: ignore[arg-type]
381 report = mpexec.getReport()
382 assert report is not None and report.exceptionInfo is not None
383 self.assertEqual(report.status, ExecutionStatus.FAILURE)
384 self.assertEqual(
385 report.exceptionInfo.className, "lsst.pipe.base.mp_graph_executor.MPGraphExecutorError"
386 )
387 self.assertEqual(_count_status(report, ExecutionStatus.FAILURE), 1)
388 self.assertTrue(any(qrep.exitCode == -signal.SIGILL for qrep in report.quantaReports))
389 self.assertTrue(all(qrep.exceptionInfo is None for qrep in report.quantaReports))
391 def test_mpexec_num_fd(self) -> None:
392 """Check that number of open files stays reasonable."""
393 helper = InMemoryRepo("base.yaml")
394 helper.add_task("a", task_class=NoMultiprocessingTask, dimensions=["detector", "visit"])
395 helper.add_task("b", task_class=NoMultiprocessingTask, dimensions=["detector", "visit"])
396 qg = helper.make_quantum_graph()
397 qexec, _ = helper.make_single_quantum_executor()
398 this_proc = psutil.Process()
399 num_fds_0 = this_proc.num_fds()
401 # run in multi-process mode, the order of results is not defined
402 mpexec = MPGraphExecutor(num_proc=3, timeout=100, quantum_executor=qexec)
403 mpexec.execute(qg) # type: ignore[arg-type]
405 num_fds_1 = this_proc.num_fds()
406 # They should be the same but allow small growth just in case.
407 # Without DM-26728 fix the difference would be equal to number of
408 # quanta (20).
409 self.assertLess(num_fds_1 - num_fds_0, 5)
412def setup_module(module):
413 """Force spawn to be used if no method given explicitly.
415 This can be removed when Python 3.14 changes the default.
417 Parameters
418 ----------
419 module : `~types.ModuleType`
420 Module to set up.
421 """
422 multiprocessing.set_start_method("spawn", force=True)
425if __name__ == "__main__":
426 # Do not need to force start mode when running standalone.
427 multiprocessing.set_start_method("spawn")
428 unittest.main()