resolving conflicts

Author: djarecka

.github/workflows/testpydra.yml

@@ -20,7 +20,7 @@ jobs:
     steps:
     - uses: actions/checkout@v2
     - name: Disable etelemetry
-      run:  echo ::set-env name=NO_ET::TRUE
+      run:  echo "NO_ET=TRUE" >> $GITHUB_ENV
     - name: Set up Python ${{ matrix.python-version }} on ${{ matrix.os }}
       uses: actions/setup-python@v2
       with:

.github/workflows/testsingularity.yml

@@ -14,8 +14,8 @@ jobs:
     steps:
       - name: Set env
         run: |
-          echo ::set-env name=RELEASE_VERSION::v3.5.0
-          echo ::set-env name=NO_ET::TRUE
+          echo "RELEASE_VERSION=v3.5.0" >> $GITHUB_ENV
+          echo "NO_ET=TRUE" >> $GITHUB_ENV
       - name: Setup Singularity
         uses: actions/checkout@v2
         with:

.github/workflows/testslurm.yml

@@ -10,7 +10,7 @@ jobs:
 
     steps:
     - name: Disable etelemetry
-      run: echo ::set-env name=NO_ET::TRUE
+      run: echo "NO_ET=TRUE" >> $GITHUB_ENV
     - uses: actions/checkout@v2
     - name: Pull docker image
       run: |

pydra/engine/core.py

@@ -6,7 +6,8 @@
 import os
 from pathlib import Path
 import typing as ty
-from copy import deepcopy, copy
+from copy import deepcopy
+from uuid import uuid4
 
 import cloudpickle as cp
 from filelock import SoftFileLock
@@ -163,7 +164,6 @@ def __init__(
 
         # checking if metadata is set properly
         self.inputs.check_metadata()
-        self.state_inputs = inputs
         # dictionary to save the connections with lazy fields
         self.inp_lf = {}
         self.state = None
@@ -184,6 +184,7 @@ def __init__(
         self.cache_locations = cache_locations
         self.allow_cache_override = True
         self._checksum = None
+        self._uid = uuid4().hex
         # if True the results are not checked (does not propagate to nodes)
         self.task_rerun = rerun
 
@@ -262,6 +263,9 @@ def checksum_states(self, state_index=None):
             TODO
 
         """
+        if is_workflow(self) and self.inputs._graph_checksums is attr.NOTHING:
+            self.inputs._graph_checksums = [nd.checksum for nd in self.graph_sorted]
+
         if state_index is not None:
             inputs_copy = deepcopy(self.inputs)
             for key, ind in self.state.inputs_ind[state_index].items():
@@ -289,6 +293,14 @@ def checksum_states(self, state_index=None):
                 checksum_list.append(self.checksum_states(state_index=ind))
             return checksum_list
 
+    @property
+    def uid(self):
+        """ the unique id number for the task
+            It will be used to create unique names for slurm scripts etc.
+            without a need to run checksum
+        """
+        return self._uid
+
     def set_state(self, splitter, combiner=None):
         """
         Set a particular state on this task.
@@ -410,7 +422,10 @@ def _run(self, rerun=False, **kwargs):
         lockfile = self.cache_dir / (checksum + ".lock")
         # Eagerly retrieve cached - see scenarios in __init__()
         self.hooks.pre_run(self)
-        # TODO add signal handler for processes killed after lock acquisition
+        # adding info file with the checksum in case the task was cancelled
+        # and the lockfile has to be removed
+        with open(self.cache_dir / f"{self.uid}_info.json", "w") as jsonfile:
+            json.dump({"checksum": self.checksum}, jsonfile)
         with SoftFileLock(lockfile):
             if not (rerun or self.task_rerun):
                 result = self.result()
@@ -444,6 +459,8 @@ def _run(self, rerun=False, **kwargs):
                 self.hooks.post_run_task(self, result)
                 self.audit.finalize_audit(result)
                 save(odir, result=result, task=self)
+                # removing the additional file with the chcksum
+                (self.cache_dir / f"{self.uid}_info.json").unlink()
                 # # function etc. shouldn't change anyway, so removing
                 orig_inputs = dict(
                     (k, v) for (k, v) in orig_inputs.items() if not k.startswith("_")
@@ -481,7 +498,6 @@ def split(self, splitter, overwrite=False, **kwargs):
             )
         if kwargs:
             self.inputs = attr.evolve(self.inputs, **kwargs)
-            self.state_inputs = kwargs
         if not self.state or splitter != self.state.splitter:
             self.set_state(splitter)
         return self
@@ -548,8 +564,8 @@ def pickle_task(self):
         """ Pickling the tasks with full inputs"""
         pkl_files = self.cache_dir / "pkl_files"
         pkl_files.mkdir(exist_ok=True, parents=True)
-        task_main_path = pkl_files / f"{self.name}_{self.checksum}_task.pklz"
-        save(task_path=pkl_files, task=self, name_prefix=f"{self.name}_{self.checksum}")
+        task_main_path = pkl_files / f"{self.name}_{self.uid}_task.pklz"
+        save(task_path=pkl_files, task=self, name_prefix=f"{self.name}_{self.uid}")
         return task_main_path
 
     @property
@@ -935,7 +951,6 @@ async def _run(self, submitter=None, rerun=False, **kwargs):
                 "Workflow output cannot be None, use set_output to define output(s)"
             )
         checksum = self.checksum
-        lockfile = self.cache_dir / (checksum + ".lock")
         # Eagerly retrieve cached
         if not (rerun or self.task_rerun):
             result = self.result()
@@ -953,6 +968,11 @@ async def _run(self, submitter=None, rerun=False, **kwargs):
             task.cache_locations = task._cache_locations + self.cache_locations
             self.create_connections(task)
         # TODO add signal handler for processes killed after lock acquisition
+        # adding info file with the checksum in case the task was cancelled
+        # and the lockfile has to be removed
+        with open(self.cache_dir / f"{self.uid}_info.json", "w") as jsonfile:
+            json.dump({"checksum": checksum}, jsonfile)
+        lockfile = self.cache_dir / (checksum + ".lock")
         self.hooks.pre_run(self)
         with SoftFileLock(lockfile):
             # # Let only one equivalent process run
@@ -977,6 +997,8 @@ async def _run(self, submitter=None, rerun=False, **kwargs):
                 self.hooks.post_run_task(self, result)
                 self.audit.finalize_audit(result=result)
                 save(odir, result=result, task=self)
+                # removing the additional file with the chcksum
+                (self.cache_dir / f"{self.uid}_info.json").unlink()
                 os.chdir(cwd)
         self.hooks.post_run(self, result)
         return result

pydra/engine/helpers.py

@@ -8,6 +8,7 @@
 import os
 import sys
 from hashlib import sha256
+from uuid import uuid4
 import subprocess as sp
 import getpass
 import re
@@ -27,6 +28,8 @@
     LazyField,
     MultiOutputObj,
     MultiInputObj,
+    MultiInputFile,
+    MultiOutputFile,
 )
 from .helpers_file import hash_file, hash_dir, copyfile, is_existing_file
 
@@ -311,7 +314,15 @@ def custom_validator(instance, attribute, value):
         or value is None
         or attribute.name.startswith("_")  # e.g. _func
         or isinstance(value, LazyField)
-        or tp_attr in [ty.Any, inspect._empty, MultiOutputObj, MultiInputObj]
+        or tp_attr
+        in [
+            ty.Any,
+            inspect._empty,
+            MultiOutputObj,
+            MultiInputObj,
+            MultiOutputFile,
+            MultiInputFile,
+        ]
     ):
         check_type = False  # no checking of the type
     elif isinstance(tp_attr, type) or tp_attr in [File, Directory]:
@@ -651,14 +662,16 @@ def hash_function(obj):
     return sha256(str(obj).encode()).hexdigest()
 
 
-def hash_value(value, tp=None, metadata=None):
+def hash_value(value, tp=None, metadata=None, precalculated=None):
     """calculating hash or returning values recursively"""
     if metadata is None:
         metadata = {}
     if isinstance(value, (tuple, list)):
-        return [hash_value(el, tp, metadata) for el in value]
+        return [hash_value(el, tp, metadata, precalculated) for el in value]
     elif isinstance(value, dict):
-        dict_hash = {k: hash_value(v, tp, metadata) for (k, v) in value.items()}
+        dict_hash = {
+            k: hash_value(v, tp, metadata, precalculated) for (k, v) in value.items()
+        }
         # returning a sorted object
         return [list(el) for el in sorted(dict_hash.items(), key=lambda x: x[0])]
     else:  # not a container
@@ -667,13 +680,13 @@ def hash_value(value, tp=None, metadata=None):
             and is_existing_file(value)
             and "container_path" not in metadata
         ):
-            return hash_file(value)
+            return hash_file(value, precalculated=precalculated)
         elif (
             (tp is File or "pydra.engine.specs.Directory" in str(tp))
             and is_existing_file(value)
             and "container_path" not in metadata
         ):
-            return hash_dir(value)
+            return hash_dir(value, precalculated=precalculated)
         else:
             return value
 
@@ -793,6 +806,8 @@ def load_task(task_pkl, ind=None):
         _, inputs_dict = task.get_input_el(ind)
         task.inputs = attr.evolve(task.inputs, **inputs_dict)
         task.state = None
+        # resetting uid for task
+        task._uid = uuid4().hex
     return task