4 # Copyright (C) 2006, 2007 Google Inc.
6 # This program is free software; you can redistribute it and/or modify
7 # it under the terms of the GNU General Public License as published by
8 # the Free Software Foundation; either version 2 of the License, or
9 # (at your option) any later version.
11 # This program is distributed in the hope that it will be useful, but
12 # WITHOUT ANY WARRANTY; without even the implied warranty of
13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
14 # General Public License for more details.
16 # You should have received a copy of the GNU General Public License
17 # along with this program; if not, write to the Free Software
18 # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
22 """Script for unittesting the locking module"""
31 from ganeti import locking
32 from ganeti import errors
35 # This is used to test the ssynchronize decorator.
36 # Since it's passed as input to a decorator it must be declared as a global.
37 _decoratorlock = locking.SharedLock()
39 #: List for looping tests
44 """Decorator for executing a function many times"""
45 def wrapper(*args, **kwargs):
51 def SafeSleep(duration):
54 delay = start + duration - time.time()
60 class _ThreadedTestCase(unittest.TestCase):
61 """Test class that supports adding/waiting on threads"""
63 unittest.TestCase.setUp(self)
64 self.done = Queue.Queue(0)
67 def _addThread(self, *args, **kwargs):
68 """Create and remember a new thread"""
69 t = threading.Thread(*args, **kwargs)
70 self.threads.append(t)
74 def _waitThreads(self):
75 """Wait for all our threads to finish"""
76 for t in self.threads:
78 self.failIf(t.isAlive())
82 class _ConditionTestCase(_ThreadedTestCase):
83 """Common test case for conditions"""
86 _ThreadedTestCase.setUp(self)
87 self.lock = threading.Lock()
88 self.cond = cls(self.lock)
90 def _testAcquireRelease(self):
91 self.assert_(not self.cond._is_owned())
92 self.assertRaises(RuntimeError, self.cond.wait)
93 self.assertRaises(RuntimeError, self.cond.notifyAll)
96 self.assert_(self.cond._is_owned())
98 self.assert_(self.cond._is_owned())
101 self.assert_(not self.cond._is_owned())
102 self.assertRaises(RuntimeError, self.cond.wait)
103 self.assertRaises(RuntimeError, self.cond.notifyAll)
105 def _testNotification(self):
110 self.cond.notifyAll()
115 self._addThread(target=_NotifyAll)
116 self.assertEqual(self.done.get(True, 1), "NE")
117 self.assertRaises(Queue.Empty, self.done.get_nowait)
119 self.assertEqual(self.done.get(True, 1), "NA")
120 self.assertEqual(self.done.get(True, 1), "NN")
121 self.assert_(self.cond._is_owned())
123 self.assert_(not self.cond._is_owned())
126 class TestSingleNotifyPipeCondition(_ConditionTestCase):
127 """SingleNotifyPipeCondition tests"""
130 _ConditionTestCase.setUp(self, locking.SingleNotifyPipeCondition)
132 def testAcquireRelease(self):
133 self._testAcquireRelease()
135 def testNotification(self):
136 self._testNotification()
138 def testWaitReuse(self):
144 def testNoNotifyReuse(self):
146 self.cond.notifyAll()
147 self.assertRaises(RuntimeError, self.cond.wait)
148 self.assertRaises(RuntimeError, self.cond.notifyAll)
152 class TestPipeCondition(_ConditionTestCase):
153 """PipeCondition tests"""
156 _ConditionTestCase.setUp(self, locking.PipeCondition)
158 def testAcquireRelease(self):
159 self._testAcquireRelease()
161 def testNotification(self):
162 self._testNotification()
164 def _TestWait(self, fn):
165 self._addThread(target=fn)
166 self._addThread(target=fn)
167 self._addThread(target=fn)
169 # Wait for threads to be waiting
170 self.assertEqual(self.done.get(True, 1), "A")
171 self.assertEqual(self.done.get(True, 1), "A")
172 self.assertEqual(self.done.get(True, 1), "A")
174 self.assertRaises(Queue.Empty, self.done.get_nowait)
177 self.assertEqual(self.cond._nwaiters, 3)
178 # This new thread can"t acquire the lock, and thus call wait, before we
180 self._addThread(target=fn)
181 self.cond.notifyAll()
182 self.assertRaises(Queue.Empty, self.done.get_nowait)
185 # We should now get 3 W and 1 A (for the new thread) in whatever order
189 got = self.done.get(True, 1)
195 self.fail("Got %s on the done queue" % got)
197 self.assertEqual(w, 3)
198 self.assertEqual(a, 1)
201 self.cond.notifyAll()
204 self.assertEqual(self.done.get_nowait(), "W")
205 self.assertRaises(Queue.Empty, self.done.get_nowait)
207 def testBlockingWait(self):
215 self._TestWait(_BlockingWait)
217 def testLongTimeoutWait(self):
225 self._TestWait(_Helper)
227 def _TimeoutWait(self, timeout, check):
229 self.cond.wait(timeout)
233 def testShortTimeoutWait(self):
234 self._addThread(target=self._TimeoutWait, args=(0.1, "T1"))
235 self._addThread(target=self._TimeoutWait, args=(0.1, "T1"))
237 self.assertEqual(self.done.get_nowait(), "T1")
238 self.assertEqual(self.done.get_nowait(), "T1")
239 self.assertRaises(Queue.Empty, self.done.get_nowait)
241 def testZeroTimeoutWait(self):
242 self._addThread(target=self._TimeoutWait, args=(0, "T0"))
243 self._addThread(target=self._TimeoutWait, args=(0, "T0"))
244 self._addThread(target=self._TimeoutWait, args=(0, "T0"))
246 self.assertEqual(self.done.get_nowait(), "T0")
247 self.assertEqual(self.done.get_nowait(), "T0")
248 self.assertEqual(self.done.get_nowait(), "T0")
249 self.assertRaises(Queue.Empty, self.done.get_nowait)
252 class TestSharedLock(_ThreadedTestCase):
253 """SharedLock tests"""
256 _ThreadedTestCase.setUp(self)
257 self.sl = locking.SharedLock()
259 def testSequenceAndOwnership(self):
260 self.assert_(not self.sl._is_owned())
261 self.sl.acquire(shared=1)
262 self.assert_(self.sl._is_owned())
263 self.assert_(self.sl._is_owned(shared=1))
264 self.assert_(not self.sl._is_owned(shared=0))
266 self.assert_(not self.sl._is_owned())
268 self.assert_(self.sl._is_owned())
269 self.assert_(not self.sl._is_owned(shared=1))
270 self.assert_(self.sl._is_owned(shared=0))
272 self.assert_(not self.sl._is_owned())
273 self.sl.acquire(shared=1)
274 self.assert_(self.sl._is_owned())
275 self.assert_(self.sl._is_owned(shared=1))
276 self.assert_(not self.sl._is_owned(shared=0))
278 self.assert_(not self.sl._is_owned())
280 def testBooleanValue(self):
281 # semaphores are supposed to return a true value on a successful acquire
282 self.assert_(self.sl.acquire(shared=1))
284 self.assert_(self.sl.acquire())
287 def testDoubleLockingStoE(self):
288 self.sl.acquire(shared=1)
289 self.assertRaises(AssertionError, self.sl.acquire)
291 def testDoubleLockingEtoS(self):
293 self.assertRaises(AssertionError, self.sl.acquire, shared=1)
295 def testDoubleLockingStoS(self):
296 self.sl.acquire(shared=1)
297 self.assertRaises(AssertionError, self.sl.acquire, shared=1)
299 def testDoubleLockingEtoE(self):
301 self.assertRaises(AssertionError, self.sl.acquire)
303 # helper functions: called in a separate thread they acquire the lock, send
304 # their identifier on the done queue, then release it.
305 def _doItSharer(self):
307 self.sl.acquire(shared=1)
310 except errors.LockError:
313 def _doItExclusive(self):
318 except errors.LockError:
321 def _doItDelete(self):
325 except errors.LockError:
328 def testSharersCanCoexist(self):
329 self.sl.acquire(shared=1)
330 threading.Thread(target=self._doItSharer).start()
331 self.assert_(self.done.get(True, 1))
335 def testExclusiveBlocksExclusive(self):
337 self._addThread(target=self._doItExclusive)
338 self.assertRaises(Queue.Empty, self.done.get_nowait)
341 self.failUnlessEqual(self.done.get_nowait(), 'EXC')
344 def testExclusiveBlocksDelete(self):
346 self._addThread(target=self._doItDelete)
347 self.assertRaises(Queue.Empty, self.done.get_nowait)
350 self.failUnlessEqual(self.done.get_nowait(), 'DEL')
351 self.sl = locking.SharedLock()
354 def testExclusiveBlocksSharer(self):
356 self._addThread(target=self._doItSharer)
357 self.assertRaises(Queue.Empty, self.done.get_nowait)
360 self.failUnlessEqual(self.done.get_nowait(), 'SHR')
363 def testSharerBlocksExclusive(self):
364 self.sl.acquire(shared=1)
365 self._addThread(target=self._doItExclusive)
366 self.assertRaises(Queue.Empty, self.done.get_nowait)
369 self.failUnlessEqual(self.done.get_nowait(), 'EXC')
372 def testSharerBlocksDelete(self):
373 self.sl.acquire(shared=1)
374 self._addThread(target=self._doItDelete)
375 self.assertRaises(Queue.Empty, self.done.get_nowait)
378 self.failUnlessEqual(self.done.get_nowait(), 'DEL')
379 self.sl = locking.SharedLock()
382 def testWaitingExclusiveBlocksSharer(self):
383 """SKIPPED testWaitingExclusiveBlockSharer"""
386 self.sl.acquire(shared=1)
387 # the lock is acquired in shared mode...
388 self._addThread(target=self._doItExclusive)
389 # ...but now an exclusive is waiting...
390 self._addThread(target=self._doItSharer)
391 # ...so the sharer should be blocked as well
392 self.assertRaises(Queue.Empty, self.done.get_nowait)
395 # The exclusive passed before
396 self.failUnlessEqual(self.done.get_nowait(), 'EXC')
397 self.failUnlessEqual(self.done.get_nowait(), 'SHR')
400 def testWaitingSharerBlocksExclusive(self):
401 """SKIPPED testWaitingSharerBlocksExclusive"""
405 # the lock is acquired in exclusive mode...
406 self._addThread(target=self._doItSharer)
407 # ...but now a sharer is waiting...
408 self._addThread(target=self._doItExclusive)
409 # ...the exclusive is waiting too...
410 self.assertRaises(Queue.Empty, self.done.get_nowait)
413 # The sharer passed before
414 self.assertEqual(self.done.get_nowait(), 'SHR')
415 self.assertEqual(self.done.get_nowait(), 'EXC')
417 def testDelete(self):
419 self.assertRaises(errors.LockError, self.sl.acquire)
420 self.assertRaises(errors.LockError, self.sl.acquire, shared=1)
421 self.assertRaises(errors.LockError, self.sl.delete)
423 def testDeleteTimeout(self):
424 self.sl.delete(timeout=60)
426 def testNoDeleteIfSharer(self):
427 self.sl.acquire(shared=1)
428 self.assertRaises(AssertionError, self.sl.delete)
431 def testDeletePendingSharersExclusiveDelete(self):
433 self._addThread(target=self._doItSharer)
434 self._addThread(target=self._doItSharer)
435 self._addThread(target=self._doItExclusive)
436 self._addThread(target=self._doItDelete)
439 # The threads who were pending return ERR
441 self.assertEqual(self.done.get_nowait(), 'ERR')
442 self.sl = locking.SharedLock()
445 def testDeletePendingDeleteExclusiveSharers(self):
447 self._addThread(target=self._doItDelete)
448 self._addThread(target=self._doItExclusive)
449 self._addThread(target=self._doItSharer)
450 self._addThread(target=self._doItSharer)
453 # The two threads who were pending return both ERR
454 self.assertEqual(self.done.get_nowait(), 'ERR')
455 self.assertEqual(self.done.get_nowait(), 'ERR')
456 self.assertEqual(self.done.get_nowait(), 'ERR')
457 self.assertEqual(self.done.get_nowait(), 'ERR')
458 self.sl = locking.SharedLock()
461 def testExclusiveAcquireTimeout(self):
462 for shared in [0, 1]:
463 on_queue = threading.Event()
464 release_exclusive = threading.Event()
466 def _LockExclusive():
467 self.sl.acquire(shared=0, test_notify=on_queue.set)
468 self.done.put("A: start wait")
469 release_exclusive.wait()
470 self.done.put("A: end wait")
473 # Start thread to hold lock in exclusive mode
474 self._addThread(target=_LockExclusive)
476 # Wait for wait to begin
477 self.assertEqual(self.done.get(timeout=60), "A: start wait")
479 # Wait up to 60s to get lock, but release exclusive lock as soon as we're
481 self.failUnless(self.sl.acquire(shared=shared, timeout=60,
482 test_notify=release_exclusive.set))
484 self.done.put("got 2nd")
489 self.assertEqual(self.done.get_nowait(), "A: end wait")
490 self.assertEqual(self.done.get_nowait(), "got 2nd")
491 self.assertRaises(Queue.Empty, self.done.get_nowait)
494 def testAcquireExpiringTimeout(self):
495 def _AcquireWithTimeout(shared, timeout):
496 if not self.sl.acquire(shared=shared, timeout=timeout):
497 self.done.put("timeout")
499 for shared in [0, 1]:
503 # Start shared acquires with timeout between 0 and 20 ms
505 self._addThread(target=_AcquireWithTimeout,
506 args=(shared, i * 2.0 / 1000.0))
508 # Wait for threads to finish (makes sure the acquire timeout expires
509 # before releasing the lock)
516 self.assertEqual(self.done.get_nowait(), "timeout")
518 self.assertRaises(Queue.Empty, self.done.get_nowait)
521 def testSharedSkipExclusiveAcquires(self):
522 # Tests whether shared acquires jump in front of exclusive acquires in the
525 def _Acquire(shared, name, notify_ev, wait_ev):
527 notify_fn = notify_ev.set
534 if not self.sl.acquire(shared=shared, test_notify=notify_fn):
540 # Get exclusive lock while we fill the queue
548 # Add acquires using threading.Event for synchronization. They'll be
549 # acquired exactly in the order defined in this list.
550 acquires = (shrcnt1 * [(1, "shared 1")] +
551 3 * [(0, "exclusive 1")] +
552 shrcnt2 * [(1, "shared 2")] +
553 shrcnt3 * [(1, "shared 3")] +
554 shrcnt4 * [(1, "shared 4")] +
555 3 * [(0, "exclusive 2")])
560 for args in acquires:
561 ev_cur = threading.Event()
562 self._addThread(target=_Acquire, args=args + (ev_cur, ev_prev))
565 # Wait for last acquire to start
568 # Expect 6 pending exclusive acquires and 1 for all shared acquires
570 self.assertEqual(self.sl._count_pending(), 7)
572 # Release exclusive lock and wait
578 for _ in range(shrcnt1 + shrcnt2 + shrcnt3 + shrcnt4):
579 # Shared locks aren't guaranteed to be notified in order, but they'll be
581 tmp = self.done.get_nowait()
582 if tmp == "shared 1":
584 elif tmp == "shared 2":
586 elif tmp == "shared 3":
588 elif tmp == "shared 4":
590 self.assertEqual(shrcnt1, 0)
591 self.assertEqual(shrcnt2, 0)
592 self.assertEqual(shrcnt3, 0)
593 self.assertEqual(shrcnt3, 0)
596 self.assertEqual(self.done.get_nowait(), "exclusive 1")
599 self.assertEqual(self.done.get_nowait(), "exclusive 2")
601 self.assertRaises(Queue.Empty, self.done.get_nowait)
604 def testMixedAcquireTimeout(self):
605 sync = threading.Condition()
607 def _AcquireShared(ev):
608 if not self.sl.acquire(shared=1, timeout=None):
611 self.done.put("shared")
616 # Wait for notification
628 ev = threading.Event()
629 self._addThread(target=_AcquireShared, args=(ev, ))
632 # Wait for all acquires to finish
636 self.assertEqual(self.sl._count_pending(), 0)
638 # Try to get exclusive lock
639 self.failIf(self.sl.acquire(shared=0, timeout=0.02))
641 # Acquire exclusive without timeout
642 exclsync = threading.Condition()
643 exclev = threading.Event()
645 def _AcquireExclusive():
646 if not self.sl.acquire(shared=0):
649 self.done.put("exclusive")
662 self._addThread(target=_AcquireExclusive)
664 # Try to get exclusive lock
665 self.failIf(self.sl.acquire(shared=0, timeout=0.02))
667 # Make all shared holders release their locks
674 # Wait for exclusive acquire to succeed
677 self.assertEqual(self.sl._count_pending(), 0)
679 # Try to get exclusive lock
680 self.failIf(self.sl.acquire(shared=0, timeout=0.02))
682 def _AcquireSharedSimple():
683 if self.sl.acquire(shared=1, timeout=None):
684 self.done.put("shared2")
688 self._addThread(target=_AcquireSharedSimple)
690 # Tell exclusive lock to release
697 # Wait for everything to finish
700 self.assertEqual(self.sl._count_pending(), 0)
704 self.assertEqual(self.done.get_nowait(), "shared")
706 self.assertEqual(self.done.get_nowait(), "exclusive")
709 self.assertEqual(self.done.get_nowait(), "shared2")
711 self.assertRaises(Queue.Empty, self.done.get_nowait)
714 class TestSSynchronizedDecorator(_ThreadedTestCase):
715 """Shared Lock Synchronized decorator test"""
718 _ThreadedTestCase.setUp(self)
720 @locking.ssynchronized(_decoratorlock)
721 def _doItExclusive(self):
722 self.assert_(_decoratorlock._is_owned())
725 @locking.ssynchronized(_decoratorlock, shared=1)
726 def _doItSharer(self):
727 self.assert_(_decoratorlock._is_owned(shared=1))
730 def testDecoratedFunctions(self):
731 self._doItExclusive()
732 self.assert_(not _decoratorlock._is_owned())
734 self.assert_(not _decoratorlock._is_owned())
736 def testSharersCanCoexist(self):
737 _decoratorlock.acquire(shared=1)
738 threading.Thread(target=self._doItSharer).start()
739 self.assert_(self.done.get(True, 1))
740 _decoratorlock.release()
743 def testExclusiveBlocksExclusive(self):
744 _decoratorlock.acquire()
745 self._addThread(target=self._doItExclusive)
746 # give it a bit of time to check that it's not actually doing anything
747 self.assertRaises(Queue.Empty, self.done.get_nowait)
748 _decoratorlock.release()
750 self.failUnlessEqual(self.done.get_nowait(), 'EXC')
753 def testExclusiveBlocksSharer(self):
754 _decoratorlock.acquire()
755 self._addThread(target=self._doItSharer)
756 self.assertRaises(Queue.Empty, self.done.get_nowait)
757 _decoratorlock.release()
759 self.failUnlessEqual(self.done.get_nowait(), 'SHR')
762 def testSharerBlocksExclusive(self):
763 _decoratorlock.acquire(shared=1)
764 self._addThread(target=self._doItExclusive)
765 self.assertRaises(Queue.Empty, self.done.get_nowait)
766 _decoratorlock.release()
768 self.failUnlessEqual(self.done.get_nowait(), 'EXC')
771 class TestLockSet(_ThreadedTestCase):
775 _ThreadedTestCase.setUp(self)
779 """Helper to (re)initialize the lock set"""
780 self.resources = ['one', 'two', 'three']
781 self.ls = locking.LockSet(members=self.resources)
783 def testResources(self):
784 self.assertEquals(self.ls._names(), set(self.resources))
785 newls = locking.LockSet()
786 self.assertEquals(newls._names(), set())
788 def testAcquireRelease(self):
789 self.assert_(self.ls.acquire('one'))
790 self.assertEquals(self.ls._list_owned(), set(['one']))
792 self.assertEquals(self.ls._list_owned(), set())
793 self.assertEquals(self.ls.acquire(['one']), set(['one']))
794 self.assertEquals(self.ls._list_owned(), set(['one']))
796 self.assertEquals(self.ls._list_owned(), set())
797 self.ls.acquire(['one', 'two', 'three'])
798 self.assertEquals(self.ls._list_owned(), set(['one', 'two', 'three']))
799 self.ls.release('one')
800 self.assertEquals(self.ls._list_owned(), set(['two', 'three']))
801 self.ls.release(['three'])
802 self.assertEquals(self.ls._list_owned(), set(['two']))
804 self.assertEquals(self.ls._list_owned(), set())
805 self.assertEquals(self.ls.acquire(['one', 'three']), set(['one', 'three']))
806 self.assertEquals(self.ls._list_owned(), set(['one', 'three']))
808 self.assertEquals(self.ls._list_owned(), set())
810 def testNoDoubleAcquire(self):
811 self.ls.acquire('one')
812 self.assertRaises(AssertionError, self.ls.acquire, 'one')
813 self.assertRaises(AssertionError, self.ls.acquire, ['two'])
814 self.assertRaises(AssertionError, self.ls.acquire, ['two', 'three'])
816 self.ls.acquire(['one', 'three'])
817 self.ls.release('one')
818 self.assertRaises(AssertionError, self.ls.acquire, ['two'])
819 self.ls.release('three')
821 def testNoWrongRelease(self):
822 self.assertRaises(AssertionError, self.ls.release)
823 self.ls.acquire('one')
824 self.assertRaises(AssertionError, self.ls.release, 'two')
826 def testAddRemove(self):
828 self.assertEquals(self.ls._list_owned(), set())
829 self.assert_('four' in self.ls._names())
830 self.ls.add(['five', 'six', 'seven'], acquired=1)
831 self.assert_('five' in self.ls._names())
832 self.assert_('six' in self.ls._names())
833 self.assert_('seven' in self.ls._names())
834 self.assertEquals(self.ls._list_owned(), set(['five', 'six', 'seven']))
835 self.assertEquals(self.ls.remove(['five', 'six']), ['five', 'six'])
836 self.assert_('five' not in self.ls._names())
837 self.assert_('six' not in self.ls._names())
838 self.assertEquals(self.ls._list_owned(), set(['seven']))
839 self.assertRaises(AssertionError, self.ls.add, 'eight', acquired=1)
840 self.ls.remove('seven')
841 self.assert_('seven' not in self.ls._names())
842 self.assertEquals(self.ls._list_owned(), set([]))
843 self.ls.acquire(None, shared=1)
844 self.assertRaises(AssertionError, self.ls.add, 'eight')
846 self.ls.acquire(None)
847 self.ls.add('eight', acquired=1)
848 self.assert_('eight' in self.ls._names())
849 self.assert_('eight' in self.ls._list_owned())
851 self.assert_('nine' in self.ls._names())
852 self.assert_('nine' not in self.ls._list_owned())
854 self.ls.remove(['two'])
855 self.assert_('two' not in self.ls._names())
856 self.ls.acquire('three')
857 self.assertEquals(self.ls.remove(['three']), ['three'])
858 self.assert_('three' not in self.ls._names())
859 self.assertEquals(self.ls.remove('three'), [])
860 self.assertEquals(self.ls.remove(['one', 'three', 'six']), ['one'])
861 self.assert_('one' not in self.ls._names())
863 def testRemoveNonBlocking(self):
864 self.ls.acquire('one')
865 self.assertEquals(self.ls.remove('one'), ['one'])
866 self.ls.acquire(['two', 'three'])
867 self.assertEquals(self.ls.remove(['two', 'three']),
870 def testNoDoubleAdd(self):
871 self.assertRaises(errors.LockError, self.ls.add, 'two')
873 self.assertRaises(errors.LockError, self.ls.add, 'four')
875 def testNoWrongRemoves(self):
876 self.ls.acquire(['one', 'three'], shared=1)
877 # Cannot remove 'two' while holding something which is not a superset
878 self.assertRaises(AssertionError, self.ls.remove, 'two')
879 # Cannot remove 'three' as we are sharing it
880 self.assertRaises(AssertionError, self.ls.remove, 'three')
882 def testAcquireSetLock(self):
883 # acquire the set-lock exclusively
884 self.assertEquals(self.ls.acquire(None), set(['one', 'two', 'three']))
885 self.assertEquals(self.ls._list_owned(), set(['one', 'two', 'three']))
886 self.assertEquals(self.ls._is_owned(), True)
887 self.assertEquals(self.ls._names(), set(['one', 'two', 'three']))
888 # I can still add/remove elements...
889 self.assertEquals(self.ls.remove(['two', 'three']), ['two', 'three'])
890 self.assert_(self.ls.add('six'))
893 self.assertEquals(self.ls.acquire(None, shared=1), set(['one', 'six']))
894 # adding new elements is not possible
895 self.assertRaises(AssertionError, self.ls.add, 'five')
898 def testAcquireWithRepetitions(self):
899 self.assertEquals(self.ls.acquire(['two', 'two', 'three'], shared=1),
900 set(['two', 'two', 'three']))
901 self.ls.release(['two', 'two'])
902 self.assertEquals(self.ls._list_owned(), set(['three']))
904 def testEmptyAcquire(self):
905 # Acquire an empty list of locks...
906 self.assertEquals(self.ls.acquire([]), set())
907 self.assertEquals(self.ls._list_owned(), set())
908 # New locks can still be addded
909 self.assert_(self.ls.add('six'))
910 # "re-acquiring" is not an issue, since we had really acquired nothing
911 self.assertEquals(self.ls.acquire([], shared=1), set())
912 self.assertEquals(self.ls._list_owned(), set())
913 # We haven't really acquired anything, so we cannot release
914 self.assertRaises(AssertionError, self.ls.release)
916 def _doLockSet(self, names, shared):
918 self.ls.acquire(names, shared=shared)
919 self.done.put('DONE')
921 except errors.LockError:
924 def _doAddSet(self, names):
926 self.ls.add(names, acquired=1)
927 self.done.put('DONE')
929 except errors.LockError:
932 def _doRemoveSet(self, names):
933 self.done.put(self.ls.remove(names))
936 def testConcurrentSharedAcquire(self):
937 self.ls.acquire(['one', 'two'], shared=1)
938 self._addThread(target=self._doLockSet, args=(['one', 'two'], 1))
940 self.assertEqual(self.done.get_nowait(), 'DONE')
941 self._addThread(target=self._doLockSet, args=(['one', 'two', 'three'], 1))
943 self.assertEqual(self.done.get_nowait(), 'DONE')
944 self._addThread(target=self._doLockSet, args=('three', 1))
946 self.assertEqual(self.done.get_nowait(), 'DONE')
947 self._addThread(target=self._doLockSet, args=(['one', 'two'], 0))
948 self._addThread(target=self._doLockSet, args=(['two', 'three'], 0))
949 self.assertRaises(Queue.Empty, self.done.get_nowait)
952 self.assertEqual(self.done.get_nowait(), 'DONE')
953 self.assertEqual(self.done.get_nowait(), 'DONE')
956 def testConcurrentExclusiveAcquire(self):
957 self.ls.acquire(['one', 'two'])
958 self._addThread(target=self._doLockSet, args=('three', 1))
960 self.assertEqual(self.done.get_nowait(), 'DONE')
961 self._addThread(target=self._doLockSet, args=('three', 0))
963 self.assertEqual(self.done.get_nowait(), 'DONE')
964 self.assertRaises(Queue.Empty, self.done.get_nowait)
965 self._addThread(target=self._doLockSet, args=(['one', 'two'], 0))
966 self._addThread(target=self._doLockSet, args=(['one', 'two'], 1))
967 self._addThread(target=self._doLockSet, args=('one', 0))
968 self._addThread(target=self._doLockSet, args=('one', 1))
969 self._addThread(target=self._doLockSet, args=(['two', 'three'], 0))
970 self._addThread(target=self._doLockSet, args=(['two', 'three'], 1))
971 self.assertRaises(Queue.Empty, self.done.get_nowait)
975 self.failUnlessEqual(self.done.get_nowait(), 'DONE')
978 def testSimpleAcquireTimeoutExpiring(self):
979 names = sorted(self.ls._names())
980 self.assert_(len(names) >= 3)
982 # Get name of first lock
985 # Get name of last lock
989 # Block first and try to lock it again
992 # Block last and try to lock all locks
995 # Block last and try to lock it again
999 for (wanted, block) in checks:
1000 # Lock in exclusive mode
1001 self.assert_(self.ls.acquire(block, shared=0))
1004 # Try to get the same lock again with a timeout (should never succeed)
1005 if self.ls.acquire(wanted, timeout=0.1, shared=0):
1006 self.done.put("acquired")
1009 self.assert_(not self.ls._list_owned())
1010 self.assert_(not self.ls._is_owned())
1011 self.done.put("not acquired")
1013 self._addThread(target=_AcquireOne)
1015 # Wait for timeout in thread to expire
1018 # Release exclusive lock again
1021 self.assertEqual(self.done.get_nowait(), "not acquired")
1022 self.assertRaises(Queue.Empty, self.done.get_nowait)
1025 def testDelayedAndExpiringLockAcquire(self):
1027 self.ls.add(['five', 'six', 'seven', 'eight', 'nine'])
1029 for expire in (False, True):
1030 names = sorted(self.ls._names())
1031 self.assertEqual(len(names), 8)
1033 lock_ev = dict([(i, threading.Event()) for i in names])
1035 # Lock all in exclusive mode
1036 self.assert_(self.ls.acquire(names, shared=0))
1039 # We'll wait at least 300ms per lock
1040 lockwait = len(names) * [0.3]
1042 # Fail if we can't acquire all locks in 400ms. There are 8 locks, so
1043 # this gives us up to 2.4s to fail.
1044 lockall_timeout = 0.4
1046 # This should finish rather quickly
1048 lockall_timeout = len(names) * 5.0
1051 def acquire_notification(name):
1053 self.done.put("getting %s" % name)
1058 if self.ls.acquire(names, shared=0, timeout=lockall_timeout,
1059 test_notify=acquire_notification):
1060 self.done.put("got all")
1063 self.done.put("timeout on all")
1066 for ev in lock_ev.values():
1069 t = self._addThread(target=_LockAll)
1071 for idx, name in enumerate(names):
1072 # Wait for actual acquire on this lock to start
1073 lock_ev[name].wait(10.0)
1075 if expire and t.isAlive():
1076 # Wait some time after getting the notification to make sure the lock
1077 # acquire will expire
1078 SafeSleep(lockwait[idx])
1080 self.ls.release(names=name)
1082 self.assert_(not self.ls._list_owned())
1087 # Not checking which locks were actually acquired. Doing so would be
1088 # too timing-dependant.
1089 self.assertEqual(self.done.get_nowait(), "timeout on all")
1092 self.assertEqual(self.done.get_nowait(), "getting %s" % i)
1093 self.assertEqual(self.done.get_nowait(), "got all")
1094 self.assertRaises(Queue.Empty, self.done.get_nowait)
1097 def testConcurrentRemove(self):
1099 self.ls.acquire(['one', 'two', 'four'])
1100 self._addThread(target=self._doLockSet, args=(['one', 'four'], 0))
1101 self._addThread(target=self._doLockSet, args=(['one', 'four'], 1))
1102 self._addThread(target=self._doLockSet, args=(['one', 'two'], 0))
1103 self._addThread(target=self._doLockSet, args=(['one', 'two'], 1))
1104 self.assertRaises(Queue.Empty, self.done.get_nowait)
1105 self.ls.remove('one')
1109 self.failUnlessEqual(self.done.get_nowait(), 'ERR')
1110 self.ls.add(['five', 'six'], acquired=1)
1111 self._addThread(target=self._doLockSet, args=(['three', 'six'], 1))
1112 self._addThread(target=self._doLockSet, args=(['three', 'six'], 0))
1113 self._addThread(target=self._doLockSet, args=(['four', 'six'], 1))
1114 self._addThread(target=self._doLockSet, args=(['four', 'six'], 0))
1115 self.ls.remove('five')
1119 self.failUnlessEqual(self.done.get_nowait(), 'DONE')
1120 self.ls.acquire(['three', 'four'])
1121 self._addThread(target=self._doRemoveSet, args=(['four', 'six'], ))
1122 self.assertRaises(Queue.Empty, self.done.get_nowait)
1123 self.ls.remove('four')
1125 self.assertEqual(self.done.get_nowait(), ['six'])
1126 self._addThread(target=self._doRemoveSet, args=(['two']))
1128 self.assertEqual(self.done.get_nowait(), ['two'])
1134 def testConcurrentSharedSetLock(self):
1135 # share the set-lock...
1136 self.ls.acquire(None, shared=1)
1137 # ...another thread can share it too
1138 self._addThread(target=self._doLockSet, args=(None, 1))
1140 self.assertEqual(self.done.get_nowait(), 'DONE')
1141 # ...or just share some elements
1142 self._addThread(target=self._doLockSet, args=(['one', 'three'], 1))
1144 self.assertEqual(self.done.get_nowait(), 'DONE')
1145 # ...but not add new ones or remove any
1146 t = self._addThread(target=self._doAddSet, args=(['nine']))
1147 self._addThread(target=self._doRemoveSet, args=(['two'], ))
1148 self.assertRaises(Queue.Empty, self.done.get_nowait)
1149 # this just releases the set-lock
1152 self.assertEqual(self.done.get_nowait(), 'DONE')
1153 # release the lock on the actual elements so remove() can proceed too
1156 self.failUnlessEqual(self.done.get_nowait(), ['two'])
1161 def testConcurrentExclusiveSetLock(self):
1162 # acquire the set-lock...
1163 self.ls.acquire(None, shared=0)
1164 # ...no one can do anything else
1165 self._addThread(target=self._doLockSet, args=(None, 1))
1166 self._addThread(target=self._doLockSet, args=(None, 0))
1167 self._addThread(target=self._doLockSet, args=(['three'], 0))
1168 self._addThread(target=self._doLockSet, args=(['two'], 1))
1169 self._addThread(target=self._doAddSet, args=(['nine']))
1170 self.assertRaises(Queue.Empty, self.done.get_nowait)
1174 self.assertEqual(self.done.get(True, 1), 'DONE')
1179 def testConcurrentSetLockAdd(self):
1180 self.ls.acquire('one')
1181 # Another thread wants the whole SetLock
1182 self._addThread(target=self._doLockSet, args=(None, 0))
1183 self._addThread(target=self._doLockSet, args=(None, 1))
1184 self.assertRaises(Queue.Empty, self.done.get_nowait)
1185 self.assertRaises(AssertionError, self.ls.add, 'four')
1188 self.assertEqual(self.done.get_nowait(), 'DONE')
1189 self.assertEqual(self.done.get_nowait(), 'DONE')
1190 self.ls.acquire(None)
1191 self._addThread(target=self._doLockSet, args=(None, 0))
1192 self._addThread(target=self._doLockSet, args=(None, 1))
1193 self.assertRaises(Queue.Empty, self.done.get_nowait)
1195 self.ls.add('five', acquired=1)
1196 self.ls.add('six', acquired=1, shared=1)
1197 self.assertEquals(self.ls._list_owned(),
1198 set(['one', 'two', 'three', 'five', 'six']))
1199 self.assertEquals(self.ls._is_owned(), True)
1200 self.assertEquals(self.ls._names(),
1201 set(['one', 'two', 'three', 'four', 'five', 'six']))
1204 self.assertEqual(self.done.get_nowait(), 'DONE')
1205 self.assertEqual(self.done.get_nowait(), 'DONE')
1209 def testEmptyLockSet(self):
1211 self.assertEqual(self.ls.acquire(None), set(['one', 'two', 'three']))
1213 self.ls.remove(['one', 'two', 'three'])
1214 # and adds/locks by another thread still wait
1215 self._addThread(target=self._doAddSet, args=(['nine']))
1216 self._addThread(target=self._doLockSet, args=(None, 1))
1217 self._addThread(target=self._doLockSet, args=(None, 0))
1218 self.assertRaises(Queue.Empty, self.done.get_nowait)
1222 self.assertEqual(self.done.get_nowait(), 'DONE')
1224 self.assertEqual(self.ls.remove(['nine']), ['nine'])
1226 self.assertEqual(self.ls.acquire(None, shared=1), set())
1227 # other sharers can go, adds still wait
1228 self._addThread(target=self._doLockSet, args=(None, 1))
1230 self.assertEqual(self.done.get_nowait(), 'DONE')
1231 self._addThread(target=self._doAddSet, args=(['nine']))
1232 self.assertRaises(Queue.Empty, self.done.get_nowait)
1235 self.assertEqual(self.done.get_nowait(), 'DONE')
1239 class TestGanetiLockManager(_ThreadedTestCase):
1242 _ThreadedTestCase.setUp(self)
1243 self.nodes=['n1', 'n2']
1244 self.instances=['i1', 'i2', 'i3']
1245 self.GL = locking.GanetiLockManager(nodes=self.nodes,
1246 instances=self.instances)
1249 # Don't try this at home...
1250 locking.GanetiLockManager._instance = None
1252 def testLockingConstants(self):
1253 # The locking library internally cheats by assuming its constants have some
1254 # relationships with each other. Check those hold true.
1255 # This relationship is also used in the Processor to recursively acquire
1256 # the right locks. Again, please don't break it.
1257 for i in range(len(locking.LEVELS)):
1258 self.assertEqual(i, locking.LEVELS[i])
1260 def testDoubleGLFails(self):
1261 self.assertRaises(AssertionError, locking.GanetiLockManager)
1263 def testLockNames(self):
1264 self.assertEqual(self.GL._names(locking.LEVEL_CLUSTER), set(['BGL']))
1265 self.assertEqual(self.GL._names(locking.LEVEL_NODE), set(self.nodes))
1266 self.assertEqual(self.GL._names(locking.LEVEL_INSTANCE),
1267 set(self.instances))
1269 def testInitAndResources(self):
1270 locking.GanetiLockManager._instance = None
1271 self.GL = locking.GanetiLockManager()
1272 self.assertEqual(self.GL._names(locking.LEVEL_CLUSTER), set(['BGL']))
1273 self.assertEqual(self.GL._names(locking.LEVEL_NODE), set())
1274 self.assertEqual(self.GL._names(locking.LEVEL_INSTANCE), set())
1276 locking.GanetiLockManager._instance = None
1277 self.GL = locking.GanetiLockManager(nodes=self.nodes)
1278 self.assertEqual(self.GL._names(locking.LEVEL_CLUSTER), set(['BGL']))
1279 self.assertEqual(self.GL._names(locking.LEVEL_NODE), set(self.nodes))
1280 self.assertEqual(self.GL._names(locking.LEVEL_INSTANCE), set())
1282 locking.GanetiLockManager._instance = None
1283 self.GL = locking.GanetiLockManager(instances=self.instances)
1284 self.assertEqual(self.GL._names(locking.LEVEL_CLUSTER), set(['BGL']))
1285 self.assertEqual(self.GL._names(locking.LEVEL_NODE), set())
1286 self.assertEqual(self.GL._names(locking.LEVEL_INSTANCE),
1287 set(self.instances))
1289 def testAcquireRelease(self):
1290 self.GL.acquire(locking.LEVEL_CLUSTER, ['BGL'], shared=1)
1291 self.assertEquals(self.GL._list_owned(locking.LEVEL_CLUSTER), set(['BGL']))
1292 self.GL.acquire(locking.LEVEL_INSTANCE, ['i1'])
1293 self.GL.acquire(locking.LEVEL_NODE, ['n1', 'n2'], shared=1)
1294 self.GL.release(locking.LEVEL_NODE, ['n2'])
1295 self.assertEquals(self.GL._list_owned(locking.LEVEL_NODE), set(['n1']))
1296 self.assertEquals(self.GL._list_owned(locking.LEVEL_INSTANCE), set(['i1']))
1297 self.GL.release(locking.LEVEL_NODE)
1298 self.assertEquals(self.GL._list_owned(locking.LEVEL_NODE), set())
1299 self.assertEquals(self.GL._list_owned(locking.LEVEL_INSTANCE), set(['i1']))
1300 self.GL.release(locking.LEVEL_INSTANCE)
1301 self.assertRaises(errors.LockError, self.GL.acquire,
1302 locking.LEVEL_INSTANCE, ['i5'])
1303 self.GL.acquire(locking.LEVEL_INSTANCE, ['i3'], shared=1)
1304 self.assertEquals(self.GL._list_owned(locking.LEVEL_INSTANCE), set(['i3']))
1306 def testAcquireWholeSets(self):
1307 self.GL.acquire(locking.LEVEL_CLUSTER, ['BGL'], shared=1)
1308 self.assertEquals(self.GL.acquire(locking.LEVEL_INSTANCE, None),
1309 set(self.instances))
1310 self.assertEquals(self.GL._list_owned(locking.LEVEL_INSTANCE),
1311 set(self.instances))
1312 self.assertEquals(self.GL.acquire(locking.LEVEL_NODE, None, shared=1),
1314 self.assertEquals(self.GL._list_owned(locking.LEVEL_NODE),
1316 self.GL.release(locking.LEVEL_NODE)
1317 self.GL.release(locking.LEVEL_INSTANCE)
1318 self.GL.release(locking.LEVEL_CLUSTER)
1320 def testAcquireWholeAndPartial(self):
1321 self.GL.acquire(locking.LEVEL_CLUSTER, ['BGL'], shared=1)
1322 self.assertEquals(self.GL.acquire(locking.LEVEL_INSTANCE, None),
1323 set(self.instances))
1324 self.assertEquals(self.GL._list_owned(locking.LEVEL_INSTANCE),
1325 set(self.instances))
1326 self.assertEquals(self.GL.acquire(locking.LEVEL_NODE, ['n2'], shared=1),
1328 self.assertEquals(self.GL._list_owned(locking.LEVEL_NODE),
1330 self.GL.release(locking.LEVEL_NODE)
1331 self.GL.release(locking.LEVEL_INSTANCE)
1332 self.GL.release(locking.LEVEL_CLUSTER)
1334 def testBGLDependency(self):
1335 self.assertRaises(AssertionError, self.GL.acquire,
1336 locking.LEVEL_NODE, ['n1', 'n2'])
1337 self.assertRaises(AssertionError, self.GL.acquire,
1338 locking.LEVEL_INSTANCE, ['i3'])
1339 self.GL.acquire(locking.LEVEL_CLUSTER, ['BGL'], shared=1)
1340 self.GL.acquire(locking.LEVEL_NODE, ['n1'])
1341 self.assertRaises(AssertionError, self.GL.release,
1342 locking.LEVEL_CLUSTER, ['BGL'])
1343 self.assertRaises(AssertionError, self.GL.release,
1344 locking.LEVEL_CLUSTER)
1345 self.GL.release(locking.LEVEL_NODE)
1346 self.GL.acquire(locking.LEVEL_INSTANCE, ['i1', 'i2'])
1347 self.assertRaises(AssertionError, self.GL.release,
1348 locking.LEVEL_CLUSTER, ['BGL'])
1349 self.assertRaises(AssertionError, self.GL.release,
1350 locking.LEVEL_CLUSTER)
1351 self.GL.release(locking.LEVEL_INSTANCE)
1353 def testWrongOrder(self):
1354 self.GL.acquire(locking.LEVEL_CLUSTER, ['BGL'], shared=1)
1355 self.GL.acquire(locking.LEVEL_NODE, ['n2'])
1356 self.assertRaises(AssertionError, self.GL.acquire,
1357 locking.LEVEL_NODE, ['n1'])
1358 self.assertRaises(AssertionError, self.GL.acquire,
1359 locking.LEVEL_INSTANCE, ['i2'])
1361 # Helper function to run as a thread that shared the BGL and then acquires
1362 # some locks at another level.
1363 def _doLock(self, level, names, shared):
1365 self.GL.acquire(locking.LEVEL_CLUSTER, ['BGL'], shared=1)
1366 self.GL.acquire(level, names, shared=shared)
1367 self.done.put('DONE')
1368 self.GL.release(level)
1369 self.GL.release(locking.LEVEL_CLUSTER)
1370 except errors.LockError:
1371 self.done.put('ERR')
1374 def testConcurrency(self):
1375 self.GL.acquire(locking.LEVEL_CLUSTER, ['BGL'], shared=1)
1376 self._addThread(target=self._doLock,
1377 args=(locking.LEVEL_INSTANCE, 'i1', 1))
1379 self.assertEqual(self.done.get_nowait(), 'DONE')
1380 self.GL.acquire(locking.LEVEL_INSTANCE, ['i3'])
1381 self._addThread(target=self._doLock,
1382 args=(locking.LEVEL_INSTANCE, 'i1', 1))
1384 self.assertEqual(self.done.get_nowait(), 'DONE')
1385 self._addThread(target=self._doLock,
1386 args=(locking.LEVEL_INSTANCE, 'i3', 1))
1387 self.assertRaises(Queue.Empty, self.done.get_nowait)
1388 self.GL.release(locking.LEVEL_INSTANCE)
1390 self.assertEqual(self.done.get_nowait(), 'DONE')
1391 self.GL.acquire(locking.LEVEL_INSTANCE, ['i2'], shared=1)
1392 self._addThread(target=self._doLock,
1393 args=(locking.LEVEL_INSTANCE, 'i2', 1))
1395 self.assertEqual(self.done.get_nowait(), 'DONE')
1396 self._addThread(target=self._doLock,
1397 args=(locking.LEVEL_INSTANCE, 'i2', 0))
1398 self.assertRaises(Queue.Empty, self.done.get_nowait)
1399 self.GL.release(locking.LEVEL_INSTANCE)
1401 self.assertEqual(self.done.get(True, 1), 'DONE')
1402 self.GL.release(locking.LEVEL_CLUSTER, ['BGL'])
1405 if __name__ == '__main__':
1407 #suite = unittest.TestLoader().loadTestsFromTestCase(TestSharedLock)
1408 #unittest.TextTestRunner(verbosity=2).run(suite)