Merge pull request #66 from lsst/tickets/DM-26136

DM-26136: Improve handling of crashes in pipetask
lsst · Aug 20, 2020 · ff154c3 · ff154c3
2 parents a1b91fd + 8cf1085
commit ff154c3
Show file tree

Hide file tree

Showing 11 changed files with 468 additions and 104 deletions.
diff --git a/python/lsst/ctrl/mpexec/cli/cmd/commands.py b/python/lsst/ctrl/mpexec/cli/cmd/commands.py
@@ -167,6 +167,7 @@ def processor(objs):
 @opt.profile_option()
 @opt.processes_option()
 @opt.timeout_option()
+@opt.fail_fast_option()
 @opt.graph_fixup_option()
 @option_section(sectionText="Meta-information output options:")
 @opt.skip_init_writes_option()

diff --git a/python/lsst/ctrl/mpexec/cli/opt/options.py b/python/lsst/ctrl/mpexec/cli/opt/options.py
@@ -232,3 +232,8 @@
 
 timeout_option = MWOptionDecorator("--timeout",
                                    help="Timeout for multiprocessing; maximum wall time (sec).")
+
+fail_fast_option = MWOptionDecorator("--fail-fast",
+                                     help=unwrap("""Stop processing at first error, default is to process
+                                                 as many tasks as possible."""),
+                                     is_flag=True)
diff --git a/python/lsst/ctrl/mpexec/cli/script/run.py b/python/lsst/ctrl/mpexec/cli/script/run.py
@@ -47,7 +47,8 @@ def run(do_raise,
         prune_replaced,
         data_query,
         skip_existing,
-        debug):
+        debug,
+        fail_fast):
     """Implements the command line interface `pipetask run` subcommand, should
     only be called by command line tools and unit test code that test this
     function.
@@ -127,6 +128,9 @@ def run(do_raise,
     debug : `bool`
         If true, enable debugging output using lsstDebug facility (imports
         debug.py).
+    fail_fast : `bool`
+        If true then stop processing at first error, otherwise process as many
+        tasks as possible.
     """
 
     if log_level is not None:
@@ -157,7 +161,8 @@ def __init__(self,
                      prune_replaced,
                      data_query,
                      skip_existing,
-                     debug):
+                     debug,
+                     fail_fast):
             self.do_raise = do_raise
             self.graph_fixup = graph_fixup
             self.init_only = init_only
@@ -177,13 +182,14 @@ def __init__(self,
             self.data_query = data_query
             self.skip_existing = skip_existing
             self.enableLsstDebug = debug
+            self.fail_fast = fail_fast
 
     args = RunArgs(do_raise=do_raise, graph_fixup=graph_fixup, init_only=init_only, no_versions=no_versions,
                    processes=processes, profile=profile, skip_init_writes=skip_init_writes, timeout=timeout,
                    register_dataset_types=register_dataset_types, butler_config=butler_config, input=input,
                    output=output, output_run=output_run, extend_run=extend_run, replace_run=replace_run,
                    prune_replaced=prune_replaced, data_query=data_query, skip_existing=skip_existing,
-                   debug=debug)
+                   debug=debug, fail_fast=fail_fast)
 
     f = CmdLineFwk()
     taskFactory = TaskFactory()

diff --git a/python/lsst/ctrl/mpexec/cmdLineFwk.py b/python/lsst/ctrl/mpexec/cmdLineFwk.py
@@ -690,6 +690,7 @@ def runPipeline(self, graph, taskFactory, args, butler=None):
             timeout = self.MP_TIMEOUT if args.timeout is None else args.timeout
             executor = MPGraphExecutor(numProc=args.processes, timeout=timeout,
                                        quantumExecutor=quantumExecutor,
+                                       failFast=args.fail_fast,
                                        executionGraphFixup=graphFixup)
             with util.profile(args.profile, _LOG):
                 executor.execute(graph, butler)

diff --git a/python/lsst/ctrl/mpexec/cmdLineParser.py b/python/lsst/ctrl/mpexec/cmdLineParser.py
@@ -393,6 +393,9 @@ def _makeExecOptions(parser):
     group.add_argument("-j", "--processes", type=int, default=1, help="Number of processes to use")
     group.add_argument("--timeout", type=float,
                        help="Timeout for multiprocessing; maximum wall time (sec)")
+    group.add_argument("--fail-fast", action="store_true", default=False,
+                       help="Stop processing at first error, default is to process as many tasks "
+                       "as possible.")
 
     # run-time graph fixup option
     group.add_argument("--graph-fixup", type=str, default=None,