Synnefo » snf-ganeti

#!/usr/bin/python

#

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

from ganeti import locking

from ganeti import errors

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

#: 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

class _ThreadedTestCase(unittest.TestCase):

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

  def setUp(self):

    unittest.TestCase.setUp(self)

    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 TestPipeCondition(_ThreadedTestCase):

  """_PipeCondition tests"""

  def setUp(self):

    _ThreadedTestCase.setUp(self)

    self.lock = threading.Lock()

    self.cond = locking._PipeCondition(self.lock)

    self.done = Queue.Queue(0)

  def testAcquireRelease(self):

    self.assert_(not 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.assert_(not self.cond._is_owned())

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

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

  def testNotification(self):

    def _NotifyAll():

      self.cond.acquire()

      self.cond.notifyAll()

      self.cond.release()

    self.cond.acquire()

    self._addThread(target=_NotifyAll)

    self.cond.wait()

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

    self.cond.release()

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

  def _TestWait(self, fn):

    self._addThread(target=fn)

    self._addThread(target=fn)

    self._addThread(target=fn)

    # Wait for threads to be waiting

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

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

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

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

    self.cond.acquire()

    self.assertEqual(self.cond._nwaiters, 3)

    # 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 TestSingleActionPipeCondition(unittest.TestCase):

  """_SingleActionPipeCondition tests"""

  def setUp(self):

    self.cond = locking._SingleActionPipeCondition()

  def testInitialization(self):

    self.assert_(self.cond._read_fd is not None)

    self.assert_(self.cond._write_fd is not None)

    self.assert_(self.cond._poller is not None)

    self.assertEqual(self.cond._nwaiters, 0)

  def testUsageCount(self):

    self.cond.StartWaiting()

    self.assert_(self.cond._read_fd is not None)

    self.assert_(self.cond._write_fd is not None)

    self.assert_(self.cond._poller is not None)

    self.assertEqual(self.cond._nwaiters, 1)

    # use again

    self.cond.StartWaiting()

    self.assertEqual(self.cond._nwaiters, 2)

    # there is more than one user

    self.assert_(not self.cond.DoneWaiting())

    self.assert_(self.cond._read_fd is not None)

    self.assert_(self.cond._write_fd is not None)

    self.assert_(self.cond._poller is not None)

    self.assertEqual(self.cond._nwaiters, 1)

    self.assert_(self.cond.DoneWaiting())

    self.assertEqual(self.cond._nwaiters, 0)

    self.assert_(self.cond._read_fd is None)

    self.assert_(self.cond._write_fd is None)

    self.assert_(self.cond._poller is None)

  def testNotify(self):

    wait1 = self.cond.StartWaiting()

    wait2 = self.cond.StartWaiting()

    self.assert_(self.cond._read_fd is not None)

    self.assert_(self.cond._write_fd is not None)

    self.assert_(self.cond._poller is not None)

    self.cond.notifyAll()

    self.assert_(self.cond._read_fd is not None)

    self.assert_(self.cond._write_fd is None)

    self.assert_(self.cond._poller is not None)

    self.assert_(not self.cond.DoneWaiting())

    self.assert_(self.cond._read_fd is not None)

    self.assert_(self.cond._write_fd is None)

    self.assert_(self.cond._poller is not None)

    self.assert_(self.cond.DoneWaiting())

    self.assert_(self.cond._read_fd is None)

    self.assert_(self.cond._write_fd is None)

    self.assert_(self.cond._poller is None)

  def testReusage(self):

    self.cond.StartWaiting()

    self.assert_(self.cond._read_fd is not None)

    self.assert_(self.cond._write_fd is not None)

    self.assert_(self.cond._poller is not None)

    self.assert_(self.cond.DoneWaiting())

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

    self.assert_(self.cond._read_fd is None)

    self.assert_(self.cond._write_fd is None)

    self.assert_(self.cond._poller is None)

  def testNotifyTwice(self):

    self.cond.notifyAll()

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

class TestSharedLock(_ThreadedTestCase):

  """SharedLock tests"""

  def setUp(self):

    _ThreadedTestCase.setUp(self)

    self.sl = locking.SharedLock()

    # helper threads use the 'done' queue to tell the master they finished.

    self.done = Queue.Queue(0)

  def testSequenceAndOwnership(self):

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

    self.sl.acquire(shared=1)

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

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

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

    self.sl.release()

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

    self.sl.acquire()

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

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

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

    self.sl.release()

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

    self.sl.acquire(shared=1)

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

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

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

    self.sl.release()

    self.assert_(not 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()

  @_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()

  @_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 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()

  @_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()

  @_Repeat

  def testExclusiveAcquireTimeout(self):

    def _LockExclusive(wait):

      self.sl.acquire(shared=0)

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

      time.sleep(wait)

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

      self.sl.release()

    for shared in [0, 1]:

      # Start thread to hold lock for 20 ms

      self._addThread(target=_LockExclusive, args=(20.0 / 1000.0, ))

      # Wait for sleep to begin

      self.assertEqual(self.done.get(), "A: start sleep")

      # Wait up to 100 ms to get lock

      self.failUnless(self.sl.acquire(shared=shared, timeout=0.1))

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

      self.sl.release()

      self._waitThreads()

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

      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 xrange(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 xrange(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.

    # Get exclusive lock while we fill the queue

    self.sl.acquire()

    def _Acquire(shared, name):

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

        return

      self.done.put(name)

      self.sl.release()

    # Start shared acquires

    for _ in xrange(5):

      self._addThread(target=_Acquire, args=(1, "shared A"))

    # Start exclusive acquires

    for _ in xrange(3):

      self._addThread(target=_Acquire, args=(0, "exclusive B"))

    # More shared acquires

    for _ in xrange(5):

      self._addThread(target=_Acquire, args=(1, "shared C"))

    # More exclusive acquires

    for _ in xrange(3):

      self._addThread(target=_Acquire, args=(0, "exclusive D"))

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

    # together. There's no way to wait for SharedLock.acquire to start

    # its work. Hence the timeout of 2 seconds.

    pending = 0

    end_time = time.time() + 2.0

    while time.time() < end_time:

      pending = self.sl._count_pending()

      self.assert_(pending >= 0 and pending <= 7)

      if pending == 7:

        break

      time.sleep(0.05)

    self.assertEqual(pending, 7)

    # Release exclusive lock and wait

    self.sl.release()

    self._waitThreads()

    # Check sequence

    shr_a = 0

    shr_c = 0

    for _ in xrange(10):

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

      # first

      tmp = self.done.get_nowait()

      if tmp == "shared A":

        shr_a += 1

      elif tmp == "shared C":

        shr_c += 1

    self.assertEqual(shr_a, 5)

    self.assertEqual(shr_c, 5)

    for _ in xrange(3):

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

    for _ in xrange(3):

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

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

  @_Repeat

  def testMixedAcquireTimeout(self):

    sync = threading.Condition()

    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

      sync.acquire()

      try:

        sync.wait()

      finally:

        sync.release()

      # Release lock

      self.sl.release()

    acquires = []

    for _ in xrange(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.Condition()

    exclev = threading.Event()

    def _AcquireExclusive():

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

        return

      self.done.put("exclusive")

      # Notify main thread

      exclev.set()

      exclsync.acquire()

      try:

        exclsync.wait()

      finally:

        exclsync.release()

      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.acquire()

    try:

      sync.notifyAll()

    finally:

      sync.release()

    # 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 xrange(10):

      self._addThread(target=_AcquireSharedSimple)

    # Tell exclusive lock to release

    exclsync.acquire()

    try:

      exclsync.notifyAll()

    finally:

      exclsync.release()

    # Wait for everything to finish

    self._waitThreads()

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

    # Check sequence

    for _ in xrange(3):

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

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

    for _ in xrange(10):

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

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

class TestSSynchronizedDecorator(_ThreadedTestCase):

  """Shared Lock Synchronized decorator test"""

  def setUp(self):

    _ThreadedTestCase.setUp(self)

    # helper threads use the 'done' queue to tell the master they finished.

    self.done = Queue.Queue(0)

  @locking.ssynchronized(_decoratorlock)

  def _doItExclusive(self):

    self.assert_(_decoratorlock._is_owned())

    self.done.put('EXC')

  @locking.ssynchronized(_decoratorlock, shared=1)

  def _doItSharer(self):

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

    self.done.put('SHR')

  def testDecoratedFunctions(self):

    self._doItExclusive()

    self.assert_(not _decoratorlock._is_owned())

    self._doItSharer()

    self.assert_(not _decoratorlock._is_owned())

  def testSharersCanCoexist(self):

    _decoratorlock.acquire(shared=1)

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

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

    _decoratorlock.release()

  @_Repeat

  def testExclusiveBlocksExclusive(self):

    _decoratorlock.acquire()

    self._addThread(target=self._doItExclusive)

    # give it a bit of time to check that it's not actually doing anything

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

    _decoratorlock.release()

    self._waitThreads()

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

  @_Repeat

  def testExclusiveBlocksSharer(self):

    _decoratorlock.acquire()

    self._addThread(target=self._doItSharer)

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

    _decoratorlock.release()

    self._waitThreads()

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

  @_Repeat

  def testSharerBlocksExclusive(self):

    _decoratorlock.acquire(shared=1)

    self._addThread(target=self._doItExclusive)

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

    _decoratorlock.release()

    self._waitThreads()

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

class TestLockSet(_ThreadedTestCase):

  """LockSet tests"""

  def setUp(self):