Merge pull request #4 from nipype/master

Merging the most updated pydra with my forked version of the repo.

Author: chasejohnson3

.github/workflows/testsingularity.yml

@@ -34,14 +34,14 @@ jobs:
       - name: Build
         run: |
           cd singularity
-          ./mconfig --without-suid -p ${{ runner.tool_cache }}/singularity/${{ env.RELEASE_VERSION }}/x64
+          ./mconfig --without-suid -p /usr/local/
           make -C ./builddir
           sudo make -C ./builddir install
           cd ..
       - name: Echo singularity version
         run: |
           echo ${{ github.ref }}
-          ${{ runner.tool_cache }}/singularity/${{ env.RELEASE_VERSION }}/x64/bin/singularity --version
+          singularity --version
 
 
       - name: Set up Python ${{ matrix.python-version }}

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]
@@ -392,26 +401,28 @@ def output_dir(self):
             return [self._cache_dir / checksum for checksum in self.checksum_states()]
         return self._cache_dir / self.checksum
 
-    def __call__(self, submitter=None, plugin=None, rerun=False, **kwargs):
+    def __call__(
+        self, submitter=None, plugin=None, plugin_kwargs=None, rerun=False, **kwargs
+    ):
         """Make tasks callable themselves."""
         from .submitter import Submitter
 
         if submitter and plugin:
             raise Exception("Specify submitter OR plugin, not both")
-        plugin = plugin or self.plugin
-        if plugin:
-            submitter = Submitter(plugin=plugin)
-        elif self.state:
-            submitter = Submitter()
+        elif submitter:
+            pass
+        # if there is plugin provided or the task is a Workflow or has a state,
+        # the submitter will be created using provided plugin, self.plugin or "cf"
+        elif plugin or self.state or is_workflow(self):
+            plugin = plugin or self.plugin or "cf"
+            if plugin_kwargs is None:
+                plugin_kwargs = {}
+            submitter = Submitter(plugin=plugin, **plugin_kwargs)
 
         if submitter:
             with submitter as sub:
                 res = sub(self)
-        else:
-            if is_workflow(self):
-                raise NotImplementedError(
-                    "TODO: linear workflow execution - assign submitter or plugin for now"
-                )
+        else:  # tasks without state could be run without a submitter
             res = self._run(rerun=rerun, **kwargs)
         return res
 

pydra/engine/tests/test_numpy_examples.py

@@ -17,15 +17,15 @@ def arrayout(val):
     return np.array([val, val])
 
 
-def test_multiout(plugin, tmpdir):
+def test_multiout(tmpdir):
     """ testing a simple function that returns a numpy array"""
     wf = Workflow("wf", input_spec=["val"], val=2)
     wf.add(arrayout(name="mo", val=wf.lzin.val))
 
     wf.set_output([("array", wf.mo.lzout.b)])
     wf.cache_dir = tmpdir
 
-    with Submitter(plugin=plugin, n_procs=2) as sub:
+    with Submitter(plugin="cf", n_procs=2) as sub:
         sub(runnable=wf)
 
     results = wf.result(return_inputs=True)
@@ -34,7 +34,7 @@ def test_multiout(plugin, tmpdir):
     assert np.array_equal(results[1].output.array, np.array([2, 2]))
 
 
-def test_multiout_st(plugin, tmpdir):
+def test_multiout_st(tmpdir):
     """ testing a simple function that returns a numpy array, adding splitter"""
     wf = Workflow("wf", input_spec=["val"], val=[0, 1, 2])
     wf.add(arrayout(name="mo", val=wf.lzin.val))
@@ -43,7 +43,7 @@ def test_multiout_st(plugin, tmpdir):
     wf.set_output([("array", wf.mo.lzout.b)])
     wf.cache_dir = tmpdir
 
-    with Submitter(plugin=plugin, n_procs=2) as sub:
+    with Submitter(plugin="cf", n_procs=2) as sub:
         sub(runnable=wf)
 
     results = wf.result(return_inputs=True)

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

pydra/engine/tests/test_submitter.py

@@ -23,17 +23,30 @@ def sleep_add_one(x):
 
 def test_callable_wf(plugin, tmpdir):
     wf = gen_basic_wf()
+    res = wf()
+    assert res.output.out == 9
+    del wf, res
 
-    with pytest.raises(NotImplementedError):
-        wf()
-
+    # providing plugin
+    wf = gen_basic_wf()
     res = wf(plugin="cf")
     assert res.output.out == 9
     del wf, res
 
+    # providing plugin_kwargs
     wf = gen_basic_wf()
-    wf.cache_dir = tmpdir
+    res = wf(plugin="cf", plugin_kwargs={"n_procs": 2})
+    assert res.output.out == 9
+    del wf, res
+
+    # providing wrong plugin_kwargs
+    wf = gen_basic_wf()
+    with pytest.raises(TypeError, match="an unexpected keyword argument"):
+        wf(plugin="cf", plugin_kwargs={"sbatch_args": "-N2"})
 
+    # providing submitter
+    wf = gen_basic_wf()
+    wf.cache_dir = tmpdir
     sub = Submitter(plugin)
     res = wf(submitter=sub)
     assert res.output.out == 9