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 # and adds/locks by another thread still wait
1610 self._addThread(target=self._doAddSet, args=(["nine"]))
1611 self._addThread(target=self._doLockSet, args=(None, 1))
1612 self._addThread(target=self._doLockSet, args=(None, 0))
1613 self.assertRaises(Queue.Empty, self.done.get_nowait)
1617 self.assertEqual(self.done.get_nowait(), "DONE")
1619 self.assertEqual(self.ls.remove(["nine"]), ["nine"])
1621 self.assertEqual(self.ls.acquire(None, shared=1), set())
1622 # other sharers can go, adds still wait
1623 self._addThread(target=self._doLockSet, args=(None, 1))
1625 self.assertEqual(self.done.get_nowait(), "DONE")
1626 self._addThread(target=self._doAddSet, args=(["nine"]))
1627 self.assertRaises(Queue.Empty, self.done.get_nowait)
1630 self.assertEqual(self.done.get_nowait(), "DONE")
1633 def testAcquireWithNamesDowngrade(self):
1634 self.assertEquals(self.ls.acquire("two", shared=0), set(["two"]))
1635 self.assertTrue(self.ls.is_owned())
1636 self.assertFalse(self.ls._get_lock().is_owned())
1638 self.assertFalse(self.ls.is_owned())
1639 self.assertFalse(self.ls._get_lock().is_owned())
1640 # Can't downgrade after releasing
1641 self.assertRaises(AssertionError, self.ls.downgrade, "two")
1643 def testDowngrade(self):
1644 # Not owning anything, must raise an exception
1645 self.assertFalse(self.ls.is_owned())
1646 self.assertRaises(AssertionError, self.ls.downgrade)
1648 self.assertFalse(compat.any(i.is_owned()
1649 for i in self.ls._get_lockdict().values()))
1650 self.assertFalse(self.ls.check_owned(self.ls._names()))
1651 for name in self.ls._names():
1652 self.assertFalse(self.ls.check_owned(name))
1654 self.assertEquals(self.ls.acquire(None, shared=0),
1655 set(["one", "two", "three"]))
1656 self.assertRaises(AssertionError, self.ls.downgrade, "unknown lock")
1658 self.assertTrue(self.ls.check_owned(self.ls._names(), shared=0))
1659 for name in self.ls._names():
1660 self.assertTrue(self.ls.check_owned(name))
1661 self.assertTrue(self.ls.check_owned(name, shared=0))
1662 self.assertFalse(self.ls.check_owned(name, shared=1))
1664 self.assertTrue(self.ls._get_lock().is_owned(shared=0))
1665 self.assertTrue(compat.all(i.is_owned(shared=0)
1666 for i in self.ls._get_lockdict().values()))
1668 # Start downgrading locks
1669 self.assertTrue(self.ls.downgrade(names=["one"]))
1670 self.assertTrue(self.ls._get_lock().is_owned(shared=0))
1671 self.assertTrue(compat.all(lock.is_owned(shared=[0, 1][int(name == "one")])
1673 self.ls._get_lockdict().items()))
1675 self.assertFalse(self.ls.check_owned("one", shared=0))
1676 self.assertTrue(self.ls.check_owned("one", shared=1))
1677 self.assertTrue(self.ls.check_owned("two", shared=0))
1678 self.assertTrue(self.ls.check_owned("three", shared=0))
1680 # Downgrade second lock
1681 self.assertTrue(self.ls.downgrade(names="two"))
1682 self.assertTrue(self.ls._get_lock().is_owned(shared=0))
1683 should_share = lambda name: [0, 1][int(name in ("one", "two"))]
1684 self.assertTrue(compat.all(lock.is_owned(shared=should_share(name))
1686 self.ls._get_lockdict().items()))
1688 self.assertFalse(self.ls.check_owned("one", shared=0))
1689 self.assertTrue(self.ls.check_owned("one", shared=1))
1690 self.assertFalse(self.ls.check_owned("two", shared=0))
1691 self.assertTrue(self.ls.check_owned("two", shared=1))
1692 self.assertTrue(self.ls.check_owned("three", shared=0))
1694 # Downgrading the last exclusive lock to shared must downgrade the
1695 # lockset-internal lock too
1696 self.assertTrue(self.ls.downgrade(names="three"))
1697 self.assertTrue(self.ls._get_lock().is_owned(shared=1))
1698 self.assertTrue(compat.all(i.is_owned(shared=1)
1699 for i in self.ls._get_lockdict().values()))
1701 # Verify owned locks
1702 for name in self.ls._names():
1703 self.assertTrue(self.ls.check_owned(name, shared=1))
1705 # Downgrading a shared lock must be a no-op
1706 self.assertTrue(self.ls.downgrade(names=["one", "three"]))
1707 self.assertTrue(self.ls._get_lock().is_owned(shared=1))
1708 self.assertTrue(compat.all(i.is_owned(shared=1)
1709 for i in self.ls._get_lockdict().values()))
1713 def testDowngradeEverything(self):
1714 self.assertEqual(self.ls.acquire(locking.ALL_SET, shared=0),
1715 set(["one", "two", "three"]))
1717 # Ensure all locks are now owned in exclusive mode
1718 for name in self.ls._names():
1719 self.assertTrue(self.ls.check_owned(name, shared=0))
1721 # Downgrade everything
1722 self.assertTrue(self.ls.downgrade())
1724 # Ensure all locks are now owned in shared mode
1725 for name in self.ls._names():
1726 self.assertTrue(self.ls.check_owned(name, shared=1))
1728 def testPriority(self):
1729 def _Acquire(prev, next, name, priority, success_fn):
1731 self.assert_(self.ls.acquire(name, shared=0,
1733 test_notify=lambda _: next.set()))
1739 # Get all in exclusive mode
1740 self.assert_(self.ls.acquire(locking.ALL_SET, shared=0))
1742 done_two = Queue.Queue(0)
1744 first = threading.Event()
1747 acquires = [("one", prio, self.done) for prio in range(1, 33)]
1748 acquires.extend([("two", prio, done_two) for prio in range(1, 33)])
1750 # Use a deterministic random generator
1751 random.Random(741).shuffle(acquires)
1753 for (name, prio, done) in acquires:
1754 ev = threading.Event()
1755 self._addThread(target=_Acquire,
1756 args=(prev, ev, name, prio,
1757 compat.partial(done.put, "Prio%s" % prio)))
1763 # Wait for last acquire to start
1766 # Let threads acquire locks
1769 # Wait for threads to finish
1772 for i in range(1, 33):
1773 self.assertEqual(self.done.get_nowait(), "Prio%s" % i)
1774 self.assertEqual(done_two.get_nowait(), "Prio%s" % i)
1776 self.assertRaises(Queue.Empty, self.done.get_nowait)
1777 self.assertRaises(Queue.Empty, done_two.get_nowait)
1780 class TestGanetiLockManager(_ThreadedTestCase):
1782 _ThreadedTestCase.setUp(self)
1783 self.nodes = ["n1", "n2"]
1784 self.nodegroups = ["g1", "g2"]
1785 self.instances = ["i1", "i2", "i3"]
1786 self.networks = ["net1", "net2", "net3"]
1787 self.GL = locking.GanetiLockManager(self.nodes, self.nodegroups,
1788 self.instances, self.networks)
1791 # Don't try this at home...
1792 locking.GanetiLockManager._instance = None
1794 def testLockingConstants(self):
1795 # The locking library internally cheats by assuming its constants have some
1796 # relationships with each other. Check those hold true.
1797 # This relationship is also used in the Processor to recursively acquire
1798 # the right locks. Again, please don't break it.
1799 for i in range(len(locking.LEVELS)):
1800 self.assertEqual(i, locking.LEVELS[i])
1802 def testDoubleGLFails(self):
1803 self.assertRaises(AssertionError, locking.GanetiLockManager, [], [], [], [])
1805 def testLockNames(self):
1806 self.assertEqual(self.GL._names(locking.LEVEL_CLUSTER), set(["BGL"]))
1807 self.assertEqual(self.GL._names(locking.LEVEL_NODE), set(self.nodes))
1808 self.assertEqual(self.GL._names(locking.LEVEL_NODEGROUP),
1809 set(self.nodegroups))
1810 self.assertEqual(self.GL._names(locking.LEVEL_INSTANCE),
1811 set(self.instances))
1812 self.assertEqual(self.GL._names(locking.LEVEL_NETWORK),
1815 def testInitAndResources(self):
1816 locking.GanetiLockManager._instance = None
1817 self.GL = locking.GanetiLockManager([], [], [], [])
1818 self.assertEqual(self.GL._names(locking.LEVEL_CLUSTER), set(["BGL"]))
1819 self.assertEqual(self.GL._names(locking.LEVEL_NODE), set())
1820 self.assertEqual(self.GL._names(locking.LEVEL_NODEGROUP), set())
1821 self.assertEqual(self.GL._names(locking.LEVEL_INSTANCE), set())
1822 self.assertEqual(self.GL._names(locking.LEVEL_NETWORK), set())
1824 locking.GanetiLockManager._instance = None
1825 self.GL = locking.GanetiLockManager(self.nodes, self.nodegroups, [], [])
1826 self.assertEqual(self.GL._names(locking.LEVEL_CLUSTER), set(["BGL"]))
1827 self.assertEqual(self.GL._names(locking.LEVEL_NODE), set(self.nodes))
1828 self.assertEqual(self.GL._names(locking.LEVEL_NODEGROUP),
1829 set(self.nodegroups))
1830 self.assertEqual(self.GL._names(locking.LEVEL_INSTANCE), set())
1831 self.assertEqual(self.GL._names(locking.LEVEL_NETWORK), set())
1833 locking.GanetiLockManager._instance = None
1834 self.GL = locking.GanetiLockManager([], [], self.instances, [])
1835 self.assertEqual(self.GL._names(locking.LEVEL_CLUSTER), set(["BGL"]))
1836 self.assertEqual(self.GL._names(locking.LEVEL_NODE), set())
1837 self.assertEqual(self.GL._names(locking.LEVEL_NODEGROUP), set())
1838 self.assertEqual(self.GL._names(locking.LEVEL_INSTANCE),
1839 set(self.instances))
1841 locking.GanetiLockManager._instance = None
1842 self.GL = locking.GanetiLockManager([], [], [], self.networks)
1843 self.assertEqual(self.GL._names(locking.LEVEL_CLUSTER), set(["BGL"]))
1844 self.assertEqual(self.GL._names(locking.LEVEL_NODE), set())
1845 self.assertEqual(self.GL._names(locking.LEVEL_NODEGROUP), set())
1846 self.assertEqual(self.GL._names(locking.LEVEL_INSTANCE), set())
1847 self.assertEqual(self.GL._names(locking.LEVEL_NETWORK),
1850 def testAcquireRelease(self):
1851 self.GL.acquire(locking.LEVEL_CLUSTER, ["BGL"], shared=1)
1852 self.assertEquals(self.GL.list_owned(locking.LEVEL_CLUSTER), set(["BGL"]))
1853 self.GL.acquire(locking.LEVEL_INSTANCE, ["i1"])
1854 self.GL.acquire(locking.LEVEL_NODEGROUP, ["g2"])
1855 self.GL.acquire(locking.LEVEL_NODE, ["n1", "n2"], shared=1)
1856 self.assertTrue(self.GL.check_owned(locking.LEVEL_NODE, ["n1", "n2"],
1858 self.assertFalse(self.GL.check_owned(locking.LEVEL_INSTANCE, ["i1", "i3"]))
1859 self.GL.release(locking.LEVEL_NODE, ["n2"])
1860 self.assertEquals(self.GL.list_owned(locking.LEVEL_NODE), set(["n1"]))
1861 self.assertEquals(self.GL.list_owned(locking.LEVEL_NODEGROUP), set(["g2"]))
1862 self.assertEquals(self.GL.list_owned(locking.LEVEL_INSTANCE), set(["i1"]))
1863 self.GL.release(locking.LEVEL_NODE)
1864 self.assertEquals(self.GL.list_owned(locking.LEVEL_NODE), set())
1865 self.assertEquals(self.GL.list_owned(locking.LEVEL_NODEGROUP), set(["g2"]))
1866 self.assertEquals(self.GL.list_owned(locking.LEVEL_INSTANCE), set(["i1"]))
1867 self.GL.release(locking.LEVEL_NODEGROUP)
1868 self.GL.release(locking.LEVEL_INSTANCE)
1869 self.assertRaises(errors.LockError, self.GL.acquire,
1870 locking.LEVEL_INSTANCE, ["i5"])
1871 self.GL.acquire(locking.LEVEL_INSTANCE, ["i3"], shared=1)
1872 self.assertEquals(self.GL.list_owned(locking.LEVEL_INSTANCE), set(["i3"]))
1874 def testAcquireWholeSets(self):
1875 self.GL.acquire(locking.LEVEL_CLUSTER, ["BGL"], shared=1)
1876 self.assertEquals(self.GL.acquire(locking.LEVEL_INSTANCE, None),
1877 set(self.instances))
1878 self.assertEquals(self.GL.list_owned(locking.LEVEL_INSTANCE),
1879 set(self.instances))
1880 self.assertEquals(self.GL.acquire(locking.LEVEL_NODEGROUP, None),
1881 set(self.nodegroups))
1882 self.assertEquals(self.GL.list_owned(locking.LEVEL_NODEGROUP),
1883 set(self.nodegroups))
1884 self.assertEquals(self.GL.acquire(locking.LEVEL_NODE, None, shared=1),
1886 self.assertEquals(self.GL.list_owned(locking.LEVEL_NODE),
1888 self.GL.release(locking.LEVEL_NODE)
1889 self.GL.release(locking.LEVEL_NODEGROUP)
1890 self.GL.release(locking.LEVEL_INSTANCE)
1891 self.GL.release(locking.LEVEL_CLUSTER)
1893 def testAcquireWholeAndPartial(self):
1894 self.GL.acquire(locking.LEVEL_CLUSTER, ["BGL"], shared=1)
1895 self.assertEquals(self.GL.acquire(locking.LEVEL_INSTANCE, None),
1896 set(self.instances))
1897 self.assertEquals(self.GL.list_owned(locking.LEVEL_INSTANCE),
1898 set(self.instances))
1899 self.assertEquals(self.GL.acquire(locking.LEVEL_NODE, ["n2"], shared=1),
1901 self.assertEquals(self.GL.list_owned(locking.LEVEL_NODE),
1903 self.GL.release(locking.LEVEL_NODE)
1904 self.GL.release(locking.LEVEL_INSTANCE)
1905 self.GL.release(locking.LEVEL_CLUSTER)
1907 def testBGLDependency(self):
1908 self.assertRaises(AssertionError, self.GL.acquire,
1909 locking.LEVEL_NODE, ["n1", "n2"])
1910 self.assertRaises(AssertionError, self.GL.acquire,
1911 locking.LEVEL_INSTANCE, ["i3"])
1912 self.assertRaises(AssertionError, self.GL.acquire,
1913 locking.LEVEL_NODEGROUP, ["g1"])
1914 self.GL.acquire(locking.LEVEL_CLUSTER, ["BGL"], shared=1)
1915 self.GL.acquire(locking.LEVEL_NODE, ["n1"])
1916 self.assertRaises(AssertionError, self.GL.release,
1917 locking.LEVEL_CLUSTER, ["BGL"])
1918 self.assertRaises(AssertionError, self.GL.release,
1919 locking.LEVEL_CLUSTER)
1920 self.GL.release(locking.LEVEL_NODE)
1921 self.GL.acquire(locking.LEVEL_INSTANCE, ["i1", "i2"])
1922 self.assertRaises(AssertionError, self.GL.release,
1923 locking.LEVEL_CLUSTER, ["BGL"])
1924 self.assertRaises(AssertionError, self.GL.release,
1925 locking.LEVEL_CLUSTER)
1926 self.GL.release(locking.LEVEL_INSTANCE)
1927 self.GL.acquire(locking.LEVEL_NODEGROUP, None)
1928 self.GL.release(locking.LEVEL_NODEGROUP, ["g1"])
1929 self.assertRaises(AssertionError, self.GL.release,
1930 locking.LEVEL_CLUSTER, ["BGL"])
1931 self.assertRaises(AssertionError, self.GL.release,
1932 locking.LEVEL_CLUSTER)
1933 self.GL.release(locking.LEVEL_NODEGROUP)
1934 self.GL.release(locking.LEVEL_CLUSTER)
1936 def testWrongOrder(self):
1937 self.GL.acquire(locking.LEVEL_CLUSTER, ["BGL"], shared=1)
1938 self.GL.acquire(locking.LEVEL_NODE, ["n2"])
1939 self.assertRaises(AssertionError, self.GL.acquire,
1940 locking.LEVEL_NODE, ["n1"])
1941 self.assertRaises(AssertionError, self.GL.acquire,
1942 locking.LEVEL_NODEGROUP, ["g1"])
1943 self.assertRaises(AssertionError, self.GL.acquire,
1944 locking.LEVEL_INSTANCE, ["i2"])
1946 def testModifiableLevels(self):
1947 self.assertRaises(AssertionError, self.GL.add, locking.LEVEL_CLUSTER,
1949 self.GL.acquire(locking.LEVEL_CLUSTER, ["BGL"])
1950 self.GL.add(locking.LEVEL_INSTANCE, ["i4"])
1951 self.GL.remove(locking.LEVEL_INSTANCE, ["i3"])
1952 self.GL.remove(locking.LEVEL_INSTANCE, ["i1"])
1953 self.assertEqual(self.GL._names(locking.LEVEL_INSTANCE), set(["i2", "i4"]))
1954 self.GL.add(locking.LEVEL_NODE, ["n3"])
1955 self.GL.remove(locking.LEVEL_NODE, ["n1"])
1956 self.assertEqual(self.GL._names(locking.LEVEL_NODE), set(["n2", "n3"]))
1957 self.GL.add(locking.LEVEL_NODEGROUP, ["g3"])
1958 self.GL.remove(locking.LEVEL_NODEGROUP, ["g2"])
1959 self.GL.remove(locking.LEVEL_NODEGROUP, ["g1"])
1960 self.assertEqual(self.GL._names(locking.LEVEL_NODEGROUP), set(["g3"]))
1961 self.assertRaises(AssertionError, self.GL.remove, locking.LEVEL_CLUSTER,
1964 # Helper function to run as a thread that shared the BGL and then acquires
1965 # some locks at another level.
1966 def _doLock(self, level, names, shared):
1968 self.GL.acquire(locking.LEVEL_CLUSTER, ["BGL"], shared=1)
1969 self.GL.acquire(level, names, shared=shared)
1970 self.done.put("DONE")
1971 self.GL.release(level)
1972 self.GL.release(locking.LEVEL_CLUSTER)
1973 except errors.LockError:
1974 self.done.put("ERR")
1977 def testConcurrency(self):
1978 self.GL.acquire(locking.LEVEL_CLUSTER, ["BGL"], shared=1)
1979 self._addThread(target=self._doLock,
1980 args=(locking.LEVEL_INSTANCE, "i1", 1))
1982 self.assertEqual(self.done.get_nowait(), "DONE")
1983 self.GL.acquire(locking.LEVEL_INSTANCE, ["i3"])
1984 self._addThread(target=self._doLock,
1985 args=(locking.LEVEL_INSTANCE, "i1", 1))
1987 self.assertEqual(self.done.get_nowait(), "DONE")
1988 self._addThread(target=self._doLock,
1989 args=(locking.LEVEL_INSTANCE, "i3", 1))
1990 self.assertRaises(Queue.Empty, self.done.get_nowait)
1991 self.GL.release(locking.LEVEL_INSTANCE)
1993 self.assertEqual(self.done.get_nowait(), "DONE")
1994 self.GL.acquire(locking.LEVEL_INSTANCE, ["i2"], shared=1)
1995 self._addThread(target=self._doLock,
1996 args=(locking.LEVEL_INSTANCE, "i2", 1))
1998 self.assertEqual(self.done.get_nowait(), "DONE")
1999 self._addThread(target=self._doLock,
2000 args=(locking.LEVEL_INSTANCE, "i2", 0))
2001 self.assertRaises(Queue.Empty, self.done.get_nowait)
2002 self.GL.release(locking.LEVEL_INSTANCE)
2004 self.assertEqual(self.done.get(True, 1), "DONE")
2005 self.GL.release(locking.LEVEL_CLUSTER, ["BGL"])
2008 class TestLockMonitor(_ThreadedTestCase):
2010 _ThreadedTestCase.setUp(self)
2011 self.lm = locking.LockMonitor()
2013 def testSingleThread(self):
2016 for i in range(100):
2017 name = "TestLock%s" % i
2018 locks.append(locking.SharedLock(name, monitor=self.lm))
2020 self.assertEqual(len(self.lm._locks), len(locks))
2021 result = objects.QueryResponse.FromDict(self.lm.QueryLocks(["name"]))
2022 self.assertEqual(len(result.fields), 1)
2023 self.assertEqual(len(result.data), 100)
2028 # The garbage collector might needs some time
2031 raise utils.RetryAgain()
2033 utils.Retry(_CheckLocks, 0.1, 30.0)
2035 self.assertFalse(self.lm._locks)
2037 def testMultiThread(self):
2040 def _CreateLock(prev, next, name):
2042 locks.append(locking.SharedLock(name, monitor=self.lm))
2048 first = threading.Event()
2051 # Use a deterministic random generator
2052 for i in random.Random(4263).sample(range(100), 33):
2053 name = "MtTestLock%s" % i
2054 expnames.append(name)
2056 ev = threading.Event()
2057 self._addThread(target=_CreateLock, args=(prev, ev, name))
2064 # Check order in which locks were added
2065 self.assertEqual([i.name for i in locks], expnames)
2067 # Check query result
2068 result = self.lm.QueryLocks(["name", "mode", "owner", "pending"])
2069 self.assert_(isinstance(result, dict))
2070 response = objects.QueryResponse.FromDict(result)
2071 self.assertEqual(response.data,
2072 [[(constants.RS_NORMAL, name),
2073 (constants.RS_NORMAL, None),
2074 (constants.RS_NORMAL, None),
2075 (constants.RS_NORMAL, [])]
2076 for name in utils.NiceSort(expnames)])
2077 self.assertEqual(len(response.fields), 4)
2078 self.assertEqual(["name", "mode", "owner", "pending"],
2079 [fdef.name for fdef in response.fields])
2081 # Test exclusive acquire
2082 for tlock in locks[::4]:
2083 tlock.acquire(shared=0)
2085 def _GetExpResult(name):
2086 if tlock.name == name:
2087 return [(constants.RS_NORMAL, name),
2088 (constants.RS_NORMAL, "exclusive"),
2089 (constants.RS_NORMAL,
2090 [threading.currentThread().getName()]),
2091 (constants.RS_NORMAL, [])]
2092 return [(constants.RS_NORMAL, name),
2093 (constants.RS_NORMAL, None),
2094 (constants.RS_NORMAL, None),
2095 (constants.RS_NORMAL, [])]
2097 result = self.lm.QueryLocks(["name", "mode", "owner", "pending"])
2098 self.assertEqual(objects.QueryResponse.FromDict(result).data,
2099 [_GetExpResult(name)
2100 for name in utils.NiceSort(expnames)])
2104 # Test shared acquire
2105 def _Acquire(lock, shared, ev, notify):
2106 lock.acquire(shared=shared)
2113 for tlock1 in locks[::11]:
2114 for tlock2 in locks[::-15]:
2115 if tlock2 == tlock1:
2119 for tlock3 in locks[::10]:
2120 if tlock3 in (tlock2, tlock1):
2124 releaseev = threading.Event()
2130 ev = threading.Event()
2131 tthreads1.append(self._addThread(target=_Acquire,
2132 args=(tlock1, 1, releaseev, ev)))
2133 acquireev.append(ev)
2135 ev = threading.Event()
2136 tthread2 = self._addThread(target=_Acquire,
2137 args=(tlock2, 1, releaseev, ev))
2138 acquireev.append(ev)
2140 ev = threading.Event()
2141 tthread3 = self._addThread(target=_Acquire,
2142 args=(tlock3, 0, releaseev, ev))
2143 acquireev.append(ev)
2145 # Wait for all locks to be acquired
2149 # Check query result
2150 result = self.lm.QueryLocks(["name", "mode", "owner"])
2151 response = objects.QueryResponse.FromDict(result)
2152 for (name, mode, owner) in response.data:
2153 (name_status, name_value) = name
2154 (owner_status, owner_value) = owner
2156 self.assertEqual(name_status, constants.RS_NORMAL)
2157 self.assertEqual(owner_status, constants.RS_NORMAL)
2159 if name_value == tlock1.name:
2160 self.assertEqual(mode, (constants.RS_NORMAL, "shared"))
2161 self.assertEqual(set(owner_value),
2162 set(i.getName() for i in tthreads1))
2165 if name_value == tlock2.name:
2166 self.assertEqual(mode, (constants.RS_NORMAL, "shared"))
2167 self.assertEqual(owner_value, [tthread2.getName()])
2170 if name_value == tlock3.name:
2171 self.assertEqual(mode, (constants.RS_NORMAL, "exclusive"))
2172 self.assertEqual(owner_value, [tthread3.getName()])
2175 self.assert_(name_value in expnames)
2176 self.assertEqual(mode, (constants.RS_NORMAL, None))
2177 self.assert_(owner_value is None)
2179 # Release locks again
2184 result = self.lm.QueryLocks(["name", "mode", "owner"])
2185 self.assertEqual(objects.QueryResponse.FromDict(result).data,
2186 [[(constants.RS_NORMAL, name),
2187 (constants.RS_NORMAL, None),
2188 (constants.RS_NORMAL, None)]
2189 for name in utils.NiceSort(expnames)])
2191 def testDelete(self):
2192 lock = locking.SharedLock("TestLock", monitor=self.lm)
2194 self.assertEqual(len(self.lm._locks), 1)
2195 result = self.lm.QueryLocks(["name", "mode", "owner"])
2196 self.assertEqual(objects.QueryResponse.FromDict(result).data,
2197 [[(constants.RS_NORMAL, lock.name),
2198 (constants.RS_NORMAL, None),
2199 (constants.RS_NORMAL, None)]])
2203 result = self.lm.QueryLocks(["name", "mode", "owner"])
2204 self.assertEqual(objects.QueryResponse.FromDict(result).data,
2205 [[(constants.RS_NORMAL, lock.name),
2206 (constants.RS_NORMAL, "deleted"),
2207 (constants.RS_NORMAL, None)]])
2208 self.assertEqual(len(self.lm._locks), 1)
2210 def testPending(self):
2211 def _Acquire(lock, shared, prev, next):
2214 lock.acquire(shared=shared, test_notify=next.set)
2220 lock = locking.SharedLock("ExcLock", monitor=self.lm)
2222 for shared in [0, 1]:
2225 self.assertEqual(len(self.lm._locks), 1)
2226 result = self.lm.QueryLocks(["name", "mode", "owner"])
2227 self.assertEqual(objects.QueryResponse.FromDict(result).data,
2228 [[(constants.RS_NORMAL, lock.name),
2229 (constants.RS_NORMAL, "exclusive"),
2230 (constants.RS_NORMAL,
2231 [threading.currentThread().getName()])]])
2235 first = threading.Event()
2239 ev = threading.Event()
2240 threads.append(self._addThread(target=_Acquire,
2241 args=(lock, shared, prev, ev)))
2247 # Wait for last acquire to start waiting
2250 # NOTE: This works only because QueryLocks will acquire the
2251 # lock-internal lock again and won't be able to get the information
2252 # until it has the lock. By then the acquire should be registered in
2253 # SharedLock.__pending (otherwise it's a bug).
2255 # All acquires are waiting now
2257 pending = [("shared", utils.NiceSort(t.getName() for t in threads))]
2259 pending = [("exclusive", [t.getName()]) for t in threads]
2261 result = self.lm.QueryLocks(["name", "mode", "owner", "pending"])
2262 self.assertEqual(objects.QueryResponse.FromDict(result).data,
2263 [[(constants.RS_NORMAL, lock.name),
2264 (constants.RS_NORMAL, "exclusive"),
2265 (constants.RS_NORMAL,
2266 [threading.currentThread().getName()]),
2267 (constants.RS_NORMAL, pending)]])
2269 self.assertEqual(len(self.lm._locks), 1)
2275 # No pending acquires
2276 result = self.lm.QueryLocks(["name", "mode", "owner", "pending"])
2277 self.assertEqual(objects.QueryResponse.FromDict(result).data,
2278 [[(constants.RS_NORMAL, lock.name),
2279 (constants.RS_NORMAL, None),
2280 (constants.RS_NORMAL, None),
2281 (constants.RS_NORMAL, [])]])
2283 self.assertEqual(len(self.lm._locks), 1)
2285 def testDeleteAndRecreate(self):
2286 lname = "TestLock101923193"
2288 # Create some locks with the same name and keep all references
2289 locks = [locking.SharedLock(lname, monitor=self.lm)
2292 self.assertEqual(len(self.lm._locks), len(locks))
2294 result = self.lm.QueryLocks(["name", "mode", "owner"])
2295 self.assertEqual(objects.QueryResponse.FromDict(result).data,
2296 [[(constants.RS_NORMAL, lname),
2297 (constants.RS_NORMAL, None),
2298 (constants.RS_NORMAL, None)]] * 5)
2302 # Check information order
2303 result = self.lm.QueryLocks(["name", "mode", "owner"])
2304 self.assertEqual(objects.QueryResponse.FromDict(result).data,
2305 [[(constants.RS_NORMAL, lname),
2306 (constants.RS_NORMAL, None),
2307 (constants.RS_NORMAL, None)]] * 2 +
2308 [[(constants.RS_NORMAL, lname),
2309 (constants.RS_NORMAL, "deleted"),
2310 (constants.RS_NORMAL, None)]] +
2311 [[(constants.RS_NORMAL, lname),
2312 (constants.RS_NORMAL, None),
2313 (constants.RS_NORMAL, None)]] * 2)
2315 locks[1].acquire(shared=0)
2318 [(constants.RS_NORMAL, lname),
2319 (constants.RS_NORMAL, None),
2320 (constants.RS_NORMAL, None)],
2321 [(constants.RS_NORMAL, lname),
2322 (constants.RS_NORMAL, "exclusive"),
2323 (constants.RS_NORMAL, [threading.currentThread().getName()])],
2324 [(constants.RS_NORMAL, lname),
2325 (constants.RS_NORMAL, "deleted"),
2326 (constants.RS_NORMAL, None)],
2327 [(constants.RS_NORMAL, lname),
2328 (constants.RS_NORMAL, None),
2329 (constants.RS_NORMAL, None)],
2330 [(constants.RS_NORMAL, lname),
2331 (constants.RS_NORMAL, None),
2332 (constants.RS_NORMAL, None)],
2335 # Check information order
2336 result = self.lm.QueryLocks(["name", "mode", "owner"])
2337 self.assertEqual(objects.QueryResponse.FromDict(result).data, last_status)
2339 self.assertEqual(len(set(self.lm._locks.values())), len(locks))
2340 self.assertEqual(len(self.lm._locks), len(locks))
2342 # Check lock deletion
2343 for idx in range(len(locks)):
2345 assert gc.isenabled()
2347 self.assertEqual(len(self.lm._locks), len(locks))
2348 result = self.lm.QueryLocks(["name", "mode", "owner"])
2349 self.assertEqual(objects.QueryResponse.FromDict(result).data,
2350 last_status[idx + 1:])
2352 # All locks should have been deleted
2354 self.assertFalse(self.lm._locks)
2356 result = self.lm.QueryLocks(["name", "mode", "owner"])
2357 self.assertEqual(objects.QueryResponse.FromDict(result).data, [])
2363 def AddResult(self, *args):
2364 self._info.append(args)
2366 def CountPending(self):
2367 return len(self._info)
2369 def GetLockInfo(self, requested):
2370 (exp_requested, result) = self._info.pop(0)
2372 if exp_requested != requested:
2373 raise Exception("Requested information (%s) does not match"
2374 " expectations (%s)" % (requested, exp_requested))
2378 def testMultipleResults(self):
2379 fl1 = self._FakeLock()
2380 fl2 = self._FakeLock()
2382 self.lm.RegisterLock(fl1)
2383 self.lm.RegisterLock(fl2)
2386 for i in [fl1, fl2]:
2387 i.AddResult(set([query.LQ_MODE, query.LQ_OWNER]), [])
2388 result = self.lm.QueryLocks(["name", "mode", "owner"])
2389 self.assertEqual(objects.QueryResponse.FromDict(result).data, [])
2390 for i in [fl1, fl2]:
2391 self.assertEqual(i.CountPending(), 0)
2394 for fn in [lambda x: x, reversed, sorted]:
2395 fl1.AddResult(set(), list(fn([
2396 ("aaa", None, None, None),
2397 ("bbb", None, None, None),
2399 fl2.AddResult(set(), [])
2400 result = self.lm.QueryLocks(["name"])
2401 self.assertEqual(objects.QueryResponse.FromDict(result).data, [
2402 [(constants.RS_NORMAL, "aaa")],
2403 [(constants.RS_NORMAL, "bbb")],
2405 for i in [fl1, fl2]:
2406 self.assertEqual(i.CountPending(), 0)
2408 for fn2 in [lambda x: x, reversed, sorted]:
2409 fl1.AddResult(set([query.LQ_MODE]), list(fn([
2410 # Same name, but different information
2411 ("aaa", "mode0", None, None),
2412 ("aaa", "mode1", None, None),
2413 ("aaa", "mode2", None, None),
2414 ("aaa", "mode3", None, None),
2416 fl2.AddResult(set([query.LQ_MODE]), [
2417 ("zzz", "end", None, None),
2418 ("000", "start", None, None),
2420 ("aaa", "b200", None, None),
2421 ("aaa", "b300", None, None),
2423 result = self.lm.QueryLocks(["name", "mode"])
2424 self.assertEqual(objects.QueryResponse.FromDict(result).data, [
2425 [(constants.RS_NORMAL, "000"), (constants.RS_NORMAL, "start")],
2427 # Name is the same, so order must be equal to incoming order
2428 [(constants.RS_NORMAL, "aaa"), (constants.RS_NORMAL, "mode0")],
2429 [(constants.RS_NORMAL, "aaa"), (constants.RS_NORMAL, "mode1")],
2430 [(constants.RS_NORMAL, "aaa"), (constants.RS_NORMAL, "mode2")],
2431 [(constants.RS_NORMAL, "aaa"), (constants.RS_NORMAL, "mode3")],
2433 [(constants.RS_NORMAL, "aaa"), (constants.RS_NORMAL, "b200")],
2434 [(constants.RS_NORMAL, "aaa"), (constants.RS_NORMAL, "b300")],
2436 [(constants.RS_NORMAL, "zzz"), (constants.RS_NORMAL, "end")],
2438 for i in [fl1, fl2]:
2439 self.assertEqual(i.CountPending(), 0)
2442 if __name__ == "__main__":
2443 testutils.GanetiTestProgram()