feat: Add from-link-prefix variables

Author: Alan Christie

tests/job-definitions/job-definitions.yaml

@@ -131,13 +131,19 @@ jobs:
 
   concatenate:
     command: >-
-      concatenate.py {% for ifile in  inputFile %}{{ ifile }} {% endfor %} --outputFile {{ outputFile }}
+      concatenate.py --inputFile {{ inputFile }} --outputFile {{ outputFile }}
     # Simulate a multiple input files Job (combiner)...
     variables:
       inputs:
         properties:
           inputFile:
             type: files
+      options:
+        type: object
+        properties:
+          inputDirPrefix:
+            title: Optional inoput directory prefix
+            type: string
       outputs:
         properties:
           outputBase:

tests/jobs/concatenate.py

@@ -2,13 +2,16 @@
 
 parser = argparse.ArgumentParser(
     prog="addcol",
-    description="Takes a list of files and writes them into single outputfile",
+    description="Takes an optional directory prefix and a file,"
+    " and combines all the input files that are found"
+    " into single outputfile",
 )
-parser.add_argument("inputFile", nargs="+", type=argparse.FileType("r"))
+parser.add_argument("--inputDirPrefix")
+parser.add_argument("--inputFile", required=True)
 parser.add_argument("-o", "--outputFile", required=True)
 args = parser.parse_args()
 
 
 with open(args.outputFile, "wt", encoding="utf8") as ofile:
-    for f in args.inputFile:
-        ofile.write(f.read())
+    with open(args.inputFile, "rt", encoding="utf8") as ifile:
+        ofile.write(ifile.read())

tests/workflow-definitions/simple-python-split-combine.yaml

@@ -46,12 +46,14 @@ steps:
     variables:
       outputFile: results.smi
   plumbing:
+  - variable: outputFile
+    from-workflow:
+      variable: combination
   - variable: inputFile
     from-step:
       name: parallel
       variable: outputFile
-  - variable: outputFile
-    from-workflow:
-      variable: combination
+  - variable: inputDirPrefix
+    from-link-prefix:
   - variable: outputFile
     to-project:

workflow/decoder.py

@@ -144,7 +144,7 @@ def get_step_workflow_variable_connections(
     return connections
 
 
-def get_step_prior_step_plumbing(
+def get_step_prior_step_connections(
     *, step_definition: dict[str, Any]
 ) -> dict[str, list[Connector]]:
     """Returns list of variable Connections, indexed by prior step name,
@@ -166,3 +166,14 @@ def get_step_prior_step_plumbing(
                         Connector(in_=step_variable, out=v_map["variable"])
                     ]
     return plumbing
+
+
+def get_step_link_prefix_variables(*, step_definition: dict[str, Any]) -> set[str]:
+    """Returns the set of variables expected to be set to the value
+    of the instance directory prefix."""
+    variables: set[str] = set()
+    if "plumbing" in step_definition:
+        for v_map in step_definition["plumbing"]:
+            if "from-link-prefix" in v_map:
+                variables.add(v_map["variable"])
+    return variables

workflow/workflow-schema.yaml

@@ -102,6 +102,29 @@ definitions:
     - variable
     - from-workflow
 
+  # A Step variable
+  # (whose value is set to the value of a directory prefix used when the DM
+  # links the instance directories of prior step instances into this
+  # step's instance directory)
+  #
+  # This _must_ be treated by the step's job as a directory prefix,
+  # typiclaly '.instance-', that can be used to identify directories in this step's
+  # execution directory where the execution directory of prior steps
+  # are hard-linked by the DM. A job can find all the prior step directory names
+  # using the selected variable (e.g. inspect any directory name
+  # that starts with "{variable}").
+  step-variable-from-link-prefix:
+    type: object
+    additionalProperties: false
+    properties:
+      variable:
+        $ref: '#/definitions/variable-name'
+      from-link-prefix:
+        type: 'null'
+    required:
+    - variable
+    - from-link-prefix
+
   # A Step variable
   # (whose value (a file) is to be copied to the project directory)
   step-variable-to-project:
@@ -181,6 +204,7 @@ definitions:
           anyOf:
           - $ref: "#/definitions/step-variable-from-step"
           - $ref: "#/definitions/step-variable-from-workflow"
+          - $ref: "#/definitions/step-variable-from-link-prefix"
           - $ref: "#/definitions/step-variable-to-project"
         minItems: 1
     required:

workflow/workflow_engine.py

@@ -43,7 +43,8 @@
 from .decoder import (
     Connector,
     get_step,
-    get_step_prior_step_plumbing,
+    get_step_link_prefix_variables,
+    get_step_prior_step_connections,
     get_step_workflow_variable_connections,
 )
 
@@ -77,10 +78,15 @@ def __init__(
         *,
         wapi_adapter: WorkflowAPIAdapter,
         instance_launcher: InstanceLauncher,
+        step_link_prefix: str = ".instance-",
     ):
+        """Initialiser, given a Workflow API adapter, Instance laucnher,
+        and a step (directory) link prefix (the directory prefix the DM uses to hard-link
+        prior step instanes into the next step, typically '.instance-')"""
         # Keep the dependent objects
         self._wapi_adapter = wapi_adapter
         self._instance_launcher = instance_launcher
+        self._step_link_prefix = step_link_prefix
 
     def handle_message(self, msg: Message) -> None:
         """Expect Workflow and Pod messages.
@@ -393,7 +399,7 @@ def _prepare_step(
         our_inputs: dict[str, Any] = job_defintion_decoder.get_inputs(
             our_job_definition
         )
-        our_plumbing: dict[str, list[Connector]] = get_step_prior_step_plumbing(
+        our_plumbing: dict[str, list[Connector]] = get_step_prior_step_connections(
             step_definition=step_definition
         )
         step_is_combiner: bool = False
@@ -483,60 +489,40 @@ def _prepare_step(
             assert connector.in_ in rwf_variables
             variables[connector.out] = rwf_variables[connector.in_]
 
+        # Process the step's "plumbing" relating to link-prefix variables.
+        #
+        # This will be a set of variable names. We just set each one
+        # to the built-in step link prefix.
+        for link_variable in get_step_link_prefix_variables(
+            step_definition=step_definition
+        ):
+            variables[link_variable] = self._step_link_prefix
+
         # Now process variables (in the "plumbing" block)
         # that relate to values used in prior steps.
         #
         # The decoder gives us a map indexed by prior step name that's a list of
-        # "in" "out" connectors as above. If this is a combiner step remember
+        # "in"/"out" connectors as above. If this is a combiner step remember
         # that the combiner_input_variable is a used as a list.
-        prior_step_plumbing: dict[str, list[Connector]] = get_step_prior_step_plumbing(
-            step_definition=step_definition
+        prior_step_plumbing: dict[str, list[Connector]] = (
+            get_step_prior_step_connections(step_definition=step_definition)
         )
         for prior_step_name, connections in prior_step_plumbing.items():
-            if step_is_combiner and prior_step_name == step_name_being_combined:
-                assert combiner_input_variable
-                input_source_list: list[str] = []
-                for replica in range(num_step_recplicas_being_combined):
-                    prior_step, _ = (
-                        self._wapi_adapter.get_running_workflow_step_by_name(
-                            name=prior_step_name,
-                            replica=replica,
-                            running_workflow_id=rwf_id,
-                        )
-                    )
-                    # Copy "in" value to "out"...
-                    # accumulating thiose for the 'combining' variable,
-                    # which will be set as a list when we're done.
-                    for connector in connections:
-                        assert connector.in_ in prior_step["variables"]
-                        if connector.out == combiner_input_variable:
-                            # Each instance may have a different value
-                            input_source_list.append(
-                                prior_step["variables"][connector.in_]
-                            )
-                        elif replica == 0:
-                            # Only the first instance value are of interest,
-                            # the rest wil be the same - only one variable
-                            # is a list of different values.
-                            variables[connector.out] = prior_step["variables"][
-                                connector.in_
-                            ]
-                # Now we have accumulated the prior steps values (files)
-                # set the combiner's corresponding input variable...
-                variables[combiner_input_variable] = input_source_list
-            else:
-                # Not a preior step for a combiner,
-                # or not a step being combined in a combiner.
-                #
-                # Retrieve the prior "running" step
-                # in order to get the variables that were set there...
-                prior_step, _ = self._wapi_adapter.get_running_workflow_step_by_name(
-                    name=prior_step_name, running_workflow_id=rwf_id
-                )
-                # Copy "in" value to "out"...
-                for connector in connections:
-                    assert connector.in_ in prior_step["variables"]
-                    variables[connector.out] = prior_step["variables"][connector.in_]
+            # Retrieve the first prior "running" step in order to get the variables
+            # that were used for it.
+            #
+            # For a combiner step we only need to inspect the first instance of
+            # the prior step (the default replica value is '0').
+            # We assume all the combiner's prior (parallel) instances
+            # have the same variables and values.
+            prior_step, _ = self._wapi_adapter.get_running_workflow_step_by_name(
+                name=prior_step_name,
+                running_workflow_id=rwf_id,
+            )
+            # Copy "in" value to "out"...
+            for connector in connections:
+                assert connector.in_ in prior_step["variables"]
+                variables[connector.out] = prior_step["variables"][connector.in_]
 
         # All variables are set ...
         # is this enough to satisfy the setp's Job command?