Merge pull request #988 from common-workflow-language/less-deep-copy

Eliminate unnecessary deep copies of job order objects.

Author: mr-c

cwltool/__init__.py

@@ -1,2 +1,2 @@
 from __future__ import absolute_import
-__author__ = 'peter.amstutz@curoverse.com'
+__author__ = 'pamstutz@veritasgenetics.com'

cwltool/argparser.py

@@ -270,8 +270,9 @@ def arg_parser():  # type: () -> argparse.ArgumentParser
                         dest="ga4gh_tool_registries", default=[])
 
     parser.add_argument("--on-error",
-                        help="Desired workflow behavior when a step fails.  One of 'stop' or 'continue'. "
-                             "Default is 'stop'.", default="stop", choices=("stop", "continue"))
+                        help="Desired workflow behavior when a step fails.  One of 'stop' (do not submit any more steps) or "
+                        "'continue' (may submit other steps that are not downstream from the error). Default is 'stop'.",
+                        default="stop", choices=("stop", "continue"))
 
     exgroup = parser.add_mutually_exclusive_group()
     exgroup.add_argument("--compute-checksum", action="store_true", default=True,

cwltool/command_line_tool.py

@@ -12,7 +12,7 @@
 import tempfile
 import threading
 from functools import cmp_to_key, partial
-from typing import (Any, Callable, Dict, Generator, List, MutableMapping,
+from typing import (Any, Callable, Dict, Generator, List, Mapping, MutableMapping,
                     MutableSequence, Optional, Set, Union, cast)
 
 import shellescape
@@ -103,7 +103,7 @@ def run(self, runtimeContext):  # type: (RuntimeContext) -> None
                 self.output_callback({}, "permanentFail")
 
     def job(self,
-            job_order,         # type: MutableMapping[Text, Text]
+            job_order,         # type: Mapping[Text, Text]
             output_callbacks,  # type: Callable[[Any, Any], Any]
             runtimeContext     # type: RuntimeContext
            ):
@@ -276,7 +276,7 @@ def updatePathmap(self, outdir, pathmap, fn):
             self.updatePathmap(os.path.join(outdir, fn["basename"]), pathmap, ls)
 
     def job(self,
-            job_order,         # type: MutableMapping[Text, Text]
+            job_order,         # type: Mapping[Text, Text]
             output_callbacks,  # type: Callable[[Any, Any], Any]
             runtimeContext     # RuntimeContext
            ):
@@ -403,7 +403,7 @@ def rm_pending_output_callback(output_callbacks, jobcachepending,
                           self.tool.get("id", ""),
                           u" as part of %s" % runtimeContext.part_of
                           if runtimeContext.part_of else "")
-            _logger.debug(u"[job %s] %s", j.name, json_dumps(job_order,
+            _logger.debug(u"[job %s] %s", j.name, json_dumps(builder.job,
                                                              indent=4))
 
         builder.pathmapper = self.make_path_mapper(

cwltool/process.py

@@ -16,7 +16,7 @@
 from collections import Iterable  # pylint: disable=unused-import
 from io import open
 from typing import (Any, Callable, Dict, Generator, Iterator, List,
-                    MutableMapping, MutableSequence, Optional, Set, Tuple,
+                    Mapping, MutableMapping, MutableSequence, Optional, Set, Tuple,
                     Union, cast)
 
 from pkg_resources import resource_stream
@@ -363,7 +363,8 @@ def cleanIntermediate(output_dirs):  # type: (Iterable[Text]) -> None
 def add_sizes(fsaccess, obj):  # type: (StdFsAccess, Dict[Text, Any]) -> None
     if 'location' in obj:
         try:
-            obj["size"] = fsaccess.size(obj["location"])
+            if "size" not in obj:
+                obj["size"] = fsaccess.size(obj["location"])
         except OSError:
             pass
     elif 'contents' in obj:
@@ -587,7 +588,7 @@ def __init__(self,
             var_spool_cwl_detector(self.tool)
 
     def _init_job(self, joborder, runtime_context):
-        # type: (MutableMapping[Text, Text], RuntimeContext) -> Builder
+        # type: (Mapping[Text, Text], RuntimeContext) -> Builder
 
         job = cast(Dict[Text, Union[Dict[Text, Any], List[Any], Text, None]],
                    copy.deepcopy(joborder))
@@ -791,7 +792,7 @@ def visit(self, op):  # type: (Callable[[MutableMapping[Text, Any]], None]) -> N
 
     @abc.abstractmethod
     def job(self,
-            job_order,         # type: MutableMapping[Text, Text]
+            job_order,         # type: Mapping[Text, Text]
             output_callbacks,  # type: Callable[[Any, Any], Any]
             runtimeContext     # type: RuntimeContext
            ):  # type: (...) -> Generator[Any, None, None]

cwltool/workflow.py

@@ -8,7 +8,7 @@
 import tempfile
 from collections import namedtuple
 from typing import (Any, Callable, Dict, Generator, Iterable, List,
-                    MutableMapping, MutableSequence, Optional, Tuple, Union)
+                    Mapping, MutableMapping, MutableSequence, Optional, Tuple, Union)
 from uuid import UUID  # pylint: disable=unused-import
 
 from ruamel.yaml.comments import CommentedMap
@@ -165,7 +165,7 @@ def object_from_state(state,                  # Dict[Text, WorkflowStateItem]
                     return None
 
         if inputobj.get(iid) is None and "default" in inp:
-            inputobj[iid] = copy.deepcopy(inp["default"])
+            inputobj[iid] = inp["default"]
 
         if iid not in inputobj and ("valueFrom" in inp or incomplete):
             inputobj[iid] = None
@@ -189,7 +189,7 @@ def __init__(self, step):
         self.parent_wf = step.parent_wf
 
     def job(self,
-            joborder,         # type: MutableMapping[Text, Text]
+            joborder,         # type: Mapping[Text, Text]
             output_callback,  # type: functools.partial[None]
             runtimeContext    # type: RuntimeContext
            ):
@@ -415,7 +415,7 @@ def run(self, runtimeContext):
         _logger.info(u"[%s] start", self.name)
 
     def job(self,
-            joborder,         # type: MutableMapping[Text, Any]
+            joborder,         # type: Mapping[Text, Any]
             output_callback,  # type: Callable[[Any, Any], Any]
             runtimeContext    # type: RuntimeContext
            ):  # type: (...) -> Generator
@@ -434,10 +434,10 @@ def job(self,
                 inp_id = shortname(inp["id"])
                 if inp_id in joborder:
                     self.state[inp["id"]] = WorkflowStateItem(
-                        inp, copy.deepcopy(joborder[inp_id]), "success")
+                        inp, joborder[inp_id], "success")
                 elif "default" in inp:
                     self.state[inp["id"]] = WorkflowStateItem(
-                        inp, copy.deepcopy(inp["default"]), "success")
+                        inp, inp["default"], "success")
                 else:
                     raise WorkflowException(
                         u"Input '%s' not in input object and does not have a "
@@ -565,7 +565,7 @@ def make_workflow_step(self,
         return WorkflowStep(toolpath_object, pos, loadingContext, parentworkflowProv)
 
     def job(self,
-            job_order,         # type: MutableMapping[Text, Text]
+            job_order,         # type: Mapping[Text, Text]
             output_callbacks,  # type: Callable[[Any, Any], Any]
             runtimeContext     # type: RuntimeContext
            ):  # type: (...) -> Generator[Any, None, None]
@@ -761,7 +761,7 @@ def receive_output(self, output_callback, jobout, processStatus):
         output_callback(output, processStatus)
 
     def job(self,
-            job_order,         # type: MutableMapping[Text, Text]
+            job_order,         # type: Mapping[Text, Text]
             output_callbacks,  # type: Callable[[Any, Any], Any]
             runtimeContext,    # type: RuntimeContext
            ):  # type: (...) -> Generator[Any, None, None]
@@ -775,15 +775,16 @@ def job(self,
             self.prov_obj.start_process(
                 process_name, datetime.datetime.now(),
                 self.embedded_tool.provenance_object.workflow_run_uri)
+
+        step_input = {}
         for inp in self.tool["inputs"]:
             field = shortname(inp["id"])
             if not inp.get("not_connected"):
-                job_order[field] = job_order[inp["id"]]
-            del job_order[inp["id"]]
+                step_input[field] = job_order[inp["id"]]
 
         try:
             for tool in self.embedded_tool.job(
-                    job_order,
+                    step_input,
                     functools.partial(self.receive_output, output_callbacks),
                     runtimeContext):
                 yield tool
@@ -878,7 +879,7 @@ def dotproduct_scatter(process,           # type: WorkflowJobStep
 
     steps = []
     for index in range(0, jobl):
-        sjobo = copy.deepcopy(joborder)
+        sjobo = copy.copy(joborder)
         for key in scatter_keys:
             sjobo[key] = joborder[key][index]
 
@@ -909,7 +910,7 @@ def nested_crossproduct_scatter(process,          # type: WorkflowJobStep
 
     steps = []
     for index in range(0, jobl):
-        sjob = copy.deepcopy(joborder)
+        sjob = copy.copy(joborder)
         sjob[scatter_key] = joborder[scatter_key][index]
 
         if len(scatter_keys) == 1:
@@ -967,7 +968,7 @@ def _flat_crossproduct_scatter(process,        # type: WorkflowJobStep
     steps = []
     put = startindex
     for index in range(0, jobl):
-        sjob = copy.deepcopy(joborder)
+        sjob = copy.copy(joborder)
         sjob[scatter_key] = joborder[scatter_key][index]
 
         if len(scatter_keys) == 1: