Synnefo » snf-ganeti

ensure opcodes with higher priorities get locks first. See

:ref:`locking priorities <locking-priorities>` for more details.

.. _locking-priorities:

Locking

+++++++

In order to support priorities in Ganeti's own lock classes,

``locking.SharedLock`` and ``locking.LockSet``, the internal structure

of the former class needs to be changed. The last major change in this

area was done for Ganeti 2.1 and can be found in the respective

:doc:`design document <design-2.1>`.

The plain list (``[]``) used as a queue is replaced by a heap queue,

similar to the `worker pool`_. The heap or priority queue does automatic

sorting, thereby automatically taking care of priorities. For each

priority there's a plain list with pending acquires, like the single

queue of pending acquires before this change.

When the lock is released, the code locates the list of pending acquires

for the highest priority waiting. The first condition (index 0) is

notified. Once all waiting threads received the notification, the

condition is removed from the list. If the list of conditions is empty

it's removed from the heap queue.

Like before, shared acquires are grouped and skip ahead of exclusive

acquires if there's already an existing shared acquire for a priority.

To accomplish this, a separate dictionary of shared acquires per

priority is maintained.

To simplify the code and reduce memory consumption, the concept of the

"active" and "inactive" condition for shared acquires is abolished. The

lock can't predict what priorities the next acquires will use and even

keeping a cache can become computationally expensive for arguable

benefit (the underlying POSIX pipe, see ``pipe(2)``, needs to be

re-created for each notification anyway).

The following diagram shows a possible state of the internal queue from

a high-level view. Conditions are shown as (waiting) threads. Assuming

no modifications are made to the queue (e.g. more acquires or timeouts),

the lock would be acquired by the threads in this order (concurrent

acquires in parentheses): ``threadE1``, ``threadE2``, (``threadS1``,

``threadS2``, ``threadS3``), (``threadS4``, ``threadS5``), ``threadE3``,

``threadS6``, ``threadE4``, ``threadE5``.

::

  [

    (0, [exc/threadE1, exc/threadE2, shr/threadS1/threadS2/threadS3]),

    (2, [shr/threadS4/threadS5]),

    (10, [exc/threadE3]),

    (33, [shr/threadS6, exc/threadE4, exc/threadE5]),

  ]

import heapq

_DEFAULT_PRIORITY = 0

class _PipeConditionWithMode(PipeCondition):

  __slots__ = [

    "shared",

    ]

  def __init__(self, lock, shared):

    """Initializes this class.

    """

    self.shared = shared

    PipeCondition.__init__(self, lock)

  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.

    "__pending_by_prio",

    "__pending_shared",

  __condition_class = _PipeConditionWithMode

    self.__pending_by_prio = {}

    self.__pending_shared = {}

          # 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())))

      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)

  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

    return cond == self.__find_first_pending_queue()[0]

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

    @type priority: integer

    @param priority: Priority for acquiring lock

    # Remove empty entries from queue

    self.__find_first_pending_queue()

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

      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

      # TODO: Decrease timeout with spurious notifications

      if not wait_condition.has_waiting():

        prioqueue.remove(wait_condition)

        if wait_condition.shared:

          del self.__pending_shared[priority]

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

              test_notify=None):

    @type priority: integer

    @param priority: Priority for acquiring lock

      return self.__acquire_unlocked(shared, timeout, priority)

      prioqueue = self.__find_first_pending_queue()

      if prioqueue:

        prioqueue[0].notifyAll()

  def delete(self, timeout=None, priority=_DEFAULT_PRIORITY):

    @type priority: integer

    @param priority: Priority for acquiring lock

        acquired = self.__acquire_unlocked(0, timeout, priority)

        for (_, prioqueue) in self.__pending:

          for cond in prioqueue:

            cond.notifyAll()

        assert self.__deleted

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

              test_notify=None):

    @type priority: integer

    @param priority: Priority for acquiring locks

        return self.__acquire_inner(names, False, shared, priority,

        # everybody else can use the instances at the same time. If we are

        if not self.__lock.acquire(shared=shared, priority=priority,

          return self.__acquire_inner(self.__names(), True, shared, priority,

  def __acquire_inner(self, names, want_all, shared, priority,

                      timeout_fn, test_notify):

    @param priority: Priority for acquiring locks

                                     priority=priority,

          # No need for priority or timeout here as this lock has just been

          # created

import itertools

from ganeti import compat

  def testPriority(self):

    # Acquire in exclusive mode

    self.assert_(self.sl.acquire(shared=0))

    # Queue acquires

    def _Acquire(prev, next, shared, priority, result):

      prev.wait()

      self.sl.acquire(shared=shared, priority=priority, test_notify=next.set)

      try:

        self.done.put(result)

      finally:

        self.sl.release()

    counter = itertools.count(0)

    priorities = range(-20, 30)

    first = threading.Event()

    prev = first

    # Data structure:

    # {

    #   priority:

    #     [(shared/exclusive, set(acquire names), set(pending threads)),

    #      (shared/exclusive, ...),

    #      ...,

    #     ],

    # }

    perprio = {}

    # References shared acquire per priority in L{perprio}. Data structure:

    # {

    #   priority: (shared=1, set(acquire names), set(pending threads)),

    # }

    prioshared = {}

    for seed in [4979, 9523, 14902, 32440]:

      # Use a deterministic random generator

      rnd = random.Random(seed)

      for priority in [rnd.choice(priorities) for _ in range(30)]:

        modes = [0, 1]

        rnd.shuffle(modes)

        for shared in modes:

          # Unique name

          acqname = "%s/shr=%s/prio=%s" % (counter.next(), shared, priority)

          ev = threading.Event()

          thread = self._addThread(target=_Acquire,

                                   args=(prev, ev, shared, priority, acqname))

          prev = ev

          # Record expected aqcuire, see above for structure

          data = (shared, set([acqname]), set([thread]))

          priolist = perprio.setdefault(priority, [])

          if shared:

            priosh = prioshared.get(priority, None)

            if priosh:

              # Shared acquires are merged

              for i, j in zip(priosh[1:], data[1:]):

                i.update(j)

              assert data[0] == priosh[0]

            else:

              prioshared[priority] = data

              priolist.append(data)

          else:

            priolist.append(data)

    # Start all acquires and wait for them

    first.set()

    prev.wait()

    # Check lock information

    self.assertEqual(self.sl.GetInfo(["name"]), [self.sl.name])

    self.assertEqual(self.sl.GetInfo(["mode", "owner"]),

                     ["exclusive", [threading.currentThread().getName()]])

    self.assertEqual(self.sl.GetInfo(["name", "pending"]),

                     [self.sl.name,

                      [(["exclusive", "shared"][int(bool(shared))],

                        sorted([t.getName() for t in threads]))

                       for acquires in [perprio[i]

                                        for i in sorted(perprio.keys())]

                       for (shared, _, threads) in acquires]])

    # Let threads acquire the lock

    self.sl.release()

    # Wait for everything to finish

    self._waitThreads()

    self.assert_(self.sl._check_empty())

    # Check acquires by priority

    for acquires in [perprio[i] for i in sorted(perprio.keys())]:

      for (_, names, _) in acquires:

        # For shared acquires, the set will contain 1..n entries. For exclusive

        # acquires only one.

        while names:

          names.remove(self.done.get_nowait())

      self.assertFalse(compat.any(names for (_, names, _) in acquires))

    self.assertRaises(Queue.Empty, self.done.get_nowait)

  def testPriority(self):

    def _Acquire(prev, next, name, priority, success_fn):

      prev.wait()

      self.assert_(self.ls.acquire(name, shared=0,

                                   priority=priority,

                                   test_notify=lambda _: next.set()))

      try:

        success_fn()

      finally:

        self.ls.release()

    # Get all in exclusive mode

    self.assert_(self.ls.acquire(locking.ALL_SET, shared=0))

    done_two = Queue.Queue(0)

    first = threading.Event()

    prev = first

    acquires = [("one", prio, self.done) for prio in range(1, 33)]

    acquires.extend([("two", prio, done_two) for prio in range(1, 33)])

    # Use a deterministic random generator

    random.Random(741).shuffle(acquires)

    for (name, prio, done) in acquires:

      ev = threading.Event()

      self._addThread(target=_Acquire,

                      args=(prev, ev, name, prio,

                            compat.partial(done.put, "Prio%s" % prio)))

      prev = ev

    # Start acquires

    first.set()

    # Wait for last acquire to start

    prev.wait()

    # Let threads acquire locks

    self.ls.release()

    # Wait for threads to finish

    self._waitThreads()

    for i in range(1, 33):

      self.assertEqual(self.done.get_nowait(), "Prio%s" % i)

      self.assertEqual(done_two.get_nowait(), "Prio%s" % i)

    self.assertRaises(Queue.Empty, self.done.get_nowait)

    self.assertRaises(Queue.Empty, done_two.get_nowait)