Improve memory usage on large inputs (WIP) (#1008)

* Improve memory efficiency by avoiding filling in "nameext" and "nameroot" strings redundantly
* Add warning recommending changing loadListing when number of input
files is very large.

Author: tetron

MANIFEST.in

@@ -4,6 +4,7 @@ include *requirements.txt
 include gittaggers.py Makefile cwltool.py
 include tests/*
 include tests/tmp1/tmp2/tmp3/.gitkeep
+include tests/tmp4/alpha/*
 include tests/wf/*
 include tests/override/*
 include tests/checker_wf/*

cwltool/command_line_tool.py

@@ -203,10 +203,17 @@ def check_adjust(builder, file_o):
     assert builder.pathmapper is not None
     file_o["path"] = docker_windows_path_adjust(
         builder.pathmapper.mapper(file_o["location"])[1])
-    file_o["dirname"], file_o["basename"] = os.path.split(file_o["path"])
+    dn, bn = os.path.split(file_o["path"])
+    if file_o.get("dirname") != dn:
+        file_o["dirname"] = Text(dn)
+    if file_o.get("basename") != bn:
+        file_o["basename"] = Text(bn)
     if file_o["class"] == "File":
-        file_o["nameroot"], file_o["nameext"] = os.path.splitext(
-            file_o["basename"])
+        nr, ne = os.path.splitext(file_o["basename"])
+        if file_o.get("nameroot") != nr:
+            file_o["nameroot"] = Text(nr)
+        if file_o.get("nameext") != ne:
+            file_o["nameext"] = Text(ne)
     if not ACCEPTLIST_RE.match(file_o["basename"]):
         raise WorkflowException(
             "Invalid filename: '{}' contains illegal characters".format(

cwltool/pathmapper.py

@@ -74,11 +74,18 @@ def addLocation(d):
             path = path.rstrip("/")
             d["location"] = urllib.parse.urlunparse((parse.scheme, parse.netloc, path, parse.params, parse.query, parse.fragment))
 
-        if "basename" not in d:
-            d["basename"] = os.path.basename(urllib.request.url2pathname(path))
+        if not d.get("basename"):
+            if path.startswith("_:"):
+                d["basename"] = Text(path[2:])
+            else:
+                d["basename"] = Text(os.path.basename(urllib.request.url2pathname(path)))
 
         if d["class"] == "File":
-            d["nameroot"], d["nameext"] = os.path.splitext(d["basename"])
+            nr, ne = os.path.splitext(d["basename"])
+            if d.get("nameroot") != nr:
+                d["nameroot"] = Text(nr)
+            if d.get("nameext") != ne:
+                d["nameext"] = Text(ne)
 
     visit_class(job, ("File", "Directory"), addLocation)
 
@@ -106,6 +113,12 @@ def mark(d):
 
 def get_listing(fs_access, rec, recursive=True):
     # type: (StdFsAccess, MutableMapping[Text, Any], bool) -> None
+    if rec.get("class") != "Directory":
+        finddirs = []  # type: List[MutableMapping]
+        visit_class(rec, ("Directory",), finddirs.append)
+        for f in finddirs:
+            get_listing(fs_access, f, recursive=recursive)
+        return
     if "listing" in rec:
         return
     listing = []

cwltool/process.py

@@ -28,7 +28,7 @@
                                Text)
 from schema_salad import schema, validate
 from schema_salad.ref_resolver import Loader, file_uri
-from schema_salad.sourceline import SourceLine
+from schema_salad.sourceline import SourceLine, strip_dup_lineno
 
 from . import expression
 from .builder import Builder, HasReqsHints
@@ -369,7 +369,6 @@ def fill_in_defaults(inputs,   # type: List[Dict[Text, Text]]
                 job[fieldname] = None
             else:
                 raise WorkflowException("Missing required input parameter '%s'" % shortname(inp["id"]))
-    visit_class(job, ("File",), functools.partial(add_sizes, fsaccess))
 
 
 def avroize_type(field_type, name_prefix=""):
@@ -434,10 +433,12 @@ def var_spool_cwl_detector(obj,           # type: Union[MutableMapping, List, Te
 
 def eval_resource(builder, resource_req):  # type: (Builder, Text) -> Any
     if expression.needs_parsing(resource_req):
-        visit_class(builder.job, ("File",), functools.partial(add_sizes, builder.fs_access))
         return builder.do_eval(resource_req)
     return resource_req
 
+# Threshold where the "too many files" warning kicks in
+FILE_COUNT_WARNING = 5000
+
 class Process(with_metaclass(abc.ABCMeta, HasReqsHints)):
     def __init__(self,
                  toolpath_object,      # type: MutableMapping[Text, Any]
@@ -581,16 +582,59 @@ def _init_job(self, joborder, runtime_context):
         make_fs_access = getdefault(runtime_context.make_fs_access, StdFsAccess)
         fs_access = make_fs_access(runtime_context.basedir)
 
+        load_listing_req, _ = self.get_requirement(
+            "http://commonwl.org/cwltool#LoadListingRequirement")
+        if load_listing_req is not None:
+            load_listing = load_listing_req.get("loadListing")
+        else:
+            load_listing = "deep_listing"   # will default to "no_listing" in CWL v1.1
+
         # Validate job order
         try:
             fill_in_defaults(self.tool[u"inputs"], job, fs_access)
+
             normalizeFilesDirs(job)
             schema = self.names.get_name("input_record_schema", "")
             if schema is None:
                 raise WorkflowException("Missing input record schema: "
                     "{}".format(self.names))
             validate.validate_ex(schema, job, strict=False,
                                  logger=_logger_validation_warnings)
+
+            if load_listing and load_listing != "no_listing":
+                get_listing(fs_access, job, recursive=(load_listing == "deep_listing"))
+
+            visit_class(job, ("File",), functools.partial(add_sizes, fs_access))
+
+            if load_listing == "deep_listing" and load_listing_req is None:
+                for i, inparm in enumerate(self.tool["inputs"]):
+                    k = shortname(inparm["id"])
+                    if k not in job:
+                        continue
+                    v = job[k]
+                    dircount = [0]
+                    def inc(d):  # type: (List[int]) -> None
+                        d[0] += 1
+                    visit_class(v, ("Directory",), lambda x: inc(dircount))
+                    if dircount[0] == 0:
+                        continue
+                    filecount = [0]
+                    visit_class(v, ("File",), lambda x: inc(filecount))
+                    if filecount[0] > FILE_COUNT_WARNING:
+                        # Long lines in this message are okay, will be reflowed based on terminal columns.
+                        _logger.warning(strip_dup_lineno(SourceLine(self.tool["inputs"], i, Text).makeError(
+                    """Recursive directory listing has resulted in a large number of File objects (%s) passed to the input parameter '%s'.  This may negatively affect workflow performance and memory use.
+
+If this is a problem, use the hint 'cwltool:LoadListingRequirement' with "shallow_listing" or "no_listing" to change the directory listing behavior:
+
+$namespaces:
+  cwltool: "http://commonwl.org/cwltool#"
+hints:
+  cwltool:LoadListingRequirement:
+    loadListing: shallow_listing
+
+""" % (filecount[0], k))))
+
         except (validate.ValidationException, WorkflowException) as err:
             raise WorkflowException("Invalid job input record:\n" + Text(err))
 
@@ -599,13 +643,6 @@ def _init_job(self, joborder, runtime_context):
         tmpdir = u""
         stagedir = u""
 
-        load_listing_req, _ = self.get_requirement(
-            "http://commonwl.org/cwltool#LoadListingRequirement")
-        if load_listing_req is not None:
-            load_listing = load_listing_req.get("loadListing")
-        else:
-            load_listing = "deep_listing"   # will default to "no_listing" in CWL v1.1
-
         docker_req, _ = self.get_requirement("DockerRequirement")
         default_docker = None
 

cwltool/workflow.py

@@ -8,7 +8,8 @@
 import tempfile
 from collections import namedtuple
 from typing import (Any, Callable, Dict, Generator, Iterable, List,
-                    Mapping, MutableMapping, MutableSequence, Optional, Tuple, Union)
+                    Mapping, MutableMapping, MutableSequence,
+                    Optional, Tuple, Union, cast)
 from uuid import UUID  # pylint: disable=unused-import
 
 from ruamel.yaml.comments import CommentedMap
@@ -290,6 +291,9 @@ def receive_output(self, step, outputparms, final_output_callback, jobout, proce
             _logger.info(u"[%s] completed %s", step.name, processStatus)
 
         step.completed = True
+        # Release the iterable related to this step to
+        # reclaim memory.
+        step.iterable = None
         self.made_progress = True
 
         completed = sum(1 for s in self.steps if s.completed)
@@ -570,7 +574,7 @@ def job(self,
             runtimeContext     # type: RuntimeContext
            ):  # type: (...) -> Generator[Any, None, None]
         builder = self._init_job(job_order, runtimeContext)
-        #relativeJob=copy.deepcopy(builder.job)
+
         if runtimeContext.research_obj is not None:
             if runtimeContext.toplevel:
                 # Record primary-job.json
@@ -810,12 +814,18 @@ def __init__(self,
         self.processStatus = u"success"
         self.total = total
         self.output_callback = output_callback
+        self.steps = []  # type: List[Optional[Generator]]
 
     def receive_scatter_output(self, index, jobout, processStatus):
         # type: (int, Dict[Text, Text], Text) -> None
         for key, val in jobout.items():
             self.dest[key][index] = val
 
+        # Release the iterable related to this step to
+        # reclaim memory.
+        if self.steps:
+            self.steps[index] = None
+
         if processStatus != "success":
             if self.processStatus != "permanentFail":
                 self.processStatus = processStatus
@@ -825,8 +835,9 @@ def receive_scatter_output(self, index, jobout, processStatus):
         if self.completed == self.total:
             self.output_callback(self.dest, self.processStatus)
 
-    def setTotal(self, total):  # type: (int) -> None
+    def setTotal(self, total, steps):  # type: (int, List[Generator]) -> None
         self.total = total
+        self.steps = cast(List[Optional[Generator]], steps)
         if self.completed == self.total:
             self.output_callback(self.dest, self.processStatus)
 
@@ -838,6 +849,8 @@ def parallel_steps(steps, rc, runtimeContext):
         for index, step in enumerate(steps):
             if getdefault(runtimeContext.on_error, "stop") == "stop" and rc.processStatus != "success":
                 break
+            if step is None:
+                continue
             try:
                 for j in step:
                     if getdefault(runtimeContext.on_error, "stop") == "stop" and rc.processStatus != "success":
@@ -847,6 +860,8 @@ def parallel_steps(steps, rc, runtimeContext):
                         yield j
                     else:
                         break
+                if made_progress:
+                    break
             except WorkflowException as exc:
                 _logger.error(u"Cannot make scatter job: %s", exc)
                 _logger.debug("", exc_info=True)
@@ -890,7 +905,7 @@ def dotproduct_scatter(process,           # type: WorkflowJobStep
             sjobo, functools.partial(rc.receive_scatter_output, index),
             runtimeContext))
 
-    rc.setTotal(jobl)
+    rc.setTotal(jobl, steps)
     return parallel_steps(steps, rc, runtimeContext)
 
 
@@ -925,7 +940,7 @@ def nested_crossproduct_scatter(process,          # type: WorkflowJobStep
                 functools.partial(rc.receive_scatter_output, index),
                 runtimeContext))
 
-    rc.setTotal(jobl)
+    rc.setTotal(jobl, steps)
     return parallel_steps(steps, rc, runtimeContext)
 
 
@@ -952,7 +967,7 @@ def flat_crossproduct_scatter(process,          # type: WorkflowJobStep
     callback = ReceiveScatterOutput(output_callback, output, 0)
     (steps, total) = _flat_crossproduct_scatter(
         process, joborder, scatter_keys, callback, 0, runtimeContext)
-    callback.setTotal(total)
+    callback.setTotal(total, steps)
     return parallel_steps(steps, callback, runtimeContext)
 
 def _flat_crossproduct_scatter(process,        # type: WorkflowJobStep

tests/listing2-job.yml

@@ -0,0 +1,3 @@
+d:
+  class: Directory
+  location: tmp4

tests/test_ext.py

@@ -3,12 +3,17 @@
 import os
 import shutil
 import tempfile
+import sys
+import re
+from io import StringIO
 
 import pytest
 
 from cwltool.main import main
+import cwltool.process
 
-from .util import get_data, needs_docker, temp_dir, windows_needs_docker
+
+from .util import get_data, needs_docker, temp_dir, windows_needs_docker, get_main_output
 
 
 @needs_docker
@@ -197,3 +202,16 @@ def test_require_prefix_timelimit():
     assert main(["--enable-ext", get_data('tests/wf/timelimit.cwl')]) == 0
     assert main([get_data('tests/wf/timelimit.cwl')]) != 0
     assert main(["--enable-ext", get_data('tests/wf/timelimit-fail.cwl')]) != 0
+
+def test_warn_large_inputs():
+    was = cwltool.process.FILE_COUNT_WARNING
+    try:
+        stream = StringIO()
+
+        cwltool.process.FILE_COUNT_WARNING = 3
+        main([get_data('tests/wf/listing_v1_0.cwl'), get_data('tests/listing2-job.yml')],
+             stderr=stream)
+
+        assert "Recursive directory listing has resulted in a large number of File objects" in re.sub("\n  *", " ", stream.getvalue())
+    finally:
+        cwltool.process.FILE_COUNT_WARNING = was