X-Git-Url: https://code.grnet.gr/git/ganeti-local/blobdiff_plain/fdfe88b12d07e398d24001661b651fa33a6021e2..784b6603e36bc7cd86c2712623cea2359f0ace91:/lib/locking.py

diff --git a/lib/locking.py b/lib/locking.py
index b67ec0d..5ac20f9 100644
--- a/lib/locking.py
+++ b/lib/locking.py
@@ -1,7 +1,7 @@
 #
 #
 
-# Copyright (C) 2006, 2007, 2008, 2009, 2010 Google Inc.
+# Copyright (C) 2006, 2007, 2008, 2009, 2010, 2011 Google Inc.
 #
 # This program is free software; you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
@@ -20,7 +20,7 @@
 
 """Module implementing the Ganeti locking code."""
 
-# pylint: disable-msg=W0212
+# pylint: disable=W0212
 
 # W0212 since e.g. LockSet methods use (a lot) the internals of
 # SharedLock
@@ -32,7 +32,6 @@ import errno
 import weakref
 import logging
 import heapq
-import operator
 import itertools
 
 from ganeti import errors
@@ -162,7 +161,7 @@ class _BaseCondition(object):
     except AttributeError:
       self._acquire_restore = self._base_acquire_restore
     try:
-      self._is_owned = lock._is_owned
+      self._is_owned = lock.is_owned
     except AttributeError:
       self._is_owned = self._base_is_owned
 
@@ -277,7 +276,7 @@ class SingleNotifyPipeCondition(_BaseCondition):
       if self._nwaiters == 0:
         self._Cleanup()
 
-  def notifyAll(self): # pylint: disable-msg=C0103
+  def notifyAll(self): # pylint: disable=C0103
     """Close the writing side of the pipe to notify all waiters.
 
     """
@@ -334,7 +333,7 @@ class PipeCondition(_BaseCondition):
       self._check_owned()
       self._waiters.remove(threading.currentThread())
 
-  def notifyAll(self): # pylint: disable-msg=C0103
+  def notifyAll(self): # pylint: disable=C0103
     """Notify all currently waiting threads.
 
     """
@@ -358,6 +357,11 @@ class PipeCondition(_BaseCondition):
 
     return bool(self._waiters)
 
+  def __repr__(self):
+    return ("<%s.%s waiters=%s at %#x>" %
+            (self.__class__.__module__, self.__class__.__name__,
+             self._waiters, id(self)))
+
 
 class _PipeConditionWithMode(PipeCondition):
   __slots__ = [
@@ -434,9 +438,15 @@ class SharedLock(object):
 
     # Register with lock monitor
     if monitor:
+      logging.debug("Adding lock %s to monitor", name)
       monitor.RegisterLock(self)
 
-  def GetInfo(self, requested):
+  def __repr__(self):
+    return ("<%s.%s name=%s at %#x>" %
+            (self.__class__.__module__, self.__class__.__name__,
+             self.name, id(self)))
+
+  def GetLockInfo(self, requested):
     """Retrieves information for querying locks.
 
     @type requested: set
@@ -489,7 +499,7 @@ class SharedLock(object):
       else:
         pending = None
 
-      return (self.name, mode, owner_names, pending)
+      return [(self.name, mode, owner_names, pending)]
     finally:
       self.__lock.release()
 
@@ -526,7 +536,7 @@ class SharedLock(object):
     else:
       return self.__is_exclusive()
 
-  def _is_owned(self, shared=-1):
+  def is_owned(self, shared=-1):
     """Is the current thread somehow owning the lock at this time?
 
     @param shared:
@@ -541,7 +551,9 @@ class SharedLock(object):
     finally:
       self.__lock.release()
 
-  is_owned = _is_owned
+  #: Necessary to remain compatible with threading.Condition, which tries to
+  #: retrieve a locks' "_is_owned" attribute
+  _is_owned = is_owned
 
   def _count_pending(self):
     """Returns the number of pending acquires.
@@ -940,17 +952,53 @@ class LockSet:
     """
     return self.__lockdict
 
-  def _is_owned(self):
-    """Is the current thread a current level owner?"""
+  def is_owned(self):
+    """Is the current thread a current level owner?
+
+    @note: Use L{check_owned} to check if a specific lock is held
+
+    """
     return threading.currentThread() in self.__owners
 
+  def check_owned(self, names, shared=-1):
+    """Check if locks are owned in a specific mode.
+
+    @type names: sequence or string
+    @param names: Lock names (or a single lock name)
+    @param shared: See L{SharedLock.is_owned}
+    @rtype: bool
+    @note: Use L{is_owned} to check if the current thread holds I{any} lock and
+      L{list_owned} to get the names of all owned locks
+
+    """
+    if isinstance(names, basestring):
+      names = [names]
+
+    # Avoid check if no locks are owned anyway
+    if names and self.is_owned():
+      candidates = []
+
+      # Gather references to all locks (in case they're deleted in the meantime)
+      for lname in names:
+        try:
+          lock = self.__lockdict[lname]
+        except KeyError:
+          raise errors.LockError("Non-existing lock '%s' in set '%s' (it may"
+                                 " have been removed)" % (lname, self.name))
+        else:
+          candidates.append(lock)
+
+      return compat.all(lock.is_owned(shared=shared) for lock in candidates)
+    else:
+      return False
+
   def _add_owned(self, name=None):
     """Note the current thread owns the given lock"""
     if name is None:
-      if not self._is_owned():
+      if not self.is_owned():
         self.__owners[threading.currentThread()] = set()
     else:
-      if self._is_owned():
+      if self.is_owned():
         self.__owners[threading.currentThread()].add(name)
       else:
         self.__owners[threading.currentThread()] = set([name])
@@ -958,29 +1006,29 @@ class LockSet:
   def _del_owned(self, name=None):
     """Note the current thread owns the given lock"""
 
-    assert not (name is None and self.__lock._is_owned()), \
+    assert not (name is None and self.__lock.is_owned()), \
            "Cannot hold internal lock when deleting owner status"
 
     if name is not None:
       self.__owners[threading.currentThread()].remove(name)
 
     # Only remove the key if we don't hold the set-lock as well
-    if (not self.__lock._is_owned() and
+    if (not self.__lock.is_owned() and
         not self.__owners[threading.currentThread()]):
       del self.__owners[threading.currentThread()]
 
-  def _list_owned(self):
+  def list_owned(self):
     """Get the set of resource names owned by the current thread"""
-    if self._is_owned():
+    if self.is_owned():
       return self.__owners[threading.currentThread()].copy()
     else:
       return set()
 
   def _release_and_delete_owned(self):
     """Release and delete all resources owned by the current thread"""
-    for lname in self._list_owned():
+    for lname in self.list_owned():
       lock = self.__lockdict[lname]
-      if lock._is_owned():
+      if lock.is_owned():
         lock.release()
       self._del_owned(name=lname)
 
@@ -1002,7 +1050,7 @@ class LockSet:
     # If we don't already own the set-level lock acquired
     # we'll get it and note we need to release it later.
     release_lock = False
-    if not self.__lock._is_owned():
+    if not self.__lock.is_owned():
       release_lock = True
       self.__lock.acquire(shared=1)
     try:
@@ -1039,8 +1087,8 @@ class LockSet:
     assert timeout is None or timeout >= 0.0
 
     # Check we don't already own locks at this level
-    assert not self._is_owned(), ("Cannot acquire locks in the same set twice"
-                                  " (lockset %s)" % self.name)
+    assert not self.is_owned(), ("Cannot acquire locks in the same set twice"
+                                 " (lockset %s)" % self.name)
 
     if priority is None:
       priority = _DEFAULT_PRIORITY
@@ -1170,7 +1218,7 @@ class LockSet:
           # We shouldn't have problems adding the lock to the owners list, but
           # if we did we'll try to release this lock and re-raise exception.
           # Of course something is going to be really wrong after this.
-          if lock._is_owned():
+          if lock.is_owned():
             lock.release()
           raise
 
@@ -1187,14 +1235,14 @@ class LockSet:
     The locks must have been acquired in exclusive mode.
 
     """
-    assert self._is_owned(), ("downgrade on lockset %s while not owning any"
-                              " lock" % self.name)
+    assert self.is_owned(), ("downgrade on lockset %s while not owning any"
+                             " lock" % self.name)
 
     # Support passing in a single resource to downgrade rather than many
     if isinstance(names, basestring):
       names = [names]
 
-    owned = self._list_owned()
+    owned = self.list_owned()
 
     if names is None:
       names = owned
@@ -1208,12 +1256,12 @@ class LockSet:
       self.__lockdict[lockname].downgrade()
 
     # Do we own the lockset in exclusive mode?
-    if self.__lock._is_owned(shared=0):
+    if self.__lock.is_owned(shared=0):
       # Have all locks been downgraded?
-      if not compat.any(lock._is_owned(shared=0)
+      if not compat.any(lock.is_owned(shared=0)
                         for lock in self.__lockdict.values()):
         self.__lock.downgrade()
-        assert self.__lock._is_owned(shared=1)
+        assert self.__lock.is_owned(shared=1)
 
     return True
 
@@ -1228,24 +1276,24 @@ class LockSet:
         (defaults to all the locks acquired at that level).
 
     """
-    assert self._is_owned(), ("release() on lock set %s while not owner" %
-                              self.name)
+    assert self.is_owned(), ("release() on lock set %s while not owner" %
+                             self.name)
 
     # Support passing in a single resource to release rather than many
     if isinstance(names, basestring):
       names = [names]
 
     if names is None:
-      names = self._list_owned()
+      names = self.list_owned()
     else:
       names = set(names)
-      assert self._list_owned().issuperset(names), (
+      assert self.list_owned().issuperset(names), (
                "release() on unheld resources %s (set %s)" %
-               (names.difference(self._list_owned()), self.name))
+               (names.difference(self.list_owned()), self.name))
 
     # First of all let's release the "all elements" lock, if set.
     # After this 'add' can work again
-    if self.__lock._is_owned():
+    if self.__lock.is_owned():
       self.__lock.release()
       self._del_owned()
 
@@ -1267,7 +1315,7 @@ class LockSet:
 
     """
     # Check we don't already own locks at this level
-    assert not self._is_owned() or self.__lock._is_owned(shared=0), \
+    assert not self.is_owned() or self.__lock.is_owned(shared=0), \
       ("Cannot add locks if the set %s is only partially owned, or shared" %
        self.name)
 
@@ -1278,7 +1326,7 @@ class LockSet:
     # If we don't already own the set-level lock acquired in an exclusive way
     # we'll get it and note we need to release it later.
     release_lock = False
-    if not self.__lock._is_owned():
+    if not self.__lock.is_owned():
       release_lock = True
       self.__lock.acquire()
 
@@ -1341,7 +1389,7 @@ class LockSet:
     # If we own any subset of this lock it must be a superset of what we want
     # to delete. The ownership must also be exclusive, but that will be checked
     # by the lock itself.
-    assert not self._is_owned() or self._list_owned().issuperset(names), (
+    assert not self.is_owned() or self.list_owned().issuperset(names), (
       "remove() on acquired lockset %s while not owning all elements" %
       self.name)
 
@@ -1358,8 +1406,8 @@ class LockSet:
         removed.append(lname)
       except (KeyError, errors.LockError):
         # This cannot happen if we were already holding it, verify:
-        assert not self._is_owned(), ("remove failed while holding lockset %s"
-                                      % self.name)
+        assert not self.is_owned(), ("remove failed while holding lockset %s" %
+                                     self.name)
       else:
         # If no LockError was raised we are the ones who deleted the lock.
         # This means we can safely remove it from lockdict, as any further or
@@ -1370,7 +1418,7 @@ class LockSet:
         # it's the job of the one who actually deleted it.
         del self.__lockdict[lname]
         # And let's remove it from our private list if we owned it.
-        if self._is_owned():
+        if self.is_owned():
           self._del_owned(name=lname)
 
     return removed
@@ -1389,20 +1437,31 @@ LEVEL_CLUSTER = 0
 LEVEL_INSTANCE = 1
 LEVEL_NODEGROUP = 2
 LEVEL_NODE = 3
+LEVEL_NODE_RES = 4
 
-LEVELS = [LEVEL_CLUSTER,
-          LEVEL_INSTANCE,
-          LEVEL_NODEGROUP,
-          LEVEL_NODE]
+LEVELS = [
+  LEVEL_CLUSTER,
+  LEVEL_INSTANCE,
+  LEVEL_NODEGROUP,
+  LEVEL_NODE,
+  LEVEL_NODE_RES,
+  ]
 
 # Lock levels which are modifiable
-LEVELS_MOD = [LEVEL_NODE, LEVEL_NODEGROUP, LEVEL_INSTANCE]
-
+LEVELS_MOD = frozenset([
+  LEVEL_NODE_RES,
+  LEVEL_NODE,
+  LEVEL_NODEGROUP,
+  LEVEL_INSTANCE,
+  ])
+
+#: Lock level names (make sure to use singular form)
 LEVEL_NAMES = {
   LEVEL_CLUSTER: "cluster",
   LEVEL_INSTANCE: "instance",
   LEVEL_NODEGROUP: "nodegroup",
   LEVEL_NODE: "node",
+  LEVEL_NODE_RES: "node-res",
   }
 
 # Constant for the big ganeti lock
@@ -1441,13 +1500,25 @@ class GanetiLockManager:
     # The keyring contains all the locks, at their level and in the correct
     # locking order.
     self.__keyring = {
-      LEVEL_CLUSTER: LockSet([BGL], "BGL", monitor=self._monitor),
-      LEVEL_NODE: LockSet(nodes, "nodes", monitor=self._monitor),
-      LEVEL_NODEGROUP: LockSet(nodegroups, "nodegroups", monitor=self._monitor),
-      LEVEL_INSTANCE: LockSet(instances, "instances",
+      LEVEL_CLUSTER: LockSet([BGL], "cluster", monitor=self._monitor),
+      LEVEL_NODE: LockSet(nodes, "node", monitor=self._monitor),
+      LEVEL_NODE_RES: LockSet(nodes, "node-res", monitor=self._monitor),
+      LEVEL_NODEGROUP: LockSet(nodegroups, "nodegroup", monitor=self._monitor),
+      LEVEL_INSTANCE: LockSet(instances, "instance",
                               monitor=self._monitor),
       }
 
+    assert compat.all(ls.name == LEVEL_NAMES[level]
+                      for (level, ls) in self.__keyring.items())
+
+  def AddToLockMonitor(self, provider):
+    """Registers a new lock with the monitor.
+
+    See L{LockMonitor.RegisterLock}.
+
+    """
+    return self._monitor.RegisterLock(provider)
+
   def QueryLocks(self, fields):
     """Queries information from all locks.
 
@@ -1475,21 +1546,25 @@ class GanetiLockManager:
     assert level in LEVELS, "Invalid locking level %s" % level
     return self.__keyring[level]._names()
 
-  def _is_owned(self, level):
+  def is_owned(self, level):
     """Check whether we are owning locks at the given level
 
     """
-    return self.__keyring[level]._is_owned()
-
-  is_owned = _is_owned
+    return self.__keyring[level].is_owned()
 
-  def _list_owned(self, level):
+  def list_owned(self, level):
     """Get the set of owned locks at the given level
 
     """
-    return self.__keyring[level]._list_owned()
+    return self.__keyring[level].list_owned()
 
-  list_owned = _list_owned
+  def check_owned(self, level, names, shared=-1):
+    """Check if locks at a certain level are owned in a specific mode.
+
+    @see: L{LockSet.check_owned}
+
+    """
+    return self.__keyring[level].check_owned(names, shared=shared)
 
   def _upper_owned(self, level):
     """Check that we don't own any lock at a level greater than the given one.
@@ -1497,18 +1572,18 @@ class GanetiLockManager:
     """
     # This way of checking only works if LEVELS[i] = i, which we check for in
     # the test cases.
-    return compat.any((self._is_owned(l) for l in LEVELS[level + 1:]))
+    return compat.any((self.is_owned(l) for l in LEVELS[level + 1:]))
 
-  def _BGL_owned(self): # pylint: disable-msg=C0103
+  def _BGL_owned(self): # pylint: disable=C0103
     """Check if the current thread owns the BGL.
 
     Both an exclusive or a shared acquisition work.
 
     """
-    return BGL in self.__keyring[LEVEL_CLUSTER]._list_owned()
+    return BGL in self.__keyring[LEVEL_CLUSTER].list_owned()
 
   @staticmethod
-  def _contains_BGL(level, names): # pylint: disable-msg=C0103
+  def _contains_BGL(level, names): # pylint: disable=C0103
     """Check if the level contains the BGL.
 
     Check if acting on the given level and set of names will change
@@ -1587,7 +1662,7 @@ class GanetiLockManager:
             not self._upper_owned(LEVEL_CLUSTER)), (
             "Cannot release the Big Ganeti Lock while holding something"
             " at upper levels (%r)" %
-            (utils.CommaJoin(["%s=%r" % (LEVEL_NAMES[i], self._list_owned(i))
+            (utils.CommaJoin(["%s=%r" % (LEVEL_NAMES[i], self.list_owned(i))
                               for i in self.__keyring.keys()]), ))
 
     # Release will complain if we don't own the locks already
@@ -1632,21 +1707,23 @@ class GanetiLockManager:
     # Check we either own the level or don't own anything from here
     # up. LockSet.remove() will check the case in which we don't own
     # all the needed resources, or we have a shared ownership.
-    assert self._is_owned(level) or not self._upper_owned(level), (
+    assert self.is_owned(level) or not self._upper_owned(level), (
            "Cannot remove locks at a level while not owning it or"
            " owning some at a greater one")
     return self.__keyring[level].remove(names)
 
 
-def _MonitorSortKey((num, item)):
+def _MonitorSortKey((item, idx, num)):
   """Sorting key function.
 
-  Sort by name, then by incoming order.
+  Sort by name, registration order and then order of information. This provides
+  a stable sort order over different providers, even if they return the same
+  name.
 
   """
   (name, _, _, _) = item
 
-  return (utils.NiceSortKey(name), num)
+  return (utils.NiceSortKey(name), num, idx)
 
 
 class LockMonitor(object):
@@ -1666,12 +1743,19 @@ class LockMonitor(object):
     self._locks = weakref.WeakKeyDictionary()
 
   @ssynchronized(_LOCK_ATTR)
-  def RegisterLock(self, lock):
+  def RegisterLock(self, provider):
     """Registers a new lock.
 
+    @param provider: Object with a callable method named C{GetLockInfo}, taking
+      a single C{set} containing the requested information items
+    @note: It would be nicer to only receive the function generating the
+      requested information but, as it turns out, weak references to bound
+      methods (e.g. C{self.GetLockInfo}) are tricky; there are several
+      workarounds, but none of the ones I found works properly in combination
+      with a standard C{WeakKeyDictionary}
+
     """
-    logging.debug("Registering lock %s", lock.name)
-    assert lock not in self._locks, "Duplicate lock registration"
+    assert provider not in self._locks, "Duplicate registration"
 
     # There used to be a check for duplicate names here. As it turned out, when
     # a lock is re-created with the same name in a very short timeframe, the
@@ -1679,14 +1763,22 @@ class LockMonitor(object):
     # By keeping track of the order of incoming registrations, a stable sort
     # ordering can still be guaranteed.
 
-    self._locks[lock] = self._counter.next()
+    self._locks[provider] = self._counter.next()
 
-  @ssynchronized(_LOCK_ATTR)
   def _GetLockInfo(self, requested):
-    """Get information from all locks while the monitor lock is held.
+    """Get information from all locks.
 
     """
-    return [(num, lock.GetInfo(requested)) for lock, num in self._locks.items()]
+    # Must hold lock while getting consistent list of tracked items
+    self._lock.acquire(shared=1)
+    try:
+      items = self._locks.items()
+    finally:
+      self._lock.release()
+
+    return [(info, idx, num)
+            for (provider, num) in items
+            for (idx, info) in enumerate(provider.GetLockInfo(requested))]
 
   def _Query(self, fields):
     """Queries information from all locks.
@@ -1703,7 +1795,7 @@ class LockMonitor(object):
                       key=_MonitorSortKey)
 
     # Extract lock information and build query data
-    return (qobj, query.LockQueryData(map(operator.itemgetter(1), lockinfo)))
+    return (qobj, query.LockQueryData(map(compat.fst, lockinfo)))
 
   def QueryLocks(self, fields):
     """Queries information from all locks.