Merge pull request #1548 from nouiz/compiledir

theano/gof/compilelock.py

-# Locking mechanism to ensure no two compilations occur simultaneously in the
-# same compilation directory (which can cause crashes).
+# Locking mechanism to ensure no two compilations occur simultaneously
+# in the same compilation directory (which can cause crashes).
 
-from theano import config
-import os, random, time, atexit
+import atexit
+import os
+import random
 import socket  # only used for gethostname()
+import time
 import logging
-_logger=logging.getLogger("theano.gof.compilelock")
-_logger.setLevel(logging.INFO) # INFO will show the the messages "Refreshing lock" message
+
+from theano import config
+
+_logger = logging.getLogger("theano.gof.compilelock")
+# INFO will show the the messages "Refreshing lock" message
+_logger.setLevel(logging.INFO)
 
 # In seconds, time that a process will wait before deciding to override an
 # existing lock. An override only happens when the existing lock is held by
@@ -82,6 +88,7 @@ def get_lock(lock_dir=None, **kw):
                 get_lock.start_time = now
     get_lock.n_lock += 1
 
+
 def release_lock():
     """
     Release lock on compilation directory.
@@ -93,6 +100,7 @@ def release_lock():
         get_lock.start_time = None
         get_lock.unlocker.unlock()
 
+
 def set_lock_status(use_lock):
     """
     Enable or disable the lock on the compilation directory (which is enabled
@@ -104,6 +112,7 @@ def set_lock_status(use_lock):
     """
     get_lock.lock_is_enabled = use_lock
 
+
 def lock(tmp_dir, timeout=120, min_wait=5, max_wait=10, verbosity=1):
     """
     Obtain lock access by creating a given temporary directory (whose base will
@@ -157,6 +166,7 @@ def lock(tmp_dir, timeout=120, min_wait=5, max_wait=10, verbosity=1):
     no_display = (verbosity == 0)
 
     # Acquire lock.
+    nb_error = 0
     while True:
         try:
             last_owner = 'no_owner'
@@ -211,7 +221,8 @@ def lock(tmp_dir, timeout=120, min_wait=5, max_wait=10, verbosity=1):
                         msg = "process '%s'" % read_owner.split('_')[0]
                     _logger.info("Waiting for existing lock by %s (I am "
                                  "process '%s')", msg, my_pid)
-                    _logger.info("To manually release the lock, delete %s", tmp_dir)
+                    _logger.info("To manually release the lock, delete %s",
+                                 tmp_dir)
                     if verbosity <= 1:
                         no_display = True
                 time.sleep(random.uniform(min_wait, max_wait))
@@ -219,9 +230,13 @@ def lock(tmp_dir, timeout=120, min_wait=5, max_wait=10, verbosity=1):
             try:
                 os.mkdir(tmp_dir)
             except OSError:
-                # Error while creating the directory: someone else must have tried
-                # at the exact same time.
+                # Error while creating the directory: someone else
+                # must have tried at the exact same time.
+                nb_error += 1
+                if nb_error < 10:
                     continue
+                else:
+                    raise
             # Safety check: the directory should be here.
             assert os.path.isdir(tmp_dir)
 
@@ -241,22 +256,28 @@ def lock(tmp_dir, timeout=120, min_wait=5, max_wait=10, verbosity=1):
         except Exception, e:
             # If something wrong happened, we try again.
             _logger.warning("Something wrong happened: %s %s", type(e), e)
+            nb_error += 1
+            if nb_error > 10:
+                raise
             time.sleep(random.uniform(min_wait, max_wait))
             continue
 
+
 def refresh_lock(lock_file):
     """
     'Refresh' an existing lock by re-writing the file containing the owner's
     unique id, using a new (randomly generated) id, which is also returned.
     """
-    unique_id = '%s_%s_%s' % (os.getpid(),
-            ''.join([str(random.randint(0,9)) for i in range(10)]),
+    unique_id = '%s_%s_%s' % (
+        os.getpid(),
+        ''.join([str(random.randint(0, 9)) for i in range(10)]),
         socket.gethostname())
     lock_write = open(lock_file, 'w')
     lock_write.write(unique_id + '\n')
     lock_write.close()
     return unique_id
 
+
 class Unlocker(object):
     """
     Class wrapper around release mechanism so that the lock is automatically
@@ -274,12 +295,14 @@ class Unlocker(object):
         self.unlock()
 
     def unlock(self):
-        """
-        Remove current lock.
-        This function does not crash if it is unable to properly delete the lock
-        file and directory. The reason is that it should be allowed for multiple
-        jobs running in parallel to unlock the same directory at the same time
-        (e.g. when reaching their timeout limit).
+        """Remove current lock.
+
+        This function does not crash if it is unable to properly
+        delete the lock file and directory. The reason is that it
+        should be allowed for multiple jobs running in parallel to
+        unlock the same directory at the same time (e.g. when reaching
+        their timeout limit).
+
         """
         # If any error occurs, we assume this is because someone else tried to
         # unlock this directory at the same time.