WIP InitialWorkdirRequirement array and null (#879)

* Allow staging array input in combination with other items

* Allow list of files in entry field

* Add InitialWorkDirRequirement conflicting names explanation

* Add InitialWorkDir allowed types

* Add InitialWorkDir allowed types

* Reformat listing types

* Add Directory[] to initwd description

* Fix isinstance list -> MutableSequence

* Improve resource error msg

* Fix parallel executor bug, was failing conformance tests.

* Add typecheck that 't' is a Mapping before checking for 'entry'

* More parallel executor work, found the actual bug.

Closures in Python are weird and easy to mess up.  Pull out runner()
so it isn't affected when the value of "job" changes in the
surrounding function.

* Shrink the default RAM request.

* Fix type annotation

Author: bogdang989

cwltool/command_line_tool.py

@@ -428,16 +428,25 @@ def rm_pending_output_callback(output_callbacks, jobcachepending,
                 ls = builder.do_eval(initialWorkdir["listing"])
             else:
                 for t in initialWorkdir["listing"]:
-                    if "entry" in t:
-                        et = {u"entry": builder.do_eval(t["entry"], strip_whitespace=False)}
-                        if "entryname" in t:
-                            et["entryname"] = builder.do_eval(t["entryname"])
-                        else:
-                            et["entryname"] = None
-                        et["writable"] = t.get("writable", False)
-                        ls.append(et)
+                    if isinstance(t, Mapping) and "entry" in t:
+                        entry_exp = builder.do_eval(t["entry"], strip_whitespace=False)
+                        for entry in aslist(entry_exp):
+                            et = {u"entry": entry}
+                            if "entryname" in t:
+                                et["entryname"] = builder.do_eval(t["entryname"])
+                            else:
+                                et["entryname"] = None
+                            et["writable"] = t.get("writable", False)
+                            if et[u"entry"]:
+                                ls.append(et)
                     else:
-                        ls.append(builder.do_eval(t))
+                        initwd_item = builder.do_eval(t)
+                        if not initwd_item:
+                            continue
+                        if isinstance(initwd_item, MutableSequence):
+                            ls.extend(initwd_item)
+                        else:
+                            ls.append(initwd_item)
             for i, t in enumerate(ls):
                 if "entry" in t:
                     if isinstance(t["entry"], string_types):

cwltool/executors.py

@@ -231,8 +231,26 @@ def select_resources(self, request, runtime_context):  # pylint: disable=unused-
 
         return result
 
+    def _runner(self, job, runtime_context):
+        """ Job running thread. """
+        try:
+            job.run(runtime_context)
+        except WorkflowException as err:
+            _logger.exception("Got workflow error")
+            self.exceptions.append(err)
+        except Exception as err:  # pylint: disable=broad-except
+            _logger.exception("Got workflow error")
+            self.exceptions.append(WorkflowException(Text(err)))
+        finally:
+            with runtime_context.workflow_eval_lock:
+                self.threads.remove(threading.current_thread())
+                if isinstance(job, JobBase):
+                    self.allocated_ram -= job.builder.resources["ram"]
+                    self.allocated_cores -= job.builder.resources["cores"]
+                runtime_context.workflow_eval_lock.notifyAll()
+
     def run_job(self,
-                job,             # type: Union[JobBase, WorkflowJob]
+                job,             # type: Union[JobBase, WorkflowJob, None]
                 runtime_context  # type: RuntimeContext
                ):  # type: (...) -> None
         """ Execute a single Job in a seperate thread. """
@@ -241,50 +259,51 @@ def run_job(self,
             with self.pending_jobs_lock:
                 self.pending_jobs.append(job)
 
-        while self.pending_jobs:
-            with self.pending_jobs_lock:
-                job = self.pending_jobs[0]
-                if isinstance(job, JobBase) \
-                        and \
-                        ((self.allocated_ram + job.builder.resources["ram"])
-                         > self.max_ram
-                         or (self.allocated_cores + job.builder.resources["cores"])
-                         > self.max_cores):
-                    _logger.warning(
-                        'Job "%s" requested more resources (%s) than are '
-                        'available (max ram is %f, max cores is %f)',
-                        job.name, job.builder.resources, self.max_ram,
-                        self.max_cores)
-                    return
+        with self.pending_jobs_lock:
+            n = 0
+            while (n+1) <= len(self.pending_jobs):
+                job = self.pending_jobs[n]
+                if isinstance(job, JobBase):
+                    if ((job.builder.resources["ram"])
+                        > self.max_ram
+                        or (job.builder.resources["cores"])
+                        > self.max_cores):
+                        _logger.error(
+                            'Job "%s" cannot be run, requests more resources (%s) '
+                            'than available on this host (max ram %d, max cores %d',
+                            job.name, job.builder.resources,
+                            self.allocated_ram,
+                            self.allocated_cores,
+                            self.max_ram,
+                            self.max_cores)
+                        self.pending_jobs.remove(job)
+                        return
+
+                    if ((self.allocated_ram + job.builder.resources["ram"])
+                        > self.max_ram
+                        or (self.allocated_cores + job.builder.resources["cores"])
+                        > self.max_cores):
+                        _logger.debug(
+                            'Job "%s" cannot run yet, resources (%s) are not '
+                            'available (already allocated ram is %d, allocated cores is %d, '
+                            'max ram %d, max cores %d',
+                            job.name, job.builder.resources,
+                            self.allocated_ram,
+                            self.allocated_cores,
+                            self.max_ram,
+                            self.max_cores)
+                        n += 1
+                        continue
+
+                thread = threading.Thread(target=self._runner, args=(job, runtime_context))
+                thread.daemon = True
+                self.threads.add(thread)
+                if isinstance(job, JobBase):
+                    self.allocated_ram += job.builder.resources["ram"]
+                    self.allocated_cores += job.builder.resources["cores"]
+                thread.start()
                 self.pending_jobs.remove(job)
 
-            def runner():
-                """ Job running thread. """
-                try:
-                    job.run(runtime_context)
-                except WorkflowException as err:
-                    _logger.exception("Got workflow error")
-                    self.exceptions.append(err)
-                except Exception as err:  # pylint: disable=broad-except
-                    _logger.exception("Got workflow error")
-                    self.exceptions.append(WorkflowException(Text(err)))
-                finally:
-                    with runtime_context.workflow_eval_lock:
-                        self.threads.remove(threading.current_thread())
-                        if isinstance(job, JobBase):
-                            self.allocated_ram -= job.builder.resources["ram"]
-                            self.allocated_cores -= job.builder.resources["cores"]
-                        runtime_context.workflow_eval_lock.notifyAll()
-
-            thread = threading.Thread(target=runner)
-            thread.daemon = True
-            self.threads.add(thread)
-            if isinstance(job, JobBase):
-                self.allocated_ram += job.builder.resources["ram"]
-                self.allocated_cores += job.builder.resources["cores"]
-            thread.start()
-
-
     def wait_for_next_completion(self, runtime_context):
         # type: (RuntimeContext) -> None
         """ Wait for jobs to finish. """
@@ -324,6 +343,9 @@ def run_jobs(self,
                     logger.error("Workflow cannot make any more progress.")
                     break
 
+        self.run_job(None, runtime_context)
         while self.threads:
             self.wait_for_next_completion(runtime_context)
+            self.run_job(None, runtime_context)
+
         runtime_context.workflow_eval_lock.release()

cwltool/job.py

@@ -202,6 +202,9 @@ def __init__(self,
         self.timelimit = None  # type: Optional[int]
         self.networkaccess = False  # type: bool
 
+    def __repr__(self):
+        return "CommandLineJob(%s)" % self.name
+
     @abstractmethod
     def run(self,
             runtimeContext  # type: RuntimeContext

cwltool/process.py

@@ -763,8 +763,8 @@ def evalResources(self, builder, runtimeContext):
         request = {
             "coresMin": 1,
             "coresMax": 1,
-            "ramMin": 1024,
-            "ramMax": 1024,
+            "ramMin": 256,
+            "ramMax": 256,
             "tmpdirMin": 1024,
             "tmpdirMax": 1024,
             "outdirMin": 1024,

cwltool/schemas/v1.1.0-dev1/CommandLineTool.yml

@@ -866,7 +866,17 @@ $graph:
     - name: listing
       type:
         - type: array
-          items: ["null", File, Directory, Dirent, string, Expression]
+          items:
+            - "null"
+            - File
+            - type: array
+              items:
+                - File
+                - Directory
+            - Directory
+            - Dirent
+            - string
+            - Expression
         - string
         - Expression
       jsonldPredicate:
@@ -875,8 +885,8 @@ $graph:
         The list of files or subdirectories that must be placed in the
         designated output directory prior to executing the command line tool.
 
-        May be an expression.  If so, the expression return value must validate
-        as `{type: array, items: ["null", File, Directory, Dirent]}`.
+        May be an expression. If so, the expression return value must validate as
+        `{type: array, items: ["null", File, File[], Directory, Directory[], Dirent]}`.
 
         Files or Directories which are listed in the input parameters and
         appear in the `InitialWorkDirRequirement` listing must have their