Synnefo » snf-ganeti » ganeti-local

#!/usr/bin/python

#

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

# 0.0510-1301, USA.

"""Script for unittesting the locking module"""

import os

import unittest

import time

import Queue

import threading

import random

import itertools

from ganeti import locking

from ganeti import errors

from ganeti import utils

from ganeti import compat

import testutils

# This is used to test the ssynchronize decorator.

# Since it's passed as input to a decorator it must be declared as a global.

_decoratorlock = locking.SharedLock("decorator lock")

#: List for looping tests

ITERATIONS = range(8)

def _Repeat(fn):

  """Decorator for executing a function many times"""

  def wrapper(*args, **kwargs):

    for i in ITERATIONS:

      fn(*args, **kwargs)

  return wrapper

def SafeSleep(duration):

  start = time.time()

  while True:

    delay = start + duration - time.time()

    if delay <= 0.0:

      break

    time.sleep(delay)

class _ThreadedTestCase(unittest.TestCase):

  """Test class that supports adding/waiting on threads"""

  def setUp(self):

    unittest.TestCase.setUp(self)

    self.done = Queue.Queue(0)

    self.threads = []

  def _addThread(self, *args, **kwargs):

    """Create and remember a new thread"""

    t = threading.Thread(*args, **kwargs)

    self.threads.append(t)

    t.start()

    return t

  def _waitThreads(self):

    """Wait for all our threads to finish"""

    for t in self.threads:

      t.join(60)

      self.failIf(t.isAlive())

    self.threads = []

class _ConditionTestCase(_ThreadedTestCase):

  """Common test case for conditions"""

  def setUp(self, cls):

    _ThreadedTestCase.setUp(self)

    self.lock = threading.Lock()

    self.cond = cls(self.lock)

  def _testAcquireRelease(self):

    self.assertFalse(self.cond._is_owned())

    self.assertRaises(RuntimeError, self.cond.wait)

    self.assertRaises(RuntimeError, self.cond.notifyAll)

    self.cond.acquire()

    self.assert_(self.cond._is_owned())

    self.cond.notifyAll()

    self.assert_(self.cond._is_owned())

    self.cond.release()

    self.assertFalse(self.cond._is_owned())

    self.assertRaises(RuntimeError, self.cond.wait)

    self.assertRaises(RuntimeError, self.cond.notifyAll)

  def _testNotification(self):

    def _NotifyAll():

      self.done.put("NE")

      self.cond.acquire()

      self.done.put("NA")

      self.cond.notifyAll()

      self.done.put("NN")

      self.cond.release()

    self.cond.acquire()

    self._addThread(target=_NotifyAll)

    self.assertEqual(self.done.get(True, 1), "NE")

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

    self.cond.wait()

    self.assertEqual(self.done.get(True, 1), "NA")

    self.assertEqual(self.done.get(True, 1), "NN")

    self.assert_(self.cond._is_owned())

    self.cond.release()

    self.assertFalse(self.cond._is_owned())

class TestSingleNotifyPipeCondition(_ConditionTestCase):

  """SingleNotifyPipeCondition tests"""

  def setUp(self):

    _ConditionTestCase.setUp(self, locking.SingleNotifyPipeCondition)

  def testAcquireRelease(self):

    self._testAcquireRelease()

  def testNotification(self):

    self._testNotification()

  def testWaitReuse(self):

    self.cond.acquire()

    self.cond.wait(0)

    self.cond.wait(0.1)

    self.cond.release()

  def testNoNotifyReuse(self):

    self.cond.acquire()

    self.cond.notifyAll()

    self.assertRaises(RuntimeError, self.cond.wait)

    self.assertRaises(RuntimeError, self.cond.notifyAll)

    self.cond.release()

class TestPipeCondition(_ConditionTestCase):

  """PipeCondition tests"""

  def setUp(self):

    _ConditionTestCase.setUp(self, locking.PipeCondition)

  def testAcquireRelease(self):

    self._testAcquireRelease()

  def testNotification(self):

    self._testNotification()

  def _TestWait(self, fn):

    threads = [

      self._addThread(target=fn),

      self._addThread(target=fn),

      self._addThread(target=fn),

      ]

    # Wait for threads to be waiting

    for _ in threads:

      self.assertEqual(self.done.get(True, 1), "A")

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

    self.cond.acquire()

    self.assertEqual(len(self.cond._waiters), 3)

    self.assertEqual(self.cond._waiters, set(threads))

    # This new thread can't acquire the lock, and thus call wait, before we

    # release it

    self._addThread(target=fn)

    self.cond.notifyAll()

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

    self.cond.release()

    # We should now get 3 W and 1 A (for the new thread) in whatever order

    w = 0

    a = 0

    for i in range(4):

      got = self.done.get(True, 1)

      if got == "W":

        w += 1

      elif got == "A":

        a += 1

      else:

        self.fail("Got %s on the done queue" % got)

    self.assertEqual(w, 3)

    self.assertEqual(a, 1)

    self.cond.acquire()

    self.cond.notifyAll()

    self.cond.release()

    self._waitThreads()

    self.assertEqual(self.done.get_nowait(), "W")

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

  def testBlockingWait(self):

    def _BlockingWait():

      self.cond.acquire()

      self.done.put("A")

      self.cond.wait()

      self.cond.release()

      self.done.put("W")

    self._TestWait(_BlockingWait)

  def testLongTimeoutWait(self):

    def _Helper():

      self.cond.acquire()

      self.done.put("A")

      self.cond.wait(15.0)

      self.cond.release()

      self.done.put("W")

    self._TestWait(_Helper)

  def _TimeoutWait(self, timeout, check):

    self.cond.acquire()

    self.cond.wait(timeout)

    self.cond.release()

    self.done.put(check)

  def testShortTimeoutWait(self):

    self._addThread(target=self._TimeoutWait, args=(0.1, "T1"))

    self._addThread(target=self._TimeoutWait, args=(0.1, "T1"))

    self._waitThreads()

    self.assertEqual(self.done.get_nowait(), "T1")

    self.assertEqual(self.done.get_nowait(), "T1")

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

  def testZeroTimeoutWait(self):

    self._addThread(target=self._TimeoutWait, args=(0, "T0"))

    self._addThread(target=self._TimeoutWait, args=(0, "T0"))

    self._addThread(target=self._TimeoutWait, args=(0, "T0"))

    self._waitThreads()

    self.assertEqual(self.done.get_nowait(), "T0")

    self.assertEqual(self.done.get_nowait(), "T0")

    self.assertEqual(self.done.get_nowait(), "T0")

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

class TestSharedLock(_ThreadedTestCase):

  """SharedLock tests"""

  def setUp(self):

    _ThreadedTestCase.setUp(self)

    self.sl = locking.SharedLock("TestSharedLock")

  def testSequenceAndOwnership(self):

    self.assertFalse(self.sl._is_owned())

    self.sl.acquire(shared=1)

    self.assert_(self.sl._is_owned())

    self.assert_(self.sl._is_owned(shared=1))

    self.assertFalse(self.sl._is_owned(shared=0))

    self.sl.release()

    self.assertFalse(self.sl._is_owned())

    self.sl.acquire()

    self.assert_(self.sl._is_owned())

    self.assertFalse(self.sl._is_owned(shared=1))

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

    self.sl.release()

    self.assertFalse(self.sl._is_owned())

    self.sl.acquire(shared=1)

    self.assert_(self.sl._is_owned())

    self.assert_(self.sl._is_owned(shared=1))

    self.assertFalse(self.sl._is_owned(shared=0))

    self.sl.release()

    self.assertFalse(self.sl._is_owned())

  def testBooleanValue(self):

    # semaphores are supposed to return a true value on a successful acquire

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

    self.sl.release()

    self.assert_(self.sl.acquire())

    self.sl.release()

  def testDoubleLockingStoE(self):

    self.sl.acquire(shared=1)

    self.assertRaises(AssertionError, self.sl.acquire)

  def testDoubleLockingEtoS(self):

    self.sl.acquire()

    self.assertRaises(AssertionError, self.sl.acquire, shared=1)

  def testDoubleLockingStoS(self):

    self.sl.acquire(shared=1)

    self.assertRaises(AssertionError, self.sl.acquire, shared=1)

  def testDoubleLockingEtoE(self):

    self.sl.acquire()

    self.assertRaises(AssertionError, self.sl.acquire)

  # helper functions: called in a separate thread they acquire the lock, send

  # their identifier on the done queue, then release it.

  def _doItSharer(self):

    try:

      self.sl.acquire(shared=1)

      self.done.put('SHR')

      self.sl.release()

    except errors.LockError:

      self.done.put('ERR')

  def _doItExclusive(self):

    try:

      self.sl.acquire()

      self.done.put('EXC')

      self.sl.release()

    except errors.LockError:

      self.done.put('ERR')

  def _doItDelete(self):

    try:

      self.sl.delete()

      self.done.put('DEL')

    except errors.LockError:

      self.done.put('ERR')

  def testSharersCanCoexist(self):

    self.sl.acquire(shared=1)

    threading.Thread(target=self._doItSharer).start()

    self.assert_(self.done.get(True, 1))

    self.sl.release()

  @_Repeat

  def testExclusiveBlocksExclusive(self):

    self.sl.acquire()

    self._addThread(target=self._doItExclusive)

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

    self.sl.release()

    self._waitThreads()

    self.failUnlessEqual(self.done.get_nowait(), 'EXC')

  @_Repeat

  def testExclusiveBlocksDelete(self):

    self.sl.acquire()

    self._addThread(target=self._doItDelete)

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

    self.sl.release()

    self._waitThreads()

    self.failUnlessEqual(self.done.get_nowait(), 'DEL')

    self.sl = locking.SharedLock(self.sl.name)

  @_Repeat

  def testExclusiveBlocksSharer(self):

    self.sl.acquire()

    self._addThread(target=self._doItSharer)

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

    self.sl.release()

    self._waitThreads()

    self.failUnlessEqual(self.done.get_nowait(), 'SHR')

  @_Repeat

  def testSharerBlocksExclusive(self):

    self.sl.acquire(shared=1)

    self._addThread(target=self._doItExclusive)

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

    self.sl.release()

    self._waitThreads()

    self.failUnlessEqual(self.done.get_nowait(), 'EXC')

  @_Repeat

  def testSharerBlocksDelete(self):

    self.sl.acquire(shared=1)

    self._addThread(target=self._doItDelete)

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

    self.sl.release()

    self._waitThreads()

    self.failUnlessEqual(self.done.get_nowait(), 'DEL')

    self.sl = locking.SharedLock(self.sl.name)

  @_Repeat

  def testWaitingExclusiveBlocksSharer(self):

    """SKIPPED testWaitingExclusiveBlockSharer"""

    return

    self.sl.acquire(shared=1)

    # the lock is acquired in shared mode...

    self._addThread(target=self._doItExclusive)

    # ...but now an exclusive is waiting...

    self._addThread(target=self._doItSharer)

    # ...so the sharer should be blocked as well

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

    self.sl.release()

    self._waitThreads()

    # The exclusive passed before

    self.failUnlessEqual(self.done.get_nowait(), 'EXC')

    self.failUnlessEqual(self.done.get_nowait(), 'SHR')

  @_Repeat

  def testWaitingSharerBlocksExclusive(self):

    """SKIPPED testWaitingSharerBlocksExclusive"""

    return

    self.sl.acquire()

    # the lock is acquired in exclusive mode...

    self._addThread(target=self._doItSharer)

    # ...but now a sharer is waiting...

    self._addThread(target=self._doItExclusive)

    # ...the exclusive is waiting too...

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

    self.sl.release()

    self._waitThreads()

    # The sharer passed before

    self.assertEqual(self.done.get_nowait(), 'SHR')

    self.assertEqual(self.done.get_nowait(), 'EXC')

  def testDelete(self):

    self.sl.delete()

    self.assertRaises(errors.LockError, self.sl.acquire)

    self.assertRaises(errors.LockError, self.sl.acquire, shared=1)

    self.assertRaises(errors.LockError, self.sl.delete)

  def testDeleteTimeout(self):

    self.sl.delete(timeout=60)

  def testNoDeleteIfSharer(self):

    self.sl.acquire(shared=1)

    self.assertRaises(AssertionError, self.sl.delete)

  @_Repeat

  def testDeletePendingSharersExclusiveDelete(self):

    self.sl.acquire()

    self._addThread(target=self._doItSharer)

    self._addThread(target=self._doItSharer)

    self._addThread(target=self._doItExclusive)

    self._addThread(target=self._doItDelete)

    self.sl.delete()

    self._waitThreads()

    # The threads who were pending return ERR

    for _ in range(4):

      self.assertEqual(self.done.get_nowait(), 'ERR')

    self.sl = locking.SharedLock(self.sl.name)

  @_Repeat

  def testDeletePendingDeleteExclusiveSharers(self):

    self.sl.acquire()

    self._addThread(target=self._doItDelete)

    self._addThread(target=self._doItExclusive)

    self._addThread(target=self._doItSharer)

    self._addThread(target=self._doItSharer)

    self.sl.delete()

    self._waitThreads()

    # The two threads who were pending return both ERR

    self.assertEqual(self.done.get_nowait(), 'ERR')

    self.assertEqual(self.done.get_nowait(), 'ERR')

    self.assertEqual(self.done.get_nowait(), 'ERR')

    self.assertEqual(self.done.get_nowait(), 'ERR')

    self.sl = locking.SharedLock(self.sl.name)

  @_Repeat

  def testExclusiveAcquireTimeout(self):

    for shared in [0, 1]:

      on_queue = threading.Event()

      release_exclusive = threading.Event()

      def _LockExclusive():

        self.sl.acquire(shared=0, test_notify=on_queue.set)

        self.done.put("A: start wait")

        release_exclusive.wait()

        self.done.put("A: end wait")

        self.sl.release()

      # Start thread to hold lock in exclusive mode

      self._addThread(target=_LockExclusive)

      # Wait for wait to begin

      self.assertEqual(self.done.get(timeout=60), "A: start wait")

      # Wait up to 60s to get lock, but release exclusive lock as soon as we're

      # on the queue

      self.failUnless(self.sl.acquire(shared=shared, timeout=60,

                                      test_notify=release_exclusive.set))

      self.done.put("got 2nd")

      self.sl.release()

      self._waitThreads()

      self.assertEqual(self.done.get_nowait(), "A: end wait")

      self.assertEqual(self.done.get_nowait(), "got 2nd")

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

  @_Repeat

  def testAcquireExpiringTimeout(self):

    def _AcquireWithTimeout(shared, timeout):

      if not self.sl.acquire(shared=shared, timeout=timeout):

        self.done.put("timeout")

    for shared in [0, 1]:

      # Lock exclusively

      self.sl.acquire()

      # Start shared acquires with timeout between 0 and 20 ms

      for i in range(11):

        self._addThread(target=_AcquireWithTimeout,

                        args=(shared, i * 2.0 / 1000.0))

      # Wait for threads to finish (makes sure the acquire timeout expires

      # before releasing the lock)

      self._waitThreads()

      # Release lock

      self.sl.release()

      for _ in range(11):

        self.assertEqual(self.done.get_nowait(), "timeout")

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

  @_Repeat

  def testSharedSkipExclusiveAcquires(self):

    # Tests whether shared acquires jump in front of exclusive acquires in the

    # queue.

    def _Acquire(shared, name, notify_ev, wait_ev):

      if notify_ev:

        notify_fn = notify_ev.set

      else:

        notify_fn = None

      if wait_ev:

        wait_ev.wait()

      if not self.sl.acquire(shared=shared, test_notify=notify_fn):

        return

      self.done.put(name)

      self.sl.release()

    # Get exclusive lock while we fill the queue

    self.sl.acquire()

    shrcnt1 = 5

    shrcnt2 = 7

    shrcnt3 = 9

    shrcnt4 = 2

    # Add acquires using threading.Event for synchronization. They'll be

    # acquired exactly in the order defined in this list.

    acquires = (shrcnt1 * [(1, "shared 1")] +

                3 * [(0, "exclusive 1")] +

                shrcnt2 * [(1, "shared 2")] +

                shrcnt3 * [(1, "shared 3")] +

                shrcnt4 * [(1, "shared 4")] +

                3 * [(0, "exclusive 2")])

    ev_cur = None

    ev_prev = None

    for args in acquires:

      ev_cur = threading.Event()

      self._addThread(target=_Acquire, args=args + (ev_cur, ev_prev))

      ev_prev = ev_cur

    # Wait for last acquire to start

    ev_prev.wait()

    # Expect 6 pending exclusive acquires and 1 for all shared acquires

    # together

    self.assertEqual(self.sl._count_pending(), 7)

    # Release exclusive lock and wait

    self.sl.release()

    self._waitThreads()

    # Check sequence

    for _ in range(shrcnt1 + shrcnt2 + shrcnt3 + shrcnt4):

      # Shared locks aren't guaranteed to be notified in order, but they'll be

      # first

      tmp = self.done.get_nowait()

      if tmp == "shared 1":

        shrcnt1 -= 1

      elif tmp == "shared 2":

        shrcnt2 -= 1

      elif tmp == "shared 3":

        shrcnt3 -= 1

      elif tmp == "shared 4":

        shrcnt4 -= 1

    self.assertEqual(shrcnt1, 0)

    self.assertEqual(shrcnt2, 0)

    self.assertEqual(shrcnt3, 0)

    self.assertEqual(shrcnt3, 0)

    for _ in range(3):

      self.assertEqual(self.done.get_nowait(), "exclusive 1")

    for _ in range(3):

      self.assertEqual(self.done.get_nowait(), "exclusive 2")

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

  @_Repeat

  def testMixedAcquireTimeout(self):

    sync = threading.Event()

    def _AcquireShared(ev):

      if not self.sl.acquire(shared=1, timeout=None):

        return

      self.done.put("shared")

      # Notify main thread

      ev.set()

      # Wait for notification from main thread

      sync.wait()

      # Release lock

      self.sl.release()

    acquires = []

    for _ in range(3):

      ev = threading.Event()

      self._addThread(target=_AcquireShared, args=(ev, ))

      acquires.append(ev)

    # Wait for all acquires to finish

    for i in acquires:

      i.wait()

    self.assertEqual(self.sl._count_pending(), 0)

    # Try to get exclusive lock

    self.failIf(self.sl.acquire(shared=0, timeout=0.02))

    # Acquire exclusive without timeout

    exclsync = threading.Event()

    exclev = threading.Event()

    def _AcquireExclusive():

      if not self.sl.acquire(shared=0):

        return

      self.done.put("exclusive")

      # Notify main thread

      exclev.set()

      # Wait for notification from main thread

      exclsync.wait()

      self.sl.release()

    self._addThread(target=_AcquireExclusive)

    # Try to get exclusive lock

    self.failIf(self.sl.acquire(shared=0, timeout=0.02))

    # Make all shared holders release their locks

    sync.set()

    # Wait for exclusive acquire to succeed

    exclev.wait()

    self.assertEqual(self.sl._count_pending(), 0)

    # Try to get exclusive lock

    self.failIf(self.sl.acquire(shared=0, timeout=0.02))

    def _AcquireSharedSimple():

      if self.sl.acquire(shared=1, timeout=None):

        self.done.put("shared2")

        self.sl.release()

    for _ in range(10):

      self._addThread(target=_AcquireSharedSimple)

    # Tell exclusive lock to release

    exclsync.set()

    # Wait for everything to finish

    self._waitThreads()

    self.assertEqual(self.sl._count_pending(), 0)

    # Check sequence

    for _ in range(3):

      self.assertEqual(self.done.get_nowait(), "shared")

    self.assertEqual(self.done.get_nowait(), "exclusive")

    for _ in range(10):

      self.assertEqual(self.done.get_nowait(), "shared2")

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

  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: