#
#
-# 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
"""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
import os
import select
import threading
-import time
import errno
import weakref
import logging
import heapq
+import itertools
from ganeti import errors
from ganeti import utils
from ganeti import compat
+from ganeti import query
_EXCLUSIVE_TEXT = "exclusive"
_SHARED_TEXT = "shared"
+_DELETED_TEXT = "deleted"
_DEFAULT_PRIORITY = 0
return wrap
-class RunningTimeout(object):
- """Class to calculate remaining timeout when doing several operations.
-
- """
- __slots__ = [
- "_allow_negative",
- "_start_time",
- "_time_fn",
- "_timeout",
- ]
-
- def __init__(self, timeout, allow_negative, _time_fn=time.time):
- """Initializes this class.
-
- @type timeout: float
- @param timeout: Timeout duration
- @type allow_negative: bool
- @param allow_negative: Whether to return values below zero
- @param _time_fn: Time function for unittests
-
- """
- object.__init__(self)
-
- if timeout is not None and timeout < 0.0:
- raise ValueError("Timeout must not be negative")
-
- self._timeout = timeout
- self._allow_negative = allow_negative
- self._time_fn = _time_fn
-
- self._start_time = None
-
- def Remaining(self):
- """Returns the remaining timeout.
-
- """
- if self._timeout is None:
- return None
-
- # Get start time on first calculation
- if self._start_time is None:
- self._start_time = self._time_fn()
-
- # Calculate remaining time
- remaining_timeout = self._start_time + self._timeout - self._time_fn()
-
- if not self._allow_negative:
- # Ensure timeout is always >= 0
- return max(0.0, remaining_timeout)
-
- return remaining_timeout
-
-
class _SingleNotifyPipeConditionWaiter(object):
"""Helper class for SingleNotifyPipeCondition
@param timeout: Timeout for waiting (can be None)
"""
- running_timeout = RunningTimeout(timeout, True)
+ running_timeout = utils.RunningTimeout(timeout, True)
while True:
remaining_time = running_timeout.Remaining()
self._write_fd = None
self._poller = None
- def wait(self, timeout=None):
+ def wait(self, timeout):
"""Wait for a notification.
@type timeout: float or None
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.
"""
self._waiters = set()
self._single_condition = self._single_condition_class(self._lock)
- def wait(self, timeout=None):
+ def wait(self, timeout):
"""Wait for a notification.
@type timeout: float or None
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.
"""
class SharedLock(object):
"""Implements a shared lock.
- Multiple threads can acquire the lock in a shared way, calling
- acquire_shared(). In order to acquire the lock in an exclusive way threads
- can call acquire_exclusive().
+ Multiple threads can acquire the lock in a shared way by calling
+ C{acquire(shared=1)}. In order to acquire the lock in an exclusive way
+ threads can call C{acquire(shared=0)}.
Notes on data structures: C{__pending} contains a priority queue (heapq) of
all pending acquires: C{[(priority1: prioqueue1), (priority2: prioqueue2),
# Register with lock monitor
if monitor:
+ logging.debug("Adding lock %s to monitor", name)
monitor.RegisterLock(self)
- def GetInfo(self, fields):
+ def GetLockInfo(self, requested):
"""Retrieves information for querying locks.
- @type fields: list of strings
- @param fields: List of fields to return
+ @type requested: set
+ @param requested: Requested information, see C{query.LQ_*}
"""
self.__lock.acquire()
try:
- info = []
-
# Note: to avoid unintentional race conditions, no references to
# modifiable objects should be returned unless they were created in this
# function.
- for fname in fields:
- if fname == "name":
- info.append(self.name)
- elif fname == "mode":
- if self.__deleted:
- info.append("deleted")
- assert not (self.__exc or self.__shr)
- elif self.__exc:
- info.append(_EXCLUSIVE_TEXT)
- elif self.__shr:
- info.append(_SHARED_TEXT)
- else:
- info.append(None)
- elif fname == "owner":
- if self.__exc:
- owner = [self.__exc]
- else:
- owner = self.__shr
-
- if owner:
- assert not self.__deleted
- info.append([i.getName() for i in owner])
- else:
- info.append(None)
- elif fname == "pending":
- data = []
-
- # Sorting instead of copying and using heaq functions for simplicity
- for (_, prioqueue) in sorted(self.__pending):
- for cond in prioqueue:
- if cond.shared:
- mode = _SHARED_TEXT
- else:
- mode = _EXCLUSIVE_TEXT
-
- # This function should be fast as it runs with the lock held.
- # Hence not using utils.NiceSort.
- data.append((mode, sorted(i.getName()
- for i in cond.get_waiting())))
-
- info.append(data)
+ mode = None
+ owner_names = None
+
+ if query.LQ_MODE in requested:
+ if self.__deleted:
+ mode = _DELETED_TEXT
+ assert not (self.__exc or self.__shr)
+ elif self.__exc:
+ mode = _EXCLUSIVE_TEXT
+ elif self.__shr:
+ mode = _SHARED_TEXT
+
+ # Current owner(s) are wanted
+ if query.LQ_OWNER in requested:
+ if self.__exc:
+ owner = [self.__exc]
else:
- raise errors.OpExecError("Invalid query field '%s'" % fname)
+ owner = self.__shr
+
+ if owner:
+ assert not self.__deleted
+ owner_names = [i.getName() for i in owner]
+
+ # Pending acquires are wanted
+ if query.LQ_PENDING in requested:
+ pending = []
- return info
+ # Sorting instead of copying and using heaq functions for simplicity
+ for (_, prioqueue) in sorted(self.__pending):
+ for cond in prioqueue:
+ if cond.shared:
+ pendmode = _SHARED_TEXT
+ else:
+ pendmode = _EXCLUSIVE_TEXT
+
+ # List of names will be sorted in L{query._GetLockPending}
+ pending.append((pendmode, [i.getName()
+ for i in cond.get_waiting()]))
+ else:
+ pending = None
+
+ return [(self.name, mode, owner_names, pending)]
finally:
self.__lock.release()
finally:
self.__lock.release()
+ is_owned = _is_owned
+
def _count_pending(self):
"""Returns the number of pending acquires.
self.__lock.acquire()
try:
# Order is important: __find_first_pending_queue modifies __pending
- return not (self.__find_first_pending_queue() or
+ (_, prioqueue) = self.__find_first_pending_queue()
+
+ return not (prioqueue or
self.__pending or
self.__pending_by_prio or
self.__pending_shared)
while self.__pending:
(priority, prioqueue) = self.__pending[0]
- if not prioqueue:
- heapq.heappop(self.__pending)
- del self.__pending_by_prio[priority]
- assert priority not in self.__pending_shared
- continue
-
if prioqueue:
- return prioqueue
+ return (priority, prioqueue)
- return None
+ # Remove empty queue
+ heapq.heappop(self.__pending)
+ del self.__pending_by_prio[priority]
+ assert priority not in self.__pending_shared
+
+ return (None, None)
def __is_on_top(self, cond):
"""Checks whether the passed condition is on top of the queue.
The caller must make sure the queue isn't empty.
"""
- return cond == self.__find_first_pending_queue()[0]
+ (_, prioqueue) = self.__find_first_pending_queue()
+
+ return cond == prioqueue[0]
def __acquire_unlocked(self, shared, timeout, priority):
"""Acquire a shared lock.
if not wait_condition.has_waiting():
prioqueue.remove(wait_condition)
if wait_condition.shared:
- del self.__pending_shared[priority]
+ # Remove from list of shared acquires if it wasn't while releasing
+ # (e.g. on lock deletion)
+ self.__pending_shared.pop(priority, None)
return False
- def acquire(self, shared=0, timeout=None, priority=_DEFAULT_PRIORITY,
+ def acquire(self, shared=0, timeout=None, priority=None,
test_notify=None):
"""Acquire a shared lock.
@param test_notify: Special callback function for unittesting
"""
+ if priority is None:
+ priority = _DEFAULT_PRIORITY
+
self.__lock.acquire()
try:
# We already got the lock, notify now
finally:
self.__lock.release()
+ def downgrade(self):
+ """Changes the lock mode from exclusive to shared.
+
+ Pending acquires in shared mode on the same priority will go ahead.
+
+ """
+ self.__lock.acquire()
+ try:
+ assert self.__is_owned(), "Lock must be owned"
+
+ if self.__is_exclusive():
+ # Do nothing if the lock is already acquired in shared mode
+ self.__exc = None
+ self.__do_acquire(1)
+
+ # Important: pending shared acquires should only jump ahead if there
+ # was a transition from exclusive to shared, otherwise an owner of a
+ # shared lock can keep calling this function to push incoming shared
+ # acquires
+ (priority, prioqueue) = self.__find_first_pending_queue()
+ if prioqueue:
+ # Is there a pending shared acquire on this priority?
+ cond = self.__pending_shared.pop(priority, None)
+ if cond:
+ assert cond.shared
+ assert cond in prioqueue
+
+ # Ensure shared acquire is on top of queue
+ if len(prioqueue) > 1:
+ prioqueue.remove(cond)
+ prioqueue.insert(0, cond)
+
+ # Notify
+ cond.notifyAll()
+
+ assert not self.__is_exclusive()
+ assert self.__is_sharer()
+
+ return True
+ finally:
+ self.__lock.release()
+
def release(self):
"""Release a Shared Lock.
self.__shr.remove(threading.currentThread())
# Notify topmost condition in queue
- prioqueue = self.__find_first_pending_queue()
+ (priority, prioqueue) = self.__find_first_pending_queue()
if prioqueue:
- prioqueue[0].notifyAll()
+ cond = prioqueue[0]
+ cond.notifyAll()
+ if cond.shared:
+ # Prevent further shared acquires from sneaking in while waiters are
+ # notified
+ self.__pending_shared.pop(priority, None)
finally:
self.__lock.release()
- def delete(self, timeout=None, priority=_DEFAULT_PRIORITY):
+ def delete(self, timeout=None, priority=None):
"""Delete a Shared Lock.
This operation will declare the lock for removal. First the lock will be
@param priority: Priority for acquiring lock
"""
+ if priority is None:
+ priority = _DEFAULT_PRIORITY
+
self.__lock.acquire()
try:
assert not self.__is_sharer(), "Cannot delete() a lock while sharing it"
# Lock monitor
self.__monitor = monitor
- # Used internally to guarantee coherency.
- self.__lock = SharedLock(name)
+ # Used internally to guarantee coherency
+ self.__lock = SharedLock(self._GetLockName("[lockset]"), monitor=monitor)
# The lockdict indexes the relationship name -> lock
# The order-of-locking is implied by the alphabetical order of names
"""
return "%s/%s" % (self.name, mname)
+ def _get_lock(self):
+ """Returns the lockset-internal lock.
+
+ """
+ return self.__lock
+
+ def _get_lockdict(self):
+ """Returns the lockset-internal lock dictionary.
+
+ Accessing this structure is only safe in single-thread usage or when the
+ lockset-internal lock is held.
+
+ """
+ return self.__lockdict
+
def _is_owned(self):
"""Is the current thread a current level owner?"""
return threading.currentThread() in self.__owners
self.__lock.release()
return set(result)
- def acquire(self, names, timeout=None, shared=0, priority=_DEFAULT_PRIORITY,
+ def acquire(self, names, timeout=None, shared=0, priority=None,
test_notify=None):
"""Acquire a set of resource locks.
assert not self._is_owned(), ("Cannot acquire locks in the same set twice"
" (lockset %s)" % self.name)
+ if priority is None:
+ priority = _DEFAULT_PRIORITY
+
# We need to keep track of how long we spent waiting for a lock. The
# timeout passed to this function is over all lock acquires.
- running_timeout = RunningTimeout(timeout, False)
+ running_timeout = utils.RunningTimeout(timeout, False)
try:
if names is not None:
# element is not there anymore.
continue
- raise errors.LockError("Non-existing lock %s in set %s" %
- (lname, self.name))
+ raise errors.LockError("Non-existing lock %s in set %s (it may have"
+ " been removed)" % (lname, self.name))
acquire_list.append((lname, lock))
# particular element is not there anymore.
continue
- raise errors.LockError("Non-existing lock %s in set %s" %
- (lname, self.name))
+ raise errors.LockError("Non-existing lock %s in set %s (it may"
+ " have been removed)" % (lname, self.name))
if not acq_success:
# Couldn't get lock or timeout occurred
return acquired
+ def downgrade(self, names=None):
+ """Downgrade a set of resource locks from exclusive to shared mode.
+
+ The locks must have been acquired in exclusive mode.
+
+ """
+ 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()
+
+ if names is None:
+ names = owned
+ else:
+ names = set(names)
+ assert owned.issuperset(names), \
+ ("downgrade() on unheld resources %s (set %s)" %
+ (names.difference(owned), self.name))
+
+ for lockname in names:
+ self.__lockdict[lockname].downgrade()
+
+ # Do we own the lockset in exclusive mode?
+ if self.__lock._is_owned(shared=0):
+ # Have all locks been downgraded?
+ 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)
+
+ return True
+
def release(self, names=None):
"""Release a set of resource locks, at the same level.
# the same time.
LEVEL_CLUSTER = 0
LEVEL_INSTANCE = 1
-LEVEL_NODE = 2
+LEVEL_NODEGROUP = 2
+LEVEL_NODE = 3
LEVELS = [LEVEL_CLUSTER,
LEVEL_INSTANCE,
+ LEVEL_NODEGROUP,
LEVEL_NODE]
# Lock levels which are modifiable
-LEVELS_MOD = [LEVEL_NODE, LEVEL_INSTANCE]
+LEVELS_MOD = [LEVEL_NODE, LEVEL_NODEGROUP, LEVEL_INSTANCE]
LEVEL_NAMES = {
LEVEL_CLUSTER: "cluster",
LEVEL_INSTANCE: "instance",
+ LEVEL_NODEGROUP: "nodegroup",
LEVEL_NODE: "node",
}
"""
_instance = None
- def __init__(self, nodes=None, instances=None):
+ def __init__(self, nodes, nodegroups, instances):
"""Constructs a new GanetiLockManager object.
There should be only a GanetiLockManager object at any time, so this
function raises an error if this is not the case.
@param nodes: list of node names
+ @param nodegroups: list of nodegroup uuids
@param instances: list of instance names
"""
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",
monitor=self._monitor),
}
- def QueryLocks(self, fields, sync):
+ 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.
See L{LockMonitor.QueryLocks}.
"""
- return self._monitor.QueryLocks(fields, sync)
+ return self._monitor.QueryLocks(fields)
+
+ def OldStyleQueryLocks(self, fields):
+ """Queries information from all locks, returning old-style data.
+
+ See L{LockMonitor.OldStyleQueryLocks}.
+
+ """
+ return self._monitor.OldStyleQueryLocks(fields)
def _names(self, level):
"""List the lock names at the given level.
"""
return self.__keyring[level]._list_owned()
+ list_owned = _list_owned
+
def _upper_owned(self, level):
"""Check that we don't own any lock at a level greater than the given one.
# the test cases.
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()
@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
"""
return level == LEVEL_CLUSTER and (names is None or BGL in names)
- def acquire(self, level, names, timeout=None, shared=0):
+ def acquire(self, level, names, timeout=None, shared=0, priority=None):
"""Acquire a set of resource locks, at the same level.
@type level: member of locking.LEVELS
an exclusive lock will be acquired
@type timeout: float
@param timeout: Maximum time to acquire all locks
+ @type priority: integer
+ @param priority: Priority for acquiring lock
"""
assert level in LEVELS, "Invalid locking level %s" % level
" while owning some at a greater one")
# Acquire the locks in the set.
- return self.__keyring[level].acquire(names, shared=shared, timeout=timeout)
+ return self.__keyring[level].acquire(names, shared=shared, timeout=timeout,
+ priority=priority)
+
+ def downgrade(self, level, names=None):
+ """Downgrade a set of resource locks from exclusive to shared mode.
+
+ You must have acquired the locks in exclusive mode.
+
+ @type level: member of locking.LEVELS
+ @param level: the level at which the locks shall be downgraded
+ @type names: list of strings, or None
+ @param names: the names of the locks which shall be downgraded
+ (defaults to all the locks acquired at the level)
+
+ """
+ assert level in LEVELS, "Invalid locking level %s" % level
+
+ return self.__keyring[level].downgrade(names=names)
def release(self, level, names=None):
"""Release a set of resource locks, at the same level.
return self.__keyring[level].remove(names)
+def _MonitorSortKey((item, idx, num)):
+ """Sorting key function.
+
+ 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, idx)
+
+
class LockMonitor(object):
_LOCK_ATTR = "_lock"
"""
self._lock = SharedLock("LockMonitor")
+ # Counter for stable sorting
+ self._counter = itertools.count(0)
+
# Tracked locks. Weak references are used to avoid issues with circular
# references and deletion.
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 not compat.any(lock.name == i.name for i in self._locks.keys()), \
- "Found duplicate lock name"
- self._locks[lock] = None
+ assert provider not in self._locks, "Duplicate registration"
- @ssynchronized(_LOCK_ATTR)
- def _GetLockInfo(self, fields):
- """Get information from all locks while the monitor lock is held.
+ # 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
+ # previous instance might not yet be removed from the weakref dictionary.
+ # By keeping track of the order of incoming registrations, a stable sort
+ # ordering can still be guaranteed.
+
+ self._locks[provider] = self._counter.next()
+
+ def _GetLockInfo(self, requested):
+ """Get information from all locks.
+
+ """
+ # 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.
+
+ @type fields: list of strings
+ @param fields: List of fields to return
"""
- result = {}
+ qobj = query.Query(query.LOCK_FIELDS, fields)
- for lock in self._locks.keys():
- assert lock.name not in result, "Found duplicate lock name"
- result[lock.name] = lock.GetInfo(fields)
+ # Get all data with internal lock held and then sort by name and incoming
+ # order
+ lockinfo = sorted(self._GetLockInfo(qobj.RequestedData()),
+ key=_MonitorSortKey)
- return result
+ # Extract lock information and build query data
+ return (qobj, query.LockQueryData(map(compat.fst, lockinfo)))
- def QueryLocks(self, fields, sync):
+ def QueryLocks(self, fields):
"""Queries information from all locks.
@type fields: list of strings
@param fields: List of fields to return
- @type sync: boolean
- @param sync: Whether to operate in synchronous mode
"""
- if sync:
- raise NotImplementedError("Synchronous queries are not implemented")
+ (qobj, ctx) = self._Query(fields)
+
+ # Prepare query response
+ return query.GetQueryResponse(qobj, ctx)
+
+ def OldStyleQueryLocks(self, fields):
+ """Queries information from all locks, returning old-style data.
- # Get all data without sorting
- result = self._GetLockInfo(fields)
+ @type fields: list of strings
+ @param fields: List of fields to return
+
+ """
+ (qobj, ctx) = self._Query(fields)
- # Sort by name
- return [result[name] for name in utils.NiceSort(result.keys())]
+ return qobj.OldStyleQuery(ctx)
projects / ganeti-local / blobdiff