import os
import select
import threading
-import time
import errno
import weakref
import logging
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()
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),
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 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"
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
# 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),
}
"""
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 release(self, level, names=None):
"""Release a set of resource locks, at the same level.
projects / ganeti-local / blobdiff