Synnefo » snf-ganeti

#

#

# Copyright (C) 2006, 2007, 2008, 2009, 2010 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

# the Free Software Foundation; either version 2 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful, but

# WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU

# General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program; if not, write to the Free Software

# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA

# 02110-1301, USA.

"""Module implementing the Ganeti locking code."""

# pylint: disable-msg=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

from ganeti import errors

from ganeti import utils

from ganeti import compat

_EXCLUSIVE_TEXT = "exclusive"

_SHARED_TEXT = "shared"

_DEFAULT_PRIORITY = 0

def ssynchronized(mylock, shared=0):

  """Shared Synchronization decorator.

  Calls the function holding the given lock, either in exclusive or shared

  mode. It requires the passed lock to be a SharedLock (or support its

  semantics).

  @type mylock: lockable object or string

  @param mylock: lock to acquire or class member name of the lock to acquire

  """

  def wrap(fn):

    def sync_function(*args, **kwargs):

      if isinstance(mylock, basestring):

        assert args, "cannot ssynchronize on non-class method: self not found"

        # args[0] is "self"

        lock = getattr(args[0], mylock)

      else:

        lock = mylock

      lock.acquire(shared=shared)

      try:

        return fn(*args, **kwargs)

      finally:

        lock.release()

    return sync_function

  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

  """

  __slots__ = [

    "_fd",

    "_poller",

    ]

  def __init__(self, poller, fd):

    """Constructor for _SingleNotifyPipeConditionWaiter

    @type poller: select.poll

    @param poller: Poller object

    @type fd: int

    @param fd: File descriptor to wait for

    """

    object.__init__(self)

    self._poller = poller

    self._fd = fd

  def __call__(self, timeout):

    """Wait for something to happen on the pipe.

    @type timeout: float or None

    @param timeout: Timeout for waiting (can be None)

    """

    running_timeout = RunningTimeout(timeout, True)

    while True:

      remaining_time = running_timeout.Remaining()

      if remaining_time is not None:

        if remaining_time < 0.0:

          break

        # Our calculation uses seconds, poll() wants milliseconds

        remaining_time *= 1000

      try:

        result = self._poller.poll(remaining_time)

      except EnvironmentError, err:

        if err.errno != errno.EINTR:

          raise

        result = None

      # Check whether we were notified

      if result and result[0][0] == self._fd:

        break

class _BaseCondition(object):

  """Base class containing common code for conditions.

  Some of this code is taken from python's threading module.

  """

  __slots__ = [

    "_lock",

    "acquire",

    "release",

    "_is_owned",

    "_acquire_restore",

    "_release_save",

    ]

  def __init__(self, lock):

    """Constructor for _BaseCondition.

    @type lock: threading.Lock

    @param lock: condition base lock

    """

    object.__init__(self)

    try:

      self._release_save = lock._release_save

    except AttributeError:

      self._release_save = self._base_release_save

    try:

      self._acquire_restore = lock._acquire_restore

    except AttributeError:

      self._acquire_restore = self._base_acquire_restore

    try:

      self._is_owned = lock._is_owned

    except AttributeError:

      self._is_owned = self._base_is_owned

    self._lock = lock

    # Export the lock's acquire() and release() methods

    self.acquire = lock.acquire

    self.release = lock.release

  def _base_is_owned(self):

    """Check whether lock is owned by current thread.

    """

    if self._lock.acquire(0):

      self._lock.release()

      return False

    return True

  def _base_release_save(self):

    self._lock.release()

  def _base_acquire_restore(self, _):

    self._lock.acquire()

  def _check_owned(self):

    """Raise an exception if the current thread doesn't own the lock.

    """

    if not self._is_owned():

      raise RuntimeError("cannot work with un-aquired lock")

class SingleNotifyPipeCondition(_BaseCondition):

  """Condition which can only be notified once.

  This condition class uses pipes and poll, internally, to be able to wait for

  notification with a timeout, without resorting to polling. It is almost

  compatible with Python's threading.Condition, with the following differences:

    - notifyAll can only be called once, and no wait can happen after that

    - notify is not supported, only notifyAll

  """

  __slots__ = [

    "_poller",

    "_read_fd",

    "_write_fd",

    "_nwaiters",

    "_notified",

    ]

  _waiter_class = _SingleNotifyPipeConditionWaiter

  def __init__(self, lock):

    """Constructor for SingleNotifyPipeCondition

    """

    _BaseCondition.__init__(self, lock)

    self._nwaiters = 0

    self._notified = False

    self._read_fd = None

    self._write_fd = None

    self._poller = None

  def _check_unnotified(self):

    """Throws an exception if already notified.

    """

    if self._notified:

      raise RuntimeError("cannot use already notified condition")

  def _Cleanup(self):

    """Cleanup open file descriptors, if any.

    """

    if self._read_fd is not None:

      os.close(self._read_fd)

      self._read_fd = None

    if self._write_fd is not None:

      os.close(self._write_fd)

      self._write_fd = None

    self._poller = None

  def wait(self, timeout=None):

    """Wait for a notification.

    @type timeout: float or None

    @param timeout: Waiting timeout (can be None)

    """

    self._check_owned()

    self._check_unnotified()

    self._nwaiters += 1

    try:

      if self._poller is None:

        (self._read_fd, self._write_fd) = os.pipe()

        self._poller = select.poll()

        self._poller.register(self._read_fd, select.POLLHUP)

      wait_fn = self._waiter_class(self._poller, self._read_fd)

      state = self._release_save()

      try:

        # Wait for notification

        wait_fn(timeout)

      finally:

        # Re-acquire lock

        self._acquire_restore(state)

    finally:

      self._nwaiters -= 1

      if self._nwaiters == 0:

        self._Cleanup()

  def notifyAll(self): # pylint: disable-msg=C0103

    """Close the writing side of the pipe to notify all waiters.

    """

    self._check_owned()

    self._check_unnotified()

    self._notified = True

    if self._write_fd is not None:

      os.close(self._write_fd)

      self._write_fd = None

class PipeCondition(_BaseCondition):

  """Group-only non-polling condition with counters.

  This condition class uses pipes and poll, internally, to be able to wait for

  notification with a timeout, without resorting to polling. It is almost

  compatible with Python's threading.Condition, but only supports notifyAll and

  non-recursive locks. As an additional features it's able to report whether

  there are any waiting threads.

  """

  __slots__ = [

    "_waiters",

    "_single_condition",

    ]

  _single_condition_class = SingleNotifyPipeCondition

  def __init__(self, lock):

    """Initializes this class.

    """

    _BaseCondition.__init__(self, lock)

    self._waiters = set()

    self._single_condition = self._single_condition_class(self._lock)

  def wait(self, timeout=None):

    """Wait for a notification.

    @type timeout: float or None

    @param timeout: Waiting timeout (can be None)

    """

    self._check_owned()

    # Keep local reference to the pipe. It could be replaced by another thread

    # notifying while we're waiting.

    cond = self._single_condition

    self._waiters.add(threading.currentThread())

    try:

      cond.wait(timeout)

    finally:

      self._check_owned()

      self._waiters.remove(threading.currentThread())

  def notifyAll(self): # pylint: disable-msg=C0103

    """Notify all currently waiting threads.

    """

    self._check_owned()

    self._single_condition.notifyAll()

    self._single_condition = self._single_condition_class(self._lock)

  def get_waiting(self):

    """Returns a list of all waiting threads.

    """

    self._check_owned()

    return self._waiters

  def has_waiting(self):

    """Returns whether there are active waiters.

    """

    self._check_owned()

    return bool(self._waiters)

class _PipeConditionWithMode(PipeCondition):

  __slots__ = [

    "shared",

    ]

  def __init__(self, lock, shared):

    """Initializes this class.

    """

    self.shared = shared

    PipeCondition.__init__(self, lock)

class SharedLock(object):

  """Implements a shared lock.

  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),

  ...]}. Each per-priority queue contains a normal in-order list of conditions

  to be notified when the lock can be acquired. Shared locks are grouped

  together by priority and the condition for them is stored in

  C{__pending_shared} if it already exists. C{__pending_by_prio} keeps

  references for the per-priority queues indexed by priority for faster access.

  @type name: string

  @ivar name: the name of the lock

  """

  __slots__ = [

    "__weakref__",

    "__deleted",

    "__exc",

    "__lock",

    "__pending",

    "__pending_by_prio",

    "__pending_shared",

    "__shr",

    "name",

    ]

  __condition_class = _PipeConditionWithMode

  def __init__(self, name, monitor=None):

    """Construct a new SharedLock.

    @param name: the name of the lock

    @type monitor: L{LockMonitor}

    @param monitor: Lock monitor with which to register

    """

    object.__init__(self)

    self.name = name

    # Internal lock

    self.__lock = threading.Lock()

    # Queue containing waiting acquires

    self.__pending = []

    self.__pending_by_prio = {}

    self.__pending_shared = {}

    # Current lock holders

    self.__shr = set()

    self.__exc = None

    # is this lock in the deleted state?

    self.__deleted = False

    # Register with lock monitor

    if monitor:

      monitor.RegisterLock(self)

  def GetInfo(self, fields):

    """Retrieves information for querying locks.

    @type fields: list of strings

    @param fields: List of fields to return

    """

    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)

        else:

          raise errors.OpExecError("Invalid query field '%s'" % fname)

      return info

    finally:

      self.__lock.release()

  def __check_deleted(self):

    """Raises an exception if the lock has been deleted.

    """

    if self.__deleted:

      raise errors.LockError("Deleted lock %s" % self.name)

  def __is_sharer(self):

    """Is the current thread sharing the lock at this time?

    """

    return threading.currentThread() in self.__shr

  def __is_exclusive(self):

    """Is the current thread holding the lock exclusively at this time?

    """

    return threading.currentThread() == self.__exc

  def __is_owned(self, shared=-1):

    """Is the current thread somehow owning the lock at this time?

    This is a private version of the function, which presumes you're holding

    the internal lock.

    """

    if shared < 0:

      return self.__is_sharer() or self.__is_exclusive()

    elif shared:

      return self.__is_sharer()

    else:

      return self.__is_exclusive()

  def _is_owned(self, shared=-1):

    """Is the current thread somehow owning the lock at this time?

    @param shared:

        - < 0: check for any type of ownership (default)

        - 0: check for exclusive ownership

        - > 0: check for shared ownership

    """

    self.__lock.acquire()

    try:

      return self.__is_owned(shared=shared)

    finally:

      self.__lock.release()

  def _count_pending(self):

    """Returns the number of pending acquires.

    @rtype: int

    """

    self.__lock.acquire()

    try:

      return sum(len(prioqueue) for (_, prioqueue) in self.__pending)

    finally:

      self.__lock.release()

  def _check_empty(self):

    """Checks whether there are any pending acquires.

    @rtype: bool

    """

    self.__lock.acquire()

    try:

      # Order is important: __find_first_pending_queue modifies __pending

      return not (self.__find_first_pending_queue() or

                  self.__pending or

                  self.__pending_by_prio or

                  self.__pending_shared)

    finally:

      self.__lock.release()

  def __do_acquire(self, shared):

    """Actually acquire the lock.

    """

    if shared:

      self.__shr.add(threading.currentThread())

    else:

      self.__exc = threading.currentThread()

  def __can_acquire(self, shared):

    """Determine whether lock can be acquired.

    """

    if shared:

      return self.__exc is None

    else:

      return len(self.__shr) == 0 and self.__exc is None

  def __find_first_pending_queue(self):

    """Tries to find the topmost queued entry with pending acquires.

    Removes empty entries while going through the list.

    """

    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 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]

  def __acquire_unlocked(self, shared, timeout, priority):

    """Acquire a shared lock.

    @param shared: whether to acquire in shared mode; by default an

        exclusive lock will be acquired

    @param timeout: maximum waiting time before giving up

    @type priority: integer

    @param priority: Priority for acquiring lock

    """

    self.__check_deleted()

    # We cannot acquire the lock if we already have it

    assert not self.__is_owned(), ("double acquire() on a non-recursive lock"

                                   " %s" % self.name)

    # Remove empty entries from queue

    self.__find_first_pending_queue()

    # Check whether someone else holds the lock or there are pending acquires.

    if not self.__pending and self.__can_acquire(shared):

      # Apparently not, can acquire lock directly.

      self.__do_acquire(shared)

      return True

    prioqueue = self.__pending_by_prio.get(priority, None)

    if shared:

      # Try to re-use condition for shared acquire

      wait_condition = self.__pending_shared.get(priority, None)

      assert (wait_condition is None or

              (wait_condition.shared and wait_condition in prioqueue))

    else:

      wait_condition = None

    if wait_condition is None:

      if prioqueue is None:

        assert priority not in self.__pending_by_prio

        prioqueue = []

        heapq.heappush(self.__pending, (priority, prioqueue))

        self.__pending_by_prio[priority] = prioqueue

      wait_condition = self.__condition_class(self.__lock, shared)

      prioqueue.append(wait_condition)

      if shared:

        # Keep reference for further shared acquires on same priority. This is

        # better than trying to find it in the list of pending acquires.

        assert priority not in self.__pending_shared

        self.__pending_shared[priority] = wait_condition

    try:

      # Wait until we become the topmost acquire in the queue or the timeout

      # expires.

      # TODO: Decrease timeout with spurious notifications

      while not (self.__is_on_top(wait_condition) and

                 self.__can_acquire(shared)):

        # Wait for notification

        wait_condition.wait(timeout)

        self.__check_deleted()

        # A lot of code assumes blocking acquires always succeed. Loop

        # internally for that case.

        if timeout is not None:

          break

      if self.__is_on_top(wait_condition) and self.__can_acquire(shared):

        self.__do_acquire(shared)

        return True

    finally:

      # Remove condition from queue if there are no more waiters

      if not wait_condition.has_waiting():

        prioqueue.remove(wait_condition)

        if wait_condition.shared:

          del self.__pending_shared[priority]

    return False

  def acquire(self, shared=0, timeout=None, priority=None,

              test_notify=None):

    """Acquire a shared lock.

    @type shared: integer (0/1) used as a boolean

    @param shared: whether to acquire in shared mode; by default an

        exclusive lock will be acquired

    @type timeout: float