Merge pull request #1363 from common-workflow-language/opaque_resources

support opaque resources

Author: mr-c

cwltool/builder.py

@@ -158,7 +158,7 @@ def __init__(
         names: Names,
         requirements: List[CWLObjectType],
         hints: List[CWLObjectType],
-        resources: Dict[str, Union[int, float]],
+        resources: Dict[str, Union[int, float, str]],
         mutation_manager: Optional[MutationManager],
         formatgraph: Optional[Graph],
         make_fs_access: Type[StdFsAccess],
@@ -635,8 +635,10 @@ def do_eval(
 
         resources = self.resources
         if self.resources and "cores" in self.resources:
-            resources = copy.copy(resources)
-            resources["cores"] = int(math.ceil(resources["cores"]))
+            cores = resources["cores"]
+            if not isinstance(cores, str):
+                resources = copy.copy(resources)
+                resources["cores"] = int(math.ceil(cores))
 
         return expression.do_eval(
             ex,

cwltool/context.py

@@ -81,7 +81,8 @@ class RuntimeContext(ContextBase):
     def __init__(self, kwargs: Optional[Dict[str, Any]] = None) -> None:
         """Initialize the RuntimeContext from the kwargs."""
         select_resources_callable = Callable[  # pylint: disable=unused-variable
-            [Dict[str, Union[int, float]], RuntimeContext], Dict[str, Union[int, float]]
+            [Dict[str, Union[int, float, str]], RuntimeContext],
+            Dict[str, Union[int, float, str]],
         ]
         self.user_space_docker_cmd = ""  # type: Optional[str]
         self.secret_store = None  # type: Optional[SecretStore]

cwltool/docker.py

@@ -446,7 +446,9 @@ def create_runtime(
             runtime.append("--env=%s=%s" % (key, value))
 
         if runtimeContext.strict_memory_limit and not user_space_docker_cmd:
-            runtime.append("--memory=%dm" % self.builder.resources["ram"])
+            ram = self.builder.resources["ram"]
+            if not isinstance(ram, str):
+                runtime.append("--memory=%dm" % ram)
         elif not user_space_docker_cmd:
             res_req, _ = self.builder.get_requirement("ResourceRequirement")
             if res_req and ("ramMin" in res_req or "ramMax" in res_req):

cwltool/executors.py

@@ -293,23 +293,33 @@ def __init__(self) -> None:
     def select_resources(
         self, request, runtime_context
     ):  # pylint: disable=unused-argument
-        # type: (Dict[str, Union[int, float]], RuntimeContext) -> Dict[str, Union[int, float]]
+        # type: (Dict[str, Union[int, float, str]], RuntimeContext) -> Dict[str, Union[int, float, str]]
         """Naïve check for available cpu cores and memory."""
-        result = {}  # type: Dict[str, Union[int, float]]
+        result = {}  # type: Dict[str, Union[int, float, str]]
         maxrsc = {"cores": self.max_cores, "ram": self.max_ram}
         for rsc in ("cores", "ram"):
-            if request[rsc + "Min"] > maxrsc[rsc]:
+            rsc_min = request[rsc + "Min"]
+            if not isinstance(rsc_min, str) and rsc_min > maxrsc[rsc]:
                 raise WorkflowException(
                     "Requested at least %d %s but only %d available"
-                    % (request[rsc + "Min"], rsc, maxrsc[rsc])
+                    % (rsc_min, rsc, maxrsc[rsc])
                 )
-            if request[rsc + "Max"] < maxrsc[rsc]:
-                result[rsc] = math.ceil(request[rsc + "Max"])
+            rsc_max = request[rsc + "Max"]
+            if not isinstance(rsc_max, str) and rsc_max < maxrsc[rsc]:
+                result[rsc] = math.ceil(rsc_max)
             else:
                 result[rsc] = maxrsc[rsc]
 
-        result["tmpdirSize"] = math.ceil(request["tmpdirMin"])
-        result["outdirSize"] = math.ceil(request["outdirMin"])
+        result["tmpdirSize"] = (
+            math.ceil(request["tmpdirMin"])
+            if not isinstance(request["tmpdirMin"], str)
+            else request["tmpdirMin"]
+        )
+        result["outdirSize"] = (
+            math.ceil(request["outdirMin"])
+            if not isinstance(request["outdirMin"], str)
+            else request["outdirMin"]
+        )
 
         return result
 
@@ -334,8 +344,12 @@ def _runner(self, job, runtime_context, TMPDIR_LOCK):
                 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"]
+                        ram = job.builder.resources["ram"]
+                        if not isinstance(ram, str):
+                            self.allocated_ram -= ram
+                        cores = job.builder.resources["cores"]
+                        if not isinstance(cores, str):
+                            self.allocated_cores -= cores
                     runtime_context.workflow_eval_lock.notifyAll()
 
     def run_job(
@@ -353,9 +367,11 @@ def run_job(
             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:
+                    ram = job.builder.resources["ram"]
+                    cores = job.builder.resources["cores"]
+                    if (not isinstance(ram, str) and ram > self.max_ram) or (
+                        not isinstance(cores, str) and 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",
@@ -370,10 +386,12 @@ def run_job(
                         return
 
                     if (
-                        self.allocated_ram + job.builder.resources["ram"]
-                    ) > self.max_ram or (
-                        self.allocated_cores + job.builder.resources["cores"]
-                    ) > self.max_cores:
+                        not isinstance(ram, str)
+                        and self.allocated_ram + ram > self.max_ram
+                    ) or (
+                        not isinstance(cores, str)
+                        and self.allocated_cores + 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, "
@@ -394,8 +412,12 @@ def run_job(
                 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"]
+                    ram = job.builder.resources["ram"]
+                    if not isinstance(ram, str):
+                        self.allocated_ram += ram
+                    cores = job.builder.resources["cores"]
+                    if not isinstance(cores, str):
+                        self.allocated_cores += cores
                 thread.start()
                 self.pending_jobs.remove(job)
 

cwltool/expression.py

@@ -345,7 +345,7 @@ def do_eval(
     requirements: List[CWLObjectType],
     outdir: Optional[str],
     tmpdir: Optional[str],
-    resources: Dict[str, Union[float, int]],
+    resources: Dict[str, Union[float, int, str]],
     context: Optional[CWLOutputType] = None,
     timeout: float = default_timeout,
     force_docker_pull: bool = False,

cwltool/main.py

@@ -101,8 +101,10 @@
     windows_default_container_id,
 )
 from .workflow import Workflow
+
 # PYTHON_ARGCOMPLETE_OK
 
+
 def _terminate_processes() -> None:
     """Kill all spawned processes.
 
@@ -397,7 +399,7 @@ def init_job_order(
                 k: v for k, v in cmd_line.items() if k.startswith(record_name)
             }
             for key, value in record_items.items():
-                record[key[len(record_name) + 1:]] = value
+                record[key[len(record_name) + 1 :]] = value
                 del cmd_line[key]
             cmd_line[str(record_name)] = record
         if "job_order" in cmd_line and cmd_line["job_order"]:

cwltool/process.py

@@ -962,7 +962,7 @@ def inc(d):  # type: (List[int]) -> None
 
     def evalResources(
         self, builder: Builder, runtimeContext: RuntimeContext
-    ) -> Dict[str, Union[int, float]]:
+    ) -> Dict[str, Union[int, float, str]]:
         resourceReq, _ = self.get_requirement("ResourceRequirement")
         if resourceReq is None:
             resourceReq = {}
@@ -973,7 +973,7 @@ def evalResources(
             ram = 1024
         else:
             ram = 256
-        request = {
+        request: Dict[str, Union[int, float, str]] = {
             "coresMin": 1,
             "coresMax": 1,
             "ramMin": ram,
@@ -982,7 +982,7 @@ def evalResources(
             "tmpdirMax": 1024,
             "outdirMin": 1024,
             "outdirMax": 1024,
-        }  # type: Dict[str, Union[int, float]]
+        }
         for a in ("cores", "ram", "tmpdir", "outdir"):
             mn = mx = None  # type: Optional[Union[int, float]]
             if resourceReq.get(a + "Min"):
@@ -1012,9 +1012,15 @@ def evalResources(
             return runtimeContext.select_resources(request, runtimeContext)
         return {
             "cores": request["coresMin"],
-            "ram": math.ceil(request["ramMin"]),
-            "tmpdirSize": math.ceil(request["tmpdirMin"]),
-            "outdirSize": math.ceil(request["outdirMin"]),
+            "ram": math.ceil(request["ramMin"])
+            if not isinstance(request["ramMin"], str)
+            else request["ramMin"],
+            "tmpdirSize": math.ceil(request["tmpdirMin"])
+            if not isinstance(request["tmpdirMin"], str)
+            else request["tmpdirMin"],
+            "outdirSize": math.ceil(request["outdirMin"])
+            if not isinstance(request["outdirMin"], str)
+            else request["outdirMin"],
         }
 
     def validate_hints(

cwltool/singularity.py

@@ -47,7 +47,10 @@ def _singularity_supports_userns() -> bool:
                 stdout=DEVNULL,
                 universal_newlines=True,
             ).communicate(timeout=60)[1]
-            _USERNS = "No valid /bin/sh" in result or "/bin/sh doesn't exist in container" in result
+            _USERNS = (
+                "No valid /bin/sh" in result
+                or "/bin/sh doesn't exist in container" in result
+            )
         except TimeoutExpired:
             _USERNS = False
     return _USERNS