enh: replace portalocker with filelock and use locking for reading/writing pickled files

Author: satra

nipype/algorithms/misc.py

@@ -799,11 +799,11 @@ def _run_interface(self, runtime):
                             '(http://pandas.pydata.org/) to run.'), e)
 
         try:
-            import lockfile as pl
+            from filelock import SoftFileLock
             self._have_lock = True
         except ImportError:
             from warnings import warn
-            warn(('Python module lockfile was not found: AddCSVRow will not be'
+            warn(('Python module filelock was not found: AddCSVRow will not be'
                   ' thread-safe in multi-processor execution'))
 
         input_dict = {}
@@ -822,7 +822,7 @@ def _run_interface(self, runtime):
         df = pd.DataFrame([input_dict])
 
         if self._have_lock:
-            self._lock = pl.FileLock(self.inputs.in_file)
+            self._lock = SoftFileLock(self.inputs.in_file + '.lock')
 
             # Acquire lock
             self._lock.acquire()
@@ -837,13 +837,6 @@ def _run_interface(self, runtime):
         if self._have_lock:
             self._lock.release()
 
-        # Using nipype.external.portalocker this might be something like:
-        # with pl.Lock(self.inputs.in_file, timeout=1) as fh:
-        #     if op.exists(fh):
-        #         formerdf = pd.read_csv(fh, index_col=0)
-        #         df = pd.concat([formerdf, df], ignore_index=True)
-        #         df.to_csv(fh)
-
         return runtime
 
     def _list_outputs(self):

nipype/external/cloghandler.py

@@ -36,10 +36,6 @@
 testing, performance was more than adequate, but if you need a high-volume or
 low-latency solution, I suggest you look elsewhere.
 
-This module currently only support the 'nt' and 'posix' platforms due to the
-usage of the portalocker module.  I do not have access to any other platforms
-for testing, patches are welcome.
-
 See the README file for an example usage of this module.
 
 """
@@ -63,13 +59,7 @@
 except ImportError:
     codecs = None
 
-# Question/TODO: Should we have a fallback mode if we can't load portalocker /
-# we should still be better off than with the standard RotattingFileHandler
-# class, right? We do some rename checking... that should prevent some file
-# clobbering that the builtin class allows.
-
-# sibling  module than handles all the ugly platform-specific details of file locking
-from .portalocker import lock, unlock, LOCK_EX, LOCK_NB, LockException
+from filelock import SoftFileLock
 
 # A client can set this to true to automatically convert relative paths to
 # absolute paths (which will also hide the absolute path warnings)
@@ -168,11 +158,8 @@ def __init__(self,
         self.maxBytes = maxBytes
         self.backupCount = backupCount
         # Prevent multiple extensions on the lock file (Only handles the normal "*.log" case.)
-        if filename.endswith(".log"):
-            lock_file = filename[:-4]
-        else:
-            lock_file = filename
-        self.stream_lock = open(lock_file + ".lock", "w")
+        self.lock_file = filename + '.lock'
+        self.stream_lock = SoftFileLock(self.lock_file)
 
         # For debug mode, swap out the "_degrade()" method with a more a verbose one.
         if debug:
@@ -189,7 +176,7 @@ def acquire(self):
         in 'degraded' mode. """
         # handle thread lock
         Handler.acquire(self)
-        lock(self.stream_lock, LOCK_EX)
+        self.stream_lock.acquire()
         if self.stream.closed:
             self._openFile(self.mode)
 
@@ -206,7 +193,7 @@ def release(self):
                 self.stream.close()
         finally:
             try:
-                unlock(self.stream_lock)
+                self.stream_lock.release()
             finally:
                 # release thread lock
                 Handler.release(self)

nipype/external/portalocker.py

@@ -25,7 +25,7 @@
 from future import standard_library
 
 from .misc import str2bool
-from ..external import portalocker
+from filelock import SoftFileLock
 
 standard_library.install_aliases()
 
@@ -209,9 +209,9 @@ def get_data(self, key):
         """Read options file"""
         if not os.path.exists(self.data_file):
             return None
-        with open(self.data_file, 'rt') as file:
-            portalocker.lock(file, portalocker.LOCK_EX)
-            datadict = load(file)
+        with SoftFileLock(self.data_file + '.lock'):
+            with open(self.data_file, 'rt') as file:
+                datadict = load(file)
         if key in datadict:
             return datadict[key]
         return None
@@ -220,17 +220,17 @@ def save_data(self, key, value):
         """Store config flie"""
         datadict = {}
         if os.path.exists(self.data_file):
-            with open(self.data_file, 'rt') as file:
-                portalocker.lock(file, portalocker.LOCK_EX)
-                datadict = load(file)
+            with SoftFileLock(self.data_file + '.lock'):
+                with open(self.data_file, 'rt') as file:
+                    datadict = load(file)
         else:
             dirname = os.path.dirname(self.data_file)
             if not os.path.exists(dirname):
                 mkdir_p(dirname)
-        with open(self.data_file, 'wt') as file:
-            portalocker.lock(file, portalocker.LOCK_EX)
-            datadict[key] = value
-            dump(datadict, file)
+        with SoftFileLock(self.data_file + '.lock'):
+            with open(self.data_file, 'wt') as file:
+                datadict[key] = value
+                dump(datadict, file)
 
     def update_config(self, config_dict):
         """Extend internal dictionary with config_dict"""

nipype/utils/config.py

@@ -21,6 +21,7 @@
 import contextlib
 import posixpath
 import simplejson as json
+from filelock import SoftFileLock
 
 from builtins import str, bytes, open
 
@@ -684,36 +685,37 @@ def loadpkl(infile):
 
     unpkl = None
     with indirectory(infile.parent):
-        pkl_file = pklopen(infile.name, 'rb')
-
-        try:  # Look if pkl file contains version file
-            pkl_metadata_line = pkl_file.readline()
-            pkl_metadata = json.loads(pkl_metadata_line)
-        except (UnicodeDecodeError, json.JSONDecodeError):
-            # Could not get version info
-            pkl_file.seek(0)
-
-        try:
-            unpkl = pickle.load(pkl_file)
-        except UnicodeDecodeError:
-            # Was this pickle created with Python 2.x?
-            unpkl = pickle.load(pkl_file, fix_imports=True, encoding='utf-8')
-            fmlogger.info('Successfully loaded pkl in compatibility mode.')
-        # Unpickling problems
-        except Exception as e:
-            if pkl_metadata and 'version' in pkl_metadata:
-                from nipype import __version__ as version
-                if pkl_metadata['version'] != version:
-                    fmlogger.error("""\
-Attempted to open a results file generated by Nipype version %s, \
-with an incompatible Nipype version (%s)""", pkl_metadata['version'], version)
-                    raise e
-            fmlogger.error("""\
-No metadata was found in the pkl file. Make sure you are currently using \
-the same Nipype version from the generated pkl.""")
-            raise e
-        finally:
-            pkl_file.close()
+        with SoftFileLock(infile.name + '.lock'):
+            pkl_file = pklopen(infile.name, 'rb')
+
+            try:  # Look if pkl file contains version file
+                pkl_metadata_line = pkl_file.readline()
+                pkl_metadata = json.loads(pkl_metadata_line)
+            except (UnicodeDecodeError, json.JSONDecodeError):
+                # Could not get version info
+                pkl_file.seek(0)
+
+            try:
+                unpkl = pickle.load(pkl_file)
+            except UnicodeDecodeError:
+                # Was this pickle created with Python 2.x?
+                unpkl = pickle.load(pkl_file, fix_imports=True, encoding='utf-8')
+                fmlogger.info('Successfully loaded pkl in compatibility mode.')
+            # Unpickling problems
+            except Exception as e:
+                if pkl_metadata and 'version' in pkl_metadata:
+                    from nipype import __version__ as version
+                    if pkl_metadata['version'] != version:
+                        fmlogger.error("""\
+    Attempted to open a results file generated by Nipype version %s, \
+    with an incompatible Nipype version (%s)""", pkl_metadata['version'], version)
+                        raise e
+                fmlogger.error("""\
+    No metadata was found in the pkl file. Make sure you are currently using \
+    the same Nipype version from the generated pkl.""")
+                raise e
+            finally:
+                pkl_file.close()
 
     if unpkl is None:
         raise ValueError('Loading %s resulted in None.' % infile)
@@ -754,20 +756,21 @@ def read_stream(stream, logger=None, encoding=None):
 
 
 def savepkl(filename, record, versioning=False):
-    if filename.endswith('pklz'):
-        pkl_file = gzip.open(filename, 'wb')
-    else:
-        pkl_file = open(filename, 'wb')
+    with SoftFileLock(filename + '.lock'):
+        if filename.endswith('pklz'):
+            pkl_file = gzip.open(filename, 'wb')
+        else:
+            pkl_file = open(filename, 'wb')
 
-    if versioning:
-        from nipype import __version__ as version
-        metadata = json.dumps({'version': version})
+        if versioning:
+            from nipype import __version__ as version
+            metadata = json.dumps({'version': version})
 
-        pkl_file.write(metadata.encode('utf-8'))
-        pkl_file.write('\n'.encode('utf-8'))
+            pkl_file.write(metadata.encode('utf-8'))
+            pkl_file.write('\n'.encode('utf-8'))
 
-    pickle.dump(record, pkl_file)
-    pkl_file.close()
+        pickle.dump(record, pkl_file)
+        pkl_file.close()
 
 
 rst_levels = ['=', '-', '~', '+']