Merge pull request #890 from oesteban/bug/fixRemoveNonEmptyDir

Not empty folder in NFS errors

Author: oesteban

nipype/pipeline/engine.py

@@ -26,6 +26,7 @@
 import os.path as op
 import re
 import shutil
+import errno
 from shutil import rmtree
 from socket import gethostname
 from string import Template
@@ -512,7 +513,7 @@ def write_graph(self, dotfilename='graph.dot', graph2use='hierarchical',
             workflow nodes;
             flat - expands workflow nodes recursively;
             hierarchical - expands workflow nodes recursively with a
-            notion on hierarchy; 
+            notion on hierarchy;
             colored - expands workflow nodes recursively with a
             notion on hierarchy in color;
             exec - expands workflows to depict iterables
@@ -529,17 +530,17 @@ def write_graph(self, dotfilename='graph.dot', graph2use='hierarchical',
         if graph2use not in graphtypes:
             raise ValueError('Unknown graph2use keyword. Must be one of: ' +
                              str(graphtypes))
-        base_dir, dotfilename = os.path.split(dotfilename)
+        base_dir, dotfilename = op.split(dotfilename)
         if base_dir == '':
             if self.base_dir:
                 base_dir = self.base_dir
                 if self.name:
-                    base_dir = os.path.join(base_dir, self.name)
+                    base_dir = op.join(base_dir, self.name)
             else:
                 base_dir = os.getcwd()
         base_dir = make_output_dir(base_dir)
         if graph2use in ['hierarchical', 'colored']:
-            dotfilename = os.path.join(base_dir, dotfilename)
+            dotfilename = op.join(base_dir, dotfilename)
             self.write_hierarchical_dotfile(dotfilename=dotfilename,
                                             colored=graph2use == "colored",
                                             simple_form=simple_form)
@@ -697,7 +698,7 @@ def run(self, plugin=None, plugin_args=None, updatehash=False):
         runner.run(execgraph, updatehash=updatehash, config=self.config)
         datestr = datetime.utcnow().strftime('%Y%m%dT%H%M%S')
         if str2bool(self.config['execution']['write_provenance']):
-            prov_base = os.path.join(self.base_dir,
+            prov_base = op.join(self.base_dir,
                                      'workflow_provenance_%s' % datestr)
             logger.info('Provenance file prefix: %s' % prov_base)
             write_workflow_prov(execgraph, prov_base, format='all')
@@ -708,17 +709,17 @@ def run(self, plugin=None, plugin_args=None, updatehash=False):
     def _write_report_info(self, workingdir, name, graph):
         if workingdir is None:
             workingdir = os.getcwd()
-        report_dir = os.path.join(workingdir, name)
-        if not os.path.exists(report_dir):
+        report_dir = op.join(workingdir, name)
+        if not op.exists(report_dir):
             os.makedirs(report_dir)
-        shutil.copyfile(os.path.join(os.path.dirname(__file__),
+        shutil.copyfile(op.join(op.dirname(__file__),
                                      'report_template.html'),
-                        os.path.join(report_dir, 'index.html'))
-        shutil.copyfile(os.path.join(os.path.dirname(__file__),
+                        op.join(report_dir, 'index.html'))
+        shutil.copyfile(op.join(op.dirname(__file__),
                                      '..', 'external', 'd3.v3.min.js'),
-                        os.path.join(report_dir, 'd3.v3.min.js'))
+                        op.join(report_dir, 'd3.v3.min.js'))
         nodes, groups = topological_sort(graph, depth_first=True)
-        graph_file = os.path.join(report_dir, 'graph1.json')
+        graph_file = op.join(report_dir, 'graph1.json')
         json_dict = {'nodes': [], 'links': [], 'groups': [], 'maxN': 0}
         for i, node in enumerate(nodes):
             report_file = "%s/_report/report.rst" % \
@@ -745,7 +746,7 @@ def _write_report_info(self, workingdir, name, graph):
                                            target=nodes.index(v),
                                            value=1))
         save_json(graph_file, json_dict)
-        graph_file = os.path.join(report_dir, 'graph.json')
+        graph_file = op.join(report_dir, 'graph.json')
         template = '%%0%dd_' % np.ceil(np.log10(len(nodes))).astype(int)
         def getname(u, i):
             name_parts = u.fullname.split('.')
@@ -790,7 +791,7 @@ def _configure_exec_nodes(self, graph):
                 data = graph.get_edge_data(*edge)
                 for sourceinfo, field in sorted(data['connect']):
                     node.input_source[field] = \
-                        (os.path.join(edge[0].output_dir(),
+                        (op.join(edge[0].output_dir(),
                          'result_%s.pklz' % edge[0].name),
                          sourceinfo)
 
@@ -1245,15 +1246,15 @@ def output_dir(self):
             self.base_dir = mkdtemp()
         outputdir = self.base_dir
         if self._hierarchy:
-            outputdir = os.path.join(outputdir, *self._hierarchy.split('.'))
+            outputdir = op.join(outputdir, *self._hierarchy.split('.'))
         if self.parameterization:
             if not str2bool(self.config['execution']['parameterize_dirs']):
                 param_dirs = [self._parameterization_dir(p) for p in
                               self.parameterization]
-                outputdir = os.path.join(outputdir, *param_dirs)
+                outputdir = op.join(outputdir, *param_dirs)
             else:
-                outputdir = os.path.join(outputdir, *self.parameterization)
-        return os.path.abspath(os.path.join(outputdir,
+                outputdir = op.join(outputdir, *self.parameterization)
+        return op.abspath(op.join(outputdir,
                                             self.name))
 
     def set_input(self, parameter, val):
@@ -1284,23 +1285,23 @@ def hash_exists(self, updatehash=False):
         # of the dictionary itself.
         hashed_inputs, hashvalue = self._get_hashval()
         outdir = self.output_dir()
-        if os.path.exists(outdir):
+        if op.exists(outdir):
             logger.debug(os.listdir(outdir))
-        hashfiles = glob(os.path.join(outdir, '_0x*.json'))
+        hashfiles = glob(op.join(outdir, '_0x*.json'))
         logger.debug(hashfiles)
         if len(hashfiles) > 1:
             logger.info(hashfiles)
             logger.info('Removing multiple hashfiles and forcing node to rerun')
             for hashfile in hashfiles:
                 os.unlink(hashfile)
-        hashfile = os.path.join(outdir, '_0x%s.json' % hashvalue)
+        hashfile = op.join(outdir, '_0x%s.json' % hashvalue)
         logger.debug(hashfile)
-        if updatehash and os.path.exists(outdir):
+        if updatehash and op.exists(outdir):
             logger.debug("Updating hash: %s" % hashvalue)
-            for file in glob(os.path.join(outdir, '_0x*.json')):
+            for file in glob(op.join(outdir, '_0x*.json')):
                 os.remove(file)
             self._save_hashfile(hashfile, hashed_inputs)
-        return os.path.exists(hashfile), hashvalue, hashfile, hashed_inputs
+        return op.exists(hashfile), hashvalue, hashfile, hashed_inputs
 
     def run(self, updatehash=False):
         """Execute the node in its directory.
@@ -1321,7 +1322,7 @@ def run(self, updatehash=False):
             self._got_inputs = True
         outdir = self.output_dir()
         logger.info("Executing node %s in dir: %s" % (self._id, outdir))
-        if os.path.exists(outdir):
+        if op.exists(outdir):
             logger.debug(os.listdir(outdir))
         hash_info = self.hash_exists(updatehash=updatehash)
         hash_exists, hashvalue, hashfile, hashed_inputs = hash_info
@@ -1337,7 +1338,7 @@ def run(self, updatehash=False):
             # by rerunning we mean only nodes that did finish to run previously
             json_pat = op.join(outdir, '_0x*.json')
             json_unfinished_pat = op.join(outdir, '_0x*_unfinished.json')
-            need_rerun = (os.path.exists(outdir)
+            need_rerun = (op.exists(outdir)
                           and not isinstance(self, MapNode)
                           and len(glob(json_pat)) != 0
                           and len(glob(json_unfinished_pat)) == 0)
@@ -1352,7 +1353,7 @@ def run(self, updatehash=False):
                               str(self.overwrite),
                               str(self._interface.always_run),
                               hashfile,
-                              str(os.path.exists(hashfile)),
+                              str(op.exists(hashfile)),
                               self.config['execution']['hash_method'].lower()))
                 log_debug = config.get('logging', 'workflow_level') == 'DEBUG'
                 if log_debug and not op.exists(hashfile):
@@ -1376,7 +1377,7 @@ def run(self, updatehash=False):
                 if cannot_rerun:
                     raise Exception(("Cannot rerun when 'stop_on_first_rerun' "
                                      "is set to True"))
-            hashfile_unfinished = os.path.join(outdir,
+            hashfile_unfinished = op.join(outdir,
                                                '_0x%s_unfinished.json' %
                                                hashvalue)
             if op.exists(hashfile):
@@ -1387,20 +1388,35 @@ def run(self, updatehash=False):
                          and not isinstance(self, MapNode))
             if rm_outdir:
                 logger.debug("Removing old %s and its contents" % outdir)
-                rmtree(outdir)
+                try:
+                    rmtree(outdir)
+                except OSError as ex:
+                    outdircont = os.listdir(outdir)
+                    if ((ex.errno == errno.ENOTEMPTY) and (len(outdircont) == 0)):
+                        logger.warn(('An exception was raised trying to remove old %s, '
+                                    'but the path seems empty. Is it an NFS mount?. '
+                                    'Passing the exception.') % outdir)
+                        pass
+                    elif ((ex.errno == errno.ENOTEMPTY) and (len(outdircont) != 0)):
+                        logger.debug(('Folder contents (%d items): '
+                                     '%s') % (len(outdircont), outdircont))
+                        raise ex
+                    else:
+                        raise ex
+
             else:
                 logger.debug(("%s found and can_resume is True or Node is a "
                               "MapNode - resuming execution") %
                              hashfile_unfinished)
                 if isinstance(self, MapNode):
                     # remove old json files
-                    for filename in glob(os.path.join(outdir, '_0x*.json')):
+                    for filename in glob(op.join(outdir, '_0x*.json')):
                         os.unlink(filename)
             outdir = make_output_dir(outdir)
             self._save_hashfile(hashfile_unfinished, hashed_inputs)
             self.write_report(report_type='preexec', cwd=outdir)
-            savepkl(os.path.join(outdir, '_node.pklz'), self)
-            savepkl(os.path.join(outdir, '_inputs.pklz'),
+            savepkl(op.join(outdir, '_node.pklz'), self)
+            savepkl(op.join(outdir, '_inputs.pklz'),
                     self.inputs.get_traitsfree())
             try:
                 self._run_interface()
@@ -1410,13 +1426,13 @@ def run(self, updatehash=False):
             shutil.move(hashfile_unfinished, hashfile)
             self.write_report(report_type='postexec', cwd=outdir)
         else:
-            if not os.path.exists(os.path.join(outdir, '_inputs.pklz')):
+            if not op.exists(op.join(outdir, '_inputs.pklz')):
                 logger.debug('%s: creating inputs file' % self.name)
-                savepkl(os.path.join(outdir, '_inputs.pklz'),
+                savepkl(op.join(outdir, '_inputs.pklz'),
                         self.inputs.get_traitsfree())
-            if not os.path.exists(os.path.join(outdir, '_node.pklz')):
+            if not op.exists(op.join(outdir, '_node.pklz')):
                 logger.debug('%s: creating node file' % self.name)
-                savepkl(os.path.join(outdir, '_node.pklz'), self)
+                savepkl(op.join(outdir, '_node.pklz'), self)
             logger.debug("Hashfile exists. Skipping execution")
             self._run_interface(execute=False, updatehash=updatehash)
         logger.debug('Finished running %s in dir: %s\n' % (self._id, outdir))
@@ -1517,7 +1533,7 @@ def _run_interface(self, execute=True, updatehash=False):
         os.chdir(old_cwd)
 
     def _save_results(self, result, cwd):
-        resultsfile = os.path.join(cwd, 'result_%s.pklz' % self.name)
+        resultsfile = op.join(cwd, 'result_%s.pklz' % self.name)
         if result.outputs:
             try:
                 outputs = result.outputs.get()
@@ -1550,10 +1566,10 @@ def _load_resultfile(self, cwd):
             rerun
         """
         aggregate = True
-        resultsoutputfile = os.path.join(cwd, 'result_%s.pklz' % self.name)
+        resultsoutputfile = op.join(cwd, 'result_%s.pklz' % self.name)
         result = None
         attribute_error = False
-        if os.path.exists(resultsoutputfile):
+        if op.exists(resultsoutputfile):
             pkl_file = gzip.open(resultsoutputfile, 'rb')
             try:
                 result = cPickle.load(pkl_file)
@@ -1589,7 +1605,7 @@ def _load_results(self, cwd):
         if aggregate:
             logger.debug('aggregating results')
             if attribute_error:
-                old_inputs = loadpkl(os.path.join(cwd, '_inputs.pklz'))
+                old_inputs = loadpkl(op.join(cwd, '_inputs.pklz'))
                 self.inputs.set(**old_inputs)
             if not isinstance(self, MapNode):
                 self._copyfiles_to_wd(cwd, True, linksonly=True)
@@ -1633,7 +1649,7 @@ def _run_command(self, execute, copyfiles=True):
                 except Exception, msg:
                     self._result.runtime.stderr = msg
                     raise
-                cmdfile = os.path.join(cwd, 'command.txt')
+                cmdfile = op.join(cwd, 'command.txt')
                 fd = open(cmdfile, 'wt')
                 fd.writelines(cmd + "\n")
                 fd.close()
@@ -1646,7 +1662,7 @@ def _run_command(self, execute, copyfiles=True):
 
             dirs2keep = None
             if isinstance(self, MapNode):
-                dirs2keep = [os.path.join(cwd, 'mapflow')]
+                dirs2keep = [op.join(cwd, 'mapflow')]
             result.outputs = clean_working_directory(result.outputs, cwd,
                                                      self._interface.inputs,
                                                      self.needed_outputs,
@@ -1670,7 +1686,7 @@ def _strip_temp(self, files, wd):
             if isinstance(f, list):
                 out.append(self._strip_temp(f, wd))
             else:
-                out.append(f.replace(os.path.join(wd, '_tempinput'), wd))
+                out.append(f.replace(op.join(wd, '_tempinput'), wd))
         return out
 
     def _copyfiles_to_wd(self, outdir, execute, linksonly=False):
@@ -1680,7 +1696,7 @@ def _copyfiles_to_wd(self, outdir, execute, linksonly=False):
                          (str(execute), str(linksonly)))
             if execute and linksonly:
                 olddir = outdir
-                outdir = os.path.join(outdir, '_tempinput')
+                outdir = op.join(outdir, '_tempinput')
                 os.makedirs(outdir)
             for info in self._interface._get_filecopy_info():
                 files = self.inputs.get().get(info['key'])
@@ -1700,7 +1716,7 @@ def _copyfiles_to_wd(self, outdir, execute, linksonly=False):
                                                             newpath=outdir)
                             newfiles = self._strip_temp(
                                 newfiles,
-                                op.abspath(olddir).split(os.path.sep)[-1])
+                                op.abspath(olddir).split(op.sep)[-1])
                         else:
                             newfiles = copyfiles(infiles,
                                                  [outdir],
@@ -1720,9 +1736,9 @@ def update(self, **opts):
     def write_report(self, report_type=None, cwd=None):
         if not str2bool(self.config['execution']['create_report']):
             return
-        report_dir = os.path.join(cwd, '_report')
-        report_file = os.path.join(report_dir, 'report.rst')
-        if not os.path.exists(report_dir):
+        report_dir = op.join(cwd, '_report')
+        report_file = op.join(report_dir, 'report.rst')
+        if not op.exists(report_dir):
             os.makedirs(report_dir)
         if report_type == 'preexec':
             logger.debug('writing pre-exec report to %s' % report_file)
@@ -2038,7 +2054,6 @@ def __init__(self, interface, iterfield, name, serial=False, **kwargs):
                                                    fields=self.iterfield)
         self._inputs.on_trait_change(self._set_mapnode_input)
         self._got_inputs = False
-        
         self._serial = serial
 
     def _create_dynamic_traits(self, basetraits, fields=None, nitems=None):
@@ -2137,7 +2152,7 @@ def _make_nodes(self, cwd=None):
                 setattr(node.inputs, field,
                         fieldvals[i])
             node.config = self.config
-            node.base_dir = os.path.join(cwd, 'mapflow')
+            node.base_dir = op.join(cwd, 'mapflow')
             yield i, node
 
     def _node_runner(self, nodes, updatehash=False):
@@ -2199,8 +2214,8 @@ def write_report(self, report_type=None, cwd=None):
             super(MapNode, self).write_report(report_type=report_type, cwd=cwd)
         if report_type == 'postexec':
             super(MapNode, self).write_report(report_type=report_type, cwd=cwd)
-            report_dir = os.path.join(cwd, '_report')
-            report_file = os.path.join(report_dir, 'report.rst')
+            report_dir = op.join(cwd, '_report')
+            report_file = op.join(report_dir, 'report.rst')
             fp = open(report_file, 'at')
             fp.writelines(write_rst_header('Subnode reports', level=1))
             nitems = len(filename_to_list(
@@ -2209,7 +2224,7 @@ def write_report(self, report_type=None, cwd=None):
             for i in range(nitems):
                 nodename = '_' + self.name + str(i)
                 subnode_report_files.insert(i, 'subnode %d' % i + ' : ' +
-                                               os.path.join(cwd,
+                                               op.join(cwd,
                                                             'mapflow',
                                                             nodename,
                                                             '_report',
@@ -2282,9 +2297,9 @@ def _run_interface(self, execute=True, updatehash=False):
             self._save_results(self._result, cwd)
             # remove any node directories no longer required
             dirs2remove = []
-            for path in glob(os.path.join(cwd, 'mapflow', '*')):
-                if os.path.isdir(path):
-                    if path.split(os.path.sep)[-1] not in nodenames:
+            for path in glob(op.join(cwd, 'mapflow', '*')):
+                if op.isdir(path):
+                    if path.split(op.sep)[-1] not in nodenames:
                         dirs2remove.append(path)
             for path in dirs2remove:
                 shutil.rmtree(path)