#!/usr/bin/python
#

# Copyright (C) 2006, 2007, 2010 Google Inc.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are
# met:
#
# 1. Redistributions of source code must retain the above copyright notice,
# this list of conditions and the following disclaimer.
#
# 2. Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in the
# documentation and/or other materials provided with the distribution.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS
# IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
# TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR
# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
# EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
# PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
# PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
# LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
# NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
# SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.


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


import os
import unittest
import time
import Queue
import threading
import random
import gc
import itertools

from ganeti import constants
from ganeti import locking
from ganeti import errors
from ganeti import utils
from ganeti import compat
from ganeti import objects
from ganeti import query

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

    self.assertTrue(repr(self.cond).startswith("<"))
    self.assertTrue("waiters=" in repr(self.cond))

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

    self.assertTrue(repr(self.sl).startswith("<"))
    self.assertTrue("name=TestSharedLock" in repr(self.sl))

  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.assertTrue(self.sl.delete(timeout=60))

  def testDeleteTimeoutFail(self):
    ready = threading.Event()
    finish = threading.Event()

    def fn():
      self.sl.acquire(shared=0)
      ready.set()

      finish.wait()
      self.sl.release()

    self._addThread(target=fn)
    ready.wait()

    # Test if deleting a lock owned in exclusive mode by another thread fails
    # to delete when a timeout is used
    self.assertFalse(self.sl.delete(timeout=0.02))

    finish.set()
    self._waitThreads()

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

  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)

  def testIllegalDowngrade(self):
    # Not yet acquired
    self.assertRaises(AssertionError, self.sl.downgrade)

    # Acquire in shared mode, downgrade should be no-op
    self.assertTrue(self.sl.acquire(shared=1))
    self.assertTrue(self.sl.is_owned(shared=1))
    self.assertTrue(self.sl.downgrade())
    self.assertTrue(self.sl.is_owned(shared=1))
    self.sl.release()

  def testDowngrade(self):
    self.assertTrue(self.sl.acquire())
    self.assertTrue(self.sl.is_owned(shared=0))
    self.assertTrue(self.sl.downgrade())
    self.assertTrue(self.sl.is_owned(shared=1))
    self.sl.release()

  @_Repeat
  def testDowngradeJumpsAheadOfExclusive(self):
    def _KeepExclusive(ev_got, ev_downgrade, ev_release):
      self.assertTrue(self.sl.acquire())
      self.assertTrue(self.sl.is_owned(shared=0))
      ev_got.set()
      ev_downgrade.wait()
      self.assertTrue(self.sl.is_owned(shared=0))
      self.assertTrue(self.sl.downgrade())
      self.assertTrue(self.sl.is_owned(shared=1))
      ev_release.wait()
      self.assertTrue(self.sl.is_owned(shared=1))
      self.sl.release()

    def _KeepExclusive2(ev_started, ev_release):
      self.assertTrue(self.sl.acquire(test_notify=ev_started.set))
      self.assertTrue(self.sl.is_owned(shared=0))
      ev_release.wait()
      self.assertTrue(self.sl.is_owned(shared=0))
      self.sl.release()

    def _KeepShared(ev_started, ev_got, ev_release):
      self.assertTrue(self.sl.acquire(shared=1, test_notify=ev_started.set))
      self.assertTrue(self.sl.is_owned(shared=1))
      ev_got.set()
      ev_release.wait()
      self.assertTrue(self.sl.is_owned(shared=1))
      self.sl.release()

    # Acquire lock in exclusive mode
    ev_got_excl1 = threading.Event()
    ev_downgrade_excl1 = threading.Event()
    ev_release_excl1 = threading.Event()
    th_excl1 = self._addThread(target=_KeepExclusive,
                               args=(ev_got_excl1, ev_downgrade_excl1,
                                     ev_release_excl1))
    ev_got_excl1.wait()

    # Start a second exclusive acquire
    ev_started_excl2 = threading.Event()
    ev_release_excl2 = threading.Event()
    th_excl2 = self._addThread(target=_KeepExclusive2,
                               args=(ev_started_excl2, ev_release_excl2))
    ev_started_excl2.wait()

    # Start shared acquires, will jump ahead of second exclusive acquire when
    # first exclusive acquire downgrades
    ev_shared = [(threading.Event(), threading.Event()) for _ in range(5)]
    ev_release_shared = threading.Event()

    th_shared = [self._addThread(target=_KeepShared,
                                 args=(ev_started, ev_got, ev_release_shared))
                 for (ev_started, ev_got) in ev_shared]

    # Wait for all shared acquires to start
    for (ev, _) in ev_shared:
      ev.wait()

    # Check lock information
    self.assertEqual(self.sl.GetLockInfo(set([query.LQ_MODE, query.LQ_OWNER])),
                     [(self.sl.name, "exclusive", [th_excl1.getName()], None)])
    [(_, _, _, pending), ] = self.sl.GetLockInfo(set([query.LQ_PENDING]))
    self.assertEqual([(pendmode, sorted(waiting))
                      for (pendmode, waiting) in pending],
                     [("exclusive", [th_excl2.getName()]),
                      ("shared", sorted(th.getName() for th in th_shared))])

    # Shared acquires won't start until the exclusive lock is downgraded
    ev_downgrade_excl1.set()

    # Wait for all shared acquires to be successful
    for (_, ev) in ev_shared:
      ev.wait()

    # Check lock information again
    self.assertEqual(self.sl.GetLockInfo(set([query.LQ_MODE,
                                              query.LQ_PENDING])),
                     [(self.sl.name, "shared", None,
                       [("exclusive", [th_excl2.getName()])])])
    [(_, _, owner, _), ] = self.sl.GetLockInfo(set([query.LQ_OWNER]))
    self.assertEqual(set(owner), set([th_excl1.getName()] +
                                     [th.getName() for th in th_shared]))

    ev_release_excl1.set()
    ev_release_excl2.set()
    ev_release_shared.set()

    self._waitThreads()

    self.assertEqual(self.sl.GetLockInfo(set([query.LQ_MODE, query.LQ_OWNER,
                                              query.LQ_PENDING])),
                     [(self.sl.name, None, None, [])])

  @_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.GetLockInfo(set()),
                     [(self.sl.name, None, None, None)])
    self.assertEqual(self.sl.GetLockInfo(set([query.LQ_MODE, query.LQ_OWNER])),
                     [(self.sl.name, "exclusive",
                       [threading.currentThread().getName()], None)])

    self._VerifyPrioPending(self.sl.GetLockInfo(set([query.LQ_PENDING])),
                            perprio)

    # 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 _VerifyPrioPending(self, ((name, mode, owner, pending), ), perprio):
    self.assertEqual(name, self.sl.name)
    self.assert_(mode is None)
    self.assert_(owner is None)

    self.assertEqual([(pendmode, sorted(waiting))
                      for (pendmode, waiting) in pending],
                     [(["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])

  class _FakeTimeForSpuriousNotifications:
    def __init__(self, now, check_end):
      self.now = now
      self.check_end = check_end

      # Deterministic random number generator
      self.rnd = random.Random(15086)

    def time(self):
      # Advance time if the random number generator thinks so (this is to test
      # multiple notifications without advancing the time)
      if self.rnd.random() < 0.3:
        self.now += self.rnd.random()

      self.check_end(self.now)

      return self.now

  @_Repeat
  def testAcquireTimeoutWithSpuriousNotifications(self):
    ready = threading.Event()
    locked = threading.Event()
    req = Queue.Queue(0)

    epoch = 4000.0
    timeout = 60.0

    def check_end(now):
      self.assertFalse(locked.isSet())

      # If we waited long enough (in virtual time), tell main thread to release
      # lock, otherwise tell it to notify once more
      req.put(now < (epoch + (timeout * 0.8)))

    time_fn = self._FakeTimeForSpuriousNotifications(epoch, check_end).time

    sl = locking.SharedLock("test", _time_fn=time_fn)

    # Acquire in exclusive mode
    sl.acquire(shared=0)

    def fn():
      self.assertTrue(sl.acquire(shared=0, timeout=timeout,
                                 test_notify=ready.set))
      locked.set()
      sl.release()
      self.done.put("success")

    # Start acquire with timeout and wait for it to be ready
    self._addThread(target=fn)
    ready.wait()

    # The separate thread is now waiting to acquire the lock, so start sending
    # spurious notifications.

    # Wait for separate thread to ask for another notification
    count = 0
    while req.get():
      # After sending the notification, the lock will take a short amount of
      # time to notice and to retrieve the current time
      sl._notify_topmost()
      count += 1

    self.assertTrue(count > 100, "Not enough notifications were sent")

    self.assertFalse(locked.isSet())

    # Some notifications have been sent, now actually release the lock
    sl.release()

    # Wait for lock to be acquired
    locked.wait()

    self._waitThreads()

    self.assertEqual(self.done.get_nowait(), "success")
    self.assertRaises(Queue.Empty, self.done.get_nowait)


class TestSharedLockInCondition(_ThreadedTestCase):
  """SharedLock as a condition lock tests"""

  def setUp(self):
    _ThreadedTestCase.setUp(self)
    self.sl = locking.SharedLock("TestSharedLockInCondition")
    self.setCondition()

  def setCondition(self):
    self.cond = threading.Condition(self.sl)

  def testKeepMode(self):
    self.cond.acquire(shared=1)
    self.assert_(self.sl.is_owned(shared=1))
    self.cond.wait(0)
    self.assert_(self.sl.is_owned(shared=1))
    self.cond.release()
    self.cond.acquire(shared=0)
    self.assert_(self.sl.is_owned(shared=0))
    self.cond.wait(0)
    self.assert_(self.sl.is_owned(shared=0))
    self.cond.release()


class TestSharedLockInPipeCondition(TestSharedLockInCondition):
  """SharedLock as a pipe condition lock tests"""

  def setCondition(self):
    self.cond = locking.PipeCondition(self.sl)


class TestSSynchronizedDecorator(_ThreadedTestCase):
  """Shared Lock Synchronized decorator test"""

  def setUp(self):
    _ThreadedTestCase.setUp(self)

  @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.assertFalse(_decoratorlock.is_owned())
    self._doItSharer()
    self.assertFalse(_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")


if __name__ == "__main__":
  testutils.GanetiTestProgram()