Coverage for python/lsst/ctrl/mpexec/cli/script/run.py: 15%
87 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-11-13 01:41 -0800
1# This file is part of ctrl_mpexec.
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
30from collections.abc import Iterable
31from typing import TYPE_CHECKING, Literal
33import astropy.units as u
35import lsst.utils.timer
36from lsst.pipe.base import ExecutionResources, TaskFactory
37from lsst.pipe.base.mp_graph_executor import MPGraphExecutor
38from lsst.pipe.base.quantum_graph import PredictedQuantumGraph
39from lsst.pipe.base.single_quantum_executor import SingleQuantumExecutor
40from lsst.resources import ResourcePath, ResourcePathExpression
41from lsst.utils.doImport import doImportType
42from lsst.utils.iteration import ensure_iterable
43from lsst.utils.logging import getLogger
44from lsst.utils.threads import disable_implicit_threading
46from ..butler_factory import ButlerFactory
47from ..utils import MP_TIMEOUT
49if TYPE_CHECKING:
50 from lsst.pipe.base.execution_graph_fixup import ExecutionGraphFixup
52_LOG = getLogger(__name__)
55def run(
56 qg: PredictedQuantumGraph,
57 *,
58 task_factory: TaskFactory | None = None,
59 pdb: str | None,
60 graph_fixup: str,
61 init_only: bool,
62 no_versions: bool,
63 processes: int,
64 start_method: Literal["spawn", "forkserver"] | None,
65 profile: str,
66 register_dataset_types: bool,
67 skip_init_writes: bool,
68 timeout: int | None,
69 butler_config: ResourcePathExpression,
70 input: Iterable[str] | str,
71 output: str | None,
72 output_run: str | None,
73 extend_run: bool,
74 replace_run: bool,
75 prune_replaced: str | None,
76 data_query: str | None,
77 skip_existing_in: Iterable[str] | None,
78 skip_existing: bool,
79 debug: bool,
80 fail_fast: bool,
81 clobber_outputs: bool,
82 summary: ResourcePathExpression | None,
83 enable_implicit_threading: bool,
84 cores_per_quantum: int,
85 memory_per_quantum: str | None,
86 rebase: bool,
87 raise_on_partial_outputs: bool,
88 **kwargs: object,
89) -> None:
90 """Implement the command line interface `pipetask run` subcommand.
92 Should only be called by command line tools and unit test code that test
93 this function.
95 Parameters
96 ----------
97 qg : `lsst.pipe.base.quantum_graph.PredictedQuantumGraph`
98 A quantum graph generated by a previous subcommand.
99 task_factory : `lsst.pipe.base.TaskFactory`, optional
100 A custom task factory to use.
101 pdb : `str`, optional
102 Debugger to import and use (via the ``post_mortem`` function) in the
103 event of an exception.
104 graph_fixup : `str`
105 The name of the class or factory method which makes an instance used
106 for execution graph fixup.
107 init_only : `bool`
108 If true, do not actually run; just register dataset types and/or save
109 init outputs.
110 no_versions : `bool`
111 If true, do not save or check package versions.
112 processes : `int`
113 The number of processes to use.
114 start_method : `str` or `None`
115 Start method from `multiprocessing` module, `None` selects the best
116 one for current platform.
117 profile : `str`
118 File name to dump cProfile information to.
119 register_dataset_types : `bool`
120 If true, register DatasetTypes that do not already exist in the
121 Registry.
122 skip_init_writes : `bool`
123 If true, do not write collection-wide 'init output' datasets (e.g.
124 schemas).
125 timeout : `int`
126 Timeout for multiprocessing; maximum wall time (sec).
127 butler_config : convertible to `lsst.resources.ResourcePath`
128 Path to butler repository configuration.
129 input : `~collections.abc.Iterable` [ `str` ] or `None`
130 List of names of the input collection(s).
131 output : `str` or `None`
132 Name of the output CHAINED collection. This may either be an existing
133 CHAINED collection to use as both input and output (if `input` is
134 `None`), or a new CHAINED collection created to include all inputs
135 (if `input` is not `None`). In both cases, the collection's children
136 will start with an output RUN collection that directly holds all new
137 datasets (see `output_run`).
138 output_run : `str` or `None`
139 Name of the new output RUN collection. If not provided then `output`
140 must be provided and a new RUN collection will be created by appending
141 a timestamp to the value passed with `output`. If this collection
142 already exists then `extend_run` must be passed.
143 extend_run : `bool`
144 Instead of creating a new RUN collection, insert datasets into either
145 the one given by `output_run` (if provided) or the first child
146 collection of `output` (which must be of type RUN).
147 replace_run : `bool`
148 Before creating a new RUN collection in an existing CHAINED collection,
149 remove the first child collection (which must be of type RUN). This can
150 be used to repeatedly write to the same (parent) collection during
151 development, but it does not delete the datasets associated with the
152 replaced run unless `prune-replaced` is also True. Requires `output`,
153 and `extend_run` must be `None`.
154 prune_replaced : `str` or `None`
155 If not `None`, delete the datasets in the collection replaced by
156 `replace_run`, either just from the datastore ("unstore") or by
157 removing them and the RUN completely ("purge"). Requires
158 ``replace_run`` to be `True`.
159 data_query : `str`
160 User query selection expression.
161 skip_existing_in : `~collections.abc.Iterable` [ `str` ] or `None`
162 Accepts list of collections, if all Quantum outputs already exist in
163 the specified list of collections then that Quantum will be excluded
164 from the QuantumGraph.
165 skip_existing : `bool`
166 Appends output RUN collection to the ``skip_existing_in`` list.
167 debug : `bool`
168 If true, enable debugging output using lsstDebug facility (imports
169 debug.py).
170 fail_fast : `bool`
171 If true then stop processing at first error, otherwise process as many
172 tasks as possible.
173 clobber_outputs : `bool`
174 Remove outputs from previous execution of the same quantum before new
175 execution. Only applies to failed quanta if skip_existing is also
176 given.
177 summary : `str`
178 File path to store job report in JSON format.
179 enable_implicit_threading : `bool`, optional
180 If `True`, do not disable implicit threading by third-party libraries.
181 Implicit threading is always disabled during actual quantum execution
182 if ``processes > 1``.
183 cores_per_quantum : `int`
184 Number of cores that can be used by each quantum.
185 memory_per_quantum : `str`
186 Amount of memory that each quantum can be allowed to use. Empty string
187 implies no limit. The string can be either a single integer (implying
188 units of MB) or a combination of number and unit.
189 rebase : `bool`
190 If `True` then reset output collection chain if it is inconsistent with
191 the ``inputs``.
192 raise_on_partial_outputs : `bool`
193 Consider partial outputs an error instead of a success.
194 **kwargs : `object`
195 Ignored; click commands may accept options for more than one script
196 function and pass all the option kwargs to each of the script functions
197 which ignore these unused kwargs.
198 """
199 # Fork option still exists for compatibility but we use spawn instead.
200 if start_method == "fork": # type: ignore[comparison-overlap]
201 start_method = "spawn" # type: ignore[unreachable]
202 _LOG.warning("Option --start-method=fork is unsafe and no longer supported, using spawn instead.")
204 if not enable_implicit_threading:
205 disable_implicit_threading()
207 skip_existing_in = tuple(skip_existing_in) if skip_existing_in is not None else ()
208 if data_query is None:
209 data_query = ""
210 inputs = list(ensure_iterable(input)) if input else []
211 del input
212 enable_lsst_debug = debug
213 del debug
215 if not output_run:
216 # If we have no output run specified, use the one from the graph rather
217 # than letting a new timestamped run be created.
218 output_run = qg.header.output_run
219 else:
220 # Check that output run defined on command line is consistent with
221 # quantum graph.
222 if qg.header.output_run != output_run:
223 raise ValueError(
224 f"Output run defined on command line ({output_run}) has to be "
225 f"identical to graph metadata ({qg.header.output_run}). "
226 "To update graph metadata run `pipetask update-graph-run` command."
227 )
229 # Make sure that --extend-run always enables --skip-existing,
230 # clobbering should be disabled if --extend-run is not specified.
231 if extend_run:
232 skip_existing = True
233 else:
234 clobber_outputs = False
236 # Make butler instance. QuantumGraph should have an output run defined,
237 # but we ignore it here and let command line decide actual output run.
238 butler = ButlerFactory.make_write_butler(
239 butler_config,
240 qg.pipeline_graph,
241 output=output,
242 output_run=output_run,
243 inputs=inputs,
244 extend_run=extend_run,
245 rebase=rebase,
246 replace_run=replace_run,
247 prune_replaced=prune_replaced,
248 )
250 assert butler.run is not None, "Guaranteed by make_write_butler."
251 if skip_existing:
252 skip_existing_in += (butler.run,)
254 # Enable lsstDebug debugging. Note that this is done once in the
255 # main process before PreExecInit and it is also repeated before
256 # running each task in SingleQuantumExecutor (which may not be
257 # needed if `multiprocessing` always uses fork start method).
258 if enable_lsst_debug:
259 try:
260 _LOG.debug("Will try to import debug.py")
261 import debug # type: ignore # noqa: F401
262 except ImportError:
263 _LOG.warning("No 'debug' module found.")
265 # Save all InitOutputs, configs, etc.
266 if register_dataset_types:
267 qg.pipeline_graph.register_dataset_types(butler, include_packages=not no_versions)
268 if not skip_init_writes:
269 qg.write_init_outputs(butler, skip_existing=skip_existing)
270 qg.write_configs(butler, compare_existing=extend_run)
271 if not no_versions:
272 qg.write_packages(butler, compare_existing=extend_run)
274 if init_only:
275 return
277 if task_factory is None:
278 task_factory = TaskFactory()
279 resources = ExecutionResources(
280 num_cores=cores_per_quantum, max_mem=memory_per_quantum, default_mem_units=u.MB
281 )
282 quantum_executor = SingleQuantumExecutor(
283 butler=butler,
284 task_factory=task_factory,
285 skip_existing_in=skip_existing_in,
286 clobber_outputs=clobber_outputs,
287 enable_lsst_debug=enable_lsst_debug,
288 resources=resources,
289 raise_on_partial_outputs=raise_on_partial_outputs,
290 )
292 if timeout is None:
293 timeout = MP_TIMEOUT
294 executor = MPGraphExecutor(
295 num_proc=processes,
296 timeout=timeout,
297 start_method=start_method,
298 quantum_executor=quantum_executor,
299 fail_fast=fail_fast,
300 pdb=pdb,
301 execution_graph_fixup=_import_graph_fixup(graph_fixup),
302 )
303 # Have to reset connection pool to avoid sharing connections with
304 # forked processes.
305 butler.registry.resetConnectionPool()
306 try:
307 with lsst.utils.timer.profile(profile, _LOG):
308 executor.execute(qg)
309 finally:
310 if summary:
311 report = executor.getReport()
312 if report:
313 with ResourcePath(summary).open("w") as out:
314 # Do not save fields that are not set.
315 out.write(report.model_dump_json(exclude_none=True, indent=2))
318def _import_graph_fixup(graph_fixup: str) -> ExecutionGraphFixup | None:
319 """Import/instantiate graph fixup object.
321 Parameters
322 ----------
323 graph_fixup : `str`
324 Graph fixup command-line argument.
326 Returns
327 -------
328 fixup : `ExecutionGraphFixup` or `None`
329 Object that imposes additional ordering constraints on the graph.
331 Raises
332 ------
333 ValueError
334 Raised if import fails, method call raises exception, or returned
335 instance has unexpected type.
336 """
337 from lsst.pipe.base.execution_graph_fixup import ExecutionGraphFixup
339 if graph_fixup:
340 try:
341 factory = doImportType(graph_fixup)
342 except Exception as exc:
343 raise ValueError("Failed to import graph fixup class/method") from exc
344 try:
345 fixup = factory()
346 except Exception as exc:
347 raise ValueError("Failed to make instance of graph fixup") from exc
348 if not isinstance(fixup, ExecutionGraphFixup):
349 raise ValueError("Graph fixup is not an instance of ExecutionGraphFixup class")
350 return fixup
351 return None