address @effigies' comments

Author: oesteban

docker/files/run_examples.sh

@@ -33,6 +33,6 @@ exit_code=$?
 
 # Collect crashfiles and generate xml report
 coverage xml -o ${WORKDIR}/tests/smoketest_${example_id}.xml
-find /work -name "crash-*" -maxdepth 1 -exec mv {} ${WORKDIR}/crashfiles/ \;
+find /work -maxdepth 1 -name "crash-*" -exec mv {} ${WORKDIR}/crashfiles/ \;
 exit $exit_code
 

docker/files/run_pytests.sh

@@ -38,7 +38,7 @@ if [[ "${PYTHON_VERSION}" -ge "30" ]]; then
 fi
 
 # Collect crashfiles
-find ${WORKDIR} -name "crash-*" -maxdepth 1 -exec mv {} ${WORKDIR}/crashfiles/ \;
+find ${WORKDIR} -maxdepth 1 -name "crash-*" -exec mv {} ${WORKDIR}/crashfiles/ \;
 
 echo "Unit tests finished with exit code ${exit_code}"
 exit ${exit_code}

nipype/interfaces/base.py

@@ -751,13 +751,26 @@ class BaseInterface(Interface):
 
     This class cannot be instantiated.
 
+
+    Relevant Interface attributes
+    -----------------------------
+
+    ``input_spec`` points to the traited class for the inputs
+    ``output_spec`` points to the traited class for the outputs
+    ``_redirect_x`` should be set to ``True`` when the interface requires
+      connecting to a ``$DISPLAY`` (default is ``False``).
+    ``resource_monitor`` if ``False`` prevents resource-monitoring this
+      interface, if ``True`` monitoring will be enabled IFF the general
+      Nipype config is set on (``resource_monitor = true``).
+
+
     """
     input_spec = BaseInterfaceInputSpec
     _version = None
     _additional_metadata = []
     _redirect_x = False
     references_ = []
-    resource_monitor = True
+    resource_monitor = True  # Enabled for this interface IFF enabled in the config
 
     def __init__(self, from_file=None, resource_monitor=None, **inputs):
         if not self.input_spec:
@@ -1133,7 +1146,7 @@ def run(self, **inputs):
 
                 # Read .prof file in and set runtime values
                 vals = np.loadtxt(mon_fname, delimiter=',')
-                if vals.tolist():
+                if vals.size:
                     vals = np.atleast_2d(vals)
                     _, mem_peak_mb, nthreads = vals.max(0).astype(float).tolist()
                     runtime.mem_peak_gb = mem_peak_mb / 1024
@@ -1310,7 +1323,6 @@ def run_command(runtime, output=None, timeout=0.01, redirect_x=False):
     """
 
     # Init variables
-    PIPE = sp.PIPE
     cmdline = runtime.cmdline
 
     if redirect_x:
@@ -1338,8 +1350,8 @@ def run_command(runtime, output=None, timeout=0.01, redirect_x=False):
                         env=env)
     else:
         proc = sp.Popen(cmdline,
-                        stdout=PIPE,
-                        stderr=PIPE,
+                        stdout=sp.PIPE,
+                        stderr=sp.PIPE,
                         shell=True,
                         cwd=runtime.cwd,
                         env=env)
@@ -1414,17 +1426,16 @@ def get_dependencies(name, environ):
     Uses otool on darwin, ldd on linux. Currently doesn't support windows.
 
     """
-    PIPE = sp.PIPE
     if sys.platform == 'darwin':
         proc = sp.Popen('otool -L `which %s`' % name,
-                        stdout=PIPE,
-                        stderr=PIPE,
+                        stdout=sp.PIPE,
+                        stderr=sp.PIPE,
                         shell=True,
                         env=environ)
     elif 'linux' in sys.platform:
         proc = sp.Popen('ldd `which %s`' % name,
-                        stdout=PIPE,
-                        stderr=PIPE,
+                        stdout=sp.PIPE,
+                        stderr=sp.PIPE,
                         shell=True,
                         env=environ)
     else:

nipype/pipeline/engine/nodes.py

@@ -55,6 +55,7 @@
 
 logger = logging.getLogger('workflow')
 
+
 class Node(EngineBase):
     """Wraps interface objects for use in pipeline
 
@@ -78,7 +79,7 @@ class Node(EngineBase):
 
     def __init__(self, interface, name, iterables=None, itersource=None,
                  synchronize=False, overwrite=None, needed_outputs=None,
-                 run_without_submitting=False, n_procs=1, mem_gb=0.25,
+                 run_without_submitting=False, n_procs=1, mem_gb=0.20,
                  **kwargs):
         """
         Parameters
@@ -200,6 +201,26 @@ def outputs(self):
         """Return the output fields of the underlying interface"""
         return self._interface._outputs()
 
+    @property
+    def mem_gb(self):
+        """Get estimated memory (GB)"""
+        if hasattr(self._interface, 'estimated_memory_gb'):
+            self._mem_gb = self._interface.estimated_memory_gb
+            logger.warning('Setting "estimated_memory_gb" on Interfaces has been '
+                           'deprecated as of nipype 1.0, please use Node.mem_gb.')
+            del self._interface.estimated_memory_gb
+        return self._mem_gb
+
+    @property
+    def n_procs(self):
+        """Get estimated number of processes"""
+        if hasattr(self._interface, 'num_threads'):
+            self._n_procs = self._interface.num_threads
+            logger.warning('Setting "num_threads" on Interfaces has been '
+                           'deprecated as of nipype 1.0, please use Node.n_procs')
+            del self._interface.num_threads
+        return self._n_procs
+
     def output_dir(self):
         """Return the location of the output directory for the node"""
         if self.base_dir is None:
@@ -685,24 +706,6 @@ def _copyfiles_to_wd(self, outdir, execute, linksonly=False):
             if execute and linksonly:
                 rmtree(outdir)
 
-    def get_mem_gb(self):
-        """Get estimated memory (GB)"""
-        if hasattr(self._interface, 'estimated_memory_gb'):
-            self._mem_gb = self._interface.estimated_memory_gb
-            logger.warning('Setting "estimated_memory_gb" on Interfaces has been '
-                           'deprecated as of nipype 1.0')
-            del self._interface.estimated_memory_gb
-        return self._mem_gb
-
-    def get_n_procs(self):
-        """Get estimated number of processes"""
-        if hasattr(self._interface, 'num_threads'):
-            self._n_procs = self._interface.num_threads
-            logger.warning('Setting "num_threads" on Interfaces has been '
-                           'deprecated as of nipype 1.0')
-            del self._interface.num_threads
-        return self._n_procs
-
     def update(self, **opts):
         self.inputs.update(**opts)
 
@@ -1129,8 +1132,8 @@ def _make_nodes(self, cwd=None):
         for i in range(nitems):
             nodename = '_' + self.name + str(i)
             node = Node(deepcopy(self._interface),
-                        n_procs=self.get_n_procs(),
-                        mem_gb=self.get_mem_gb(),
+                        n_procs=self.n_procs,
+                        mem_gb=self.mem_gb,
                         overwrite=self.overwrite,
                         needed_outputs=self.needed_outputs,
                         run_without_submitting=self.run_without_submitting,

nipype/pipeline/plugins/base.py

@@ -31,27 +31,28 @@ class PluginBase(object):
     """
     Base class for plugins
 
-    Execution plugin API
-    ====================
-
-    Current status::
-
-        class plugin_runner(PluginBase):
-
-            def run(graph, config, updatehash)
-
     """
 
     def __init__(self, plugin_args=None):
         if plugin_args is None:
             plugin_args = {}
         self.plugin_args = plugin_args
         self._config = None
-
         self._status_callback = plugin_args.get('status_callback')
-        return
 
     def run(self, graph, config, updatehash=False):
+        """
+        The core plugin member that should be implemented by
+        all plugins.
+
+        graph: a networkx, flattened :abbr:`DAG (Directed Acyclic Graph)`
+          to be executed
+
+        config: a nipype.config object
+
+        updatehash:
+
+        """
         raise NotImplementedError
 
 
@@ -63,9 +64,9 @@ def __init__(self, plugin_args=None):
         """Initialize runtime attributes to none
 
         procs: list (N) of underlying interface elements to be processed
-        proc_done: a boolean numpy array (N) signifying whether a process has been
+        proc_done: a boolean numpy array (N,) signifying whether a process has been
             executed
-        proc_pending: a boolean numpy array (N) signifying whether a
+        proc_pending: a boolean numpy array (N,) signifying whether a
             process is currently running. Note: A process is finished only when
             both proc_done==True and
         proc_pending==False
@@ -84,7 +85,7 @@ def __init__(self, plugin_args=None):
         self.max_jobs = self.plugin_args.get('max_jobs', np.inf)
 
     def _prerun_check(self, graph):
-        """Stub."""
+        """Stub method to validate/massage graph and nodes before running"""
 
     def run(self, graph, config, updatehash=False):
         """
@@ -227,9 +228,10 @@ def _send_procs_to_workers(self, updatehash=False, graph=None):
             logger.debug('Slots available: %s' % slots)
             if (num_jobs >= self.max_jobs) or (slots == 0):
                 break
-            # Check to see if a job is available
-            jobids = np.flatnonzero(
-                ~self.proc_done & (self.depidx.sum(axis=0) == 0).__array__())
+
+            # Check to see if a job is available (jobs without dependencies not run)
+            # See https://github.com/nipy/nipype/pull/2200#discussion_r141605722
+            jobids = np.nonzero(~self.proc_done & (self.depidx.sum(0) == 0))[1]
 
             if len(jobids) > 0:
                 # send all available jobs

nipype/pipeline/plugins/condor.py

@@ -8,9 +8,10 @@
 
 from ...interfaces.base import CommandLine
 from ... import logging
-from .base import (SGELikeBatchManagerBase, logger, logging)
+from .base import SGELikeBatchManagerBase, logger
 iflogger = logging.getLogger('interface')
 
+
 class CondorPlugin(SGELikeBatchManagerBase):
     """Execute using Condor
 

nipype/pipeline/plugins/dagman.py

@@ -10,7 +10,7 @@
 import time
 from warnings import warn
 
-from .base import (GraphPluginBase, logger)
+from .base import GraphPluginBase, logger
 from ...interfaces.base import CommandLine
 
 

nipype/pipeline/plugins/lsf.py

@@ -7,11 +7,12 @@
 import re
 from time import sleep
 
-from .base import (SGELikeBatchManagerBase, logger, logging)
 from ... import logging
 from ...interfaces.base import CommandLine
+from .base import SGELikeBatchManagerBase, logger
 iflogger = logging.getLogger('interface')
 
+
 class LSFPlugin(SGELikeBatchManagerBase):
     """Execute using LSF Cluster Submission
 

nipype/pipeline/plugins/multiproc.py

@@ -16,7 +16,7 @@
 from copy import deepcopy
 import numpy as np
 
-from ... import logging, config
+from ... import logging
 from ...utils.misc import str2bool
 from ...utils.profiler import get_system_total_memory_gb
 from ..engine import MapNode
@@ -155,8 +155,8 @@ def _prerun_check(self, graph):
         tasks_mem_gb = []
         tasks_num_th = []
         for node in graph.nodes():
-            tasks_mem_gb.append(node.get_mem_gb())
-            tasks_num_th.append(node.get_n_procs())
+            tasks_mem_gb.append(node.mem_gb)
+            tasks_num_th.append(node.n_procs)
 
         if np.any(np.array(tasks_mem_gb) > self.memory_gb):
             logger.warning(
@@ -179,8 +179,8 @@ def _check_resources(self, running_tasks):
         free_memory_gb = self.memory_gb
         free_processors = self.processors
         for _, jobid in running_tasks:
-            free_memory_gb -= min(self.procs[jobid].get_mem_gb(), self.memory_gb)
-            free_processors -= min(self.procs[jobid].get_n_procs(), self.processors)
+            free_memory_gb -= min(self.procs[jobid].get_mem_gb(), free_memory_gb)
+            free_processors -= min(self.procs[jobid].get_n_procs(), free_processors)
 
         return free_memory_gb, free_processors
 
@@ -189,29 +189,30 @@ def _send_procs_to_workers(self, updatehash=False, graph=None):
         Sends jobs to workers when system resources are available.
         """
 
-        # Check all jobs without dependency not run
+        # Check to see if a job is available (jobs without dependencies not run)
+        # See https://github.com/nipy/nipype/pull/2200#discussion_r141605722
+        jobids = np.nonzero(~self.proc_done & (self.depidx.sum(0) == 0))[1]
         jobids = np.flatnonzero(
             ~self.proc_done & (self.depidx.sum(axis=0) == 0).__array__())
 
         # Check available system resources by summing all threads and memory used
         free_memory_gb, free_processors = self._check_resources(self.pending_tasks)
 
         logger.info('Currently running %d tasks, and %d jobs ready. '
-                     'Free memory (GB): %0.2f/%0.2f, Free processors: %d/%d',
-                     len(self.pending_tasks), len(jobids),
-                     free_memory_gb, self.memory_gb, free_processors, self.processors)
+                    'Free memory (GB): %0.2f/%0.2f, Free processors: %d/%d',
+                    len(self.pending_tasks), len(jobids),
+                    free_memory_gb, self.memory_gb, free_processors, self.processors)
 
-
-        if (len(jobids) + len(self.pending_tasks)) == 0:
+        if len(jobids) + len(self.pending_tasks) == 0:
             logger.debug('No tasks are being run, and no jobs can '
                          'be submitted to the queue. Potential deadlock')
             return
 
         # Sort jobs ready to run first by memory and then by number of threads
         # The most resource consuming jobs run first
         # jobids = sorted(jobids,
-        #                 key=lambda item: (self.procs[item]._get_mem_gb(),
-        #                                   self.procs[item]._get_n_procs()))
+        #                 key=lambda item: (self.procs[item]._mem_gb,
+        #                                   self.procs[item]._n_procs))
 
         # While have enough memory and processors for first job
         # Submit first job on the list
@@ -232,8 +233,8 @@ def _send_procs_to_workers(self, updatehash=False, graph=None):
                         continue
 
             # Check requirements of this job
-            next_job_gb = min(self.procs[jobid].get_mem_gb(), self.memory_gb)
-            next_job_th = min(self.procs[jobid].get_n_procs(), self.processors)
+            next_job_gb = min(self.procs[jobid].mem_gb, self.memory_gb)
+            next_job_th = min(self.procs[jobid].n_procs, self.processors)
 
             # If node does not fit, skip at this moment
             if next_job_th > free_processors or next_job_gb > free_memory_gb:
@@ -257,8 +258,8 @@ def _send_procs_to_workers(self, updatehash=False, graph=None):
                     hash_exists, _, _, _ = self.procs[jobid].hash_exists()
                     overwrite = self.procs[jobid].overwrite
                     always_run = self.procs[jobid]._interface.always_run
-                    if (hash_exists and (overwrite is False or
-                                         (overwrite is None and not always_run))):
+                    if hash_exists and (overwrite is False or
+                                        overwrite is None and not always_run):
                         logger.debug('Skipping cached node %s with ID %s.',
                                      self.procs[jobid]._id, jobid)
                         self._task_finished_cb(jobid)

nipype/pipeline/plugins/oar.py

@@ -10,9 +10,9 @@
 import subprocess
 import simplejson as json
 
-from .base import (SGELikeBatchManagerBase, logger, logging)
 from ... import logging
 from ...interfaces.base import CommandLine
+from .base import SGELikeBatchManagerBase, logger
 iflogger = logging.getLogger('interface')
 
 class OARPlugin(SGELikeBatchManagerBase):