X-Git-Url: https://code.grnet.gr/git/ganeti-local/blobdiff_plain/8b537bb0fd52a71c6f260f1f8ca0549423aa3d82..ba55d062da8dfb89a37afc2f13f2e689d0094829:/lib/jstore.py

diff --git a/lib/jstore.py b/lib/jstore.py
index 5065e86..4d9189e 100644
--- a/lib/jstore.py
+++ b/lib/jstore.py
@@ -69,14 +69,13 @@ def ReadVersion():
   return _ReadNumericFile(constants.JOB_QUEUE_VERSION_FILE)
 
 
-def InitAndVerifyQueue(exclusive):
+def InitAndVerifyQueue(must_lock):
   """Open and lock job queue.
 
   If necessary, the queue is automatically initialized.
 
-  @type exclusive: bool
-  @param exclusive: Whether to lock the queue in exclusive mode. Shared
-                    mode otherwise.
+  @type must_lock: bool
+  @param must_lock: Whether an exclusive lock must be held.
   @rtype: utils.FileLock
   @return: Lock object for the queue. This can be used to change the
            locking mode.
@@ -93,40 +92,52 @@ def InitAndVerifyQueue(exclusive):
   # Lock queue
   queue_lock = utils.FileLock(constants.JOB_QUEUE_LOCK_FILE)
   try:
-    # Determine locking function and call it
-    if exclusive:
-      fn = queue_lock.Exclusive
+    # The queue needs to be locked in exclusive mode to write to the serial and
+    # version files.
+    if must_lock:
+      queue_lock.Exclusive(blocking=True)
+      holding_lock = True
     else:
-      fn = queue_lock.Shared
-
-    fn(blocking=False)
-
-    # Verify version
-    version = ReadVersion()
-    if version is None:
-      # Write new version file
-      utils.WriteFile(constants.JOB_QUEUE_VERSION_FILE,
-                      data="%s\n" % constants.JOB_QUEUE_VERSION)
-
-      # Read again
+      try:
+        queue_lock.Exclusive(blocking=False)
+        holding_lock = True
+      except errors.LockError:
+        # Ignore errors and assume the process keeping the lock checked
+        # everything.
+        holding_lock = False
+
+    if holding_lock:
+      # Verify version
       version = ReadVersion()
+      if version is None:
+        # Write new version file
+        utils.WriteFile(constants.JOB_QUEUE_VERSION_FILE,
+                        data="%s\n" % constants.JOB_QUEUE_VERSION)
 
-    if version != constants.JOB_QUEUE_VERSION:
-      raise errors.JobQueueError("Found job queue version %s, expected %s",
-                                 version, constants.JOB_QUEUE_VERSION)
+        # Read again
+        version = ReadVersion()
 
-    serial = ReadSerial()
-    if serial is None:
-      # Write new serial file
-      utils.WriteFile(constants.JOB_QUEUE_SERIAL_FILE,
-                      data="%s\n" % 0)
+      if version != constants.JOB_QUEUE_VERSION:
+        raise errors.JobQueueError("Found job queue version %s, expected %s",
+                                   version, constants.JOB_QUEUE_VERSION)
 
-      # Read again
       serial = ReadSerial()
-
-    if serial is None:
-      # There must be a serious problem
-      raise errors.JobQueueError("Can't read/parse the job queue serial file")
+      if serial is None:
+        # Write new serial file
+        utils.WriteFile(constants.JOB_QUEUE_SERIAL_FILE,
+                        data="%s\n" % 0)
+
+        # Read again
+        serial = ReadSerial()
+
+      if serial is None:
+        # There must be a serious problem
+        raise errors.JobQueueError("Can't read/parse the job queue serial file")
+
+      if not must_lock:
+        # There's no need for more error handling. Closing the lock file below in
+        # case of an error will unlock it anyway.
+        queue_lock.Unlock()
 
   except:
     queue_lock.Close()