Merge branch '13301-cwl-resource-scatter' closes #13301

Arvados-DCO-1.1-Signed-off-by: Peter Amstutz <pamstutz@veritasgenetics.com>

diff --git a/sdk/cwl/arvados_cwl/arvworkflow.py b/sdk/cwl/arvados_cwl/arvworkflow.py
index 5aed871a12bc58d1a747efd2035dbf9d2a23b5a4..bdc2e274b0daa695e9b0f2cdcfe698f53f502730 100644 (file)
--- a/sdk/cwl/arvados_cwl/arvworkflow.py
+++ b/sdk/cwl/arvados_cwl/arvworkflow.py
@@ -89,7 +89,7 @@ def get_overall_res_req(res_reqs):
                 if isinstance(res_req[a], int): # integer check
                     all_res_req[a].append(res_req[a])
                 else:
-                    msg = SourceLine(res_req).makeError(
+                    msg = SourceLine(res_req, a).makeError(
                     "Non-top-level ResourceRequirement in single container cannot have expressions")
                     exception_msgs.append(msg)
     if exception_msgs:
@@ -114,6 +114,8 @@ class ArvadosWorkflow(Workflow):
         self.arvrunner = arvrunner
         self.work_api = kwargs["work_api"]
         self.wf_pdh = None
+        self.dynamic_resource_req = []
+        self.static_resource_req = []
 
     def job(self, joborder, output_callback, **kwargs):
         kwargs["work_api"] = self.work_api
@@ -146,32 +148,31 @@ class ArvadosWorkflow(Workflow):
                     builder.hints = workflowobj["hints"]
                     builder.resources = {}
 
-                    res_reqs = {"requirements": [], "hints": []}
-                    for t in ("requirements", "hints"):
-                        for item in packed["$graph"]:
-                            if t in item:
-                                if item["id"] == "#main": # evaluate potential expressions in the top-level requirements/hints
-                                    for req in item[t]:
-                                        if req["class"] == "ResourceRequirement":
-                                            eval_req = {"class": "ResourceRequirement"}
-                                            for a in max_res_pars + sum_res_pars:
-                                                if a in req:
-                                                    eval_req[a] = builder.do_eval(req[a])
-                                            res_reqs[t].append(eval_req)
-                                else:
-                                    for req in item[t]:
-                                        if req["class"] == "ResourceRequirement":
-                                            res_reqs[t].append(req)
-                    overall_res_req = {"requirements": get_overall_res_req(res_reqs["requirements"]),
-                                       "hints": get_overall_res_req(res_reqs["hints"])}
-
-                    new_spec = {"requirements": self.requirements, "hints": self.hints}
-                    for t in ("requirements", "hints"):
-                        for req in new_spec[t]:
-                            if req["class"] == "ResourceRequirement":
-                                new_spec[t].remove(req)
-                        if overall_res_req[t]:
-                            new_spec[t].append(overall_res_req[t])
+                    def visit(item):
+                        for t in ("hints", "requirements"):
+                            if t not in item:
+                                continue
+                            for req in item[t]:
+                                if req["class"] == "ResourceRequirement":
+                                    dyn = False
+                                    for k in max_res_pars + sum_res_pars:
+                                        if k in req:
+                                            if isinstance(req[k], basestring):
+                                                if item["id"] == "#main":
+                                                    # only the top-level requirements/hints may contain expressions
+                                                    self.dynamic_resource_req.append(req)
+                                                    dyn = True
+                                                    break
+                                                else:
+                                                    with SourceLine(req, k, WorkflowException):
+                                                        raise WorkflowException("Non-top-level ResourceRequirement in single container cannot have expressions")
+                                    if not dyn:
+                                        self.static_resource_req.append(req)
+
+                    visit_class(packed["$graph"], ("Workflow", "CommandLineTool"), visit)
+
+                    if self.static_resource_req:
+                        self.static_resource_req = [get_overall_res_req(self.static_resource_req)]
 
                     upload_dependencies(self.arvrunner,
                                         kwargs.get("name", ""),
@@ -180,6 +181,25 @@ class ArvadosWorkflow(Workflow):
                                         uri,
                                         False)
 
+            if self.dynamic_resource_req:
+                builder = Builder()
+                builder.job = joborder
+                builder.requirements = self.requirements
+                builder.hints = self.hints
+                builder.resources = {}
+
+                # Evaluate dynamic resource requirements using current builder
+                rs = copy.copy(self.static_resource_req)
+                for dyn_rs in self.dynamic_resource_req:
+                    eval_req = {"class": "ResourceRequirement"}
+                    for a in max_res_pars + sum_res_pars:
+                        if a in dyn_rs:
+                            eval_req[a] = builder.do_eval(dyn_rs[a])
+                    rs.append(eval_req)
+                job_res_reqs = [get_overall_res_req(rs)]
+            else:
+                job_res_reqs = self.static_resource_req
+
             with Perf(metrics, "subworkflow adjust"):
                 joborder_resolved = copy.deepcopy(joborder)
                 joborder_keepmount = copy.deepcopy(joborder)
@@ -226,7 +246,7 @@ class ArvadosWorkflow(Workflow):
                 "inputs": self.tool["inputs"],
                 "outputs": self.tool["outputs"],
                 "stdout": "cwl.output.json",
-                "requirements": self.requirements+[
+                "requirements": self.requirements+job_res_reqs+[
                     {
                     "class": "InitialWorkDirRequirement",
                     "listing": [{

diff --git a/sdk/cwl/tests/arvados-tests.yml b/sdk/cwl/tests/arvados-tests.yml
index ea6477cfe8b2c58facd80602c01786da1f0a9677..87db44b094f9c234280c7c7e37bc5be5e9d5d313 100644 (file)
--- a/sdk/cwl/tests/arvados-tests.yml
+++ b/sdk/cwl/tests/arvados-tests.yml
@@ -146,3 +146,28 @@
   }
   tool: wf/secret_wf.cwl
   doc: "Test secret input parameters"
+
+- job: null
+  output:
+    out: null
+  tool: wf/runin-reqs-wf.cwl
+  doc: "RunInSingleContainer handles dynamic resource requests on step"
+
+- job: null
+  output:
+    out: null
+  tool: wf/runin-reqs-wf2.cwl
+  doc: "RunInSingleContainer handles dynamic resource requests on embedded subworkflow"
+
+- job: null
+  output:
+    out: null
+  tool: wf/runin-reqs-wf3.cwl
+  should_fail: true
+  doc: "RunInSingleContainer disallows dynamic resource request on subworkflow steps"
+
+- job: null
+  output:
+    out: null
+  tool: wf/runin-reqs-wf4.cwl
+  doc: "RunInSingleContainer discovers static resource request in subworkflow steps"

diff --git a/sdk/cwl/tests/wf/runin-reqs-wf.cwl b/sdk/cwl/tests/wf/runin-reqs-wf.cwl
new file mode 100644 (file)
index 0000000..9032e26
--- /dev/null
+++ b/sdk/cwl/tests/wf/runin-reqs-wf.cwl
@@ -0,0 +1,58 @@
+# Copyright (C) The Arvados Authors. All rights reserved.
+#
+# SPDX-License-Identifier: Apache-2.0
+
+class: Workflow
+cwlVersion: v1.0
+$namespaces:
+  arv: "http://arvados.org/cwl#"
+inputs:
+  count:
+    type: int[]
+    default: [1, 2, 3, 4]
+  script:
+    type: File
+    default:
+      class: File
+      location: check_mem.py
+outputs:
+  out: []
+requirements:
+  SubworkflowFeatureRequirement: {}
+  ScatterFeatureRequirement: {}
+  InlineJavascriptRequirement: {}
+  StepInputExpressionRequirement: {}
+steps:
+  substep:
+    in:
+      count: count
+      script: script
+    out: []
+    hints:
+      - class: arv:RunInSingleContainer
+      - class: ResourceRequirement
+        ramMin: $(inputs.count*4)
+      - class: arv:APIRequirement
+    scatter: count
+    run:
+      class: Workflow
+      id: mysub
+      inputs:
+        count: int
+        script: File
+      outputs: []
+      steps:
+        sleep1:
+          in:
+            count: count
+            script: script
+          out: []
+          run:
+            class: CommandLineTool
+            id: subtool
+            inputs:
+              count:
+                type: int
+              script: File
+            outputs: []
+            arguments: [python, $(inputs.script), $(inputs.count * 4)]

diff --git a/sdk/cwl/tests/wf/runin-reqs-wf2.cwl b/sdk/cwl/tests/wf/runin-reqs-wf2.cwl
new file mode 100644 (file)
index 0000000..cc1321a
--- /dev/null
+++ b/sdk/cwl/tests/wf/runin-reqs-wf2.cwl
@@ -0,0 +1,59 @@
+# Copyright (C) The Arvados Authors. All rights reserved.
+#
+# SPDX-License-Identifier: Apache-2.0
+
+class: Workflow
+cwlVersion: v1.0
+$namespaces:
+  arv: "http://arvados.org/cwl#"
+inputs:
+  count:
+    type: int[]
+    default: [1, 2, 3, 4]
+  script:
+    type: File
+    default:
+      class: File
+      location: check_mem.py
+outputs:
+  out: []
+requirements:
+  SubworkflowFeatureRequirement: {}
+  ScatterFeatureRequirement: {}
+  InlineJavascriptRequirement: {}
+  StepInputExpressionRequirement: {}
+steps:
+  substep:
+    in:
+      count: count
+      script: script
+    out: []
+    hints:
+      - class: arv:RunInSingleContainer
+      - class: arv:APIRequirement
+    scatter: count
+    run:
+      class: Workflow
+      id: mysub
+      inputs:
+        count: int
+        script: File
+      outputs: []
+      hints:
+        - class: ResourceRequirement
+          ramMin: $(inputs.count*4)
+      steps:
+        sleep1:
+          in:
+            count: count
+            script: script
+          out: []
+          run:
+            class: CommandLineTool
+            id: subtool
+            inputs:
+              count:
+                type: int
+              script: File
+            outputs: []
+            arguments: [python, $(inputs.script), $(inputs.count * 4)]

diff --git a/sdk/cwl/tests/wf/runin-reqs-wf3.cwl b/sdk/cwl/tests/wf/runin-reqs-wf3.cwl
new file mode 100644 (file)
index 0000000..92bf482
--- /dev/null
+++ b/sdk/cwl/tests/wf/runin-reqs-wf3.cwl
@@ -0,0 +1,59 @@
+# Copyright (C) The Arvados Authors. All rights reserved.
+#
+# SPDX-License-Identifier: Apache-2.0
+
+class: Workflow
+cwlVersion: v1.0
+$namespaces:
+  arv: "http://arvados.org/cwl#"
+inputs:
+  count:
+    type: int[]
+    default: [1, 2, 3, 4]
+  script:
+    type: File
+    default:
+      class: File
+      location: check_mem.py
+outputs:
+  out: []
+requirements:
+  SubworkflowFeatureRequirement: {}
+  ScatterFeatureRequirement: {}
+  InlineJavascriptRequirement: {}
+  StepInputExpressionRequirement: {}
+steps:
+  substep:
+    in:
+      count: count
+      script: script
+    out: []
+    hints:
+      - class: arv:RunInSingleContainer
+      - class: arv:APIRequirement
+    scatter: count
+    run:
+      class: Workflow
+      id: mysub
+      inputs:
+        count: int
+        script: File
+      outputs: []
+      steps:
+        sleep1:
+          in:
+            count: count
+            script: script
+          out: []
+          run:
+            class: CommandLineTool
+            id: subtool
+            hints:
+              - class: ResourceRequirement
+                ramMin: $(inputs.count*4)
+            inputs:
+              count:
+                type: int
+              script: File
+            outputs: []
+            arguments: [python, $(inputs.script), $(inputs.count * 4)]

diff --git a/sdk/cwl/tests/wf/runin-reqs-wf4.cwl b/sdk/cwl/tests/wf/runin-reqs-wf4.cwl
new file mode 100644 (file)
index 0000000..b7a9779
--- /dev/null
+++ b/sdk/cwl/tests/wf/runin-reqs-wf4.cwl
@@ -0,0 +1,59 @@
+# Copyright (C) The Arvados Authors. All rights reserved.
+#
+# SPDX-License-Identifier: Apache-2.0
+
+class: Workflow
+cwlVersion: v1.0
+$namespaces:
+  arv: "http://arvados.org/cwl#"
+inputs:
+  count:
+    type: int[]
+    default: [1, 2, 3, 4]
+  script:
+    type: File
+    default:
+      class: File
+      location: check_mem.py
+outputs:
+  out: []
+requirements:
+  SubworkflowFeatureRequirement: {}
+  ScatterFeatureRequirement: {}
+  InlineJavascriptRequirement: {}
+  StepInputExpressionRequirement: {}
+steps:
+  substep:
+    in:
+      count: count
+      script: script
+    out: []
+    hints:
+      - class: arv:RunInSingleContainer
+      - class: arv:APIRequirement
+    scatter: count
+    run:
+      class: Workflow
+      id: mysub
+      inputs:
+        count: int
+        script: File
+      outputs: []
+      steps:
+        sleep1:
+          in:
+            count: count
+            script: script
+          out: []
+          run:
+            class: CommandLineTool
+            id: subtool
+            hints:
+              - class: ResourceRequirement
+                ramMin: 8
+            inputs:
+              count:
+                type: int
+              script: File
+            outputs: []
+            arguments: [python, $(inputs.script), "8"]