Coverage for python / lsst / pipe / base / mp_graph_executor.py: 14%
384 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-09 01:22 -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# (http://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 <http://www.gnu.org/licenses/>.
28from __future__ import annotations
30__all__ = ["MPGraphExecutor", "MPGraphExecutorError", "MPTimeoutError"]
32import enum
33import importlib
34import logging
35import multiprocessing
36import pickle
37import signal
38import sys
39import threading
40import time
41import uuid
42from typing import Literal, cast
44import networkx
46from lsst.daf.butler import DataCoordinate, Quantum
47from lsst.daf.butler.cli.cliLog import CliLog
48from lsst.utils.threads import disable_implicit_threading
50from ._status import InvalidQuantumError, RepeatableQuantumError
51from .execution_graph_fixup import ExecutionGraphFixup
52from .graph import QuantumGraph
53from .graph_walker import GraphWalker
54from .pipeline_graph import TaskNode
55from .quantum_graph import PredictedQuantumGraph, PredictedQuantumInfo
56from .quantum_graph_executor import QuantumExecutor, QuantumGraphExecutor
57from .quantum_reports import ExecutionStatus, QuantumReport, Report
59_LOG = logging.getLogger(__name__)
62class JobState(enum.Enum):
63 """Possible state for an executing task."""
65 PENDING = enum.auto()
66 """The job has not started yet."""
68 RUNNING = enum.auto()
69 """The job is currently executing."""
71 FINISHED = enum.auto()
72 """The job finished successfully."""
74 FAILED = enum.auto()
75 """The job execution failed (process returned non-zero status)."""
77 TIMED_OUT = enum.auto()
78 """The job was killed due to too long execution time."""
80 FAILED_DEP = enum.auto()
81 """One of the dependencies of this job failed or timed out."""
84class _Job:
85 """Class representing a job running single task.
87 Parameters
88 ----------
89 quantum_id : `uuid.UUID`
90 ID of the quantum this job executes.
91 quantum : `lsst.daf.butler.Quantum`
92 Description of the inputs and outputs.
93 task_node : `.pipeline_graph.TaskNode`
94 Description of the task and configuration.
95 """
97 def __init__(self, quantum_id: uuid.UUID, quantum: Quantum, task_node: TaskNode):
98 self.quantum_id = quantum_id
99 self.quantum = quantum
100 self.task_node = task_node
101 self.process: multiprocessing.process.BaseProcess | None = None
102 self._state = JobState.PENDING
103 self.started: float = 0.0
104 self._rcv_conn: multiprocessing.connection.Connection | None = None
105 self._terminated = False
107 @property
108 def state(self) -> JobState:
109 """Job processing state (JobState)."""
110 return self._state
112 @property
113 def terminated(self) -> bool:
114 """Return `True` if job was killed by stop() method and negative exit
115 code is returned from child process (`bool`).
116 """
117 if self._terminated:
118 assert self.process is not None, "Process must be started"
119 if self.process.exitcode is not None:
120 return self.process.exitcode < 0
121 return False
123 def start(
124 self,
125 quantumExecutor: QuantumExecutor,
126 startMethod: Literal["spawn"] | Literal["forkserver"],
127 fail_fast: bool,
128 ) -> None:
129 """Start process which runs the task.
131 Parameters
132 ----------
133 quantumExecutor : `QuantumExecutor`
134 Executor for single quantum.
135 startMethod : `str`, optional
136 Start method from `multiprocessing` module.
137 fail_fast : `bool`, optional
138 If `True` then kill subprocess on RepeatableQuantumError.
139 """
140 # Unpickling of quantum has to happen after butler/executor, also we
141 # want to setup logging before unpickling anything that can generate
142 # messages, this is why things are pickled manually here.
143 qe_pickle = pickle.dumps(quantumExecutor)
144 task_node_pickle = pickle.dumps(self.task_node)
145 quantum_pickle = pickle.dumps(self.quantum)
146 self._rcv_conn, snd_conn = multiprocessing.Pipe(False)
147 logConfigState = CliLog.configState
149 mp_ctx = multiprocessing.get_context(startMethod)
150 self.process = mp_ctx.Process( # type: ignore[attr-defined]
151 target=_Job._executeJob,
152 args=(
153 qe_pickle,
154 task_node_pickle,
155 quantum_pickle,
156 self.quantum_id,
157 logConfigState,
158 snd_conn,
159 fail_fast,
160 ),
161 name=f"task-{self.quantum.dataId}",
162 )
163 # mypy is getting confused by multiprocessing.
164 assert self.process is not None
165 self.process.start()
166 self.started = time.time()
167 self._state = JobState.RUNNING
169 @staticmethod
170 def _executeJob(
171 quantumExecutor_pickle: bytes,
172 task_node_pickle: bytes,
173 quantum_pickle: bytes,
174 quantum_id: uuid.UUID,
175 logConfigState: list,
176 snd_conn: multiprocessing.connection.Connection,
177 fail_fast: bool,
178 ) -> None:
179 """Execute a job with arguments.
181 Parameters
182 ----------
183 quantumExecutor_pickle : `bytes`
184 Executor for single quantum, pickled.
185 task_node_pickle : `bytes`
186 Task definition structure, pickled.
187 quantum_pickle : `bytes`
188 Quantum for this task execution in pickled form.
189 quantum_id : `uuid.UUID`
190 Unique ID for the quantum.
191 logConfigState : `list`
192 Logging state from parent process.
193 snd_conn : `multiprocessing.Connection`
194 Connection to send job report to parent process.
195 fail_fast : `bool`
196 If `True` then kill subprocess on RepeatableQuantumError.
197 """
198 # This terrible hack is a workaround for Python threading bug:
199 # https://github.com/python/cpython/issues/102512. Should be removed
200 # when fix for that bug is deployed. Inspired by
201 # https://github.com/QubesOS/qubes-core-admin-client/pull/236/files.
202 thread = threading.current_thread()
203 if isinstance(thread, threading._DummyThread):
204 if getattr(thread, "_tstate_lock", "") is None:
205 thread._set_tstate_lock() # type: ignore[attr-defined]
207 if logConfigState and not CliLog.configState:
208 # means that we are in a new spawned Python process and we have to
209 # re-initialize logging
210 CliLog.replayConfigState(logConfigState)
212 quantumExecutor: QuantumExecutor = pickle.loads(quantumExecutor_pickle)
213 task_node: TaskNode = pickle.loads(task_node_pickle)
214 quantum = pickle.loads(quantum_pickle)
215 report: QuantumReport | None = None
216 # Catch a few known failure modes and stop the process immediately,
217 # with exception-specific exit code.
218 try:
219 _, report = quantumExecutor.execute(task_node, quantum, quantum_id=quantum_id)
220 except RepeatableQuantumError as exc:
221 report = QuantumReport.from_exception(
222 quantumId=quantum_id,
223 exception=exc,
224 dataId=quantum.dataId,
225 taskLabel=task_node.label,
226 exitCode=exc.EXIT_CODE if fail_fast else None,
227 )
228 if fail_fast:
229 _LOG.warning("Caught repeatable quantum error for %s (%s):", task_node.label, quantum.dataId)
230 _LOG.warning(exc, exc_info=True)
231 sys.exit(exc.EXIT_CODE)
232 else:
233 raise
234 except InvalidQuantumError as exc:
235 _LOG.fatal("Invalid quantum error for %s (%s): %s", task_node.label, quantum.dataId)
236 _LOG.fatal(exc, exc_info=True)
237 report = QuantumReport.from_exception(
238 quantumId=quantum_id,
239 exception=exc,
240 dataId=quantum.dataId,
241 taskLabel=task_node.label,
242 exitCode=exc.EXIT_CODE,
243 )
244 sys.exit(exc.EXIT_CODE)
245 except Exception as exc:
246 _LOG.debug("exception from task %s dataId %s: %s", task_node.label, quantum.dataId, exc)
247 report = QuantumReport.from_exception(
248 quantumId=quantum_id,
249 exception=exc,
250 dataId=quantum.dataId,
251 taskLabel=task_node.label,
252 )
253 raise
254 finally:
255 if report is not None:
256 # If sending fails we do not want this new exception to be
257 # exposed.
258 try:
259 _LOG.debug("sending report for task %s dataId %s", task_node.label, quantum.dataId)
260 snd_conn.send(report)
261 except Exception:
262 pass
264 def stop(self) -> None:
265 """Stop the process."""
266 assert self.process is not None, "Process must be started"
267 self.process.terminate()
268 # give it 1 second to finish or KILL
269 for _ in range(10):
270 time.sleep(0.1)
271 if not self.process.is_alive():
272 break
273 else:
274 _LOG.debug("Killing process %s", self.process.name)
275 self.process.kill()
276 self._terminated = True
278 def cleanup(self) -> None:
279 """Release processes resources, has to be called for each finished
280 process.
281 """
282 if self.process and not self.process.is_alive():
283 self.process.close()
284 self.process = None
285 self._rcv_conn = None
287 def report(self) -> QuantumReport:
288 """Return task report, should be called after process finishes and
289 before cleanup().
290 """
291 assert self.process is not None, "Process must be started"
292 assert self._rcv_conn is not None, "Process must be started"
293 try:
294 report = self._rcv_conn.recv()
295 report.exitCode = self.process.exitcode
296 except Exception:
297 # Likely due to the process killed, but there may be other reasons.
298 # Exit code should not be None, this is to keep mypy happy.
299 exitcode = self.process.exitcode if self.process.exitcode is not None else -1
300 assert self.quantum.dataId is not None, "Quantum DataId cannot be None"
301 report = QuantumReport.from_exit_code(
302 quantumId=self.quantum_id,
303 exitCode=exitcode,
304 dataId=self.quantum.dataId,
305 taskLabel=self.task_node.label,
306 )
307 if self.terminated:
308 # Means it was killed, assume it's due to timeout
309 report.status = ExecutionStatus.TIMEOUT
310 return report
312 def failMessage(self) -> str:
313 """Return a message describing task failure."""
314 assert self.process is not None, "Process must be started"
315 assert self.process.exitcode is not None, "Process has to finish"
316 exitcode = self.process.exitcode
317 if exitcode < 0:
318 # Negative exit code means it is killed by signal
319 signum = -exitcode
320 msg = f"Task {self} failed, killed by signal {signum}"
321 # Just in case this is some very odd signal, expect ValueError
322 try:
323 strsignal = signal.strsignal(signum)
324 msg = f"{msg} ({strsignal})"
325 except ValueError:
326 pass
327 elif exitcode > 0:
328 msg = f"Task {self} failed, exit code={exitcode}"
329 else:
330 msg = ""
331 return msg
333 def __str__(self) -> str:
334 return f"<{self.task_node.label} dataId={self.quantum.dataId}>"
337class _JobList:
338 """Simple list of _Job instances with few convenience methods.
340 Parameters
341 ----------
342 xgraph : `networkx.DiGraph`
343 Directed acyclic graph of quantum IDs.
344 """
346 def __init__(self, xgraph: networkx.DiGraph):
347 self.jobs = {
348 quantum_id: _Job(
349 quantum_id=quantum_id,
350 quantum=xgraph.nodes[quantum_id]["quantum"],
351 task_node=xgraph.nodes[quantum_id]["pipeline_node"],
352 )
353 for quantum_id in xgraph
354 }
355 self.walker: GraphWalker[uuid.UUID] = GraphWalker(xgraph.copy())
356 self.pending = set(next(self.walker, ()))
357 self.running: set[uuid.UUID] = set()
358 self.finished: set[uuid.UUID] = set()
359 self.failed: set[uuid.UUID] = set()
360 self.timed_out: set[uuid.UUID] = set()
362 def submit(
363 self,
364 quantumExecutor: QuantumExecutor,
365 startMethod: Literal["spawn"] | Literal["forkserver"],
366 fail_fast: bool = False,
367 ) -> _Job:
368 """Submit a pending job for execution.
370 Parameters
371 ----------
372 quantumExecutor : `QuantumExecutor`
373 Executor for single quantum.
374 startMethod : `str`, optional
375 Start method from `multiprocessing` module.
376 fail_fast : `bool`, optional
377 If `True` then kill subprocess on RepeatableQuantumError.
379 Returns
380 -------
381 job : `_Job`
382 The job that was submitted.
383 """
384 quantum_id = self.pending.pop()
385 job = self.jobs[quantum_id]
386 job.start(quantumExecutor, startMethod, fail_fast=fail_fast)
387 self.running.add(job.quantum_id)
388 return job
390 def setJobState(self, job: _Job, state: JobState) -> list[_Job]:
391 """Update job state.
393 Parameters
394 ----------
395 job : `_Job`
396 Job to submit.
397 state : `JobState`
398 New job state; note that only the FINISHED, FAILED, and TIMED_OUT
399 states are acceptable.
401 Returns
402 -------
403 blocked : `list` [ `_Job` ]
404 Additional jobs that have been marked as failed because this job
405 was upstream of them and failed or timed out.
406 """
407 allowedStates = (JobState.FINISHED, JobState.FAILED, JobState.TIMED_OUT)
408 assert state in allowedStates, f"State {state} not allowed here"
410 # remove job from pending/running lists
411 if job.state == JobState.PENDING:
412 self.pending.remove(job.quantum_id)
413 elif job.state == JobState.RUNNING:
414 self.running.remove(job.quantum_id)
416 quantum_id = job.quantum_id
417 # it should not be in any of these, but just in case
418 self.finished.discard(quantum_id)
419 self.failed.discard(quantum_id)
420 self.timed_out.discard(quantum_id)
421 job._state = state
422 match job.state:
423 case JobState.FINISHED:
424 self.finished.add(quantum_id)
425 self.walker.finish(quantum_id)
426 self.pending.update(next(self.walker, ()))
427 return []
428 case JobState.FAILED:
429 self.failed.add(quantum_id)
430 case JobState.TIMED_OUT:
431 self.failed.add(quantum_id)
432 self.timed_out.add(quantum_id)
433 case _:
434 raise ValueError(f"Unexpected state value: {state}")
435 blocked: list[_Job] = []
436 for downstream_quantum_id in self.walker.fail(quantum_id):
437 self.failed.add(downstream_quantum_id)
438 blocked.append(self.jobs[downstream_quantum_id])
439 self.jobs[downstream_quantum_id]._state = JobState.FAILED_DEP
440 return blocked
442 def cleanup(self) -> None:
443 """Do periodic cleanup for jobs that did not finish correctly.
445 If timed out jobs are killed but take too long to stop then regular
446 cleanup will not work for them. Here we check all timed out jobs
447 periodically and do cleanup if they managed to die by this time.
448 """
449 for quantum_id in self.timed_out:
450 job = self.jobs[quantum_id]
451 assert job.state == JobState.TIMED_OUT, "Job state should be consistent with the set it's in."
452 if job.process is not None:
453 job.cleanup()
456class MPGraphExecutorError(Exception):
457 """Exception class for errors raised by MPGraphExecutor."""
459 pass
462class MPTimeoutError(MPGraphExecutorError):
463 """Exception raised when task execution times out."""
465 pass
468class MPGraphExecutor(QuantumGraphExecutor):
469 """Implementation of QuantumGraphExecutor using same-host multiprocess
470 execution of Quanta.
472 Parameters
473 ----------
474 num_proc : `int`
475 Number of processes to use for executing tasks.
476 timeout : `float`
477 Time in seconds to wait for tasks to finish.
478 quantum_executor : `.quantum_graph_executor.QuantumExecutor`
479 Executor for single quantum. For multiprocess-style execution when
480 ``num_proc`` is greater than one this instance must support pickle.
481 start_method : `str`, optional
482 Start method from `multiprocessing` module, `None` selects the best
483 one for current platform.
484 fail_fast : `bool`, optional
485 If set to ``True`` then stop processing on first error from any task.
486 pdb : `str`, optional
487 Debugger to import and use (via the ``post_mortem`` function) in the
488 event of an exception.
489 execution_graph_fixup : `.execution_graph_fixup.ExecutionGraphFixup`, \
490 optional
491 Instance used for modification of execution graph.
492 """
494 def __init__(
495 self,
496 *,
497 num_proc: int,
498 timeout: float,
499 quantum_executor: QuantumExecutor,
500 start_method: Literal["spawn"] | Literal["forkserver"] | None = None,
501 fail_fast: bool = False,
502 pdb: str | None = None,
503 execution_graph_fixup: ExecutionGraphFixup | None = None,
504 ):
505 self._num_proc = num_proc
506 self._timeout = timeout
507 self._quantum_executor = quantum_executor
508 self._fail_fast = fail_fast
509 self._pdb = pdb
510 self._execution_graph_fixup = execution_graph_fixup
511 self._report: Report | None = None
513 # We set default start method as spawn for all platforms.
514 if start_method is None:
515 start_method = "spawn"
516 self._start_method = start_method
518 def execute(self, graph: QuantumGraph | PredictedQuantumGraph) -> None:
519 # Docstring inherited from QuantumGraphExecutor.execute
520 old_graph: QuantumGraph | None = None
521 if isinstance(graph, QuantumGraph):
522 old_graph = graph
523 new_graph = PredictedQuantumGraph.from_old_quantum_graph(old_graph)
524 else:
525 new_graph = graph
526 xgraph = self._make_xgraph(new_graph, old_graph)
527 self._report = Report(qgraphSummary=new_graph._make_summary())
528 try:
529 if self._num_proc > 1:
530 self._execute_quanta_mp(xgraph, self._report)
531 else:
532 self._execute_quanta_in_process(xgraph, self._report)
533 except Exception as exc:
534 self._report.set_exception(exc)
535 raise
537 def _make_xgraph(
538 self, new_graph: PredictedQuantumGraph, old_graph: QuantumGraph | None
539 ) -> networkx.DiGraph:
540 """Obtain a networkx DAG from a quantum graph, applying any fixup and
541 adding `lsst.daf.butler.Quantum` and `~.pipeline_graph.TaskNode`
542 attributes.
544 Parameters
545 ----------
546 new_graph : `.quantum_graph.PredictedQuantumGraph`
547 New quantum graph object.
548 old_graph : `.QuantumGraph` or `None`
549 Equivalent old quantum graph object.
551 Returns
552 -------
553 xgraph : `networkx.DiGraph`
554 NetworkX DAG with quantum IDs as node keys.
556 Raises
557 ------
558 MPGraphExecutorError
559 Raised if execution graph cannot be ordered after modification,
560 i.e. it has dependency cycles.
561 """
562 new_graph.build_execution_quanta()
563 xgraph = new_graph.quantum_only_xgraph.copy()
564 if self._execution_graph_fixup:
565 try:
566 self._execution_graph_fixup.fixup_graph(xgraph, new_graph.quanta_by_task)
567 except NotImplementedError:
568 # Backwards compatibility.
569 if old_graph is None:
570 old_graph = new_graph.to_old_quantum_graph()
571 old_graph = self._execution_graph_fixup.fixupQuanta(old_graph)
572 # Adding all of the edges from old_graph is overkill, but the
573 # only option we really have to make sure we add any new ones.
574 xgraph.update([(a.nodeId, b.nodeId) for a, b in old_graph.graph.edges])
575 if networkx.dag.has_cycle(xgraph):
576 raise MPGraphExecutorError("Updated execution graph has dependency cycle.")
577 return xgraph
579 def _execute_quanta_in_process(self, xgraph: networkx.DiGraph, report: Report) -> None:
580 """Execute all Quanta in current process.
582 Parameters
583 ----------
584 xgraph : `networkx.DiGraph`
585 DAG to execute. Should have quantum IDs for nodes and ``quantum``
586 (`lsst.daf.butler.Quantum`) and ``pipeline_node``
587 (`lsst.pipe.base.pipeline_graph.TaskNode`) attributes in addition
588 to those provided by
589 `.quantum_graph.PredictedQuantumGraph.quantum_only_xgraph`.
590 report : `Report`
591 Object for reporting execution status.
592 """
594 def tiebreaker_sort_key(quantum_id: uuid.UUID) -> tuple:
595 node_state = xgraph.nodes[quantum_id]
596 return (node_state["task_label"],) + node_state["data_id"].required_values
598 success_count, failed_count, total_count = 0, 0, len(xgraph.nodes)
599 walker = GraphWalker[uuid.UUID](xgraph.copy())
600 for unblocked_quanta in walker:
601 for quantum_id in sorted(unblocked_quanta, key=tiebreaker_sort_key):
602 node_state: PredictedQuantumInfo = xgraph.nodes[quantum_id]
603 data_id = node_state["data_id"]
604 task_node = node_state["pipeline_node"]
605 quantum = node_state["quantum"]
607 _LOG.debug("Executing %s (%s@%s)", quantum_id, task_node.label, data_id)
608 fail_exit_code: int | None = None
609 try:
610 # For some exception types we want to exit immediately with
611 # exception-specific exit code, but we still want to start
612 # debugger before exiting if debugging is enabled.
613 try:
614 _, quantum_report = self._quantum_executor.execute(
615 task_node, quantum, quantum_id=quantum_id
616 )
617 if quantum_report:
618 report.quantaReports.append(quantum_report)
619 success_count += 1
620 walker.finish(quantum_id)
621 except RepeatableQuantumError as exc:
622 if self._fail_fast:
623 _LOG.warning(
624 "Caught repeatable quantum error for %s (%s@%s):",
625 quantum_id,
626 task_node.label,
627 data_id,
628 )
629 _LOG.warning(exc, exc_info=True)
630 fail_exit_code = exc.EXIT_CODE
631 raise
632 except InvalidQuantumError as exc:
633 _LOG.fatal(
634 "Invalid quantum error for %s (%s@%s):", quantum_id, task_node.label, data_id
635 )
636 _LOG.fatal(exc, exc_info=True)
637 fail_exit_code = exc.EXIT_CODE
638 raise
639 except Exception as exc:
640 quantum_report = QuantumReport.from_exception(
641 exception=exc,
642 dataId=data_id,
643 taskLabel=task_node.label,
644 )
645 report.quantaReports.append(quantum_report)
647 if self._pdb and sys.stdin.isatty() and sys.stdout.isatty():
648 _LOG.error(
649 "%s (%s@%s) failed; dropping into pdb.",
650 quantum_id,
651 task_node.label,
652 data_id,
653 exc_info=exc,
654 )
655 try:
656 pdb = importlib.import_module(self._pdb)
657 except ImportError as imp_exc:
658 raise MPGraphExecutorError(
659 f"Unable to import specified debugger module ({self._pdb}): {imp_exc}"
660 ) from exc
661 if not hasattr(pdb, "post_mortem"):
662 raise MPGraphExecutorError(
663 f"Specified debugger module ({self._pdb}) can't debug with post_mortem",
664 ) from exc
665 pdb.post_mortem(exc.__traceback__)
667 report.status = ExecutionStatus.FAILURE
668 failed_count += 1
670 # If exception specified an exit code then just exit with
671 # that code, otherwise crash if fail-fast option is
672 # enabled.
673 if fail_exit_code is not None:
674 sys.exit(fail_exit_code)
675 if self._fail_fast:
676 raise MPGraphExecutorError(
677 f"Quantum {quantum_id} ({task_node.label}@{data_id}) failed."
678 ) from exc
679 else:
680 _LOG.error(
681 "%s (%s@%s) failed; processing will continue for remaining tasks.",
682 quantum_id,
683 task_node.label,
684 data_id,
685 exc_info=exc,
686 )
688 for downstream_quantum_id in walker.fail(quantum_id):
689 downstream_node_state = xgraph.nodes[downstream_quantum_id]
690 failed_quantum_report = QuantumReport(
691 status=ExecutionStatus.SKIPPED,
692 dataId=downstream_node_state["data_id"],
693 taskLabel=downstream_node_state["task_label"],
694 )
695 report.quantaReports.append(failed_quantum_report)
696 _LOG.error(
697 "Upstream job failed for task %s (%s@%s), skipping this quantum.",
698 downstream_quantum_id,
699 downstream_node_state["task_label"],
700 downstream_node_state["data_id"],
701 )
702 failed_count += 1
704 _LOG.info(
705 "Executed %d quanta successfully, %d failed and %d remain out of total %d quanta.",
706 success_count,
707 failed_count,
708 total_count - success_count - failed_count,
709 total_count,
710 )
712 # Raise an exception if there were any failures.
713 if failed_count:
714 raise MPGraphExecutorError("One or more tasks failed during execution.")
716 def _execute_quanta_mp(self, xgraph: networkx.DiGraph, report: Report) -> None:
717 """Execute all Quanta in separate processes.
719 Parameters
720 ----------
721 xgraph : `networkx.DiGraph`
722 DAG to execute. Should have quantum IDs for nodes and ``quantum``
723 (`lsst.daf.butler.Quantum`) and ``task_node``
724 (`lsst.pipe.base.pipeline_graph.TaskNode`) attributes in addition
725 to those provided by
726 `.quantum_graph.PredictedQuantumGraph.quantum_only_xgraph`.
727 report : `Report`
728 Object for reporting execution status.
729 """
730 disable_implicit_threading() # To prevent thread contention
732 _LOG.debug("Using %r for multiprocessing start method", self._start_method)
734 # re-pack input quantum data into jobs list
735 jobs = _JobList(xgraph)
737 # check that all tasks can run in sub-process
738 for job in jobs.jobs.values():
739 if not job.task_node.task_class.canMultiprocess:
740 raise MPGraphExecutorError(
741 f"Task {job.task_node.label!r} does not support multiprocessing; use single process"
742 )
744 finishedCount, failedCount = 0, 0
745 while jobs.pending or jobs.running:
746 _LOG.debug("#pendingJobs: %s", len(jobs.pending))
747 _LOG.debug("#runningJobs: %s", len(jobs.running))
749 # See if any jobs have finished
750 for quantum_id in list(jobs.running): # iterate over a copy so we can remove.
751 job = jobs.jobs[quantum_id]
752 assert job.process is not None, "Process cannot be None"
753 blocked: list[_Job] = []
754 if not job.process.is_alive():
755 _LOG.debug("finished: %s", job)
756 # finished
757 exitcode = job.process.exitcode
758 quantum_report = job.report()
759 report.quantaReports.append(quantum_report)
760 if exitcode == 0:
761 jobs.setJobState(job, JobState.FINISHED)
762 job.cleanup()
763 _LOG.debug("success: %s took %.3f seconds", job, time.time() - job.started)
764 else:
765 if job.terminated:
766 # Was killed due to timeout.
767 if report.status == ExecutionStatus.SUCCESS:
768 # Do not override global FAILURE status
769 report.status = ExecutionStatus.TIMEOUT
770 message = f"Timeout ({self._timeout} sec) for task {job}, task is killed"
771 blocked = jobs.setJobState(job, JobState.TIMED_OUT)
772 else:
773 report.status = ExecutionStatus.FAILURE
774 # failMessage() has to be called before cleanup()
775 message = job.failMessage()
776 blocked = jobs.setJobState(job, JobState.FAILED)
778 job.cleanup()
779 _LOG.debug("failed: %s", job)
780 if self._fail_fast or exitcode == InvalidQuantumError.EXIT_CODE:
781 # stop all running jobs
782 for stop_quantum_id in jobs.running:
783 stop_job = jobs.jobs[stop_quantum_id]
784 if stop_job is not job:
785 stop_job.stop()
786 if job.state is JobState.TIMED_OUT:
787 raise MPTimeoutError(f"Timeout ({self._timeout} sec) for task {job}.")
788 else:
789 raise MPGraphExecutorError(message)
790 else:
791 _LOG.error("%s; processing will continue for remaining tasks.", message)
792 else:
793 # check for timeout
794 now = time.time()
795 if now - job.started > self._timeout:
796 # Try to kill it, and there is a chance that it
797 # finishes successfully before it gets killed. Exit
798 # status is handled by the code above on next
799 # iteration.
800 _LOG.debug("Terminating job %s due to timeout", job)
801 job.stop()
803 for downstream_job in blocked:
804 quantum_report = QuantumReport(
805 quantumId=downstream_job.quantum_id,
806 status=ExecutionStatus.SKIPPED,
807 dataId=cast(DataCoordinate, downstream_job.quantum.dataId),
808 taskLabel=downstream_job.task_node.label,
809 )
810 report.quantaReports.append(quantum_report)
811 _LOG.error("Upstream job failed for task %s, skipping this task.", downstream_job)
813 # see if we can start more jobs
814 while len(jobs.running) < self._num_proc and jobs.pending:
815 job = jobs.submit(self._quantum_executor, self._start_method)
816 _LOG.debug("Submitted %s", job)
818 # Do cleanup for timed out jobs if necessary.
819 jobs.cleanup()
821 # Print progress message if something changed.
822 newFinished, newFailed = len(jobs.finished), len(jobs.failed)
823 if (finishedCount, failedCount) != (newFinished, newFailed):
824 finishedCount, failedCount = newFinished, newFailed
825 totalCount = len(jobs.jobs)
826 _LOG.info(
827 "Executed %d quanta successfully, %d failed and %d remain out of total %d quanta.",
828 finishedCount,
829 failedCount,
830 totalCount - finishedCount - failedCount,
831 totalCount,
832 )
834 # Here we want to wait until one of the running jobs completes
835 # but multiprocessing does not provide an API for that, for now
836 # just sleep a little bit and go back to the loop.
837 if jobs.running:
838 time.sleep(0.1)
840 if jobs.failed:
841 # print list of failed jobs
842 _LOG.error("Failed jobs:")
843 for quantum_id in jobs.failed:
844 job = jobs.jobs[quantum_id]
845 _LOG.error(" - %s: %s", job.state.name, job)
847 # if any job failed raise an exception
848 if jobs.failed == jobs.timed_out:
849 raise MPTimeoutError("One or more tasks timed out during execution.")
850 else:
851 raise MPGraphExecutorError("One or more tasks failed or timed out during execution.")
853 def getReport(self) -> Report:
854 # Docstring inherited from base class
855 if self._report is None:
856 raise RuntimeError("getReport() called before execute()")
857 return self._report