Statistics
| Branch: | Tag: | Revision:

root / test / ganeti.locking_unittest.py @ 8716b1db

History | View | Annotate | Download (83.7 kB)

1
#!/usr/bin/python
2
#
3

    
4
# Copyright (C) 2006, 2007, 2010 Google Inc.
5
#
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.
10
#
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.
15
#
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
19
# 02110-1301, USA.
20

    
21

    
22
"""Script for unittesting the locking module"""
23

    
24

    
25
import os
26
import unittest
27
import time
28
import Queue
29
import threading
30
import random
31
import gc
32
import itertools
33

    
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
41

    
42
import testutils
43

    
44

    
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")
48

    
49
#: List for looping tests
50
ITERATIONS = range(8)
51

    
52

    
53
def _Repeat(fn):
54
  """Decorator for executing a function many times"""
55
  def wrapper(*args, **kwargs):
56
    for i in ITERATIONS:
57
      fn(*args, **kwargs)
58
  return wrapper
59

    
60

    
61
def SafeSleep(duration):
62
  start = time.time()
63
  while True:
64
    delay = start + duration - time.time()
65
    if delay <= 0.0:
66
      break
67
    time.sleep(delay)
68

    
69

    
70
class _ThreadedTestCase(unittest.TestCase):
71
  """Test class that supports adding/waiting on threads"""
72
  def setUp(self):
73
    unittest.TestCase.setUp(self)
74
    self.done = Queue.Queue(0)
75
    self.threads = []
76

    
77
  def _addThread(self, *args, **kwargs):
78
    """Create and remember a new thread"""
79
    t = threading.Thread(*args, **kwargs)
80
    self.threads.append(t)
81
    t.start()
82
    return t
83

    
84
  def _waitThreads(self):
85
    """Wait for all our threads to finish"""
86
    for t in self.threads:
87
      t.join(60)
88
      self.failIf(t.isAlive())
89
    self.threads = []
90

    
91

    
92
class _ConditionTestCase(_ThreadedTestCase):
93
  """Common test case for conditions"""
94

    
95
  def setUp(self, cls):
96
    _ThreadedTestCase.setUp(self)
97
    self.lock = threading.Lock()
98
    self.cond = cls(self.lock)
99

    
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)
104

    
105
    self.cond.acquire()
106
    self.assert_(self.cond._is_owned())
107
    self.cond.notifyAll()
108
    self.assert_(self.cond._is_owned())
109
    self.cond.release()
110

    
111
    self.assertFalse(self.cond._is_owned())
112
    self.assertRaises(RuntimeError, self.cond.wait, None)
113
    self.assertRaises(RuntimeError, self.cond.notifyAll)
114

    
115
  def _testNotification(self):
116
    def _NotifyAll():
117
      self.done.put("NE")
118
      self.cond.acquire()
119
      self.done.put("NA")
120
      self.cond.notifyAll()
121
      self.done.put("NN")
122
      self.cond.release()
123

    
124
    self.cond.acquire()
125
    self._addThread(target=_NotifyAll)
126
    self.assertEqual(self.done.get(True, 1), "NE")
127
    self.assertRaises(Queue.Empty, self.done.get_nowait)
128
    self.cond.wait(None)
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())
132
    self.cond.release()
133
    self.assertFalse(self.cond._is_owned())
134

    
135

    
136
class TestSingleNotifyPipeCondition(_ConditionTestCase):
137
  """SingleNotifyPipeCondition tests"""
138

    
139
  def setUp(self):
140
    _ConditionTestCase.setUp(self, locking.SingleNotifyPipeCondition)
141

    
142
  def testAcquireRelease(self):
143
    self._testAcquireRelease()
144

    
145
  def testNotification(self):
146
    self._testNotification()
147

    
148
  def testWaitReuse(self):
149
    self.cond.acquire()
150
    self.cond.wait(0)
151
    self.cond.wait(0.1)
152
    self.cond.release()
153

    
154
  def testNoNotifyReuse(self):
155
    self.cond.acquire()
156
    self.cond.notifyAll()
157
    self.assertRaises(RuntimeError, self.cond.wait, None)
158
    self.assertRaises(RuntimeError, self.cond.notifyAll)
159
    self.cond.release()
160

    
161

    
162
class TestPipeCondition(_ConditionTestCase):
163
  """PipeCondition tests"""
164

    
165
  def setUp(self):
166
    _ConditionTestCase.setUp(self, locking.PipeCondition)
167

    
168
  def testAcquireRelease(self):
169
    self._testAcquireRelease()
170

    
171
  def testNotification(self):
172
    self._testNotification()
173

    
174
  def _TestWait(self, fn):
175
    threads = [
176
      self._addThread(target=fn),
177
      self._addThread(target=fn),
178
      self._addThread(target=fn),
179
      ]
180

    
181
    # Wait for threads to be waiting
182
    for _ in threads:
183
      self.assertEqual(self.done.get(True, 1), "A")
184

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

    
187
    self.cond.acquire()
188
    self.assertEqual(len(self.cond._waiters), 3)
189
    self.assertEqual(self.cond._waiters, set(threads))
190

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

    
194
    # This new thread can't acquire the lock, and thus call wait, before we
195
    # release it
196
    self._addThread(target=fn)
197
    self.cond.notifyAll()
198
    self.assertRaises(Queue.Empty, self.done.get_nowait)
199
    self.cond.release()
200

    
201
    # We should now get 3 W and 1 A (for the new thread) in whatever order
202
    w = 0
203
    a = 0
204
    for i in range(4):
205
      got = self.done.get(True, 1)
206
      if got == "W":
207
        w += 1
208
      elif got == "A":
209
        a += 1
210
      else:
211
        self.fail("Got %s on the done queue" % got)
212

    
213
    self.assertEqual(w, 3)
214
    self.assertEqual(a, 1)
215

    
216
    self.cond.acquire()
217
    self.cond.notifyAll()
218
    self.cond.release()
219
    self._waitThreads()
220
    self.assertEqual(self.done.get_nowait(), "W")
221
    self.assertRaises(Queue.Empty, self.done.get_nowait)
222

    
223
  def testBlockingWait(self):
224
    def _BlockingWait():
225
      self.cond.acquire()
226
      self.done.put("A")
227
      self.cond.wait(None)
228
      self.cond.release()
229
      self.done.put("W")
230

    
231
    self._TestWait(_BlockingWait)
232

    
233
  def testLongTimeoutWait(self):
234
    def _Helper():
235
      self.cond.acquire()
236
      self.done.put("A")
237
      self.cond.wait(15.0)
238
      self.cond.release()
239
      self.done.put("W")
240

    
241
    self._TestWait(_Helper)
242

    
243
  def _TimeoutWait(self, timeout, check):
244
    self.cond.acquire()
245
    self.cond.wait(timeout)
246
    self.cond.release()
247
    self.done.put(check)
248

    
249
  def testShortTimeoutWait(self):
250
    self._addThread(target=self._TimeoutWait, args=(0.1, "T1"))
251
    self._addThread(target=self._TimeoutWait, args=(0.1, "T1"))
252
    self._waitThreads()
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)
256

    
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"))
261
    self._waitThreads()
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)
266

    
267

    
268
class TestSharedLock(_ThreadedTestCase):
269
  """SharedLock tests"""
270

    
271
  def setUp(self):
272
    _ThreadedTestCase.setUp(self)
273
    self.sl = locking.SharedLock("TestSharedLock")
274

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

    
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))
284
    self.sl.release()
285
    self.assertFalse(self.sl.is_owned())
286
    self.sl.acquire()
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))
290
    self.sl.release()
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))
296
    self.sl.release()
297
    self.assertFalse(self.sl.is_owned())
298

    
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))
302
    self.sl.release()
303
    self.assert_(self.sl.acquire())
304
    self.sl.release()
305

    
306
  def testDoubleLockingStoE(self):
307
    self.sl.acquire(shared=1)
308
    self.assertRaises(AssertionError, self.sl.acquire)
309

    
310
  def testDoubleLockingEtoS(self):
311
    self.sl.acquire()
312
    self.assertRaises(AssertionError, self.sl.acquire, shared=1)
313

    
314
  def testDoubleLockingStoS(self):
315
    self.sl.acquire(shared=1)
316
    self.assertRaises(AssertionError, self.sl.acquire, shared=1)
317

    
318
  def testDoubleLockingEtoE(self):
319
    self.sl.acquire()
320
    self.assertRaises(AssertionError, self.sl.acquire)
321

    
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):
325
    try:
326
      self.sl.acquire(shared=1)
327
      self.done.put("SHR")
328
      self.sl.release()
329
    except errors.LockError:
330
      self.done.put("ERR")
331

    
332
  def _doItExclusive(self):
333
    try:
334
      self.sl.acquire()
335
      self.done.put("EXC")
336
      self.sl.release()
337
    except errors.LockError:
338
      self.done.put("ERR")
339

    
340
  def _doItDelete(self):
341
    try:
342
      self.sl.delete()
343
      self.done.put("DEL")
344
    except errors.LockError:
345
      self.done.put("ERR")
346

    
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))
351
    self.sl.release()
352

    
353
  @_Repeat
354
  def testExclusiveBlocksExclusive(self):
355
    self.sl.acquire()
356
    self._addThread(target=self._doItExclusive)
357
    self.assertRaises(Queue.Empty, self.done.get_nowait)
358
    self.sl.release()
359
    self._waitThreads()
360
    self.failUnlessEqual(self.done.get_nowait(), "EXC")
361

    
362
  @_Repeat
363
  def testExclusiveBlocksDelete(self):
364
    self.sl.acquire()
365
    self._addThread(target=self._doItDelete)
366
    self.assertRaises(Queue.Empty, self.done.get_nowait)
367
    self.sl.release()
368
    self._waitThreads()
369
    self.failUnlessEqual(self.done.get_nowait(), "DEL")
370
    self.sl = locking.SharedLock(self.sl.name)
371

    
372
  @_Repeat
373
  def testExclusiveBlocksSharer(self):
374
    self.sl.acquire()
375
    self._addThread(target=self._doItSharer)
376
    self.assertRaises(Queue.Empty, self.done.get_nowait)
377
    self.sl.release()
378
    self._waitThreads()
379
    self.failUnlessEqual(self.done.get_nowait(), "SHR")
380

    
381
  @_Repeat
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)
386
    self.sl.release()
387
    self._waitThreads()
388
    self.failUnlessEqual(self.done.get_nowait(), "EXC")
389

    
390
  @_Repeat
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)
395
    self.sl.release()
396
    self._waitThreads()
397
    self.failUnlessEqual(self.done.get_nowait(), "DEL")
398
    self.sl = locking.SharedLock(self.sl.name)
399

    
400
  @_Repeat
401
  def testWaitingExclusiveBlocksSharer(self):
402
    """SKIPPED testWaitingExclusiveBlockSharer"""
403
    return
404

    
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)
412
    self.sl.release()
413
    self._waitThreads()
414
    # The exclusive passed before
415
    self.failUnlessEqual(self.done.get_nowait(), "EXC")
416
    self.failUnlessEqual(self.done.get_nowait(), "SHR")
417

    
418
  @_Repeat
419
  def testWaitingSharerBlocksExclusive(self):
420
    """SKIPPED testWaitingSharerBlocksExclusive"""
421
    return
422

    
423
    self.sl.acquire()
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)
430
    self.sl.release()
431
    self._waitThreads()
432
    # The sharer passed before
433
    self.assertEqual(self.done.get_nowait(), "SHR")
434
    self.assertEqual(self.done.get_nowait(), "EXC")
435

    
436
  def testDelete(self):
437
    self.sl.delete()
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)
441

    
442
  def testDeleteTimeout(self):
443
    self.assertTrue(self.sl.delete(timeout=60))
444

    
445
  def testDeleteTimeoutFail(self):
446
    ready = threading.Event()
447
    finish = threading.Event()
448

    
449
    def fn():
450
      self.sl.acquire(shared=0)
451
      ready.set()
452

    
453
      finish.wait()
454
      self.sl.release()
455

    
456
    self._addThread(target=fn)
457
    ready.wait()
458

    
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))
462

    
463
    finish.set()
464
    self._waitThreads()
465

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

    
469
  def testNoDeleteIfSharer(self):
470
    self.sl.acquire(shared=1)
471
    self.assertRaises(AssertionError, self.sl.delete)
472

    
473
  @_Repeat
474
  def testDeletePendingSharersExclusiveDelete(self):
475
    self.sl.acquire()
476
    self._addThread(target=self._doItSharer)
477
    self._addThread(target=self._doItSharer)
478
    self._addThread(target=self._doItExclusive)
479
    self._addThread(target=self._doItDelete)
480
    self.sl.delete()
481
    self._waitThreads()
482
    # The threads who were pending return ERR
483
    for _ in range(4):
484
      self.assertEqual(self.done.get_nowait(), "ERR")
485
    self.sl = locking.SharedLock(self.sl.name)
486

    
487
  @_Repeat
488
  def testDeletePendingDeleteExclusiveSharers(self):
489
    self.sl.acquire()
490
    self._addThread(target=self._doItDelete)
491
    self._addThread(target=self._doItExclusive)
492
    self._addThread(target=self._doItSharer)
493
    self._addThread(target=self._doItSharer)
494
    self.sl.delete()
495
    self._waitThreads()
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)
502

    
503
  @_Repeat
504
  def testExclusiveAcquireTimeout(self):
505
    for shared in [0, 1]:
506
      on_queue = threading.Event()
507
      release_exclusive = threading.Event()
508

    
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")
514
        self.sl.release()
515

    
516
      # Start thread to hold lock in exclusive mode
517
      self._addThread(target=_LockExclusive)
518

    
519
      # Wait for wait to begin
520
      self.assertEqual(self.done.get(timeout=60), "A: start wait")
521

    
522
      # Wait up to 60s to get lock, but release exclusive lock as soon as we're
523
      # on the queue
524
      self.failUnless(self.sl.acquire(shared=shared, timeout=60,
525
                                      test_notify=release_exclusive.set))
526

    
527
      self.done.put("got 2nd")
528
      self.sl.release()
529

    
530
      self._waitThreads()
531

    
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)
535

    
536
  @_Repeat
537
  def testAcquireExpiringTimeout(self):
538
    def _AcquireWithTimeout(shared, timeout):
539
      if not self.sl.acquire(shared=shared, timeout=timeout):
540
        self.done.put("timeout")
541

    
542
    for shared in [0, 1]:
543
      # Lock exclusively
544
      self.sl.acquire()
545

    
546
      # Start shared acquires with timeout between 0 and 20 ms
547
      for i in range(11):
548
        self._addThread(target=_AcquireWithTimeout,
549
                        args=(shared, i * 2.0 / 1000.0))
550

    
551
      # Wait for threads to finish (makes sure the acquire timeout expires
552
      # before releasing the lock)
553
      self._waitThreads()
554

    
555
      # Release lock
556
      self.sl.release()
557

    
558
      for _ in range(11):
559
        self.assertEqual(self.done.get_nowait(), "timeout")
560

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

    
563
  @_Repeat
564
  def testSharedSkipExclusiveAcquires(self):
565
    # Tests whether shared acquires jump in front of exclusive acquires in the
566
    # queue.
567

    
568
    def _Acquire(shared, name, notify_ev, wait_ev):
569
      if notify_ev:
570
        notify_fn = notify_ev.set
571
      else:
572
        notify_fn = None
573

    
574
      if wait_ev:
575
        wait_ev.wait()
576

    
577
      if not self.sl.acquire(shared=shared, test_notify=notify_fn):
578
        return
579

    
580
      self.done.put(name)
581
      self.sl.release()
582

    
583
    # Get exclusive lock while we fill the queue
584
    self.sl.acquire()
585

    
586
    shrcnt1 = 5
587
    shrcnt2 = 7
588
    shrcnt3 = 9
589
    shrcnt4 = 2
590

    
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")])
599

    
600
    ev_cur = None
601
    ev_prev = None
602

    
603
    for args in acquires:
604
      ev_cur = threading.Event()
605
      self._addThread(target=_Acquire, args=args + (ev_cur, ev_prev))
606
      ev_prev = ev_cur
607

    
608
    # Wait for last acquire to start
609
    ev_prev.wait()
610

    
611
    # Expect 6 pending exclusive acquires and 1 for all shared acquires
612
    # together
613
    self.assertEqual(self.sl._count_pending(), 7)
614

    
615
    # Release exclusive lock and wait
616
    self.sl.release()
617

    
618
    self._waitThreads()
619

    
620
    # Check sequence
621
    for _ in range(shrcnt1 + shrcnt2 + shrcnt3 + shrcnt4):
622
      # Shared locks aren't guaranteed to be notified in order, but they'll be
623
      # first
624
      tmp = self.done.get_nowait()
625
      if tmp == "shared 1":
626
        shrcnt1 -= 1
627
      elif tmp == "shared 2":
628
        shrcnt2 -= 1
629
      elif tmp == "shared 3":
630
        shrcnt3 -= 1
631
      elif tmp == "shared 4":
632
        shrcnt4 -= 1
633
    self.assertEqual(shrcnt1, 0)
634
    self.assertEqual(shrcnt2, 0)
635
    self.assertEqual(shrcnt3, 0)
636
    self.assertEqual(shrcnt3, 0)
637

    
638
    for _ in range(3):
639
      self.assertEqual(self.done.get_nowait(), "exclusive 1")
640

    
641
    for _ in range(3):
642
      self.assertEqual(self.done.get_nowait(), "exclusive 2")
643

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

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

    
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))
655
    self.sl.release()
656

    
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))
662
    self.sl.release()
663

    
664
  @_Repeat
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))
669
      ev_got.set()
670
      ev_downgrade.wait()
671
      self.assertTrue(self.sl.is_owned(shared=0))
672
      self.assertTrue(self.sl.downgrade())
673
      self.assertTrue(self.sl.is_owned(shared=1))
674
      ev_release.wait()
675
      self.assertTrue(self.sl.is_owned(shared=1))
676
      self.sl.release()
677

    
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))
681
      ev_release.wait()
682
      self.assertTrue(self.sl.is_owned(shared=0))
683
      self.sl.release()
684

    
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))
688
      ev_got.set()
689
      ev_release.wait()
690
      self.assertTrue(self.sl.is_owned(shared=1))
691
      self.sl.release()
692

    
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,
699
                                     ev_release_excl1))
700
    ev_got_excl1.wait()
701

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

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

    
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]
717

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

    
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))])
730

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

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

    
738
    # Check lock information again
739
    self.assertEqual(self.sl.GetLockInfo(set([query.LQ_MODE,
740
                                              query.LQ_PENDING])),
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]))
746

    
747
    ev_release_excl1.set()
748
    ev_release_excl2.set()
749
    ev_release_shared.set()
750

    
751
    self._waitThreads()
752

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

    
757
  @_Repeat
758
  def testMixedAcquireTimeout(self):
759
    sync = threading.Event()
760

    
761
    def _AcquireShared(ev):
762
      if not self.sl.acquire(shared=1, timeout=None):
763
        return
764

    
765
      self.done.put("shared")
766

    
767
      # Notify main thread
768
      ev.set()
769

    
770
      # Wait for notification from main thread
771
      sync.wait()
772

    
773
      # Release lock
774
      self.sl.release()
775

    
776
    acquires = []
777
    for _ in range(3):
778
      ev = threading.Event()
779
      self._addThread(target=_AcquireShared, args=(ev, ))
780
      acquires.append(ev)
781

    
782
    # Wait for all acquires to finish
783
    for i in acquires:
784
      i.wait()
785

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

    
788
    # Try to get exclusive lock
789
    self.failIf(self.sl.acquire(shared=0, timeout=0.02))
790

    
791
    # Acquire exclusive without timeout
792
    exclsync = threading.Event()
793
    exclev = threading.Event()
794

    
795
    def _AcquireExclusive():
796
      if not self.sl.acquire(shared=0):
797
        return
798

    
799
      self.done.put("exclusive")
800

    
801
      # Notify main thread
802
      exclev.set()
803

    
804
      # Wait for notification from main thread
805
      exclsync.wait()
806

    
807
      self.sl.release()
808

    
809
    self._addThread(target=_AcquireExclusive)
810

    
811
    # Try to get exclusive lock
812
    self.failIf(self.sl.acquire(shared=0, timeout=0.02))
813

    
814
    # Make all shared holders release their locks
815
    sync.set()
816

    
817
    # Wait for exclusive acquire to succeed
818
    exclev.wait()
819

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

    
822
    # Try to get exclusive lock
823
    self.failIf(self.sl.acquire(shared=0, timeout=0.02))
824

    
825
    def _AcquireSharedSimple():
826
      if self.sl.acquire(shared=1, timeout=None):
827
        self.done.put("shared2")
828
        self.sl.release()
829

    
830
    for _ in range(10):
831
      self._addThread(target=_AcquireSharedSimple)
832

    
833
    # Tell exclusive lock to release
834
    exclsync.set()
835

    
836
    # Wait for everything to finish
837
    self._waitThreads()
838

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

    
841
    # Check sequence
842
    for _ in range(3):
843
      self.assertEqual(self.done.get_nowait(), "shared")
844

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

    
847
    for _ in range(10):
848
      self.assertEqual(self.done.get_nowait(), "shared2")
849

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

    
852
  def testPriority(self):
853
    # Acquire in exclusive mode
854
    self.assert_(self.sl.acquire(shared=0))
855

    
856
    # Queue acquires
857
    def _Acquire(prev, next, shared, priority, result):
858
      prev.wait()
859
      self.sl.acquire(shared=shared, priority=priority, test_notify=next.set)
860
      try:
861
        self.done.put(result)
862
      finally:
863
        self.sl.release()
864

    
865
    counter = itertools.count(0)
866
    priorities = range(-20, 30)
867
    first = threading.Event()
868
    prev = first
869

    
870
    # Data structure:
871
    # {
872
    #   priority:
873
    #     [(shared/exclusive, set(acquire names), set(pending threads)),
874
    #      (shared/exclusive, ...),
875
    #      ...,
876
    #     ],
877
    # }
878
    perprio = {}
879

    
880
    # References shared acquire per priority in L{perprio}. Data structure:
881
    # {
882
    #   priority: (shared=1, set(acquire names), set(pending threads)),
883
    # }
884
    prioshared = {}
885

    
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)]:
890
        modes = [0, 1]
891
        rnd.shuffle(modes)
892
        for shared in modes:
893
          # Unique name
894
          acqname = "%s/shr=%s/prio=%s" % (counter.next(), shared, priority)
895

    
896
          ev = threading.Event()
897
          thread = self._addThread(target=_Acquire,
898
                                   args=(prev, ev, shared, priority, acqname))
899
          prev = ev
900

    
901
          # Record expected aqcuire, see above for structure
902
          data = (shared, set([acqname]), set([thread]))
903
          priolist = perprio.setdefault(priority, [])
904
          if shared:
905
            priosh = prioshared.get(priority, None)
906
            if priosh:
907
              # Shared acquires are merged
908
              for i, j in zip(priosh[1:], data[1:]):
909
                i.update(j)
910
              assert data[0] == priosh[0]
911
            else:
912
              prioshared[priority] = data
913
              priolist.append(data)
914
          else:
915
            priolist.append(data)
916

    
917
    # Start all acquires and wait for them
918
    first.set()
919
    prev.wait()
920

    
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)])
927

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

    
931
    # Let threads acquire the lock
932
    self.sl.release()
933

    
934
    # Wait for everything to finish
935
    self._waitThreads()
936

    
937
    self.assert_(self.sl._check_empty())
938

    
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
943
        # acquires only one.
944
        while names:
945
          names.remove(self.done.get_nowait())
946
      self.assertFalse(compat.any(names for (_, names, _) in acquires))
947

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

    
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)
954

    
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])
962

    
963
  class _FakeTimeForSpuriousNotifications:
964
    def __init__(self, now, check_end):
965
      self.now = now
966
      self.check_end = check_end
967

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

    
971
    def time(self):
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()
976

    
977
      self.check_end(self.now)
978

    
979
      return self.now
980

    
981
  @_Repeat
982
  def testAcquireTimeoutWithSpuriousNotifications(self):
983
    ready = threading.Event()
984
    locked = threading.Event()
985
    req = Queue.Queue(0)
986

    
987
    epoch = 4000.0
988
    timeout = 60.0
989

    
990
    def check_end(now):
991
      self.assertFalse(locked.isSet())
992

    
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)))
996

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

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

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

    
1004
    def fn():
1005
      self.assertTrue(sl.acquire(shared=0, timeout=timeout,
1006
                                 test_notify=ready.set))
1007
      locked.set()
1008
      sl.release()
1009
      self.done.put("success")
1010

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

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

    
1018
    # Wait for separate thread to ask for another notification
1019
    count = 0
1020
    while req.get():
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()
1024
      count += 1
1025

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

    
1028
    self.assertFalse(locked.isSet())
1029

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

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

    
1036
    self._waitThreads()
1037

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

    
1041

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

    
1045
  def setUp(self):
1046
    _ThreadedTestCase.setUp(self)
1047
    self.sl = locking.SharedLock("TestSharedLockInCondition")
1048
    self.setCondition()
1049

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

    
1053
  def testKeepMode(self):
1054
    self.cond.acquire(shared=1)
1055
    self.assert_(self.sl.is_owned(shared=1))
1056
    self.cond.wait(0)
1057
    self.assert_(self.sl.is_owned(shared=1))
1058
    self.cond.release()
1059
    self.cond.acquire(shared=0)
1060
    self.assert_(self.sl.is_owned(shared=0))
1061
    self.cond.wait(0)
1062
    self.assert_(self.sl.is_owned(shared=0))
1063
    self.cond.release()
1064

    
1065

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

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

    
1072

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

    
1076
  def setUp(self):
1077
    _ThreadedTestCase.setUp(self)
1078

    
1079
  @locking.ssynchronized(_decoratorlock)
1080
  def _doItExclusive(self):
1081
    self.assert_(_decoratorlock.is_owned())
1082
    self.done.put("EXC")
1083

    
1084
  @locking.ssynchronized(_decoratorlock, shared=1)
1085
  def _doItSharer(self):
1086
    self.assert_(_decoratorlock.is_owned(shared=1))
1087
    self.done.put("SHR")
1088

    
1089
  def testDecoratedFunctions(self):
1090
    self._doItExclusive()
1091
    self.assertFalse(_decoratorlock.is_owned())
1092
    self._doItSharer()
1093
    self.assertFalse(_decoratorlock.is_owned())
1094

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

    
1101
  @_Repeat
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()
1108
    self._waitThreads()
1109
    self.failUnlessEqual(self.done.get_nowait(), "EXC")
1110

    
1111
  @_Repeat
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()
1117
    self._waitThreads()
1118
    self.failUnlessEqual(self.done.get_nowait(), "SHR")
1119

    
1120
  @_Repeat
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()
1126
    self._waitThreads()
1127
    self.failUnlessEqual(self.done.get_nowait(), "EXC")
1128

    
1129

    
1130
class TestLockSet(_ThreadedTestCase):
1131
  """LockSet tests"""
1132

    
1133
  def setUp(self):
1134
    _ThreadedTestCase.setUp(self)
1135
    self._setUpLS()
1136

    
1137
  def _setUpLS(self):
1138
    """Helper to (re)initialize the lock set"""
1139
    self.resources = ["one", "two", "three"]
1140
    self.ls = locking.LockSet(self.resources, "TestLockSet")
1141

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

    
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",
1151
                                           shared=shared))
1152

    
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, "")
1163
    self.ls.release()
1164
    self.assertFalse(self.ls.check_owned([]))
1165
    self.assertFalse(self.ls.check_owned("one"))
1166

    
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))
1174
    self.ls.release()
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"]))
1179
    self.ls.release()
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"]))
1194
    self.ls.release()
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"]))
1198
    self.ls.release()
1199
    self.assertEquals(self.ls.list_owned(), set())
1200
    for name in self.ls._names():
1201
      self.assertFalse(self.ls.check_owned(name))
1202

    
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"])
1208
    self.ls.release()
1209
    self.ls.acquire(["one", "three"])
1210
    self.ls.release("one")
1211
    self.assertRaises(AssertionError, self.ls.acquire, ["two"])
1212
    self.ls.release("three")
1213

    
1214
  def testNoWrongRelease(self):
1215
    self.assertRaises(AssertionError, self.ls.release)
1216
    self.ls.acquire("one")
1217
    self.assertRaises(AssertionError, self.ls.release, "two")
1218

    
1219
  def testAddRemove(self):
1220
    self.ls.add("four")
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")
1238
    self.ls.release()
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())
1243
    self.ls.add("nine")
1244
    self.assert_("nine" in self.ls._names())
1245
    self.assert_("nine" not in self.ls.list_owned())
1246
    self.ls.release()
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())
1255

    
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"]),
1261
                      ["two", "three"])
1262

    
1263
  def testNoDoubleAdd(self):
1264
    self.assertRaises(errors.LockError, self.ls.add, "two")
1265
    self.ls.add("four")
1266
    self.assertRaises(errors.LockError, self.ls.add, "four")
1267

    
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")
1274

    
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"))
1284
    self.ls.release()
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")
1289
    self.ls.release()
1290

    
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"]))
1296

    
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)
1308

    
1309
  def _doLockSet(self, names, shared):
1310
    try:
1311
      self.ls.acquire(names, shared=shared)
1312
      self.done.put("DONE")
1313
      self.ls.release()
1314
    except errors.LockError:
1315
      self.done.put("ERR")
1316

    
1317
  def _doAddSet(self, names):
1318
    try:
1319
      self.ls.add(names, acquired=1)
1320
      self.done.put("DONE")
1321
      self.ls.release()
1322
    except errors.LockError:
1323
      self.done.put("ERR")
1324

    
1325
  def _doRemoveSet(self, names):
1326
    self.done.put(self.ls.remove(names))
1327

    
1328
  @_Repeat
1329
  def testConcurrentSharedAcquire(self):
1330
    self.ls.acquire(["one", "two"], shared=1)
1331
    self._addThread(target=self._doLockSet, args=(["one", "two"], 1))
1332
    self._waitThreads()
1333
    self.assertEqual(self.done.get_nowait(), "DONE")
1334
    self._addThread(target=self._doLockSet, args=(["one", "two", "three"], 1))
1335
    self._waitThreads()
1336
    self.assertEqual(self.done.get_nowait(), "DONE")
1337
    self._addThread(target=self._doLockSet, args=("three", 1))
1338
    self._waitThreads()
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)
1343
    self.ls.release()
1344
    self._waitThreads()
1345
    self.assertEqual(self.done.get_nowait(), "DONE")
1346
    self.assertEqual(self.done.get_nowait(), "DONE")
1347

    
1348
  @_Repeat
1349
  def testConcurrentExclusiveAcquire(self):
1350
    self.ls.acquire(["one", "two"])
1351
    self._addThread(target=self._doLockSet, args=("three", 1))
1352
    self._waitThreads()
1353
    self.assertEqual(self.done.get_nowait(), "DONE")
1354
    self._addThread(target=self._doLockSet, args=("three", 0))
1355
    self._waitThreads()
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)
1365
    self.ls.release()
1366
    self._waitThreads()
1367
    for _ in range(6):
1368
      self.failUnlessEqual(self.done.get_nowait(), "DONE")
1369

    
1370
  @_Repeat
1371
  def testSimpleAcquireTimeoutExpiring(self):
1372
    names = sorted(self.ls._names())
1373
    self.assert_(len(names) >= 3)
1374

    
1375
    # Get name of first lock
1376
    first = names[0]
1377

    
1378
    # Get name of last lock
1379
    last = names.pop()
1380

    
1381
    checks = [
1382
      # Block first and try to lock it again
1383
      (first, first),
1384

    
1385
      # Block last and try to lock all locks
1386
      (None, first),
1387

    
1388
      # Block last and try to lock it again
1389
      (last, last),
1390
      ]
1391

    
1392
    for (wanted, block) in checks:
1393
      # Lock in exclusive mode
1394
      self.assert_(self.ls.acquire(block, shared=0))
1395

    
1396
      def _AcquireOne():
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)
1399
        if acquired:
1400
          self.done.put("acquired")
1401
          self.ls.release()
1402
        else:
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")
1407

    
1408
      self._addThread(target=_AcquireOne)
1409

    
1410
      # Wait for timeout in thread to expire
1411
      self._waitThreads()
1412

    
1413
      # Release exclusive lock again
1414
      self.ls.release()
1415

    
1416
      self.assertEqual(self.done.get_nowait(), "not acquired")
1417
      self.assertRaises(Queue.Empty, self.done.get_nowait)
1418

    
1419
  @_Repeat
1420
  def testDelayedAndExpiringLockAcquire(self):
1421
    self._setUpLS()
1422
    self.ls.add(["five", "six", "seven", "eight", "nine"])
1423

    
1424
    for expire in (False, True):
1425
      names = sorted(self.ls._names())
1426
      self.assertEqual(len(names), 8)
1427

    
1428
      lock_ev = dict([(i, threading.Event()) for i in names])
1429

    
1430
      # Lock all in exclusive mode
1431
      self.assert_(self.ls.acquire(names, shared=0))
1432

    
1433
      if expire:
1434
        # We'll wait at least 300ms per lock
1435
        lockwait = len(names) * [0.3]
1436

    
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
1440
      else:
1441
        # This should finish rather quickly
1442
        lockwait = None
1443
        lockall_timeout = len(names) * 5.0
1444

    
1445
      def _LockAll():
1446
        def acquire_notification(name):
1447
          if not expire:
1448
            self.done.put("getting %s" % name)
1449

    
1450
          # Kick next lock
1451
          lock_ev[name].set()
1452

    
1453
        if self.ls.acquire(names, shared=0, timeout=lockall_timeout,
1454
                           test_notify=acquire_notification):
1455
          self.done.put("got all")
1456
          self.ls.release()
1457
        else:
1458
          self.done.put("timeout on all")
1459

    
1460
        # Notify all locks
1461
        for ev in lock_ev.values():
1462
          ev.set()
1463

    
1464
      t = self._addThread(target=_LockAll)
1465

    
1466
      for idx, name in enumerate(names):
1467
        # Wait for actual acquire on this lock to start
1468
        lock_ev[name].wait(10.0)
1469

    
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])
1474

    
1475
        self.ls.release(names=name)
1476

    
1477
      self.assertFalse(self.ls.list_owned())
1478

    
1479
      self._waitThreads()
1480

    
1481
      if expire:
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")
1485
      else:
1486
        for i in names:
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)
1490

    
1491
  @_Repeat
1492
  def testConcurrentRemove(self):
1493
    self.ls.add("four")
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")
1501
    self.ls.release()
1502
    self._waitThreads()
1503
    for i in range(4):
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")
1511
    self.ls.release()
1512
    self._waitThreads()
1513
    for i in range(4):
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")
1519
    self._waitThreads()
1520
    self.assertEqual(self.done.get_nowait(), ["six"])
1521
    self._addThread(target=self._doRemoveSet, args=(["two"]))
1522
    self._waitThreads()
1523
    self.assertEqual(self.done.get_nowait(), ["two"])
1524
    self.ls.release()
1525
    # reset lockset
1526
    self._setUpLS()
1527

    
1528
  @_Repeat
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))
1534
    self._waitThreads()
1535
    self.assertEqual(self.done.get_nowait(), "DONE")
1536
    # ...or just share some elements
1537
    self._addThread(target=self._doLockSet, args=(["one", "three"], 1))
1538
    self._waitThreads()
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
1545
    self.ls.release([])
1546
    t.join(60)
1547
    self.assertEqual(self.done.get_nowait(), "DONE")
1548
    # release the lock on the actual elements so remove() can proceed too
1549
    self.ls.release()
1550
    self._waitThreads()
1551
    self.failUnlessEqual(self.done.get_nowait(), ["two"])
1552
    # reset lockset
1553
    self._setUpLS()
1554

    
1555
  @_Repeat
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)
1566
    self.ls.release()
1567
    self._waitThreads()
1568
    for _ in range(5):
1569
      self.assertEqual(self.done.get(True, 1), "DONE")
1570
    # cleanup
1571
    self._setUpLS()
1572

    
1573
  @_Repeat
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")
1581
    self.ls.release()
1582
    self._waitThreads()
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)
1589
    self.ls.add("four")
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"]))
1597
    self.ls.release()
1598
    self._waitThreads()
1599
    self.assertEqual(self.done.get_nowait(), "DONE")
1600
    self.assertEqual(self.done.get_nowait(), "DONE")
1601
    self._setUpLS()
1602

    
1603
  @_Repeat
1604
  def testEmptyLockSet(self):
1605
    # get the set-lock
1606
    self.assertEqual(self.ls.acquire(None), set(["one", "two", "three"]))
1607
    # now empty it...
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)
1614
    self.ls.release()
1615
    self._waitThreads()
1616
    for _ in range(3):
1617
      self.assertEqual(self.done.get_nowait(), "DONE")
1618
    # empty it again...
1619
    self.assertEqual(self.ls.remove(["nine"]), ["nine"])
1620
    # now share it...
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))
1624
    self._waitThreads()
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)
1628
    self.ls.release()
1629
    self._waitThreads()
1630
    self.assertEqual(self.done.get_nowait(), "DONE")
1631
    self._setUpLS()
1632

    
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())
1637
    self.ls.release()
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")
1642

    
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)
1647

    
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))
1653

    
1654
    self.assertEquals(self.ls.acquire(None, shared=0),
1655
                      set(["one", "two", "three"]))
1656
    self.assertRaises(AssertionError, self.ls.downgrade, "unknown lock")
1657

    
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))
1663

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

    
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")])
1672
                               for name, lock in
1673
                                 self.ls._get_lockdict().items()))
1674

    
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))
1679

    
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))
1685
                               for name, lock in
1686
                                 self.ls._get_lockdict().items()))
1687

    
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))
1693

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

    
1701
    # Verify owned locks
1702
    for name in self.ls._names():
1703
      self.assertTrue(self.ls.check_owned(name, shared=1))
1704

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

    
1711
    self.ls.release()
1712

    
1713
  def testDowngradeEverything(self):
1714
    self.assertEqual(self.ls.acquire(locking.ALL_SET, shared=0),
1715
                     set(["one", "two", "three"]))
1716

    
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))
1720

    
1721
    # Downgrade everything
1722
    self.assertTrue(self.ls.downgrade())
1723

    
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))
1727

    
1728
  def testPriority(self):
1729
    def _Acquire(prev, next, name, priority, success_fn):
1730
      prev.wait()
1731
      self.assert_(self.ls.acquire(name, shared=0,
1732
                                   priority=priority,
1733
                                   test_notify=lambda _: next.set()))
1734
      try:
1735
        success_fn()
1736
      finally:
1737
        self.ls.release()
1738

    
1739
    # Get all in exclusive mode
1740
    self.assert_(self.ls.acquire(locking.ALL_SET, shared=0))
1741

    
1742
    done_two = Queue.Queue(0)
1743

    
1744
    first = threading.Event()
1745
    prev = first
1746

    
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)])
1749

    
1750
    # Use a deterministic random generator
1751
    random.Random(741).shuffle(acquires)
1752

    
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)))
1758
      prev = ev
1759

    
1760
    # Start acquires
1761
    first.set()
1762

    
1763
    # Wait for last acquire to start
1764
    prev.wait()
1765

    
1766
    # Let threads acquire locks
1767
    self.ls.release()
1768

    
1769
    # Wait for threads to finish
1770
    self._waitThreads()
1771

    
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)
1775

    
1776
    self.assertRaises(Queue.Empty, self.done.get_nowait)
1777
    self.assertRaises(Queue.Empty, done_two.get_nowait)
1778

    
1779

    
1780
class TestGanetiLockManager(_ThreadedTestCase):
1781
  def setUp(self):
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)
1789

    
1790
  def tearDown(self):
1791
    # Don't try this at home...
1792
    locking.GanetiLockManager._instance = None
1793

    
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])
1801

    
1802
  def testDoubleGLFails(self):
1803
    self.assertRaises(AssertionError, locking.GanetiLockManager, [], [], [], [])
1804

    
1805
  def testLockNames(self):
1806
    self.assertEqual(self.GL._names(locking.LEVEL_CLUSTER), set(["BGL"]))
1807
    self.assertEqual(self.GL._names(locking.LEVEL_NODE_ALLOC), set(["NAL"]))
1808
    self.assertEqual(self.GL._names(locking.LEVEL_NODE), set(self.nodes))
1809
    self.assertEqual(self.GL._names(locking.LEVEL_NODEGROUP),
1810
                     set(self.nodegroups))
1811
    self.assertEqual(self.GL._names(locking.LEVEL_INSTANCE),
1812
                     set(self.instances))
1813
    self.assertEqual(self.GL._names(locking.LEVEL_NETWORK),
1814
                     set(self.networks))
1815

    
1816
  def testInitAndResources(self):
1817
    locking.GanetiLockManager._instance = None
1818
    self.GL = locking.GanetiLockManager([], [], [], [])
1819
    self.assertEqual(self.GL._names(locking.LEVEL_CLUSTER), set(["BGL"]))
1820
    self.assertEqual(self.GL._names(locking.LEVEL_NODE_ALLOC), set(["NAL"]))
1821
    self.assertEqual(self.GL._names(locking.LEVEL_NODE), set())
1822
    self.assertEqual(self.GL._names(locking.LEVEL_NODEGROUP), set())
1823
    self.assertEqual(self.GL._names(locking.LEVEL_INSTANCE), set())
1824
    self.assertEqual(self.GL._names(locking.LEVEL_NETWORK), set())
1825

    
1826
    locking.GanetiLockManager._instance = None
1827
    self.GL = locking.GanetiLockManager(self.nodes, self.nodegroups, [], [])
1828
    self.assertEqual(self.GL._names(locking.LEVEL_CLUSTER), set(["BGL"]))
1829
    self.assertEqual(self.GL._names(locking.LEVEL_NODE_ALLOC), set(["NAL"]))
1830
    self.assertEqual(self.GL._names(locking.LEVEL_NODE), set(self.nodes))
1831
    self.assertEqual(self.GL._names(locking.LEVEL_NODEGROUP),
1832
                                    set(self.nodegroups))
1833
    self.assertEqual(self.GL._names(locking.LEVEL_INSTANCE), set())
1834
    self.assertEqual(self.GL._names(locking.LEVEL_NETWORK), set())
1835

    
1836
    locking.GanetiLockManager._instance = None
1837
    self.GL = locking.GanetiLockManager([], [], self.instances, [])
1838
    self.assertEqual(self.GL._names(locking.LEVEL_CLUSTER), set(["BGL"]))
1839
    self.assertEqual(self.GL._names(locking.LEVEL_NODE_ALLOC), set(["NAL"]))
1840
    self.assertEqual(self.GL._names(locking.LEVEL_NODE), set())
1841
    self.assertEqual(self.GL._names(locking.LEVEL_NODEGROUP), set())
1842
    self.assertEqual(self.GL._names(locking.LEVEL_INSTANCE),
1843
                     set(self.instances))
1844

    
1845
    locking.GanetiLockManager._instance = None
1846
    self.GL = locking.GanetiLockManager([], [], [], self.networks)
1847
    self.assertEqual(self.GL._names(locking.LEVEL_CLUSTER), set(["BGL"]))
1848
    self.assertEqual(self.GL._names(locking.LEVEL_NODE_ALLOC), set(["NAL"]))
1849
    self.assertEqual(self.GL._names(locking.LEVEL_NODE), set())
1850
    self.assertEqual(self.GL._names(locking.LEVEL_NODEGROUP), set())
1851
    self.assertEqual(self.GL._names(locking.LEVEL_INSTANCE), set())
1852
    self.assertEqual(self.GL._names(locking.LEVEL_NETWORK),
1853
                     set(self.networks))
1854

    
1855
  def testAcquireRelease(self):
1856
    self.GL.acquire(locking.LEVEL_CLUSTER, ["BGL"], shared=1)
1857
    self.assertEquals(self.GL.list_owned(locking.LEVEL_CLUSTER), set(["BGL"]))
1858
    self.GL.acquire(locking.LEVEL_INSTANCE, ["i1"])
1859
    self.GL.acquire(locking.LEVEL_NODEGROUP, ["g2"])
1860
    self.GL.acquire(locking.LEVEL_NODE, ["n1", "n2"], shared=1)
1861
    self.assertTrue(self.GL.check_owned(locking.LEVEL_NODE, ["n1", "n2"],
1862
                                        shared=1))
1863
    self.assertFalse(self.GL.check_owned(locking.LEVEL_INSTANCE, ["i1", "i3"]))
1864
    self.GL.release(locking.LEVEL_NODE, ["n2"])
1865
    self.assertEquals(self.GL.list_owned(locking.LEVEL_NODE), set(["n1"]))
1866
    self.assertEquals(self.GL.list_owned(locking.LEVEL_NODEGROUP), set(["g2"]))
1867
    self.assertEquals(self.GL.list_owned(locking.LEVEL_INSTANCE), set(["i1"]))
1868
    self.GL.release(locking.LEVEL_NODE)
1869
    self.assertEquals(self.GL.list_owned(locking.LEVEL_NODE), set())
1870
    self.assertEquals(self.GL.list_owned(locking.LEVEL_NODEGROUP), set(["g2"]))
1871
    self.assertEquals(self.GL.list_owned(locking.LEVEL_INSTANCE), set(["i1"]))
1872
    self.GL.release(locking.LEVEL_NODEGROUP)
1873
    self.GL.release(locking.LEVEL_INSTANCE)
1874
    self.assertRaises(errors.LockError, self.GL.acquire,
1875
                      locking.LEVEL_INSTANCE, ["i5"])
1876
    self.GL.acquire(locking.LEVEL_INSTANCE, ["i3"], shared=1)
1877
    self.assertEquals(self.GL.list_owned(locking.LEVEL_INSTANCE), set(["i3"]))
1878

    
1879
  def testAcquireWholeSets(self):
1880
    self.GL.acquire(locking.LEVEL_CLUSTER, ["BGL"], shared=1)
1881
    self.assertEquals(self.GL.acquire(locking.LEVEL_INSTANCE, None),
1882
                      set(self.instances))
1883
    self.assertEquals(self.GL.list_owned(locking.LEVEL_INSTANCE),
1884
                      set(self.instances))
1885
    self.assertEquals(self.GL.acquire(locking.LEVEL_NODEGROUP, None),
1886
                      set(self.nodegroups))
1887
    self.assertEquals(self.GL.list_owned(locking.LEVEL_NODEGROUP),
1888
                      set(self.nodegroups))
1889
    self.assertEquals(self.GL.acquire(locking.LEVEL_NODE, None, shared=1),
1890
                      set(self.nodes))
1891
    self.assertEquals(self.GL.list_owned(locking.LEVEL_NODE),
1892
                      set(self.nodes))
1893
    self.GL.release(locking.LEVEL_NODE)
1894
    self.GL.release(locking.LEVEL_NODEGROUP)
1895
    self.GL.release(locking.LEVEL_INSTANCE)
1896
    self.GL.release(locking.LEVEL_CLUSTER)
1897

    
1898
  def testAcquireWholeAndPartial(self):
1899
    self.GL.acquire(locking.LEVEL_CLUSTER, ["BGL"], shared=1)
1900
    self.assertEquals(self.GL.acquire(locking.LEVEL_INSTANCE, None),
1901
                      set(self.instances))
1902
    self.assertEquals(self.GL.list_owned(locking.LEVEL_INSTANCE),
1903
                      set(self.instances))
1904
    self.assertEquals(self.GL.acquire(locking.LEVEL_NODE, ["n2"], shared=1),
1905
                      set(["n2"]))
1906
    self.assertEquals(self.GL.list_owned(locking.LEVEL_NODE),
1907
                      set(["n2"]))
1908
    self.GL.release(locking.LEVEL_NODE)
1909
    self.GL.release(locking.LEVEL_INSTANCE)
1910
    self.GL.release(locking.LEVEL_CLUSTER)
1911

    
1912
  def testBGLDependency(self):
1913
    self.assertRaises(AssertionError, self.GL.acquire,
1914
                      locking.LEVEL_NODE, ["n1", "n2"])
1915
    self.assertRaises(AssertionError, self.GL.acquire,
1916
                      locking.LEVEL_INSTANCE, ["i3"])
1917
    self.assertRaises(AssertionError, self.GL.acquire,
1918
                      locking.LEVEL_NODEGROUP, ["g1"])
1919
    self.GL.acquire(locking.LEVEL_CLUSTER, ["BGL"], shared=1)
1920
    self.GL.acquire(locking.LEVEL_NODE, ["n1"])
1921
    self.assertRaises(AssertionError, self.GL.release,
1922
                      locking.LEVEL_CLUSTER, ["BGL"])
1923
    self.assertRaises(AssertionError, self.GL.release,
1924
                      locking.LEVEL_CLUSTER)
1925
    self.GL.release(locking.LEVEL_NODE)
1926
    self.GL.acquire(locking.LEVEL_INSTANCE, ["i1", "i2"])
1927
    self.assertRaises(AssertionError, self.GL.release,
1928
                      locking.LEVEL_CLUSTER, ["BGL"])
1929
    self.assertRaises(AssertionError, self.GL.release,
1930
                      locking.LEVEL_CLUSTER)
1931
    self.GL.release(locking.LEVEL_INSTANCE)
1932
    self.GL.acquire(locking.LEVEL_NODEGROUP, None)
1933
    self.GL.release(locking.LEVEL_NODEGROUP, ["g1"])
1934
    self.assertRaises(AssertionError, self.GL.release,
1935
                      locking.LEVEL_CLUSTER, ["BGL"])
1936
    self.assertRaises(AssertionError, self.GL.release,
1937
                      locking.LEVEL_CLUSTER)
1938
    self.GL.release(locking.LEVEL_NODEGROUP)
1939
    self.GL.release(locking.LEVEL_CLUSTER)
1940

    
1941
  def testWrongOrder(self):
1942
    self.GL.acquire(locking.LEVEL_CLUSTER, ["BGL"], shared=1)
1943
    self.GL.acquire(locking.LEVEL_NODE, ["n2"])
1944
    self.assertRaises(AssertionError, self.GL.acquire,
1945
                      locking.LEVEL_NODE, ["n1"])
1946
    self.assertRaises(AssertionError, self.GL.acquire,
1947
                      locking.LEVEL_NODEGROUP, ["g1"])
1948
    self.assertRaises(AssertionError, self.GL.acquire,
1949
                      locking.LEVEL_INSTANCE, ["i2"])
1950

    
1951
  def testModifiableLevels(self):
1952
    self.assertRaises(AssertionError, self.GL.add, locking.LEVEL_CLUSTER,
1953
                      ["BGL2"])
1954
    self.assertRaises(AssertionError, self.GL.add, locking.LEVEL_NODE_ALLOC,
1955
                      ["NAL2"])
1956
    self.GL.acquire(locking.LEVEL_CLUSTER, ["BGL"])
1957
    self.GL.add(locking.LEVEL_INSTANCE, ["i4"])
1958
    self.GL.remove(locking.LEVEL_INSTANCE, ["i3"])
1959
    self.GL.remove(locking.LEVEL_INSTANCE, ["i1"])
1960
    self.assertEqual(self.GL._names(locking.LEVEL_INSTANCE), set(["i2", "i4"]))
1961
    self.GL.add(locking.LEVEL_NODE, ["n3"])
1962
    self.GL.remove(locking.LEVEL_NODE, ["n1"])
1963
    self.assertEqual(self.GL._names(locking.LEVEL_NODE), set(["n2", "n3"]))
1964
    self.GL.add(locking.LEVEL_NODEGROUP, ["g3"])
1965
    self.GL.remove(locking.LEVEL_NODEGROUP, ["g2"])
1966
    self.GL.remove(locking.LEVEL_NODEGROUP, ["g1"])
1967
    self.assertEqual(self.GL._names(locking.LEVEL_NODEGROUP), set(["g3"]))
1968
    self.assertRaises(AssertionError, self.GL.remove, locking.LEVEL_CLUSTER,
1969
                      ["BGL2"])
1970

    
1971
  # Helper function to run as a thread that shared the BGL and then acquires
1972
  # some locks at another level.
1973
  def _doLock(self, level, names, shared):
1974
    try:
1975
      self.GL.acquire(locking.LEVEL_CLUSTER, ["BGL"], shared=1)
1976
      self.GL.acquire(level, names, shared=shared)
1977
      self.done.put("DONE")
1978
      self.GL.release(level)
1979
      self.GL.release(locking.LEVEL_CLUSTER)
1980
    except errors.LockError:
1981
      self.done.put("ERR")
1982

    
1983
  @_Repeat
1984
  def testConcurrency(self):
1985
    self.GL.acquire(locking.LEVEL_CLUSTER, ["BGL"], shared=1)
1986
    self._addThread(target=self._doLock,
1987
                    args=(locking.LEVEL_INSTANCE, "i1", 1))
1988
    self._waitThreads()
1989
    self.assertEqual(self.done.get_nowait(), "DONE")
1990
    self.GL.acquire(locking.LEVEL_INSTANCE, ["i3"])
1991
    self._addThread(target=self._doLock,
1992
                    args=(locking.LEVEL_INSTANCE, "i1", 1))
1993
    self._waitThreads()
1994
    self.assertEqual(self.done.get_nowait(), "DONE")
1995
    self._addThread(target=self._doLock,
1996
                    args=(locking.LEVEL_INSTANCE, "i3", 1))
1997
    self.assertRaises(Queue.Empty, self.done.get_nowait)
1998
    self.GL.release(locking.LEVEL_INSTANCE)
1999
    self._waitThreads()
2000
    self.assertEqual(self.done.get_nowait(), "DONE")
2001
    self.GL.acquire(locking.LEVEL_INSTANCE, ["i2"], shared=1)
2002
    self._addThread(target=self._doLock,
2003
                    args=(locking.LEVEL_INSTANCE, "i2", 1))
2004
    self._waitThreads()
2005
    self.assertEqual(self.done.get_nowait(), "DONE")
2006
    self._addThread(target=self._doLock,
2007
                    args=(locking.LEVEL_INSTANCE, "i2", 0))
2008
    self.assertRaises(Queue.Empty, self.done.get_nowait)
2009
    self.GL.release(locking.LEVEL_INSTANCE)
2010
    self._waitThreads()
2011
    self.assertEqual(self.done.get(True, 1), "DONE")
2012
    self.GL.release(locking.LEVEL_CLUSTER, ["BGL"])
2013

    
2014

    
2015
class TestLockMonitor(_ThreadedTestCase):
2016
  def setUp(self):
2017
    _ThreadedTestCase.setUp(self)
2018
    self.lm = locking.LockMonitor()
2019

    
2020
  def testSingleThread(self):
2021
    locks = []
2022

    
2023
    for i in range(100):
2024
      name = "TestLock%s" % i
2025
      locks.append(locking.SharedLock(name, monitor=self.lm))
2026

    
2027
    self.assertEqual(len(self.lm._locks), len(locks))
2028
    result = objects.QueryResponse.FromDict(self.lm.QueryLocks(["name"]))
2029
    self.assertEqual(len(result.fields), 1)
2030
    self.assertEqual(len(result.data), 100)
2031

    
2032
    # Delete all locks
2033
    del locks[:]
2034

    
2035
    # The garbage collector might needs some time
2036
    def _CheckLocks():
2037
      if self.lm._locks:
2038
        raise utils.RetryAgain()
2039

    
2040
    utils.Retry(_CheckLocks, 0.1, 30.0)
2041

    
2042
    self.assertFalse(self.lm._locks)
2043

    
2044
  def testMultiThread(self):
2045
    locks = []
2046

    
2047
    def _CreateLock(prev, next, name):
2048
      prev.wait()
2049
      locks.append(locking.SharedLock(name, monitor=self.lm))
2050
      if next:
2051
        next.set()
2052

    
2053
    expnames = []
2054

    
2055
    first = threading.Event()
2056
    prev = first
2057

    
2058
    # Use a deterministic random generator
2059
    for i in random.Random(4263).sample(range(100), 33):
2060
      name = "MtTestLock%s" % i
2061
      expnames.append(name)
2062

    
2063
      ev = threading.Event()
2064
      self._addThread(target=_CreateLock, args=(prev, ev, name))
2065
      prev = ev
2066

    
2067
    # Add locks
2068
    first.set()
2069
    self._waitThreads()
2070

    
2071
    # Check order in which locks were added
2072
    self.assertEqual([i.name for i in locks], expnames)
2073

    
2074
    # Check query result
2075
    result = self.lm.QueryLocks(["name", "mode", "owner", "pending"])
2076
    self.assert_(isinstance(result, dict))
2077
    response = objects.QueryResponse.FromDict(result)
2078
    self.assertEqual(response.data,
2079
                     [[(constants.RS_NORMAL, name),
2080
                       (constants.RS_NORMAL, None),
2081
                       (constants.RS_NORMAL, None),
2082
                       (constants.RS_NORMAL, [])]
2083
                      for name in utils.NiceSort(expnames)])
2084
    self.assertEqual(len(response.fields), 4)
2085
    self.assertEqual(["name", "mode", "owner", "pending"],
2086
                     [fdef.name for fdef in response.fields])
2087

    
2088
    # Test exclusive acquire
2089
    for tlock in locks[::4]:
2090
      tlock.acquire(shared=0)
2091
      try:
2092
        def _GetExpResult(name):
2093
          if tlock.name == name:
2094
            return [(constants.RS_NORMAL, name),
2095
                    (constants.RS_NORMAL, "exclusive"),
2096
                    (constants.RS_NORMAL,
2097
                     [threading.currentThread().getName()]),
2098
                    (constants.RS_NORMAL, [])]
2099
          return [(constants.RS_NORMAL, name),
2100
                  (constants.RS_NORMAL, None),
2101
                  (constants.RS_NORMAL, None),
2102
                  (constants.RS_NORMAL, [])]
2103

    
2104
        result = self.lm.QueryLocks(["name", "mode", "owner", "pending"])
2105
        self.assertEqual(objects.QueryResponse.FromDict(result).data,
2106
                         [_GetExpResult(name)
2107
                          for name in utils.NiceSort(expnames)])
2108
      finally:
2109
        tlock.release()
2110

    
2111
    # Test shared acquire
2112
    def _Acquire(lock, shared, ev, notify):
2113
      lock.acquire(shared=shared)
2114
      try:
2115
        notify.set()
2116
        ev.wait()
2117
      finally:
2118
        lock.release()
2119

    
2120
    for tlock1 in locks[::11]:
2121
      for tlock2 in locks[::-15]:
2122
        if tlock2 == tlock1:
2123
          # Avoid deadlocks
2124
          continue
2125

    
2126
        for tlock3 in locks[::10]:
2127
          if tlock3 in (tlock2, tlock1):
2128
            # Avoid deadlocks
2129
            continue
2130

    
2131
          releaseev = threading.Event()
2132

    
2133
          # Acquire locks
2134
          acquireev = []
2135
          tthreads1 = []
2136
          for i in range(3):
2137
            ev = threading.Event()
2138
            tthreads1.append(self._addThread(target=_Acquire,
2139
                                             args=(tlock1, 1, releaseev, ev)))
2140
            acquireev.append(ev)
2141

    
2142
          ev = threading.Event()
2143
          tthread2 = self._addThread(target=_Acquire,
2144
                                     args=(tlock2, 1, releaseev, ev))
2145
          acquireev.append(ev)
2146

    
2147
          ev = threading.Event()
2148
          tthread3 = self._addThread(target=_Acquire,
2149
                                     args=(tlock3, 0, releaseev, ev))
2150
          acquireev.append(ev)
2151

    
2152
          # Wait for all locks to be acquired
2153
          for i in acquireev:
2154
            i.wait()
2155

    
2156
          # Check query result
2157
          result = self.lm.QueryLocks(["name", "mode", "owner"])
2158
          response = objects.QueryResponse.FromDict(result)
2159
          for (name, mode, owner) in response.data:
2160
            (name_status, name_value) = name
2161
            (owner_status, owner_value) = owner
2162

    
2163
            self.assertEqual(name_status, constants.RS_NORMAL)
2164
            self.assertEqual(owner_status, constants.RS_NORMAL)
2165

    
2166
            if name_value == tlock1.name:
2167
              self.assertEqual(mode, (constants.RS_NORMAL, "shared"))
2168
              self.assertEqual(set(owner_value),
2169
                               set(i.getName() for i in tthreads1))
2170
              continue
2171

    
2172
            if name_value == tlock2.name:
2173
              self.assertEqual(mode, (constants.RS_NORMAL, "shared"))
2174
              self.assertEqual(owner_value, [tthread2.getName()])
2175
              continue
2176

    
2177
            if name_value == tlock3.name:
2178
              self.assertEqual(mode, (constants.RS_NORMAL, "exclusive"))
2179
              self.assertEqual(owner_value, [tthread3.getName()])
2180
              continue
2181

    
2182
            self.assert_(name_value in expnames)
2183
            self.assertEqual(mode, (constants.RS_NORMAL, None))
2184
            self.assert_(owner_value is None)
2185

    
2186
          # Release locks again
2187
          releaseev.set()
2188

    
2189
          self._waitThreads()
2190

    
2191
          result = self.lm.QueryLocks(["name", "mode", "owner"])
2192
          self.assertEqual(objects.QueryResponse.FromDict(result).data,
2193
                           [[(constants.RS_NORMAL, name),
2194
                             (constants.RS_NORMAL, None),
2195
                             (constants.RS_NORMAL, None)]
2196
                            for name in utils.NiceSort(expnames)])
2197

    
2198
  def testDelete(self):
2199
    lock = locking.SharedLock("TestLock", monitor=self.lm)
2200

    
2201
    self.assertEqual(len(self.lm._locks), 1)
2202
    result = self.lm.QueryLocks(["name", "mode", "owner"])
2203
    self.assertEqual(objects.QueryResponse.FromDict(result).data,
2204
                     [[(constants.RS_NORMAL, lock.name),
2205
                       (constants.RS_NORMAL, None),
2206
                       (constants.RS_NORMAL, None)]])
2207

    
2208
    lock.delete()
2209

    
2210
    result = self.lm.QueryLocks(["name", "mode", "owner"])
2211
    self.assertEqual(objects.QueryResponse.FromDict(result).data,
2212
                     [[(constants.RS_NORMAL, lock.name),
2213
                       (constants.RS_NORMAL, "deleted"),
2214
                       (constants.RS_NORMAL, None)]])
2215
    self.assertEqual(len(self.lm._locks), 1)
2216

    
2217
  def testPending(self):
2218
    def _Acquire(lock, shared, prev, next):
2219
      prev.wait()
2220

    
2221
      lock.acquire(shared=shared, test_notify=next.set)
2222
      try:
2223
        pass
2224
      finally:
2225
        lock.release()
2226

    
2227
    lock = locking.SharedLock("ExcLock", monitor=self.lm)
2228

    
2229
    for shared in [0, 1]:
2230
      lock.acquire()
2231
      try:
2232
        self.assertEqual(len(self.lm._locks), 1)
2233
        result = self.lm.QueryLocks(["name", "mode", "owner"])
2234
        self.assertEqual(objects.QueryResponse.FromDict(result).data,
2235
                         [[(constants.RS_NORMAL, lock.name),
2236
                           (constants.RS_NORMAL, "exclusive"),
2237
                           (constants.RS_NORMAL,
2238
                            [threading.currentThread().getName()])]])
2239

    
2240
        threads = []
2241

    
2242
        first = threading.Event()
2243
        prev = first
2244

    
2245
        for i in range(5):
2246
          ev = threading.Event()
2247
          threads.append(self._addThread(target=_Acquire,
2248
                                          args=(lock, shared, prev, ev)))
2249
          prev = ev
2250

    
2251
        # Start acquires
2252
        first.set()
2253

    
2254
        # Wait for last acquire to start waiting
2255
        prev.wait()
2256

    
2257
        # NOTE: This works only because QueryLocks will acquire the
2258
        # lock-internal lock again and won't be able to get the information
2259
        # until it has the lock. By then the acquire should be registered in
2260
        # SharedLock.__pending (otherwise it's a bug).
2261

    
2262
        # All acquires are waiting now
2263
        if shared:
2264
          pending = [("shared", utils.NiceSort(t.getName() for t in threads))]
2265
        else:
2266
          pending = [("exclusive", [t.getName()]) for t in threads]
2267

    
2268
        result = self.lm.QueryLocks(["name", "mode", "owner", "pending"])
2269
        self.assertEqual(objects.QueryResponse.FromDict(result).data,
2270
                         [[(constants.RS_NORMAL, lock.name),
2271
                           (constants.RS_NORMAL, "exclusive"),
2272
                           (constants.RS_NORMAL,
2273
                            [threading.currentThread().getName()]),
2274
                           (constants.RS_NORMAL, pending)]])
2275

    
2276
        self.assertEqual(len(self.lm._locks), 1)
2277
      finally:
2278
        lock.release()
2279

    
2280
      self._waitThreads()
2281

    
2282
      # No pending acquires
2283
      result = self.lm.QueryLocks(["name", "mode", "owner", "pending"])
2284
      self.assertEqual(objects.QueryResponse.FromDict(result).data,
2285
                       [[(constants.RS_NORMAL, lock.name),
2286
                         (constants.RS_NORMAL, None),
2287
                         (constants.RS_NORMAL, None),
2288
                         (constants.RS_NORMAL, [])]])
2289

    
2290
      self.assertEqual(len(self.lm._locks), 1)
2291

    
2292
  def testDeleteAndRecreate(self):
2293
    lname = "TestLock101923193"
2294

    
2295
    # Create some locks with the same name and keep all references
2296
    locks = [locking.SharedLock(lname, monitor=self.lm)
2297
             for _ in range(5)]
2298

    
2299
    self.assertEqual(len(self.lm._locks), len(locks))
2300

    
2301
    result = self.lm.QueryLocks(["name", "mode", "owner"])
2302
    self.assertEqual(objects.QueryResponse.FromDict(result).data,
2303
                     [[(constants.RS_NORMAL, lname),
2304
                       (constants.RS_NORMAL, None),
2305
                       (constants.RS_NORMAL, None)]] * 5)
2306

    
2307
    locks[2].delete()
2308

    
2309
    # Check information order
2310
    result = self.lm.QueryLocks(["name", "mode", "owner"])
2311
    self.assertEqual(objects.QueryResponse.FromDict(result).data,
2312
                     [[(constants.RS_NORMAL, lname),
2313
                       (constants.RS_NORMAL, None),
2314
                       (constants.RS_NORMAL, None)]] * 2 +
2315
                     [[(constants.RS_NORMAL, lname),
2316
                       (constants.RS_NORMAL, "deleted"),
2317
                       (constants.RS_NORMAL, None)]] +
2318
                     [[(constants.RS_NORMAL, lname),
2319
                       (constants.RS_NORMAL, None),
2320
                       (constants.RS_NORMAL, None)]] * 2)
2321

    
2322
    locks[1].acquire(shared=0)
2323

    
2324
    last_status = [
2325
      [(constants.RS_NORMAL, lname),
2326
       (constants.RS_NORMAL, None),
2327
       (constants.RS_NORMAL, None)],
2328
      [(constants.RS_NORMAL, lname),
2329
       (constants.RS_NORMAL, "exclusive"),
2330
       (constants.RS_NORMAL, [threading.currentThread().getName()])],
2331
      [(constants.RS_NORMAL, lname),
2332
       (constants.RS_NORMAL, "deleted"),
2333
       (constants.RS_NORMAL, None)],
2334
      [(constants.RS_NORMAL, lname),
2335
       (constants.RS_NORMAL, None),
2336
       (constants.RS_NORMAL, None)],
2337
      [(constants.RS_NORMAL, lname),
2338
       (constants.RS_NORMAL, None),
2339
       (constants.RS_NORMAL, None)],
2340
      ]
2341

    
2342
    # Check information order
2343
    result = self.lm.QueryLocks(["name", "mode", "owner"])
2344
    self.assertEqual(objects.QueryResponse.FromDict(result).data, last_status)
2345

    
2346
    self.assertEqual(len(set(self.lm._locks.values())), len(locks))
2347
    self.assertEqual(len(self.lm._locks), len(locks))
2348

    
2349
    # Check lock deletion
2350
    for idx in range(len(locks)):
2351
      del locks[0]
2352
      assert gc.isenabled()
2353
      gc.collect()
2354
      self.assertEqual(len(self.lm._locks), len(locks))
2355
      result = self.lm.QueryLocks(["name", "mode", "owner"])
2356
      self.assertEqual(objects.QueryResponse.FromDict(result).data,
2357
                       last_status[idx + 1:])
2358

    
2359
    # All locks should have been deleted
2360
    assert not locks
2361
    self.assertFalse(self.lm._locks)
2362

    
2363
    result = self.lm.QueryLocks(["name", "mode", "owner"])
2364
    self.assertEqual(objects.QueryResponse.FromDict(result).data, [])
2365

    
2366
  class _FakeLock:
2367
    def __init__(self):
2368
      self._info = []
2369

    
2370
    def AddResult(self, *args):
2371
      self._info.append(args)
2372

    
2373
    def CountPending(self):
2374
      return len(self._info)
2375

    
2376
    def GetLockInfo(self, requested):
2377
      (exp_requested, result) = self._info.pop(0)
2378

    
2379
      if exp_requested != requested:
2380
        raise Exception("Requested information (%s) does not match"
2381
                        " expectations (%s)" % (requested, exp_requested))
2382

    
2383
      return result
2384

    
2385
  def testMultipleResults(self):
2386
    fl1 = self._FakeLock()
2387
    fl2 = self._FakeLock()
2388

    
2389
    self.lm.RegisterLock(fl1)
2390
    self.lm.RegisterLock(fl2)
2391

    
2392
    # Empty information
2393
    for i in [fl1, fl2]:
2394
      i.AddResult(set([query.LQ_MODE, query.LQ_OWNER]), [])
2395
    result = self.lm.QueryLocks(["name", "mode", "owner"])
2396
    self.assertEqual(objects.QueryResponse.FromDict(result).data, [])
2397
    for i in [fl1, fl2]:
2398
      self.assertEqual(i.CountPending(), 0)
2399

    
2400
    # Check ordering
2401
    for fn in [lambda x: x, reversed, sorted]:
2402
      fl1.AddResult(set(), list(fn([
2403
        ("aaa", None, None, None),
2404
        ("bbb", None, None, None),
2405
        ])))
2406
      fl2.AddResult(set(), [])
2407
      result = self.lm.QueryLocks(["name"])
2408
      self.assertEqual(objects.QueryResponse.FromDict(result).data, [
2409
        [(constants.RS_NORMAL, "aaa")],
2410
        [(constants.RS_NORMAL, "bbb")],
2411
        ])
2412
      for i in [fl1, fl2]:
2413
        self.assertEqual(i.CountPending(), 0)
2414

    
2415
      for fn2 in [lambda x: x, reversed, sorted]:
2416
        fl1.AddResult(set([query.LQ_MODE]), list(fn([
2417
          # Same name, but different information
2418
          ("aaa", "mode0", None, None),
2419
          ("aaa", "mode1", None, None),
2420
          ("aaa", "mode2", None, None),
2421
          ("aaa", "mode3", None, None),
2422
          ])))
2423
        fl2.AddResult(set([query.LQ_MODE]), [
2424
          ("zzz", "end", None, None),
2425
          ("000", "start", None, None),
2426
          ] + list(fn2([
2427
          ("aaa", "b200", None, None),
2428
          ("aaa", "b300", None, None),
2429
          ])))
2430
        result = self.lm.QueryLocks(["name", "mode"])
2431
        self.assertEqual(objects.QueryResponse.FromDict(result).data, [
2432
          [(constants.RS_NORMAL, "000"), (constants.RS_NORMAL, "start")],
2433
          ] + list(fn([
2434
          # Name is the same, so order must be equal to incoming order
2435
          [(constants.RS_NORMAL, "aaa"), (constants.RS_NORMAL, "mode0")],
2436
          [(constants.RS_NORMAL, "aaa"), (constants.RS_NORMAL, "mode1")],
2437
          [(constants.RS_NORMAL, "aaa"), (constants.RS_NORMAL, "mode2")],
2438
          [(constants.RS_NORMAL, "aaa"), (constants.RS_NORMAL, "mode3")],
2439
          ])) + list(fn2([
2440
          [(constants.RS_NORMAL, "aaa"), (constants.RS_NORMAL, "b200")],
2441
          [(constants.RS_NORMAL, "aaa"), (constants.RS_NORMAL, "b300")],
2442
          ])) + [
2443
          [(constants.RS_NORMAL, "zzz"), (constants.RS_NORMAL, "end")],
2444
          ])
2445
        for i in [fl1, fl2]:
2446
          self.assertEqual(i.CountPending(), 0)
2447

    
2448

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