4 # Copyright (C) 2006, 2007, 2010 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"""
34 from ganeti import constants
35 from ganeti import locking
36 from ganeti import errors
37 from ganeti import utils
38 from ganeti import compat
39 from ganeti import objects
40 from ganeti import query
45 # This is used to test the ssynchronize decorator.
46 # Since it's passed as input to a decorator it must be declared as a global.
47 _decoratorlock = locking.SharedLock("decorator lock")
49 #: List for looping tests
54 """Decorator for executing a function many times"""
55 def wrapper(*args, **kwargs):
61 def SafeSleep(duration):
64 delay = start + duration - time.time()
70 class _ThreadedTestCase(unittest.TestCase):
71 """Test class that supports adding/waiting on threads"""
73 unittest.TestCase.setUp(self)
74 self.done = Queue.Queue(0)
77 def _addThread(self, *args, **kwargs):
78 """Create and remember a new thread"""
79 t = threading.Thread(*args, **kwargs)
80 self.threads.append(t)
84 def _waitThreads(self):
85 """Wait for all our threads to finish"""
86 for t in self.threads:
88 self.failIf(t.isAlive())
92 class _ConditionTestCase(_ThreadedTestCase):
93 """Common test case for conditions"""
96 _ThreadedTestCase.setUp(self)
97 self.lock = threading.Lock()
98 self.cond = cls(self.lock)
100 def _testAcquireRelease(self):
101 self.assertFalse(self.cond._is_owned())
102 self.assertRaises(RuntimeError, self.cond.wait, None)
103 self.assertRaises(RuntimeError, self.cond.notifyAll)
106 self.assert_(self.cond._is_owned())
107 self.cond.notifyAll()
108 self.assert_(self.cond._is_owned())
111 self.assertFalse(self.cond._is_owned())
112 self.assertRaises(RuntimeError, self.cond.wait, None)
113 self.assertRaises(RuntimeError, self.cond.notifyAll)
115 def _testNotification(self):
120 self.cond.notifyAll()
125 self._addThread(target=_NotifyAll)
126 self.assertEqual(self.done.get(True, 1), "NE")
127 self.assertRaises(Queue.Empty, self.done.get_nowait)
129 self.assertEqual(self.done.get(True, 1), "NA")
130 self.assertEqual(self.done.get(True, 1), "NN")
131 self.assert_(self.cond._is_owned())
133 self.assertFalse(self.cond._is_owned())
136 class TestSingleNotifyPipeCondition(_ConditionTestCase):
137 """SingleNotifyPipeCondition tests"""
140 _ConditionTestCase.setUp(self, locking.SingleNotifyPipeCondition)
142 def testAcquireRelease(self):
143 self._testAcquireRelease()
145 def testNotification(self):
146 self._testNotification()
148 def testWaitReuse(self):
154 def testNoNotifyReuse(self):
156 self.cond.notifyAll()
157 self.assertRaises(RuntimeError, self.cond.wait, None)
158 self.assertRaises(RuntimeError, self.cond.notifyAll)
162 class TestPipeCondition(_ConditionTestCase):
163 """PipeCondition tests"""
166 _ConditionTestCase.setUp(self, locking.PipeCondition)
168 def testAcquireRelease(self):
169 self._testAcquireRelease()
171 def testNotification(self):
172 self._testNotification()
174 def _TestWait(self, fn):
176 self._addThread(target=fn),
177 self._addThread(target=fn),
178 self._addThread(target=fn),
181 # Wait for threads to be waiting
183 self.assertEqual(self.done.get(True, 1), "A")
185 self.assertRaises(Queue.Empty, self.done.get_nowait)
188 self.assertEqual(len(self.cond._waiters), 3)
189 self.assertEqual(self.cond._waiters, set(threads))
191 self.assertTrue(repr(self.cond).startswith("<"))
192 self.assertTrue("waiters=" in repr(self.cond))
194 # This new thread can't acquire the lock, and thus call wait, before we
196 self._addThread(target=fn)
197 self.cond.notifyAll()
198 self.assertRaises(Queue.Empty, self.done.get_nowait)
201 # We should now get 3 W and 1 A (for the new thread) in whatever order
205 got = self.done.get(True, 1)
211 self.fail("Got %s on the done queue" % got)
213 self.assertEqual(w, 3)
214 self.assertEqual(a, 1)
217 self.cond.notifyAll()
220 self.assertEqual(self.done.get_nowait(), "W")
221 self.assertRaises(Queue.Empty, self.done.get_nowait)
223 def testBlockingWait(self):
231 self._TestWait(_BlockingWait)
233 def testLongTimeoutWait(self):
241 self._TestWait(_Helper)
243 def _TimeoutWait(self, timeout, check):
245 self.cond.wait(timeout)
249 def testShortTimeoutWait(self):
250 self._addThread(target=self._TimeoutWait, args=(0.1, "T1"))
251 self._addThread(target=self._TimeoutWait, args=(0.1, "T1"))
253 self.assertEqual(self.done.get_nowait(), "T1")
254 self.assertEqual(self.done.get_nowait(), "T1")
255 self.assertRaises(Queue.Empty, self.done.get_nowait)
257 def testZeroTimeoutWait(self):
258 self._addThread(target=self._TimeoutWait, args=(0, "T0"))
259 self._addThread(target=self._TimeoutWait, args=(0, "T0"))
260 self._addThread(target=self._TimeoutWait, args=(0, "T0"))
262 self.assertEqual(self.done.get_nowait(), "T0")
263 self.assertEqual(self.done.get_nowait(), "T0")
264 self.assertEqual(self.done.get_nowait(), "T0")
265 self.assertRaises(Queue.Empty, self.done.get_nowait)
268 class TestSharedLock(_ThreadedTestCase):
269 """SharedLock tests"""
272 _ThreadedTestCase.setUp(self)
273 self.sl = locking.SharedLock("TestSharedLock")
275 self.assertTrue(repr(self.sl).startswith("<"))
276 self.assertTrue("name=TestSharedLock" in repr(self.sl))
278 def testSequenceAndOwnership(self):
279 self.assertFalse(self.sl.is_owned())
280 self.sl.acquire(shared=1)
281 self.assert_(self.sl.is_owned())
282 self.assert_(self.sl.is_owned(shared=1))
283 self.assertFalse(self.sl.is_owned(shared=0))
285 self.assertFalse(self.sl.is_owned())
287 self.assert_(self.sl.is_owned())
288 self.assertFalse(self.sl.is_owned(shared=1))
289 self.assert_(self.sl.is_owned(shared=0))
291 self.assertFalse(self.sl.is_owned())
292 self.sl.acquire(shared=1)
293 self.assert_(self.sl.is_owned())
294 self.assert_(self.sl.is_owned(shared=1))
295 self.assertFalse(self.sl.is_owned(shared=0))
297 self.assertFalse(self.sl.is_owned())
299 def testBooleanValue(self):
300 # semaphores are supposed to return a true value on a successful acquire
301 self.assert_(self.sl.acquire(shared=1))
303 self.assert_(self.sl.acquire())
306 def testDoubleLockingStoE(self):
307 self.sl.acquire(shared=1)
308 self.assertRaises(AssertionError, self.sl.acquire)
310 def testDoubleLockingEtoS(self):
312 self.assertRaises(AssertionError, self.sl.acquire, shared=1)
314 def testDoubleLockingStoS(self):
315 self.sl.acquire(shared=1)
316 self.assertRaises(AssertionError, self.sl.acquire, shared=1)
318 def testDoubleLockingEtoE(self):
320 self.assertRaises(AssertionError, self.sl.acquire)
322 # helper functions: called in a separate thread they acquire the lock, send
323 # their identifier on the done queue, then release it.
324 def _doItSharer(self):
326 self.sl.acquire(shared=1)
329 except errors.LockError:
332 def _doItExclusive(self):
337 except errors.LockError:
340 def _doItDelete(self):
344 except errors.LockError:
347 def testSharersCanCoexist(self):
348 self.sl.acquire(shared=1)
349 threading.Thread(target=self._doItSharer).start()
350 self.assert_(self.done.get(True, 1))
354 def testExclusiveBlocksExclusive(self):
356 self._addThread(target=self._doItExclusive)
357 self.assertRaises(Queue.Empty, self.done.get_nowait)
360 self.failUnlessEqual(self.done.get_nowait(), "EXC")
363 def testExclusiveBlocksDelete(self):
365 self._addThread(target=self._doItDelete)
366 self.assertRaises(Queue.Empty, self.done.get_nowait)
369 self.failUnlessEqual(self.done.get_nowait(), "DEL")
370 self.sl = locking.SharedLock(self.sl.name)
373 def testExclusiveBlocksSharer(self):
375 self._addThread(target=self._doItSharer)
376 self.assertRaises(Queue.Empty, self.done.get_nowait)
379 self.failUnlessEqual(self.done.get_nowait(), "SHR")
382 def testSharerBlocksExclusive(self):
383 self.sl.acquire(shared=1)
384 self._addThread(target=self._doItExclusive)
385 self.assertRaises(Queue.Empty, self.done.get_nowait)
388 self.failUnlessEqual(self.done.get_nowait(), "EXC")
391 def testSharerBlocksDelete(self):
392 self.sl.acquire(shared=1)
393 self._addThread(target=self._doItDelete)
394 self.assertRaises(Queue.Empty, self.done.get_nowait)
397 self.failUnlessEqual(self.done.get_nowait(), "DEL")
398 self.sl = locking.SharedLock(self.sl.name)
401 def testWaitingExclusiveBlocksSharer(self):
402 """SKIPPED testWaitingExclusiveBlockSharer"""
405 self.sl.acquire(shared=1)
406 # the lock is acquired in shared mode...
407 self._addThread(target=self._doItExclusive)
408 # ...but now an exclusive is waiting...
409 self._addThread(target=self._doItSharer)
410 # ...so the sharer should be blocked as well
411 self.assertRaises(Queue.Empty, self.done.get_nowait)
414 # The exclusive passed before
415 self.failUnlessEqual(self.done.get_nowait(), "EXC")
416 self.failUnlessEqual(self.done.get_nowait(), "SHR")
419 def testWaitingSharerBlocksExclusive(self):
420 """SKIPPED testWaitingSharerBlocksExclusive"""
424 # the lock is acquired in exclusive mode...
425 self._addThread(target=self._doItSharer)
426 # ...but now a sharer is waiting...
427 self._addThread(target=self._doItExclusive)
428 # ...the exclusive is waiting too...
429 self.assertRaises(Queue.Empty, self.done.get_nowait)
432 # The sharer passed before
433 self.assertEqual(self.done.get_nowait(), "SHR")
434 self.assertEqual(self.done.get_nowait(), "EXC")
436 def testDelete(self):
438 self.assertRaises(errors.LockError, self.sl.acquire)
439 self.assertRaises(errors.LockError, self.sl.acquire, shared=1)
440 self.assertRaises(errors.LockError, self.sl.delete)
442 def testDeleteTimeout(self):
443 self.assertTrue(self.sl.delete(timeout=60))
445 def testDeleteTimeoutFail(self):
446 ready = threading.Event()
447 finish = threading.Event()
450 self.sl.acquire(shared=0)
456 self._addThread(target=fn)
459 # Test if deleting a lock owned in exclusive mode by another thread fails
460 # to delete when a timeout is used
461 self.assertFalse(self.sl.delete(timeout=0.02))
466 self.assertTrue(self.sl.delete())
467 self.assertRaises(errors.LockError, self.sl.acquire)
469 def testNoDeleteIfSharer(self):
470 self.sl.acquire(shared=1)
471 self.assertRaises(AssertionError, self.sl.delete)
474 def testDeletePendingSharersExclusiveDelete(self):
476 self._addThread(target=self._doItSharer)
477 self._addThread(target=self._doItSharer)
478 self._addThread(target=self._doItExclusive)
479 self._addThread(target=self._doItDelete)
482 # The threads who were pending return ERR
484 self.assertEqual(self.done.get_nowait(), "ERR")
485 self.sl = locking.SharedLock(self.sl.name)
488 def testDeletePendingDeleteExclusiveSharers(self):
490 self._addThread(target=self._doItDelete)
491 self._addThread(target=self._doItExclusive)
492 self._addThread(target=self._doItSharer)
493 self._addThread(target=self._doItSharer)
496 # The two threads who were pending return both ERR
497 self.assertEqual(self.done.get_nowait(), "ERR")
498 self.assertEqual(self.done.get_nowait(), "ERR")
499 self.assertEqual(self.done.get_nowait(), "ERR")
500 self.assertEqual(self.done.get_nowait(), "ERR")
501 self.sl = locking.SharedLock(self.sl.name)
504 def testExclusiveAcquireTimeout(self):
505 for shared in [0, 1]:
506 on_queue = threading.Event()
507 release_exclusive = threading.Event()
509 def _LockExclusive():
510 self.sl.acquire(shared=0, test_notify=on_queue.set)
511 self.done.put("A: start wait")
512 release_exclusive.wait()
513 self.done.put("A: end wait")
516 # Start thread to hold lock in exclusive mode
517 self._addThread(target=_LockExclusive)
519 # Wait for wait to begin
520 self.assertEqual(self.done.get(timeout=60), "A: start wait")
522 # Wait up to 60s to get lock, but release exclusive lock as soon as we're
524 self.failUnless(self.sl.acquire(shared=shared, timeout=60,
525 test_notify=release_exclusive.set))
527 self.done.put("got 2nd")
532 self.assertEqual(self.done.get_nowait(), "A: end wait")
533 self.assertEqual(self.done.get_nowait(), "got 2nd")
534 self.assertRaises(Queue.Empty, self.done.get_nowait)
537 def testAcquireExpiringTimeout(self):
538 def _AcquireWithTimeout(shared, timeout):
539 if not self.sl.acquire(shared=shared, timeout=timeout):
540 self.done.put("timeout")
542 for shared in [0, 1]:
546 # Start shared acquires with timeout between 0 and 20 ms
548 self._addThread(target=_AcquireWithTimeout,
549 args=(shared, i * 2.0 / 1000.0))
551 # Wait for threads to finish (makes sure the acquire timeout expires
552 # before releasing the lock)
559 self.assertEqual(self.done.get_nowait(), "timeout")
561 self.assertRaises(Queue.Empty, self.done.get_nowait)
564 def testSharedSkipExclusiveAcquires(self):
565 # Tests whether shared acquires jump in front of exclusive acquires in the
568 def _Acquire(shared, name, notify_ev, wait_ev):
570 notify_fn = notify_ev.set
577 if not self.sl.acquire(shared=shared, test_notify=notify_fn):
583 # Get exclusive lock while we fill the queue
591 # Add acquires using threading.Event for synchronization. They'll be
592 # acquired exactly in the order defined in this list.
593 acquires = (shrcnt1 * [(1, "shared 1")] +
594 3 * [(0, "exclusive 1")] +
595 shrcnt2 * [(1, "shared 2")] +
596 shrcnt3 * [(1, "shared 3")] +
597 shrcnt4 * [(1, "shared 4")] +
598 3 * [(0, "exclusive 2")])
603 for args in acquires:
604 ev_cur = threading.Event()
605 self._addThread(target=_Acquire, args=args + (ev_cur, ev_prev))
608 # Wait for last acquire to start
611 # Expect 6 pending exclusive acquires and 1 for all shared acquires
613 self.assertEqual(self.sl._count_pending(), 7)
615 # Release exclusive lock and wait
621 for _ in range(shrcnt1 + shrcnt2 + shrcnt3 + shrcnt4):
622 # Shared locks aren't guaranteed to be notified in order, but they'll be
624 tmp = self.done.get_nowait()
625 if tmp == "shared 1":
627 elif tmp == "shared 2":
629 elif tmp == "shared 3":
631 elif tmp == "shared 4":
633 self.assertEqual(shrcnt1, 0)
634 self.assertEqual(shrcnt2, 0)
635 self.assertEqual(shrcnt3, 0)
636 self.assertEqual(shrcnt3, 0)
639 self.assertEqual(self.done.get_nowait(), "exclusive 1")
642 self.assertEqual(self.done.get_nowait(), "exclusive 2")
644 self.assertRaises(Queue.Empty, self.done.get_nowait)
646 def testIllegalDowngrade(self):
648 self.assertRaises(AssertionError, self.sl.downgrade)
650 # Acquire in shared mode, downgrade should be no-op
651 self.assertTrue(self.sl.acquire(shared=1))
652 self.assertTrue(self.sl.is_owned(shared=1))
653 self.assertTrue(self.sl.downgrade())
654 self.assertTrue(self.sl.is_owned(shared=1))
657 def testDowngrade(self):
658 self.assertTrue(self.sl.acquire())
659 self.assertTrue(self.sl.is_owned(shared=0))
660 self.assertTrue(self.sl.downgrade())
661 self.assertTrue(self.sl.is_owned(shared=1))
665 def testDowngradeJumpsAheadOfExclusive(self):
666 def _KeepExclusive(ev_got, ev_downgrade, ev_release):
667 self.assertTrue(self.sl.acquire())
668 self.assertTrue(self.sl.is_owned(shared=0))
671 self.assertTrue(self.sl.is_owned(shared=0))
672 self.assertTrue(self.sl.downgrade())
673 self.assertTrue(self.sl.is_owned(shared=1))
675 self.assertTrue(self.sl.is_owned(shared=1))
678 def _KeepExclusive2(ev_started, ev_release):
679 self.assertTrue(self.sl.acquire(test_notify=ev_started.set))
680 self.assertTrue(self.sl.is_owned(shared=0))
682 self.assertTrue(self.sl.is_owned(shared=0))
685 def _KeepShared(ev_started, ev_got, ev_release):
686 self.assertTrue(self.sl.acquire(shared=1, test_notify=ev_started.set))
687 self.assertTrue(self.sl.is_owned(shared=1))
690 self.assertTrue(self.sl.is_owned(shared=1))
693 # Acquire lock in exclusive mode
694 ev_got_excl1 = threading.Event()
695 ev_downgrade_excl1 = threading.Event()
696 ev_release_excl1 = threading.Event()
697 th_excl1 = self._addThread(target=_KeepExclusive,
698 args=(ev_got_excl1, ev_downgrade_excl1,
702 # Start a second exclusive acquire
703 ev_started_excl2 = threading.Event()
704 ev_release_excl2 = threading.Event()
705 th_excl2 = self._addThread(target=_KeepExclusive2,
706 args=(ev_started_excl2, ev_release_excl2))
707 ev_started_excl2.wait()
709 # Start shared acquires, will jump ahead of second exclusive acquire when
710 # first exclusive acquire downgrades
711 ev_shared = [(threading.Event(), threading.Event()) for _ in range(5)]
712 ev_release_shared = threading.Event()
714 th_shared = [self._addThread(target=_KeepShared,
715 args=(ev_started, ev_got, ev_release_shared))
716 for (ev_started, ev_got) in ev_shared]
718 # Wait for all shared acquires to start
719 for (ev, _) in ev_shared:
722 # Check lock information
723 self.assertEqual(self.sl.GetLockInfo(set([query.LQ_MODE, query.LQ_OWNER])),
724 [(self.sl.name, "exclusive", [th_excl1.getName()], None)])
725 [(_, _, _, pending), ] = self.sl.GetLockInfo(set([query.LQ_PENDING]))
726 self.assertEqual([(pendmode, sorted(waiting))
727 for (pendmode, waiting) in pending],
728 [("exclusive", [th_excl2.getName()]),
729 ("shared", sorted(th.getName() for th in th_shared))])
731 # Shared acquires won't start until the exclusive lock is downgraded
732 ev_downgrade_excl1.set()
734 # Wait for all shared acquires to be successful
735 for (_, ev) in ev_shared:
738 # Check lock information again
739 self.assertEqual(self.sl.GetLockInfo(set([query.LQ_MODE,
741 [(self.sl.name, "shared", None,
742 [("exclusive", [th_excl2.getName()])])])
743 [(_, _, owner, _), ] = self.sl.GetLockInfo(set([query.LQ_OWNER]))
744 self.assertEqual(set(owner), set([th_excl1.getName()] +
745 [th.getName() for th in th_shared]))
747 ev_release_excl1.set()
748 ev_release_excl2.set()
749 ev_release_shared.set()
753 self.assertEqual(self.sl.GetLockInfo(set([query.LQ_MODE, query.LQ_OWNER,
755 [(self.sl.name, None, None, [])])
758 def testMixedAcquireTimeout(self):
759 sync = threading.Event()
761 def _AcquireShared(ev):
762 if not self.sl.acquire(shared=1, timeout=None):
765 self.done.put("shared")
770 # Wait for notification from main thread
778 ev = threading.Event()
779 self._addThread(target=_AcquireShared, args=(ev, ))
782 # Wait for all acquires to finish
786 self.assertEqual(self.sl._count_pending(), 0)
788 # Try to get exclusive lock
789 self.failIf(self.sl.acquire(shared=0, timeout=0.02))
791 # Acquire exclusive without timeout
792 exclsync = threading.Event()
793 exclev = threading.Event()
795 def _AcquireExclusive():
796 if not self.sl.acquire(shared=0):
799 self.done.put("exclusive")
804 # Wait for notification from main thread
809 self._addThread(target=_AcquireExclusive)
811 # Try to get exclusive lock
812 self.failIf(self.sl.acquire(shared=0, timeout=0.02))
814 # Make all shared holders release their locks
817 # Wait for exclusive acquire to succeed
820 self.assertEqual(self.sl._count_pending(), 0)
822 # Try to get exclusive lock
823 self.failIf(self.sl.acquire(shared=0, timeout=0.02))
825 def _AcquireSharedSimple():
826 if self.sl.acquire(shared=1, timeout=None):
827 self.done.put("shared2")
831 self._addThread(target=_AcquireSharedSimple)
833 # Tell exclusive lock to release
836 # Wait for everything to finish
839 self.assertEqual(self.sl._count_pending(), 0)
843 self.assertEqual(self.done.get_nowait(), "shared")
845 self.assertEqual(self.done.get_nowait(), "exclusive")
848 self.assertEqual(self.done.get_nowait(), "shared2")
850 self.assertRaises(Queue.Empty, self.done.get_nowait)
852 def testPriority(self):
853 # Acquire in exclusive mode
854 self.assert_(self.sl.acquire(shared=0))
857 def _Acquire(prev, next, shared, priority, result):
859 self.sl.acquire(shared=shared, priority=priority, test_notify=next.set)
861 self.done.put(result)
865 counter = itertools.count(0)
866 priorities = range(-20, 30)
867 first = threading.Event()
873 # [(shared/exclusive, set(acquire names), set(pending threads)),
874 # (shared/exclusive, ...),
880 # References shared acquire per priority in L{perprio}. Data structure:
882 # priority: (shared=1, set(acquire names), set(pending threads)),
886 for seed in [4979, 9523, 14902, 32440]:
887 # Use a deterministic random generator
888 rnd = random.Random(seed)
889 for priority in [rnd.choice(priorities) for _ in range(30)]:
894 acqname = "%s/shr=%s/prio=%s" % (counter.next(), shared, priority)
896 ev = threading.Event()
897 thread = self._addThread(target=_Acquire,
898 args=(prev, ev, shared, priority, acqname))
901 # Record expected aqcuire, see above for structure
902 data = (shared, set([acqname]), set([thread]))
903 priolist = perprio.setdefault(priority, [])
905 priosh = prioshared.get(priority, None)
907 # Shared acquires are merged
908 for i, j in zip(priosh[1:], data[1:]):
910 assert data[0] == priosh[0]
912 prioshared[priority] = data
913 priolist.append(data)
915 priolist.append(data)
917 # Start all acquires and wait for them
921 # Check lock information
922 self.assertEqual(self.sl.GetLockInfo(set()),
923 [(self.sl.name, None, None, None)])
924 self.assertEqual(self.sl.GetLockInfo(set([query.LQ_MODE, query.LQ_OWNER])),
925 [(self.sl.name, "exclusive",
926 [threading.currentThread().getName()], None)])
928 self._VerifyPrioPending(self.sl.GetLockInfo(set([query.LQ_PENDING])),
931 # Let threads acquire the lock
934 # Wait for everything to finish
937 self.assert_(self.sl._check_empty())
939 # Check acquires by priority
940 for acquires in [perprio[i] for i in sorted(perprio.keys())]:
941 for (_, names, _) in acquires:
942 # For shared acquires, the set will contain 1..n entries. For exclusive
945 names.remove(self.done.get_nowait())
946 self.assertFalse(compat.any(names for (_, names, _) in acquires))
948 self.assertRaises(Queue.Empty, self.done.get_nowait)
950 def _VerifyPrioPending(self, ((name, mode, owner, pending), ), perprio):
951 self.assertEqual(name, self.sl.name)
952 self.assert_(mode is None)
953 self.assert_(owner is None)
955 self.assertEqual([(pendmode, sorted(waiting))
956 for (pendmode, waiting) in pending],
957 [(["exclusive", "shared"][int(bool(shared))],
958 sorted(t.getName() for t in threads))
959 for acquires in [perprio[i]
960 for i in sorted(perprio.keys())]
961 for (shared, _, threads) in acquires])
963 class _FakeTimeForSpuriousNotifications:
964 def __init__(self, now, check_end):
966 self.check_end = check_end
968 # Deterministic random number generator
969 self.rnd = random.Random(15086)
972 # Advance time if the random number generator thinks so (this is to test
973 # multiple notifications without advancing the time)
974 if self.rnd.random() < 0.3:
975 self.now += self.rnd.random()
977 self.check_end(self.now)
982 def testAcquireTimeoutWithSpuriousNotifications(self):
983 ready = threading.Event()
984 locked = threading.Event()
991 self.assertFalse(locked.isSet())
993 # If we waited long enough (in virtual time), tell main thread to release
994 # lock, otherwise tell it to notify once more
995 req.put(now < (epoch + (timeout * 0.8)))
997 time_fn = self._FakeTimeForSpuriousNotifications(epoch, check_end).time
999 sl = locking.SharedLock("test", _time_fn=time_fn)
1001 # Acquire in exclusive mode
1002 sl.acquire(shared=0)
1005 self.assertTrue(sl.acquire(shared=0, timeout=timeout,
1006 test_notify=ready.set))
1009 self.done.put("success")
1011 # Start acquire with timeout and wait for it to be ready
1012 self._addThread(target=fn)
1015 # The separate thread is now waiting to acquire the lock, so start sending
1016 # spurious notifications.
1018 # Wait for separate thread to ask for another notification
1021 # After sending the notification, the lock will take a short amount of
1022 # time to notice and to retrieve the current time
1023 sl._notify_topmost()
1026 self.assertTrue(count > 100, "Not enough notifications were sent")
1028 self.assertFalse(locked.isSet())
1030 # Some notifications have been sent, now actually release the lock
1033 # Wait for lock to be acquired
1038 self.assertEqual(self.done.get_nowait(), "success")
1039 self.assertRaises(Queue.Empty, self.done.get_nowait)
1042 class TestSharedLockInCondition(_ThreadedTestCase):
1043 """SharedLock as a condition lock tests"""
1046 _ThreadedTestCase.setUp(self)
1047 self.sl = locking.SharedLock("TestSharedLockInCondition")
1050 def setCondition(self):
1051 self.cond = threading.Condition(self.sl)
1053 def testKeepMode(self):
1054 self.cond.acquire(shared=1)
1055 self.assert_(self.sl.is_owned(shared=1))
1057 self.assert_(self.sl.is_owned(shared=1))
1059 self.cond.acquire(shared=0)
1060 self.assert_(self.sl.is_owned(shared=0))
1062 self.assert_(self.sl.is_owned(shared=0))
1066 class TestSharedLockInPipeCondition(TestSharedLockInCondition):
1067 """SharedLock as a pipe condition lock tests"""
1069 def setCondition(self):
1070 self.cond = locking.PipeCondition(self.sl)
1073 class TestSSynchronizedDecorator(_ThreadedTestCase):
1074 """Shared Lock Synchronized decorator test"""
1077 _ThreadedTestCase.setUp(self)
1079 @locking.ssynchronized(_decoratorlock)
1080 def _doItExclusive(self):
1081 self.assert_(_decoratorlock.is_owned())
1082 self.done.put("EXC")
1084 @locking.ssynchronized(_decoratorlock, shared=1)
1085 def _doItSharer(self):
1086 self.assert_(_decoratorlock.is_owned(shared=1))
1087 self.done.put("SHR")
1089 def testDecoratedFunctions(self):
1090 self._doItExclusive()
1091 self.assertFalse(_decoratorlock.is_owned())
1093 self.assertFalse(_decoratorlock.is_owned())
1095 def testSharersCanCoexist(self):
1096 _decoratorlock.acquire(shared=1)
1097 threading.Thread(target=self._doItSharer).start()
1098 self.assert_(self.done.get(True, 1))
1099 _decoratorlock.release()
1102 def testExclusiveBlocksExclusive(self):
1103 _decoratorlock.acquire()
1104 self._addThread(target=self._doItExclusive)
1105 # give it a bit of time to check that it's not actually doing anything
1106 self.assertRaises(Queue.Empty, self.done.get_nowait)
1107 _decoratorlock.release()
1109 self.failUnlessEqual(self.done.get_nowait(), "EXC")
1112 def testExclusiveBlocksSharer(self):
1113 _decoratorlock.acquire()
1114 self._addThread(target=self._doItSharer)
1115 self.assertRaises(Queue.Empty, self.done.get_nowait)
1116 _decoratorlock.release()
1118 self.failUnlessEqual(self.done.get_nowait(), "SHR")
1121 def testSharerBlocksExclusive(self):
1122 _decoratorlock.acquire(shared=1)
1123 self._addThread(target=self._doItExclusive)
1124 self.assertRaises(Queue.Empty, self.done.get_nowait)
1125 _decoratorlock.release()
1127 self.failUnlessEqual(self.done.get_nowait(), "EXC")
1130 class TestLockSet(_ThreadedTestCase):
1134 _ThreadedTestCase.setUp(self)
1138 """Helper to (re)initialize the lock set"""
1139 self.resources = ["one", "two", "three"]
1140 self.ls = locking.LockSet(self.resources, "TestLockSet")
1142 def testResources(self):
1143 self.assertEquals(self.ls._names(), set(self.resources))
1144 newls = locking.LockSet([], "TestLockSet.testResources")
1145 self.assertEquals(newls._names(), set())
1147 def testCheckOwnedUnknown(self):
1148 self.assertFalse(self.ls.check_owned("certainly-not-owning-this-one"))
1149 for shared in [-1, 0, 1, 6378, 24255]:
1150 self.assertFalse(self.ls.check_owned("certainly-not-owning-this-one",
1153 def testCheckOwnedUnknownWhileHolding(self):
1154 self.assertFalse(self.ls.check_owned([]))
1155 self.ls.acquire("one", shared=1)
1156 self.assertRaises(errors.LockError, self.ls.check_owned, "nonexist")
1157 self.assertTrue(self.ls.check_owned("one", shared=1))
1158 self.assertFalse(self.ls.check_owned("one", shared=0))
1159 self.assertFalse(self.ls.check_owned(["one", "two"]))
1160 self.assertRaises(errors.LockError, self.ls.check_owned,
1161 ["one", "nonexist"])
1162 self.assertRaises(errors.LockError, self.ls.check_owned, "")
1164 self.assertFalse(self.ls.check_owned([]))
1165 self.assertFalse(self.ls.check_owned("one"))
1167 def testAcquireRelease(self):
1168 self.assertFalse(self.ls.check_owned(self.ls._names()))
1169 self.assert_(self.ls.acquire("one"))
1170 self.assertEquals(self.ls.list_owned(), set(["one"]))
1171 self.assertTrue(self.ls.check_owned("one"))
1172 self.assertTrue(self.ls.check_owned("one", shared=0))
1173 self.assertFalse(self.ls.check_owned("one", shared=1))
1175 self.assertEquals(self.ls.list_owned(), set())
1176 self.assertFalse(self.ls.check_owned(self.ls._names()))
1177 self.assertEquals(self.ls.acquire(["one"]), set(["one"]))
1178 self.assertEquals(self.ls.list_owned(), set(["one"]))
1180 self.assertEquals(self.ls.list_owned(), set())
1181 self.ls.acquire(["one", "two", "three"])
1182 self.assertEquals(self.ls.list_owned(), set(["one", "two", "three"]))
1183 self.assertTrue(self.ls.check_owned(self.ls._names()))
1184 self.assertTrue(self.ls.check_owned(self.ls._names(), shared=0))
1185 self.assertFalse(self.ls.check_owned(self.ls._names(), shared=1))
1186 self.ls.release("one")
1187 self.assertFalse(self.ls.check_owned(["one"]))
1188 self.assertTrue(self.ls.check_owned(["two", "three"]))
1189 self.assertTrue(self.ls.check_owned(["two", "three"], shared=0))
1190 self.assertFalse(self.ls.check_owned(["two", "three"], shared=1))
1191 self.assertEquals(self.ls.list_owned(), set(["two", "three"]))
1192 self.ls.release(["three"])
1193 self.assertEquals(self.ls.list_owned(), set(["two"]))
1195 self.assertEquals(self.ls.list_owned(), set())
1196 self.assertEquals(self.ls.acquire(["one", "three"]), set(["one", "three"]))
1197 self.assertEquals(self.ls.list_owned(), set(["one", "three"]))
1199 self.assertEquals(self.ls.list_owned(), set())
1200 for name in self.ls._names():
1201 self.assertFalse(self.ls.check_owned(name))
1203 def testNoDoubleAcquire(self):
1204 self.ls.acquire("one")
1205 self.assertRaises(AssertionError, self.ls.acquire, "one")
1206 self.assertRaises(AssertionError, self.ls.acquire, ["two"])
1207 self.assertRaises(AssertionError, self.ls.acquire, ["two", "three"])
1209 self.ls.acquire(["one", "three"])
1210 self.ls.release("one")
1211 self.assertRaises(AssertionError, self.ls.acquire, ["two"])
1212 self.ls.release("three")
1214 def testNoWrongRelease(self):
1215 self.assertRaises(AssertionError, self.ls.release)
1216 self.ls.acquire("one")
1217 self.assertRaises(AssertionError, self.ls.release, "two")
1219 def testAddRemove(self):
1221 self.assertEquals(self.ls.list_owned(), set())
1222 self.assert_("four" in self.ls._names())
1223 self.ls.add(["five", "six", "seven"], acquired=1)
1224 self.assert_("five" in self.ls._names())
1225 self.assert_("six" in self.ls._names())
1226 self.assert_("seven" in self.ls._names())
1227 self.assertEquals(self.ls.list_owned(), set(["five", "six", "seven"]))
1228 self.assertEquals(self.ls.remove(["five", "six"]), ["five", "six"])
1229 self.assert_("five" not in self.ls._names())
1230 self.assert_("six" not in self.ls._names())
1231 self.assertEquals(self.ls.list_owned(), set(["seven"]))
1232 self.assertRaises(AssertionError, self.ls.add, "eight", acquired=1)
1233 self.ls.remove("seven")
1234 self.assert_("seven" not in self.ls._names())
1235 self.assertEquals(self.ls.list_owned(), set([]))
1236 self.ls.acquire(None, shared=1)
1237 self.assertRaises(AssertionError, self.ls.add, "eight")
1239 self.ls.acquire(None)
1240 self.ls.add("eight", acquired=1)
1241 self.assert_("eight" in self.ls._names())
1242 self.assert_("eight" in self.ls.list_owned())
1244 self.assert_("nine" in self.ls._names())
1245 self.assert_("nine" not in self.ls.list_owned())
1247 self.ls.remove(["two"])
1248 self.assert_("two" not in self.ls._names())
1249 self.ls.acquire("three")
1250 self.assertEquals(self.ls.remove(["three"]), ["three"])
1251 self.assert_("three" not in self.ls._names())
1252 self.assertEquals(self.ls.remove("three"), [])
1253 self.assertEquals(self.ls.remove(["one", "three", "six"]), ["one"])
1254 self.assert_("one" not in self.ls._names())
1256 def testRemoveNonBlocking(self):
1257 self.ls.acquire("one")
1258 self.assertEquals(self.ls.remove("one"), ["one"])
1259 self.ls.acquire(["two", "three"])
1260 self.assertEquals(self.ls.remove(["two", "three"]),
1263 def testNoDoubleAdd(self):
1264 self.assertRaises(errors.LockError, self.ls.add, "two")
1266 self.assertRaises(errors.LockError, self.ls.add, "four")
1268 def testNoWrongRemoves(self):
1269 self.ls.acquire(["one", "three"], shared=1)
1270 # Cannot remove "two" while holding something which is not a superset
1271 self.assertRaises(AssertionError, self.ls.remove, "two")
1272 # Cannot remove "three" as we are sharing it
1273 self.assertRaises(AssertionError, self.ls.remove, "three")
1275 def testAcquireSetLock(self):
1276 # acquire the set-lock exclusively
1277 self.assertEquals(self.ls.acquire(None), set(["one", "two", "three"]))
1278 self.assertEquals(self.ls.list_owned(), set(["one", "two", "three"]))
1279 self.assertEquals(self.ls.is_owned(), True)
1280 self.assertEquals(self.ls._names(), set(["one", "two", "three"]))
1281 # I can still add/remove elements...
1282 self.assertEquals(self.ls.remove(["two", "three"]), ["two", "three"])
1283 self.assert_(self.ls.add("six"))
1285 # share the set-lock
1286 self.assertEquals(self.ls.acquire(None, shared=1), set(["one", "six"]))
1287 # adding new elements is not possible
1288 self.assertRaises(AssertionError, self.ls.add, "five")
1291 def testAcquireWithRepetitions(self):
1292 self.assertEquals(self.ls.acquire(["two", "two", "three"], shared=1),
1293 set(["two", "two", "three"]))
1294 self.ls.release(["two", "two"])
1295 self.assertEquals(self.ls.list_owned(), set(["three"]))
1297 def testEmptyAcquire(self):
1298 # Acquire an empty list of locks...
1299 self.assertEquals(self.ls.acquire([]), set())
1300 self.assertEquals(self.ls.list_owned(), set())
1301 # New locks can still be addded
1302 self.assert_(self.ls.add("six"))
1303 # "re-acquiring" is not an issue, since we had really acquired nothing
1304 self.assertEquals(self.ls.acquire([], shared=1), set())
1305 self.assertEquals(self.ls.list_owned(), set())
1306 # We haven't really acquired anything, so we cannot release
1307 self.assertRaises(AssertionError, self.ls.release)
1309 def _doLockSet(self, names, shared):
1311 self.ls.acquire(names, shared=shared)
1312 self.done.put("DONE")
1314 except errors.LockError:
1315 self.done.put("ERR")
1317 def _doAddSet(self, names):
1319 self.ls.add(names, acquired=1)
1320 self.done.put("DONE")
1322 except errors.LockError:
1323 self.done.put("ERR")
1325 def _doRemoveSet(self, names):
1326 self.done.put(self.ls.remove(names))
1329 def testConcurrentSharedAcquire(self):
1330 self.ls.acquire(["one", "two"], shared=1)
1331 self._addThread(target=self._doLockSet, args=(["one", "two"], 1))
1333 self.assertEqual(self.done.get_nowait(), "DONE")
1334 self._addThread(target=self._doLockSet, args=(["one", "two", "three"], 1))
1336 self.assertEqual(self.done.get_nowait(), "DONE")
1337 self._addThread(target=self._doLockSet, args=("three", 1))
1339 self.assertEqual(self.done.get_nowait(), "DONE")
1340 self._addThread(target=self._doLockSet, args=(["one", "two"], 0))
1341 self._addThread(target=self._doLockSet, args=(["two", "three"], 0))
1342 self.assertRaises(Queue.Empty, self.done.get_nowait)
1345 self.assertEqual(self.done.get_nowait(), "DONE")
1346 self.assertEqual(self.done.get_nowait(), "DONE")
1349 def testConcurrentExclusiveAcquire(self):
1350 self.ls.acquire(["one", "two"])
1351 self._addThread(target=self._doLockSet, args=("three", 1))
1353 self.assertEqual(self.done.get_nowait(), "DONE")
1354 self._addThread(target=self._doLockSet, args=("three", 0))
1356 self.assertEqual(self.done.get_nowait(), "DONE")
1357 self.assertRaises(Queue.Empty, self.done.get_nowait)
1358 self._addThread(target=self._doLockSet, args=(["one", "two"], 0))
1359 self._addThread(target=self._doLockSet, args=(["one", "two"], 1))
1360 self._addThread(target=self._doLockSet, args=("one", 0))
1361 self._addThread(target=self._doLockSet, args=("one", 1))
1362 self._addThread(target=self._doLockSet, args=(["two", "three"], 0))
1363 self._addThread(target=self._doLockSet, args=(["two", "three"], 1))
1364 self.assertRaises(Queue.Empty, self.done.get_nowait)
1368 self.failUnlessEqual(self.done.get_nowait(), "DONE")
1371 def testSimpleAcquireTimeoutExpiring(self):
1372 names = sorted(self.ls._names())
1373 self.assert_(len(names) >= 3)
1375 # Get name of first lock
1378 # Get name of last lock
1382 # Block first and try to lock it again
1385 # Block last and try to lock all locks
1388 # Block last and try to lock it again
1392 for (wanted, block) in checks:
1393 # Lock in exclusive mode
1394 self.assert_(self.ls.acquire(block, shared=0))
1397 # Try to get the same lock again with a timeout (should never succeed)
1398 acquired = self.ls.acquire(wanted, timeout=0.1, shared=0)
1400 self.done.put("acquired")
1403 self.assert_(acquired is None)
1404 self.assertFalse(self.ls.list_owned())
1405 self.assertFalse(self.ls.is_owned())
1406 self.done.put("not acquired")
1408 self._addThread(target=_AcquireOne)
1410 # Wait for timeout in thread to expire
1413 # Release exclusive lock again
1416 self.assertEqual(self.done.get_nowait(), "not acquired")
1417 self.assertRaises(Queue.Empty, self.done.get_nowait)
1420 def testDelayedAndExpiringLockAcquire(self):
1422 self.ls.add(["five", "six", "seven", "eight", "nine"])
1424 for expire in (False, True):
1425 names = sorted(self.ls._names())
1426 self.assertEqual(len(names), 8)
1428 lock_ev = dict([(i, threading.Event()) for i in names])
1430 # Lock all in exclusive mode
1431 self.assert_(self.ls.acquire(names, shared=0))
1434 # We'll wait at least 300ms per lock
1435 lockwait = len(names) * [0.3]
1437 # Fail if we can't acquire all locks in 400ms. There are 8 locks, so
1438 # this gives us up to 2.4s to fail.
1439 lockall_timeout = 0.4
1441 # This should finish rather quickly
1443 lockall_timeout = len(names) * 5.0
1446 def acquire_notification(name):
1448 self.done.put("getting %s" % name)
1453 if self.ls.acquire(names, shared=0, timeout=lockall_timeout,
1454 test_notify=acquire_notification):
1455 self.done.put("got all")
1458 self.done.put("timeout on all")
1461 for ev in lock_ev.values():
1464 t = self._addThread(target=_LockAll)
1466 for idx, name in enumerate(names):
1467 # Wait for actual acquire on this lock to start
1468 lock_ev[name].wait(10.0)
1470 if expire and t.isAlive():
1471 # Wait some time after getting the notification to make sure the lock
1472 # acquire will expire
1473 SafeSleep(lockwait[idx])
1475 self.ls.release(names=name)
1477 self.assertFalse(self.ls.list_owned())
1482 # Not checking which locks were actually acquired. Doing so would be
1483 # too timing-dependant.
1484 self.assertEqual(self.done.get_nowait(), "timeout on all")
1487 self.assertEqual(self.done.get_nowait(), "getting %s" % i)
1488 self.assertEqual(self.done.get_nowait(), "got all")
1489 self.assertRaises(Queue.Empty, self.done.get_nowait)
1492 def testConcurrentRemove(self):
1494 self.ls.acquire(["one", "two", "four"])
1495 self._addThread(target=self._doLockSet, args=(["one", "four"], 0))
1496 self._addThread(target=self._doLockSet, args=(["one", "four"], 1))
1497 self._addThread(target=self._doLockSet, args=(["one", "two"], 0))
1498 self._addThread(target=self._doLockSet, args=(["one", "two"], 1))
1499 self.assertRaises(Queue.Empty, self.done.get_nowait)
1500 self.ls.remove("one")
1504 self.failUnlessEqual(self.done.get_nowait(), "ERR")
1505 self.ls.add(["five", "six"], acquired=1)
1506 self._addThread(target=self._doLockSet, args=(["three", "six"], 1))
1507 self._addThread(target=self._doLockSet, args=(["three", "six"], 0))
1508 self._addThread(target=self._doLockSet, args=(["four", "six"], 1))
1509 self._addThread(target=self._doLockSet, args=(["four", "six"], 0))
1510 self.ls.remove("five")
1514 self.failUnlessEqual(self.done.get_nowait(), "DONE")
1515 self.ls.acquire(["three", "four"])
1516 self._addThread(target=self._doRemoveSet, args=(["four", "six"], ))
1517 self.assertRaises(Queue.Empty, self.done.get_nowait)
1518 self.ls.remove("four")
1520 self.assertEqual(self.done.get_nowait(), ["six"])
1521 self._addThread(target=self._doRemoveSet, args=(["two"]))
1523 self.assertEqual(self.done.get_nowait(), ["two"])
1529 def testConcurrentSharedSetLock(self):
1530 # share the set-lock...
1531 self.ls.acquire(None, shared=1)
1532 # ...another thread can share it too
1533 self._addThread(target=self._doLockSet, args=(None, 1))
1535 self.assertEqual(self.done.get_nowait(), "DONE")
1536 # ...or just share some elements
1537 self._addThread(target=self._doLockSet, args=(["one", "three"], 1))
1539 self.assertEqual(self.done.get_nowait(), "DONE")
1540 # ...but not add new ones or remove any
1541 t = self._addThread(target=self._doAddSet, args=(["nine"]))
1542 self._addThread(target=self._doRemoveSet, args=(["two"], ))
1543 self.assertRaises(Queue.Empty, self.done.get_nowait)
1544 # this just releases the set-lock
1547 self.assertEqual(self.done.get_nowait(), "DONE")
1548 # release the lock on the actual elements so remove() can proceed too
1551 self.failUnlessEqual(self.done.get_nowait(), ["two"])
1556 def testConcurrentExclusiveSetLock(self):
1557 # acquire the set-lock...
1558 self.ls.acquire(None, shared=0)
1559 # ...no one can do anything else
1560 self._addThread(target=self._doLockSet, args=(None, 1))
1561 self._addThread(target=self._doLockSet, args=(None, 0))
1562 self._addThread(target=self._doLockSet, args=(["three"], 0))
1563 self._addThread(target=self._doLockSet, args=(["two"], 1))
1564 self._addThread(target=self._doAddSet, args=(["nine"]))
1565 self.assertRaises(Queue.Empty, self.done.get_nowait)
1569 self.assertEqual(self.done.get(True, 1), "DONE")
1574 def testConcurrentSetLockAdd(self):
1575 self.ls.acquire("one")
1576 # Another thread wants the whole SetLock
1577 self._addThread(target=self._doLockSet, args=(None, 0))
1578 self._addThread(target=self._doLockSet, args=(None, 1))
1579 self.assertRaises(Queue.Empty, self.done.get_nowait)
1580 self.assertRaises(AssertionError, self.ls.add, "four")
1583 self.assertEqual(self.done.get_nowait(), "DONE")
1584 self.assertEqual(self.done.get_nowait(), "DONE")
1585 self.ls.acquire(None)
1586 self._addThread(target=self._doLockSet, args=(None, 0))
1587 self._addThread(target=self._doLockSet, args=(None, 1))
1588 self.assertRaises(Queue.Empty, self.done.get_nowait)
1590 self.ls.add("five", acquired=1)
1591 self.ls.add("six", acquired=1, shared=1)
1592 self.assertEquals(self.ls.list_owned(),
1593 set(["one", "two", "three", "five", "six"]))
1594 self.assertEquals(self.ls.is_owned(), True)
1595 self.assertEquals(self.ls._names(),
1596 set(["one", "two", "three", "four", "five", "six"]))
1599 self.assertEqual(self.done.get_nowait(), "DONE")
1600 self.assertEqual(self.done.get_nowait(), "DONE")
1604 def testEmptyLockSet(self):
1606 self.assertEqual(self.ls.acquire(None), set(["one", "two", "three"]))
1608 self.ls.remove(["one", "two", "three"])
1609 self.assertFalse(self.ls._names())
1610 # and adds/locks by another thread still wait
1611 self._addThread(target=self._doAddSet, args=(["nine"]))
1612 self._addThread(target=self._doLockSet, args=(None, 1))
1613 self._addThread(target=self._doLockSet, args=(None, 0))
1614 self.assertRaises(Queue.Empty, self.done.get_nowait)
1618 self.assertEqual(self.done.get_nowait(), "DONE")
1620 self.assertEqual(self.ls.remove(["nine"]), ["nine"])
1622 self.assertEqual(self.ls.acquire(None, shared=1), set())
1623 # other sharers can go, adds still wait
1624 self._addThread(target=self._doLockSet, args=(None, 1))
1626 self.assertEqual(self.done.get_nowait(), "DONE")
1627 self._addThread(target=self._doAddSet, args=(["nine"]))
1628 self.assertRaises(Queue.Empty, self.done.get_nowait)
1631 self.assertEqual(self.done.get_nowait(), "DONE")
1634 def testAcquireWithNamesDowngrade(self):
1635 self.assertEquals(self.ls.acquire("two", shared=0), set(["two"]))
1636 self.assertTrue(self.ls.is_owned())
1637 self.assertFalse(self.ls._get_lock().is_owned())
1639 self.assertFalse(self.ls.is_owned())
1640 self.assertFalse(self.ls._get_lock().is_owned())
1641 # Can't downgrade after releasing
1642 self.assertRaises(AssertionError, self.ls.downgrade, "two")
1644 def testDowngrade(self):
1645 # Not owning anything, must raise an exception
1646 self.assertFalse(self.ls.is_owned())
1647 self.assertRaises(AssertionError, self.ls.downgrade)
1649 self.assertFalse(compat.any(i.is_owned()
1650 for i in self.ls._get_lockdict().values()))
1651 self.assertFalse(self.ls.check_owned(self.ls._names()))
1652 for name in self.ls._names():
1653 self.assertFalse(self.ls.check_owned(name))
1655 self.assertEquals(self.ls.acquire(None, shared=0),
1656 set(["one", "two", "three"]))
1657 self.assertRaises(AssertionError, self.ls.downgrade, "unknown lock")
1659 self.assertTrue(self.ls.check_owned(self.ls._names(), shared=0))
1660 for name in self.ls._names():
1661 self.assertTrue(self.ls.check_owned(name))
1662 self.assertTrue(self.ls.check_owned(name, shared=0))
1663 self.assertFalse(self.ls.check_owned(name, shared=1))
1665 self.assertTrue(self.ls._get_lock().is_owned(shared=0))
1666 self.assertTrue(compat.all(i.is_owned(shared=0)
1667 for i in self.ls._get_lockdict().values()))
1669 # Start downgrading locks
1670 self.assertTrue(self.ls.downgrade(names=["one"]))
1671 self.assertTrue(self.ls._get_lock().is_owned(shared=0))
1672 self.assertTrue(compat.all(lock.is_owned(shared=[0, 1][int(name == "one")])
1674 self.ls._get_lockdict().items()))
1676 self.assertFalse(self.ls.check_owned("one", shared=0))
1677 self.assertTrue(self.ls.check_owned("one", shared=1))
1678 self.assertTrue(self.ls.check_owned("two", shared=0))
1679 self.assertTrue(self.ls.check_owned("three", shared=0))
1681 # Downgrade second lock
1682 self.assertTrue(self.ls.downgrade(names="two"))
1683 self.assertTrue(self.ls._get_lock().is_owned(shared=0))
1684 should_share = lambda name: [0, 1][int(name in ("one", "two"))]
1685 self.assertTrue(compat.all(lock.is_owned(shared=should_share(name))
1687 self.ls._get_lockdict().items()))
1689 self.assertFalse(self.ls.check_owned("one", shared=0))
1690 self.assertTrue(self.ls.check_owned("one", shared=1))
1691 self.assertFalse(self.ls.check_owned("two", shared=0))
1692 self.assertTrue(self.ls.check_owned("two", shared=1))
1693 self.assertTrue(self.ls.check_owned("three", shared=0))
1695 # Downgrading the last exclusive lock to shared must downgrade the
1696 # lockset-internal lock too
1697 self.assertTrue(self.ls.downgrade(names="three"))
1698 self.assertTrue(self.ls._get_lock().is_owned(shared=1))
1699 self.assertTrue(compat.all(i.is_owned(shared=1)
1700 for i in self.ls._get_lockdict().values()))
1702 # Verify owned locks
1703 for name in self.ls._names():
1704 self.assertTrue(self.ls.check_owned(name, shared=1))
1706 # Downgrading a shared lock must be a no-op
1707 self.assertTrue(self.ls.downgrade(names=["one", "three"]))
1708 self.assertTrue(self.ls._get_lock().is_owned(shared=1))
1709 self.assertTrue(compat.all(i.is_owned(shared=1)
1710 for i in self.ls._get_lockdict().values()))
1714 def testDowngradeEverything(self):
1715 self.assertEqual(self.ls.acquire(locking.ALL_SET, shared=0),
1716 set(["one", "two", "three"]))
1717 self.assertTrue(self.ls.owning_all())
1719 # Ensure all locks are now owned in exclusive mode
1720 for name in self.ls._names():
1721 self.assertTrue(self.ls.check_owned(name, shared=0))
1723 # Downgrade everything
1724 self.assertTrue(self.ls.downgrade())
1726 # Ensure all locks are now owned in shared mode
1727 for name in self.ls._names():
1728 self.assertTrue(self.ls.check_owned(name, shared=1))
1730 self.assertTrue(self.ls.owning_all())
1732 def testPriority(self):
1733 def _Acquire(prev, next, name, priority, success_fn):
1735 self.assert_(self.ls.acquire(name, shared=0,
1737 test_notify=lambda _: next.set()))
1743 # Get all in exclusive mode
1744 self.assert_(self.ls.acquire(locking.ALL_SET, shared=0))
1746 done_two = Queue.Queue(0)
1748 first = threading.Event()
1751 acquires = [("one", prio, self.done) for prio in range(1, 33)]
1752 acquires.extend([("two", prio, done_two) for prio in range(1, 33)])
1754 # Use a deterministic random generator
1755 random.Random(741).shuffle(acquires)
1757 for (name, prio, done) in acquires:
1758 ev = threading.Event()
1759 self._addThread(target=_Acquire,
1760 args=(prev, ev, name, prio,
1761 compat.partial(done.put, "Prio%s" % prio)))
1767 # Wait for last acquire to start
1770 # Let threads acquire locks
1773 # Wait for threads to finish
1776 for i in range(1, 33):
1777 self.assertEqual(self.done.get_nowait(), "Prio%s" % i)
1778 self.assertEqual(done_two.get_nowait(), "Prio%s" % i)
1780 self.assertRaises(Queue.Empty, self.done.get_nowait)
1781 self.assertRaises(Queue.Empty, done_two.get_nowait)
1783 def testNamesWithOpportunisticAndTimeout(self):
1784 self.assertRaises(AssertionError, self.ls.acquire,
1785 ["one", "two"], timeout=1.0, opportunistic=True)
1787 def testOpportunisticWithUnknownName(self):
1789 self.assertFalse(name in self.ls._names())
1790 result = self.ls.acquire(name, opportunistic=True)
1791 self.assertFalse(result)
1792 self.assertFalse(self.ls.list_owned())
1794 result = self.ls.acquire(["two", name], opportunistic=True)
1795 self.assertEqual(result, set(["two"]))
1796 self.assertEqual(self.ls.list_owned(), set(["two"]))
1800 def testSimpleOpportunisticAcquisition(self):
1801 self.assertEquals(self.ls._names(), set(["one", "two", "three"]))
1803 # Hold a lock in main thread
1804 self.assertEqual(self.ls.acquire("two", shared=0), set(["two"]))
1807 # The lock "two" is held by the main thread
1808 result = self.ls.acquire(["one", "two"], shared=0, opportunistic=True)
1809 self.assertEqual(result, set(["one"]))
1810 self.assertEqual(self.ls.list_owned(), set(["one"]))
1811 self.assertFalse(self.ls._get_lock().is_owned())
1814 self.assertFalse(self.ls.list_owned())
1816 # Try to acquire the lock held by the main thread
1817 result = self.ls.acquire(["two"], shared=0, opportunistic=True)
1818 self.assertFalse(self.ls._get_lock().is_owned())
1819 self.assertFalse(result)
1820 self.assertFalse(self.ls.list_owned())
1822 # Try to acquire all locks
1823 result = self.ls.acquire(locking.ALL_SET, shared=0, opportunistic=True)
1824 self.assertTrue(self.ls._get_lock().is_owned(),
1825 msg="Internal lock is not owned")
1826 self.assertEqual(result, set(["one", "three"]))
1827 self.assertEqual(self.ls.list_owned(), set(["one", "three"]))
1831 self.assertFalse(self.ls.list_owned())
1835 self._addThread(target=fn)
1837 # Wait for threads to finish
1840 self.assertEqual(self.ls.list_owned(), set(["two"]))
1843 self.assertFalse(self.ls.list_owned())
1844 self.assertFalse(self.ls._get_lock().is_owned())
1846 self.assertTrue(self.done.get_nowait())
1847 self.assertRaises(Queue.Empty, self.done.get_nowait)
1849 def testOpportunisticAcquisitionWithoutNamesExpires(self):
1850 self.assertEquals(self.ls._names(), set(["one", "two", "three"]))
1852 # Hold all locks in main thread
1853 self.ls.acquire(locking.ALL_SET, shared=0)
1854 self.assertTrue(self.ls._get_lock().is_owned())
1857 # Try to acquire all locks in separate thread
1858 result = self.ls.acquire(locking.ALL_SET, shared=0, opportunistic=True,
1860 self.assertFalse(result)
1861 self.assertFalse(self.ls._get_lock().is_owned())
1862 self.assertFalse(self.ls.list_owned())
1864 # Try once more without a timeout
1865 self.assertFalse(self.ls.acquire("one", shared=0, opportunistic=True))
1869 self._addThread(target=fn)
1871 # Wait for threads to finish
1874 self.assertEqual(self.ls.list_owned(), set(["one", "two", "three"]))
1877 self.assertFalse(self.ls.list_owned())
1878 self.assertFalse(self.ls._get_lock().is_owned(shared=0))
1880 self.assertTrue(self.done.get_nowait())
1881 self.assertRaises(Queue.Empty, self.done.get_nowait)
1883 def testSharedOpportunisticAcquisitionWithoutNames(self):
1884 self.assertEquals(self.ls._names(), set(["one", "two", "three"]))
1886 # Hold all locks in main thread
1887 self.ls.acquire(locking.ALL_SET, shared=1)
1888 self.assertTrue(self.ls._get_lock().is_owned(shared=1))
1891 # Try to acquire all locks in separate thread in shared mode
1892 result = self.ls.acquire(locking.ALL_SET, shared=1, opportunistic=True,
1894 self.assertEqual(result, set(["one", "two", "three"]))
1895 self.assertTrue(self.ls._get_lock().is_owned(shared=1))
1897 self.assertFalse(self.ls._get_lock().is_owned())
1899 # Try one in exclusive mode
1900 self.assertFalse(self.ls.acquire("one", shared=0, opportunistic=True))
1904 self._addThread(target=fn)
1906 # Wait for threads to finish
1909 self.assertEqual(self.ls.list_owned(), set(["one", "two", "three"]))
1912 self.assertFalse(self.ls.list_owned())
1913 self.assertFalse(self.ls._get_lock().is_owned())
1915 self.assertTrue(self.done.get_nowait())
1916 self.assertRaises(Queue.Empty, self.done.get_nowait)
1918 def testLockDeleteWithOpportunisticAcquisition(self):
1919 # This test exercises some code handling LockError on acquisition, that is
1920 # after all lock names have been gathered. This shouldn't happen in reality
1921 # as removing locks from the set requires the lockset-internal lock, but
1922 # the code should handle the situation anyway.
1923 ready = threading.Event()
1924 finished = threading.Event()
1926 self.assertEquals(self.ls._names(), set(["one", "two", "three"]))
1928 # Thread function to delete lock
1930 # Wait for notification
1933 # Delete lock named "two" by accessing lockset-internal data
1934 ld = self.ls._get_lockdict()
1935 self.assertTrue(ld["two"].delete())
1937 self.done.put("deleted.two")
1942 self._addThread(target=fn)
1944 # Notification helper, called when lock already holds internal lock.
1945 # Therefore only one of the locks not yet locked can be deleted.
1947 self.done.put("notify.%s" % name)
1950 # Tell helper thread to delete lock "two"
1954 # Hold all locks in main thread
1955 self.ls.acquire(locking.ALL_SET, shared=0, test_notify=notify)
1956 self.assertEqual(self.ls.list_owned(), set(["one", "three"]))
1958 # Wait for threads to finish
1963 self.assertFalse(self.ls.list_owned())
1964 self.assertFalse(self.ls._get_lock().is_owned())
1966 self.assertEqual(self.done.get_nowait(), "notify.one")
1967 self.assertEqual(self.done.get_nowait(), "deleted.two")
1968 self.assertEqual(self.done.get_nowait(), "notify.three")
1969 self.assertEqual(self.done.get_nowait(), "notify.two")
1970 self.assertRaises(Queue.Empty, self.done.get_nowait)
1973 class TestGetLsAcquireModeAndTimeouts(unittest.TestCase):
1975 self.fn = locking._GetLsAcquireModeAndTimeouts
1977 def testOpportunisticWithoutNames(self):
1978 (mode, ls_timeout_fn, timeout_fn) = self.fn(False, None, True)
1979 self.assertEqual(mode, locking._LS_ACQUIRE_OPPORTUNISTIC)
1980 self.assertTrue(ls_timeout_fn is None)
1981 self.assertEqual(timeout_fn(), 0)
1983 def testAllInputCombinations(self):
1984 for want_all in [False, True]:
1985 for timeout in [None, 0, 100]:
1986 for opportunistic in [False, True]:
1987 if (opportunistic and
1989 timeout is not None):
1990 # Can't accept a timeout when acquiring opportunistically
1991 self.assertRaises(AssertionError, self.fn,
1992 want_all, timeout, opportunistic)
1994 (mode, ls_timeout_fn, timeout_fn) = \
1995 self.fn(want_all, timeout, opportunistic)
1998 self.assertEqual(mode, locking._LS_ACQUIRE_OPPORTUNISTIC)
1999 self.assertEqual(timeout_fn(), 0)
2001 self.assertTrue(callable(timeout_fn))
2003 self.assertEqual(mode, locking._LS_ACQUIRE_ALL)
2005 self.assertEqual(mode, locking._LS_ACQUIRE_EXACT)
2008 self.assertTrue(callable(ls_timeout_fn))
2010 self.assertTrue(ls_timeout_fn is None)
2013 class TestGanetiLockManager(_ThreadedTestCase):
2015 _ThreadedTestCase.setUp(self)
2016 self.nodes = ["n1", "n2"]
2017 self.nodegroups = ["g1", "g2"]
2018 self.instances = ["i1", "i2", "i3"]
2019 self.networks = ["net1", "net2", "net3"]
2020 self.GL = locking.GanetiLockManager(self.nodes, self.nodegroups,
2021 self.instances, self.networks)
2024 # Don't try this at home...
2025 locking.GanetiLockManager._instance = None
2027 def testLockingConstants(self):
2028 # The locking library internally cheats by assuming its constants have some
2029 # relationships with each other. Check those hold true.
2030 # This relationship is also used in the Processor to recursively acquire
2031 # the right locks. Again, please don't break it.
2032 for i in range(len(locking.LEVELS)):
2033 self.assertEqual(i, locking.LEVELS[i])
2035 def testDoubleGLFails(self):
2036 self.assertRaises(AssertionError, locking.GanetiLockManager, [], [], [], [])
2038 def testLockNames(self):
2039 self.assertEqual(self.GL._names(locking.LEVEL_CLUSTER), set(["BGL"]))
2040 self.assertEqual(self.GL._names(locking.LEVEL_NODE_ALLOC), set(["NAL"]))
2041 self.assertEqual(self.GL._names(locking.LEVEL_NODE), set(self.nodes))
2042 self.assertEqual(self.GL._names(locking.LEVEL_NODEGROUP),
2043 set(self.nodegroups))
2044 self.assertEqual(self.GL._names(locking.LEVEL_INSTANCE),
2045 set(self.instances))
2046 self.assertEqual(self.GL._names(locking.LEVEL_NETWORK),
2049 def testInitAndResources(self):
2050 locking.GanetiLockManager._instance = None
2051 self.GL = locking.GanetiLockManager([], [], [], [])
2052 self.assertEqual(self.GL._names(locking.LEVEL_CLUSTER), set(["BGL"]))
2053 self.assertEqual(self.GL._names(locking.LEVEL_NODE_ALLOC), set(["NAL"]))
2054 self.assertEqual(self.GL._names(locking.LEVEL_NODE), set())
2055 self.assertEqual(self.GL._names(locking.LEVEL_NODEGROUP), set())
2056 self.assertEqual(self.GL._names(locking.LEVEL_INSTANCE), set())
2057 self.assertEqual(self.GL._names(locking.LEVEL_NETWORK), set())
2059 locking.GanetiLockManager._instance = None
2060 self.GL = locking.GanetiLockManager(self.nodes, self.nodegroups, [], [])
2061 self.assertEqual(self.GL._names(locking.LEVEL_CLUSTER), set(["BGL"]))
2062 self.assertEqual(self.GL._names(locking.LEVEL_NODE_ALLOC), set(["NAL"]))
2063 self.assertEqual(self.GL._names(locking.LEVEL_NODE), set(self.nodes))
2064 self.assertEqual(self.GL._names(locking.LEVEL_NODEGROUP),
2065 set(self.nodegroups))
2066 self.assertEqual(self.GL._names(locking.LEVEL_INSTANCE), set())
2067 self.assertEqual(self.GL._names(locking.LEVEL_NETWORK), set())
2069 locking.GanetiLockManager._instance = None
2070 self.GL = locking.GanetiLockManager([], [], self.instances, [])
2071 self.assertEqual(self.GL._names(locking.LEVEL_CLUSTER), set(["BGL"]))
2072 self.assertEqual(self.GL._names(locking.LEVEL_NODE_ALLOC), set(["NAL"]))
2073 self.assertEqual(self.GL._names(locking.LEVEL_NODE), set())
2074 self.assertEqual(self.GL._names(locking.LEVEL_NODEGROUP), set())
2075 self.assertEqual(self.GL._names(locking.LEVEL_INSTANCE),
2076 set(self.instances))
2078 locking.GanetiLockManager._instance = None
2079 self.GL = locking.GanetiLockManager([], [], [], self.networks)
2080 self.assertEqual(self.GL._names(locking.LEVEL_CLUSTER), set(["BGL"]))
2081 self.assertEqual(self.GL._names(locking.LEVEL_NODE_ALLOC), set(["NAL"]))
2082 self.assertEqual(self.GL._names(locking.LEVEL_NODE), set())
2083 self.assertEqual(self.GL._names(locking.LEVEL_NODEGROUP), set())
2084 self.assertEqual(self.GL._names(locking.LEVEL_INSTANCE), set())
2085 self.assertEqual(self.GL._names(locking.LEVEL_NETWORK),
2088 def testAcquireRelease(self):
2089 self.GL.acquire(locking.LEVEL_CLUSTER, ["BGL"], shared=1)
2090 self.assertEquals(self.GL.list_owned(locking.LEVEL_CLUSTER), set(["BGL"]))
2091 self.GL.acquire(locking.LEVEL_INSTANCE, ["i1"])
2092 self.GL.acquire(locking.LEVEL_NODEGROUP, ["g2"])
2093 self.GL.acquire(locking.LEVEL_NODE, ["n1", "n2"], shared=1)
2094 self.assertTrue(self.GL.check_owned(locking.LEVEL_NODE, ["n1", "n2"],
2096 self.assertFalse(self.GL.check_owned(locking.LEVEL_INSTANCE, ["i1", "i3"]))
2097 self.GL.release(locking.LEVEL_NODE, ["n2"])
2098 self.assertEquals(self.GL.list_owned(locking.LEVEL_NODE), set(["n1"]))
2099 self.assertEquals(self.GL.list_owned(locking.LEVEL_NODEGROUP), set(["g2"]))
2100 self.assertEquals(self.GL.list_owned(locking.LEVEL_INSTANCE), set(["i1"]))
2101 self.GL.release(locking.LEVEL_NODE)
2102 self.assertEquals(self.GL.list_owned(locking.LEVEL_NODE), set())
2103 self.assertEquals(self.GL.list_owned(locking.LEVEL_NODEGROUP), set(["g2"]))
2104 self.assertEquals(self.GL.list_owned(locking.LEVEL_INSTANCE), set(["i1"]))
2105 self.GL.release(locking.LEVEL_NODEGROUP)
2106 self.GL.release(locking.LEVEL_INSTANCE)
2107 self.assertRaises(errors.LockError, self.GL.acquire,
2108 locking.LEVEL_INSTANCE, ["i5"])
2109 self.GL.acquire(locking.LEVEL_INSTANCE, ["i3"], shared=1)
2110 self.assertEquals(self.GL.list_owned(locking.LEVEL_INSTANCE), set(["i3"]))
2112 def testAcquireWholeSets(self):
2113 self.GL.acquire(locking.LEVEL_CLUSTER, ["BGL"], shared=1)
2114 self.assertEquals(self.GL.acquire(locking.LEVEL_INSTANCE, None),
2115 set(self.instances))
2116 self.assertEquals(self.GL.list_owned(locking.LEVEL_INSTANCE),
2117 set(self.instances))
2118 self.assertEquals(self.GL.acquire(locking.LEVEL_NODEGROUP, None),
2119 set(self.nodegroups))
2120 self.assertEquals(self.GL.list_owned(locking.LEVEL_NODEGROUP),
2121 set(self.nodegroups))
2122 self.assertEquals(self.GL.acquire(locking.LEVEL_NODE, None, shared=1),
2124 self.assertEquals(self.GL.list_owned(locking.LEVEL_NODE),
2126 self.assertTrue(self.GL.owning_all(locking.LEVEL_INSTANCE))
2127 self.assertTrue(self.GL.owning_all(locking.LEVEL_NODEGROUP))
2128 self.assertTrue(self.GL.owning_all(locking.LEVEL_NODE))
2129 self.GL.release(locking.LEVEL_NODE)
2130 self.GL.release(locking.LEVEL_NODEGROUP)
2131 self.GL.release(locking.LEVEL_INSTANCE)
2132 self.GL.release(locking.LEVEL_CLUSTER)
2134 def testAcquireWholeAndPartial(self):
2135 self.assertFalse(self.GL.owning_all(locking.LEVEL_INSTANCE))
2136 self.GL.acquire(locking.LEVEL_CLUSTER, ["BGL"], shared=1)
2137 self.assertEquals(self.GL.acquire(locking.LEVEL_INSTANCE, None),
2138 set(self.instances))
2139 self.assertEquals(self.GL.list_owned(locking.LEVEL_INSTANCE),
2140 set(self.instances))
2141 self.assertEquals(self.GL.acquire(locking.LEVEL_NODE, ["n2"], shared=1),
2143 self.assertEquals(self.GL.list_owned(locking.LEVEL_NODE),
2145 self.assertTrue(self.GL.owning_all(locking.LEVEL_INSTANCE))
2146 self.assertFalse(self.GL.owning_all(locking.LEVEL_NODE))
2147 self.GL.release(locking.LEVEL_NODE)
2148 self.GL.release(locking.LEVEL_INSTANCE)
2149 self.GL.release(locking.LEVEL_CLUSTER)
2151 def testBGLDependency(self):
2152 self.assertRaises(AssertionError, self.GL.acquire,
2153 locking.LEVEL_NODE, ["n1", "n2"])
2154 self.assertRaises(AssertionError, self.GL.acquire,
2155 locking.LEVEL_INSTANCE, ["i3"])
2156 self.assertRaises(AssertionError, self.GL.acquire,
2157 locking.LEVEL_NODEGROUP, ["g1"])
2158 self.GL.acquire(locking.LEVEL_CLUSTER, ["BGL"], shared=1)
2159 self.GL.acquire(locking.LEVEL_NODE, ["n1"])
2160 self.assertRaises(AssertionError, self.GL.release,
2161 locking.LEVEL_CLUSTER, ["BGL"])
2162 self.assertRaises(AssertionError, self.GL.release,
2163 locking.LEVEL_CLUSTER)
2164 self.GL.release(locking.LEVEL_NODE)
2165 self.GL.acquire(locking.LEVEL_INSTANCE, ["i1", "i2"])
2166 self.assertRaises(AssertionError, self.GL.release,
2167 locking.LEVEL_CLUSTER, ["BGL"])
2168 self.assertRaises(AssertionError, self.GL.release,
2169 locking.LEVEL_CLUSTER)
2170 self.GL.release(locking.LEVEL_INSTANCE)
2171 self.GL.acquire(locking.LEVEL_NODEGROUP, None)
2172 self.GL.release(locking.LEVEL_NODEGROUP, ["g1"])
2173 self.assertRaises(AssertionError, self.GL.release,
2174 locking.LEVEL_CLUSTER, ["BGL"])
2175 self.assertRaises(AssertionError, self.GL.release,
2176 locking.LEVEL_CLUSTER)
2177 self.GL.release(locking.LEVEL_NODEGROUP)
2178 self.GL.release(locking.LEVEL_CLUSTER)
2180 def testWrongOrder(self):
2181 self.GL.acquire(locking.LEVEL_CLUSTER, ["BGL"], shared=1)
2182 self.GL.acquire(locking.LEVEL_NODE, ["n2"])
2183 self.assertRaises(AssertionError, self.GL.acquire,
2184 locking.LEVEL_NODE, ["n1"])
2185 self.assertRaises(AssertionError, self.GL.acquire,
2186 locking.LEVEL_NODEGROUP, ["g1"])
2187 self.assertRaises(AssertionError, self.GL.acquire,
2188 locking.LEVEL_INSTANCE, ["i2"])
2190 def testModifiableLevels(self):
2191 self.assertRaises(AssertionError, self.GL.add, locking.LEVEL_CLUSTER,
2193 self.assertRaises(AssertionError, self.GL.add, locking.LEVEL_NODE_ALLOC,
2195 self.GL.acquire(locking.LEVEL_CLUSTER, ["BGL"])
2196 self.GL.add(locking.LEVEL_INSTANCE, ["i4"])
2197 self.GL.remove(locking.LEVEL_INSTANCE, ["i3"])
2198 self.GL.remove(locking.LEVEL_INSTANCE, ["i1"])
2199 self.assertEqual(self.GL._names(locking.LEVEL_INSTANCE), set(["i2", "i4"]))
2200 self.GL.add(locking.LEVEL_NODE, ["n3"])
2201 self.GL.remove(locking.LEVEL_NODE, ["n1"])
2202 self.assertEqual(self.GL._names(locking.LEVEL_NODE), set(["n2", "n3"]))
2203 self.GL.add(locking.LEVEL_NODEGROUP, ["g3"])
2204 self.GL.remove(locking.LEVEL_NODEGROUP, ["g2"])
2205 self.GL.remove(locking.LEVEL_NODEGROUP, ["g1"])
2206 self.assertEqual(self.GL._names(locking.LEVEL_NODEGROUP), set(["g3"]))
2207 self.assertRaises(AssertionError, self.GL.remove, locking.LEVEL_CLUSTER,
2210 # Helper function to run as a thread that shared the BGL and then acquires
2211 # some locks at another level.
2212 def _doLock(self, level, names, shared):
2214 self.GL.acquire(locking.LEVEL_CLUSTER, ["BGL"], shared=1)
2215 self.GL.acquire(level, names, shared=shared)
2216 self.done.put("DONE")
2217 self.GL.release(level)
2218 self.GL.release(locking.LEVEL_CLUSTER)
2219 except errors.LockError:
2220 self.done.put("ERR")
2223 def testConcurrency(self):
2224 self.GL.acquire(locking.LEVEL_CLUSTER, ["BGL"], shared=1)
2225 self._addThread(target=self._doLock,
2226 args=(locking.LEVEL_INSTANCE, "i1", 1))
2228 self.assertEqual(self.done.get_nowait(), "DONE")
2229 self.GL.acquire(locking.LEVEL_INSTANCE, ["i3"])
2230 self._addThread(target=self._doLock,
2231 args=(locking.LEVEL_INSTANCE, "i1", 1))
2233 self.assertEqual(self.done.get_nowait(), "DONE")
2234 self._addThread(target=self._doLock,
2235 args=(locking.LEVEL_INSTANCE, "i3", 1))
2236 self.assertRaises(Queue.Empty, self.done.get_nowait)
2237 self.GL.release(locking.LEVEL_INSTANCE)
2239 self.assertEqual(self.done.get_nowait(), "DONE")
2240 self.GL.acquire(locking.LEVEL_INSTANCE, ["i2"], shared=1)
2241 self._addThread(target=self._doLock,
2242 args=(locking.LEVEL_INSTANCE, "i2", 1))
2244 self.assertEqual(self.done.get_nowait(), "DONE")
2245 self._addThread(target=self._doLock,
2246 args=(locking.LEVEL_INSTANCE, "i2", 0))
2247 self.assertRaises(Queue.Empty, self.done.get_nowait)
2248 self.GL.release(locking.LEVEL_INSTANCE)
2250 self.assertEqual(self.done.get(True, 1), "DONE")
2251 self.GL.release(locking.LEVEL_CLUSTER, ["BGL"])
2254 class TestLockMonitor(_ThreadedTestCase):
2256 _ThreadedTestCase.setUp(self)
2257 self.lm = locking.LockMonitor()
2259 def testSingleThread(self):
2262 for i in range(100):
2263 name = "TestLock%s" % i
2264 locks.append(locking.SharedLock(name, monitor=self.lm))
2266 self.assertEqual(len(self.lm._locks), len(locks))
2267 result = objects.QueryResponse.FromDict(self.lm.QueryLocks(["name"]))
2268 self.assertEqual(len(result.fields), 1)
2269 self.assertEqual(len(result.data), 100)
2274 # The garbage collector might needs some time
2277 raise utils.RetryAgain()
2279 utils.Retry(_CheckLocks, 0.1, 30.0)
2281 self.assertFalse(self.lm._locks)
2283 def testMultiThread(self):
2286 def _CreateLock(prev, next, name):
2288 locks.append(locking.SharedLock(name, monitor=self.lm))
2294 first = threading.Event()
2297 # Use a deterministic random generator
2298 for i in random.Random(4263).sample(range(100), 33):
2299 name = "MtTestLock%s" % i
2300 expnames.append(name)
2302 ev = threading.Event()
2303 self._addThread(target=_CreateLock, args=(prev, ev, name))
2310 # Check order in which locks were added
2311 self.assertEqual([i.name for i in locks], expnames)
2313 # Check query result
2314 result = self.lm.QueryLocks(["name", "mode", "owner", "pending"])
2315 self.assert_(isinstance(result, dict))
2316 response = objects.QueryResponse.FromDict(result)
2317 self.assertEqual(response.data,
2318 [[(constants.RS_NORMAL, name),
2319 (constants.RS_NORMAL, None),
2320 (constants.RS_NORMAL, None),
2321 (constants.RS_NORMAL, [])]
2322 for name in utils.NiceSort(expnames)])
2323 self.assertEqual(len(response.fields), 4)
2324 self.assertEqual(["name", "mode", "owner", "pending"],
2325 [fdef.name for fdef in response.fields])
2327 # Test exclusive acquire
2328 for tlock in locks[::4]:
2329 tlock.acquire(shared=0)
2331 def _GetExpResult(name):
2332 if tlock.name == name:
2333 return [(constants.RS_NORMAL, name),
2334 (constants.RS_NORMAL, "exclusive"),
2335 (constants.RS_NORMAL,
2336 [threading.currentThread().getName()]),
2337 (constants.RS_NORMAL, [])]
2338 return [(constants.RS_NORMAL, name),
2339 (constants.RS_NORMAL, None),
2340 (constants.RS_NORMAL, None),
2341 (constants.RS_NORMAL, [])]
2343 result = self.lm.QueryLocks(["name", "mode", "owner", "pending"])
2344 self.assertEqual(objects.QueryResponse.FromDict(result).data,
2345 [_GetExpResult(name)
2346 for name in utils.NiceSort(expnames)])
2350 # Test shared acquire
2351 def _Acquire(lock, shared, ev, notify):
2352 lock.acquire(shared=shared)
2359 for tlock1 in locks[::11]:
2360 for tlock2 in locks[::-15]:
2361 if tlock2 == tlock1:
2365 for tlock3 in locks[::10]:
2366 if tlock3 in (tlock2, tlock1):
2370 releaseev = threading.Event()
2376 ev = threading.Event()
2377 tthreads1.append(self._addThread(target=_Acquire,
2378 args=(tlock1, 1, releaseev, ev)))
2379 acquireev.append(ev)
2381 ev = threading.Event()
2382 tthread2 = self._addThread(target=_Acquire,
2383 args=(tlock2, 1, releaseev, ev))
2384 acquireev.append(ev)
2386 ev = threading.Event()
2387 tthread3 = self._addThread(target=_Acquire,
2388 args=(tlock3, 0, releaseev, ev))
2389 acquireev.append(ev)
2391 # Wait for all locks to be acquired
2395 # Check query result
2396 result = self.lm.QueryLocks(["name", "mode", "owner"])
2397 response = objects.QueryResponse.FromDict(result)
2398 for (name, mode, owner) in response.data:
2399 (name_status, name_value) = name
2400 (owner_status, owner_value) = owner
2402 self.assertEqual(name_status, constants.RS_NORMAL)
2403 self.assertEqual(owner_status, constants.RS_NORMAL)
2405 if name_value == tlock1.name:
2406 self.assertEqual(mode, (constants.RS_NORMAL, "shared"))
2407 self.assertEqual(set(owner_value),
2408 set(i.getName() for i in tthreads1))
2411 if name_value == tlock2.name:
2412 self.assertEqual(mode, (constants.RS_NORMAL, "shared"))
2413 self.assertEqual(owner_value, [tthread2.getName()])
2416 if name_value == tlock3.name:
2417 self.assertEqual(mode, (constants.RS_NORMAL, "exclusive"))
2418 self.assertEqual(owner_value, [tthread3.getName()])
2421 self.assert_(name_value in expnames)
2422 self.assertEqual(mode, (constants.RS_NORMAL, None))
2423 self.assert_(owner_value is None)
2425 # Release locks again
2430 result = self.lm.QueryLocks(["name", "mode", "owner"])
2431 self.assertEqual(objects.QueryResponse.FromDict(result).data,
2432 [[(constants.RS_NORMAL, name),
2433 (constants.RS_NORMAL, None),
2434 (constants.RS_NORMAL, None)]
2435 for name in utils.NiceSort(expnames)])
2437 def testDelete(self):
2438 lock = locking.SharedLock("TestLock", monitor=self.lm)
2440 self.assertEqual(len(self.lm._locks), 1)
2441 result = self.lm.QueryLocks(["name", "mode", "owner"])
2442 self.assertEqual(objects.QueryResponse.FromDict(result).data,
2443 [[(constants.RS_NORMAL, lock.name),
2444 (constants.RS_NORMAL, None),
2445 (constants.RS_NORMAL, None)]])
2449 result = self.lm.QueryLocks(["name", "mode", "owner"])
2450 self.assertEqual(objects.QueryResponse.FromDict(result).data,
2451 [[(constants.RS_NORMAL, lock.name),
2452 (constants.RS_NORMAL, "deleted"),
2453 (constants.RS_NORMAL, None)]])
2454 self.assertEqual(len(self.lm._locks), 1)
2456 def testPending(self):
2457 def _Acquire(lock, shared, prev, next):
2460 lock.acquire(shared=shared, test_notify=next.set)
2466 lock = locking.SharedLock("ExcLock", monitor=self.lm)
2468 for shared in [0, 1]:
2471 self.assertEqual(len(self.lm._locks), 1)
2472 result = self.lm.QueryLocks(["name", "mode", "owner"])
2473 self.assertEqual(objects.QueryResponse.FromDict(result).data,
2474 [[(constants.RS_NORMAL, lock.name),
2475 (constants.RS_NORMAL, "exclusive"),
2476 (constants.RS_NORMAL,
2477 [threading.currentThread().getName()])]])
2481 first = threading.Event()
2485 ev = threading.Event()
2486 threads.append(self._addThread(target=_Acquire,
2487 args=(lock, shared, prev, ev)))
2493 # Wait for last acquire to start waiting
2496 # NOTE: This works only because QueryLocks will acquire the
2497 # lock-internal lock again and won't be able to get the information
2498 # until it has the lock. By then the acquire should be registered in
2499 # SharedLock.__pending (otherwise it's a bug).
2501 # All acquires are waiting now
2503 pending = [("shared", utils.NiceSort(t.getName() for t in threads))]
2505 pending = [("exclusive", [t.getName()]) for t in threads]
2507 result = self.lm.QueryLocks(["name", "mode", "owner", "pending"])
2508 self.assertEqual(objects.QueryResponse.FromDict(result).data,
2509 [[(constants.RS_NORMAL, lock.name),
2510 (constants.RS_NORMAL, "exclusive"),
2511 (constants.RS_NORMAL,
2512 [threading.currentThread().getName()]),
2513 (constants.RS_NORMAL, pending)]])
2515 self.assertEqual(len(self.lm._locks), 1)
2521 # No pending acquires
2522 result = self.lm.QueryLocks(["name", "mode", "owner", "pending"])
2523 self.assertEqual(objects.QueryResponse.FromDict(result).data,
2524 [[(constants.RS_NORMAL, lock.name),
2525 (constants.RS_NORMAL, None),
2526 (constants.RS_NORMAL, None),
2527 (constants.RS_NORMAL, [])]])
2529 self.assertEqual(len(self.lm._locks), 1)
2531 def testDeleteAndRecreate(self):
2532 lname = "TestLock101923193"
2534 # Create some locks with the same name and keep all references
2535 locks = [locking.SharedLock(lname, monitor=self.lm)
2538 self.assertEqual(len(self.lm._locks), len(locks))
2540 result = self.lm.QueryLocks(["name", "mode", "owner"])
2541 self.assertEqual(objects.QueryResponse.FromDict(result).data,
2542 [[(constants.RS_NORMAL, lname),
2543 (constants.RS_NORMAL, None),
2544 (constants.RS_NORMAL, None)]] * 5)
2548 # Check information order
2549 result = self.lm.QueryLocks(["name", "mode", "owner"])
2550 self.assertEqual(objects.QueryResponse.FromDict(result).data,
2551 [[(constants.RS_NORMAL, lname),
2552 (constants.RS_NORMAL, None),
2553 (constants.RS_NORMAL, None)]] * 2 +
2554 [[(constants.RS_NORMAL, lname),
2555 (constants.RS_NORMAL, "deleted"),
2556 (constants.RS_NORMAL, None)]] +
2557 [[(constants.RS_NORMAL, lname),
2558 (constants.RS_NORMAL, None),
2559 (constants.RS_NORMAL, None)]] * 2)
2561 locks[1].acquire(shared=0)
2564 [(constants.RS_NORMAL, lname),
2565 (constants.RS_NORMAL, None),
2566 (constants.RS_NORMAL, None)],
2567 [(constants.RS_NORMAL, lname),
2568 (constants.RS_NORMAL, "exclusive"),
2569 (constants.RS_NORMAL, [threading.currentThread().getName()])],
2570 [(constants.RS_NORMAL, lname),
2571 (constants.RS_NORMAL, "deleted"),
2572 (constants.RS_NORMAL, None)],
2573 [(constants.RS_NORMAL, lname),
2574 (constants.RS_NORMAL, None),
2575 (constants.RS_NORMAL, None)],
2576 [(constants.RS_NORMAL, lname),
2577 (constants.RS_NORMAL, None),
2578 (constants.RS_NORMAL, None)],
2581 # Check information order
2582 result = self.lm.QueryLocks(["name", "mode", "owner"])
2583 self.assertEqual(objects.QueryResponse.FromDict(result).data, last_status)
2585 self.assertEqual(len(set(self.lm._locks.values())), len(locks))
2586 self.assertEqual(len(self.lm._locks), len(locks))
2588 # Check lock deletion
2589 for idx in range(len(locks)):
2591 assert gc.isenabled()
2593 self.assertEqual(len(self.lm._locks), len(locks))
2594 result = self.lm.QueryLocks(["name", "mode", "owner"])
2595 self.assertEqual(objects.QueryResponse.FromDict(result).data,
2596 last_status[idx + 1:])
2598 # All locks should have been deleted
2600 self.assertFalse(self.lm._locks)
2602 result = self.lm.QueryLocks(["name", "mode", "owner"])
2603 self.assertEqual(objects.QueryResponse.FromDict(result).data, [])
2609 def AddResult(self, *args):
2610 self._info.append(args)
2612 def CountPending(self):
2613 return len(self._info)
2615 def GetLockInfo(self, requested):
2616 (exp_requested, result) = self._info.pop(0)
2618 if exp_requested != requested:
2619 raise Exception("Requested information (%s) does not match"
2620 " expectations (%s)" % (requested, exp_requested))
2624 def testMultipleResults(self):
2625 fl1 = self._FakeLock()
2626 fl2 = self._FakeLock()
2628 self.lm.RegisterLock(fl1)
2629 self.lm.RegisterLock(fl2)
2632 for i in [fl1, fl2]:
2633 i.AddResult(set([query.LQ_MODE, query.LQ_OWNER]), [])
2634 result = self.lm.QueryLocks(["name", "mode", "owner"])
2635 self.assertEqual(objects.QueryResponse.FromDict(result).data, [])
2636 for i in [fl1, fl2]:
2637 self.assertEqual(i.CountPending(), 0)
2640 for fn in [lambda x: x, reversed, sorted]:
2641 fl1.AddResult(set(), list(fn([
2642 ("aaa", None, None, None),
2643 ("bbb", None, None, None),
2645 fl2.AddResult(set(), [])
2646 result = self.lm.QueryLocks(["name"])
2647 self.assertEqual(objects.QueryResponse.FromDict(result).data, [
2648 [(constants.RS_NORMAL, "aaa")],
2649 [(constants.RS_NORMAL, "bbb")],
2651 for i in [fl1, fl2]:
2652 self.assertEqual(i.CountPending(), 0)
2654 for fn2 in [lambda x: x, reversed, sorted]:
2655 fl1.AddResult(set([query.LQ_MODE]), list(fn([
2656 # Same name, but different information
2657 ("aaa", "mode0", None, None),
2658 ("aaa", "mode1", None, None),
2659 ("aaa", "mode2", None, None),
2660 ("aaa", "mode3", None, None),
2662 fl2.AddResult(set([query.LQ_MODE]), [
2663 ("zzz", "end", None, None),
2664 ("000", "start", None, None),
2666 ("aaa", "b200", None, None),
2667 ("aaa", "b300", None, None),
2669 result = self.lm.QueryLocks(["name", "mode"])
2670 self.assertEqual(objects.QueryResponse.FromDict(result).data, [
2671 [(constants.RS_NORMAL, "000"), (constants.RS_NORMAL, "start")],
2673 # Name is the same, so order must be equal to incoming order
2674 [(constants.RS_NORMAL, "aaa"), (constants.RS_NORMAL, "mode0")],
2675 [(constants.RS_NORMAL, "aaa"), (constants.RS_NORMAL, "mode1")],
2676 [(constants.RS_NORMAL, "aaa"), (constants.RS_NORMAL, "mode2")],
2677 [(constants.RS_NORMAL, "aaa"), (constants.RS_NORMAL, "mode3")],
2679 [(constants.RS_NORMAL, "aaa"), (constants.RS_NORMAL, "b200")],
2680 [(constants.RS_NORMAL, "aaa"), (constants.RS_NORMAL, "b300")],
2682 [(constants.RS_NORMAL, "zzz"), (constants.RS_NORMAL, "end")],
2684 for i in [fl1, fl2]:
2685 self.assertEqual(i.CountPending(), 0)
2688 if __name__ == "__main__":
2689 testutils.GanetiTestProgram()