remove logging from run_node which blocked mriqc, improve logging of both Nodes and MultiProc

Author: oesteban

nipype/pipeline/engine/nodes.py

@@ -314,7 +314,7 @@ def run(self, updatehash=False):
             self._get_inputs()
             self._got_inputs = True
         outdir = self.output_dir()
-        logger.info("Executing node %s in dir: %s", self._id, outdir)
+        logger.info("Executing node %s in dir: %s", self.fullname, outdir)
         if op.exists(outdir):
             logger.debug('Output dir: %s', to_str(os.listdir(outdir)))
         hash_info = self.hash_exists(updatehash=updatehash)
@@ -630,20 +630,21 @@ def _run_command(self, execute, copyfiles=True):
                 runtime=runtime,
                 inputs=self._interface.inputs.get_traitsfree())
             self._result = result
-            logger.debug('Executing node')
             if copyfiles:
                 self._copyfiles_to_wd(cwd, execute)
+
+            message = 'Running a "%s" interface'
             if issubclass(self._interface.__class__, CommandLine):
                 try:
                     cmd = self._interface.cmdline
                 except Exception as msg:
                     self._result.runtime.stderr = msg
                     raise
                 cmdfile = op.join(cwd, 'command.txt')
-                fd = open(cmdfile, 'wt')
-                fd.writelines(cmd + "\n")
-                fd.close()
-                logger.info('Running: %s' % cmd)
+                with open(cmdfile, 'wt') as fd:
+                    print(cmd + "\n", file=fd)
+                message += ', a CommandLine Interface with command:\n%s' % cmd
+            logger.info(message + '.', self._interface.__class__.__name__)
             try:
                 result = self._interface.run()
             except Exception as msg:

nipype/pipeline/plugins/multiproc.py

@@ -17,7 +17,6 @@
 import numpy as np
 
 from ... import logging
-from ...utils.misc import str2bool
 from ...utils.profiler import get_system_total_memory_gb
 from ..engine import MapNode
 from .base import DistributedPluginBase
@@ -44,10 +43,6 @@ def run_node(node, updatehash, taskid):
         dictionary containing the node runtime results and stats
     """
 
-    from nipype import logging
-    logger = logging.getLogger('workflow')
-
-    logger.debug('run_node called on %s', node.name)
     # Init variables
     result = dict(result=None, traceback=None, taskid=taskid)
 
@@ -148,6 +143,9 @@ def _submit_job(self, node, updatehash=False):
         self._task_obj[self._taskid] = self.pool.apply_async(
             run_node, (node, updatehash, self._taskid),
             callback=self._async_callback)
+
+        logger.debug('MultiProc submitted task %s (taskid=%d).',
+                     node.fullname, self._taskid)
         return self._taskid
 
     def _prerun_check(self, graph):
@@ -245,7 +243,7 @@ def _send_procs_to_workers(self, updatehash=False, graph=None):
             free_memory_gb -= next_job_gb
             free_processors -= next_job_th
             logger.debug('Allocating %s ID=%d (%0.2fGB, %d threads). Free: %0.2fGB, %d threads.',
-                         self.procs[jobid]._id, jobid, next_job_gb, next_job_th,
+                         self.procs[jobid].fullname, jobid, next_job_gb, next_job_th,
                          free_memory_gb, free_processors)
 
             # change job status in appropriate queues
@@ -274,7 +272,6 @@ def _send_procs_to_workers(self, updatehash=False, graph=None):
 
             # Task should be submitted to workers
             # Send job to task manager and add to pending tasks
-            logger.debug('MultiProc submitting job ID %d', jobid)
             if self._status_callback:
                 self._status_callback(self.procs[jobid], 'start')
             tid = self._submit_job(deepcopy(self.procs[jobid]),