Merge pull request #390 from djarecka/enh/faster_checksum_states

checksum_states improvements

Author: satra

pydra/engine/core.py

@@ -274,7 +274,16 @@ def checksum_states(self, state_index=None):
                     key.split(".")[1],
                     getattr(inputs_copy, key.split(".")[1])[ind],
                 )
+            # setting files_hash again in case it was cleaned by setting specific element
+            # that might be important for outer splitter of input variable with big files
+            # the file can be changed with every single index even if there are only two files
+            inputs_copy.files_hash = self.inputs.files_hash
             input_hash = inputs_copy.hash
+            # updating self.inputs.files_hash, so big files hashes
+            # doesn't have to be recompute for the next element
+            for key, val in inputs_copy.files_hash.items():
+                if val:
+                    self.inputs.files_hash[key].update(val)
             if is_workflow(self):
                 con_hash = hash_function(self._connections)
                 hash_list = [input_hash, con_hash]

pydra/engine/tests/test_singularity.py

@@ -17,6 +17,10 @@
     shutil.which("singularity") is None, reason="no singularity available"
 )
 
+need_slurm = pytest.mark.skipif(
+    not bool(shutil.which("sbatch")), reason="no singularity available"
+)
+
 
 @need_singularity
 def test_singularity_1_nosubm(tmpdir):
@@ -255,6 +259,26 @@ def test_singularity_st_3(plugin, tmpdir):
     assert "Ubuntu" in res[3].output.stdout
 
 
+@need_singularity
+@need_slurm
+@pytest.mark.xfail(
+    reason="slurm can complain if the number of submitted jobs exceeds the limit"
+)
+@pytest.mark.parametrize("n", [10, 50, 100])
+def test_singularity_st_4(tmpdir, n):
+    """ splitter over args (checking bigger splitters if slurm available)"""
+    args_n = list(range(n))
+    image = "library://sylabsed/linux/alpine"
+    singu = SingularityTask(
+        name="singu", executable="echo", image=image, cache_dir=tmpdir, args=args_n
+    ).split("args")
+    assert singu.state.splitter == "singu.args"
+    res = singu(plugin="slurm")
+    assert "1" in res[1].output.stdout
+    assert str(n - 1) in res[-1].output.stdout
+    assert res[0].output.return_code == res[1].output.return_code == 0
+
+
 @need_singularity
 def test_wf_singularity_1(plugin, tmpdir):
     """ a workflow with two connected task