Statistics
| Branch: | Tag: | Revision:

root / test / ganeti.locking_unittest.py @ 3dbe3ddf

History | View | Annotate | Download (73.1 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
# 0.0510-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
    # This new thread can't acquire the lock, and thus call wait, before we
191
    # release it
192
    self._addThread(target=fn)
193
    self.cond.notifyAll()
194
    self.assertRaises(Queue.Empty, self.done.get_nowait)
195
    self.cond.release()
196

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

    
209
    self.assertEqual(w, 3)
210
    self.assertEqual(a, 1)
211

    
212
    self.cond.acquire()
213
    self.cond.notifyAll()
214
    self.cond.release()
215
    self._waitThreads()
216
    self.assertEqual(self.done.get_nowait(), "W")
217
    self.assertRaises(Queue.Empty, self.done.get_nowait)
218

    
219
  def testBlockingWait(self):
220
    def _BlockingWait():
221
      self.cond.acquire()
222
      self.done.put("A")
223
      self.cond.wait(None)
224
      self.cond.release()
225
      self.done.put("W")
226

    
227
    self._TestWait(_BlockingWait)
228

    
229
  def testLongTimeoutWait(self):
230
    def _Helper():
231
      self.cond.acquire()
232
      self.done.put("A")
233
      self.cond.wait(15.0)
234
      self.cond.release()
235
      self.done.put("W")
236

    
237
    self._TestWait(_Helper)
238

    
239
  def _TimeoutWait(self, timeout, check):
240
    self.cond.acquire()
241
    self.cond.wait(timeout)
242
    self.cond.release()
243
    self.done.put(check)
244

    
245
  def testShortTimeoutWait(self):
246
    self._addThread(target=self._TimeoutWait, args=(0.1, "T1"))
247
    self._addThread(target=self._TimeoutWait, args=(0.1, "T1"))
248
    self._waitThreads()
249
    self.assertEqual(self.done.get_nowait(), "T1")
250
    self.assertEqual(self.done.get_nowait(), "T1")
251
    self.assertRaises(Queue.Empty, self.done.get_nowait)
252

    
253
  def testZeroTimeoutWait(self):
254
    self._addThread(target=self._TimeoutWait, args=(0, "T0"))
255
    self._addThread(target=self._TimeoutWait, args=(0, "T0"))
256
    self._addThread(target=self._TimeoutWait, args=(0, "T0"))
257
    self._waitThreads()
258
    self.assertEqual(self.done.get_nowait(), "T0")
259
    self.assertEqual(self.done.get_nowait(), "T0")
260
    self.assertEqual(self.done.get_nowait(), "T0")
261
    self.assertRaises(Queue.Empty, self.done.get_nowait)
262

    
263

    
264
class TestSharedLock(_ThreadedTestCase):
265
  """SharedLock tests"""
266

    
267
  def setUp(self):
268
    _ThreadedTestCase.setUp(self)
269
    self.sl = locking.SharedLock("TestSharedLock")
270

    
271
  def testSequenceAndOwnership(self):
272
    self.assertFalse(self.sl._is_owned())
273
    self.sl.acquire(shared=1)
274
    self.assert_(self.sl._is_owned())
275
    self.assert_(self.sl._is_owned(shared=1))
276
    self.assertFalse(self.sl._is_owned(shared=0))
277
    self.sl.release()
278
    self.assertFalse(self.sl._is_owned())
279
    self.sl.acquire()
280
    self.assert_(self.sl._is_owned())
281
    self.assertFalse(self.sl._is_owned(shared=1))
282
    self.assert_(self.sl._is_owned(shared=0))
283
    self.sl.release()
284
    self.assertFalse(self.sl._is_owned())
285
    self.sl.acquire(shared=1)
286
    self.assert_(self.sl._is_owned())
287
    self.assert_(self.sl._is_owned(shared=1))
288
    self.assertFalse(self.sl._is_owned(shared=0))
289
    self.sl.release()
290
    self.assertFalse(self.sl._is_owned())
291

    
292
  def testBooleanValue(self):
293
    # semaphores are supposed to return a true value on a successful acquire
294
    self.assert_(self.sl.acquire(shared=1))
295
    self.sl.release()
296
    self.assert_(self.sl.acquire())
297
    self.sl.release()
298

    
299
  def testDoubleLockingStoE(self):
300
    self.sl.acquire(shared=1)
301
    self.assertRaises(AssertionError, self.sl.acquire)
302

    
303
  def testDoubleLockingEtoS(self):
304
    self.sl.acquire()
305
    self.assertRaises(AssertionError, self.sl.acquire, shared=1)
306

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

    
311
  def testDoubleLockingEtoE(self):
312
    self.sl.acquire()
313
    self.assertRaises(AssertionError, self.sl.acquire)
314

    
315
  # helper functions: called in a separate thread they acquire the lock, send
316
  # their identifier on the done queue, then release it.
317
  def _doItSharer(self):
318
    try:
319
      self.sl.acquire(shared=1)
320
      self.done.put('SHR')
321
      self.sl.release()
322
    except errors.LockError:
323
      self.done.put('ERR')
324

    
325
  def _doItExclusive(self):
326
    try:
327
      self.sl.acquire()
328
      self.done.put('EXC')
329
      self.sl.release()
330
    except errors.LockError:
331
      self.done.put('ERR')
332

    
333
  def _doItDelete(self):
334
    try:
335
      self.sl.delete()
336
      self.done.put('DEL')
337
    except errors.LockError:
338
      self.done.put('ERR')
339

    
340
  def testSharersCanCoexist(self):
341
    self.sl.acquire(shared=1)
342
    threading.Thread(target=self._doItSharer).start()
343
    self.assert_(self.done.get(True, 1))
344
    self.sl.release()
345

    
346
  @_Repeat
347
  def testExclusiveBlocksExclusive(self):
348
    self.sl.acquire()
349
    self._addThread(target=self._doItExclusive)
350
    self.assertRaises(Queue.Empty, self.done.get_nowait)
351
    self.sl.release()
352
    self._waitThreads()
353
    self.failUnlessEqual(self.done.get_nowait(), 'EXC')
354

    
355
  @_Repeat
356
  def testExclusiveBlocksDelete(self):
357
    self.sl.acquire()
358
    self._addThread(target=self._doItDelete)
359
    self.assertRaises(Queue.Empty, self.done.get_nowait)
360
    self.sl.release()
361
    self._waitThreads()
362
    self.failUnlessEqual(self.done.get_nowait(), 'DEL')
363
    self.sl = locking.SharedLock(self.sl.name)
364

    
365
  @_Repeat
366
  def testExclusiveBlocksSharer(self):
367
    self.sl.acquire()
368
    self._addThread(target=self._doItSharer)
369
    self.assertRaises(Queue.Empty, self.done.get_nowait)
370
    self.sl.release()
371
    self._waitThreads()
372
    self.failUnlessEqual(self.done.get_nowait(), 'SHR')
373

    
374
  @_Repeat
375
  def testSharerBlocksExclusive(self):
376
    self.sl.acquire(shared=1)
377
    self._addThread(target=self._doItExclusive)
378
    self.assertRaises(Queue.Empty, self.done.get_nowait)
379
    self.sl.release()
380
    self._waitThreads()
381
    self.failUnlessEqual(self.done.get_nowait(), 'EXC')
382

    
383
  @_Repeat
384
  def testSharerBlocksDelete(self):
385
    self.sl.acquire(shared=1)
386
    self._addThread(target=self._doItDelete)
387
    self.assertRaises(Queue.Empty, self.done.get_nowait)
388
    self.sl.release()
389
    self._waitThreads()
390
    self.failUnlessEqual(self.done.get_nowait(), 'DEL')
391
    self.sl = locking.SharedLock(self.sl.name)
392

    
393
  @_Repeat
394
  def testWaitingExclusiveBlocksSharer(self):
395
    """SKIPPED testWaitingExclusiveBlockSharer"""
396
    return
397

    
398
    self.sl.acquire(shared=1)
399
    # the lock is acquired in shared mode...
400
    self._addThread(target=self._doItExclusive)
401
    # ...but now an exclusive is waiting...
402
    self._addThread(target=self._doItSharer)
403
    # ...so the sharer should be blocked as well
404
    self.assertRaises(Queue.Empty, self.done.get_nowait)
405
    self.sl.release()
406
    self._waitThreads()
407
    # The exclusive passed before
408
    self.failUnlessEqual(self.done.get_nowait(), 'EXC')
409
    self.failUnlessEqual(self.done.get_nowait(), 'SHR')
410

    
411
  @_Repeat
412
  def testWaitingSharerBlocksExclusive(self):
413
    """SKIPPED testWaitingSharerBlocksExclusive"""
414
    return
415

    
416
    self.sl.acquire()
417
    # the lock is acquired in exclusive mode...
418
    self._addThread(target=self._doItSharer)
419
    # ...but now a sharer is waiting...
420
    self._addThread(target=self._doItExclusive)
421
    # ...the exclusive is waiting too...
422
    self.assertRaises(Queue.Empty, self.done.get_nowait)
423
    self.sl.release()
424
    self._waitThreads()
425
    # The sharer passed before
426
    self.assertEqual(self.done.get_nowait(), 'SHR')
427
    self.assertEqual(self.done.get_nowait(), 'EXC')
428

    
429
  def testDelete(self):
430
    self.sl.delete()
431
    self.assertRaises(errors.LockError, self.sl.acquire)
432
    self.assertRaises(errors.LockError, self.sl.acquire, shared=1)
433
    self.assertRaises(errors.LockError, self.sl.delete)
434

    
435
  def testDeleteTimeout(self):
436
    self.sl.delete(timeout=60)
437

    
438
  def testNoDeleteIfSharer(self):
439
    self.sl.acquire(shared=1)
440
    self.assertRaises(AssertionError, self.sl.delete)
441

    
442
  @_Repeat
443
  def testDeletePendingSharersExclusiveDelete(self):
444
    self.sl.acquire()
445
    self._addThread(target=self._doItSharer)
446
    self._addThread(target=self._doItSharer)
447
    self._addThread(target=self._doItExclusive)
448
    self._addThread(target=self._doItDelete)
449
    self.sl.delete()
450
    self._waitThreads()
451
    # The threads who were pending return ERR
452
    for _ in range(4):
453
      self.assertEqual(self.done.get_nowait(), 'ERR')
454
    self.sl = locking.SharedLock(self.sl.name)
455

    
456
  @_Repeat
457
  def testDeletePendingDeleteExclusiveSharers(self):
458
    self.sl.acquire()
459
    self._addThread(target=self._doItDelete)
460
    self._addThread(target=self._doItExclusive)
461
    self._addThread(target=self._doItSharer)
462
    self._addThread(target=self._doItSharer)
463
    self.sl.delete()
464
    self._waitThreads()
465
    # The two threads who were pending return both ERR
466
    self.assertEqual(self.done.get_nowait(), 'ERR')
467
    self.assertEqual(self.done.get_nowait(), 'ERR')
468
    self.assertEqual(self.done.get_nowait(), 'ERR')
469
    self.assertEqual(self.done.get_nowait(), 'ERR')
470
    self.sl = locking.SharedLock(self.sl.name)
471

    
472
  @_Repeat
473
  def testExclusiveAcquireTimeout(self):
474
    for shared in [0, 1]:
475
      on_queue = threading.Event()
476
      release_exclusive = threading.Event()
477

    
478
      def _LockExclusive():
479
        self.sl.acquire(shared=0, test_notify=on_queue.set)
480
        self.done.put("A: start wait")
481
        release_exclusive.wait()
482
        self.done.put("A: end wait")
483
        self.sl.release()
484

    
485
      # Start thread to hold lock in exclusive mode
486
      self._addThread(target=_LockExclusive)
487

    
488
      # Wait for wait to begin
489
      self.assertEqual(self.done.get(timeout=60), "A: start wait")
490

    
491
      # Wait up to 60s to get lock, but release exclusive lock as soon as we're
492
      # on the queue
493
      self.failUnless(self.sl.acquire(shared=shared, timeout=60,
494
                                      test_notify=release_exclusive.set))
495

    
496
      self.done.put("got 2nd")
497
      self.sl.release()
498

    
499
      self._waitThreads()
500

    
501
      self.assertEqual(self.done.get_nowait(), "A: end wait")
502
      self.assertEqual(self.done.get_nowait(), "got 2nd")
503
      self.assertRaises(Queue.Empty, self.done.get_nowait)
504

    
505
  @_Repeat
506
  def testAcquireExpiringTimeout(self):
507
    def _AcquireWithTimeout(shared, timeout):
508
      if not self.sl.acquire(shared=shared, timeout=timeout):
509
        self.done.put("timeout")
510

    
511
    for shared in [0, 1]:
512
      # Lock exclusively
513
      self.sl.acquire()
514

    
515
      # Start shared acquires with timeout between 0 and 20 ms
516
      for i in range(11):
517
        self._addThread(target=_AcquireWithTimeout,
518
                        args=(shared, i * 2.0 / 1000.0))
519

    
520
      # Wait for threads to finish (makes sure the acquire timeout expires
521
      # before releasing the lock)
522
      self._waitThreads()
523

    
524
      # Release lock
525
      self.sl.release()
526

    
527
      for _ in range(11):
528
        self.assertEqual(self.done.get_nowait(), "timeout")
529

    
530
      self.assertRaises(Queue.Empty, self.done.get_nowait)
531

    
532
  @_Repeat
533
  def testSharedSkipExclusiveAcquires(self):
534
    # Tests whether shared acquires jump in front of exclusive acquires in the
535
    # queue.
536

    
537
    def _Acquire(shared, name, notify_ev, wait_ev):
538
      if notify_ev:
539
        notify_fn = notify_ev.set
540
      else:
541
        notify_fn = None
542

    
543
      if wait_ev:
544
        wait_ev.wait()
545

    
546
      if not self.sl.acquire(shared=shared, test_notify=notify_fn):
547
        return
548

    
549
      self.done.put(name)
550
      self.sl.release()
551

    
552
    # Get exclusive lock while we fill the queue
553
    self.sl.acquire()
554

    
555
    shrcnt1 = 5
556
    shrcnt2 = 7
557
    shrcnt3 = 9
558
    shrcnt4 = 2
559

    
560
    # Add acquires using threading.Event for synchronization. They'll be
561
    # acquired exactly in the order defined in this list.
562
    acquires = (shrcnt1 * [(1, "shared 1")] +
563
                3 * [(0, "exclusive 1")] +
564
                shrcnt2 * [(1, "shared 2")] +
565
                shrcnt3 * [(1, "shared 3")] +
566
                shrcnt4 * [(1, "shared 4")] +
567
                3 * [(0, "exclusive 2")])
568

    
569
    ev_cur = None
570
    ev_prev = None
571

    
572
    for args in acquires:
573
      ev_cur = threading.Event()
574
      self._addThread(target=_Acquire, args=args + (ev_cur, ev_prev))
575
      ev_prev = ev_cur
576

    
577
    # Wait for last acquire to start
578
    ev_prev.wait()
579

    
580
    # Expect 6 pending exclusive acquires and 1 for all shared acquires
581
    # together
582
    self.assertEqual(self.sl._count_pending(), 7)
583

    
584
    # Release exclusive lock and wait
585
    self.sl.release()
586

    
587
    self._waitThreads()
588

    
589
    # Check sequence
590
    for _ in range(shrcnt1 + shrcnt2 + shrcnt3 + shrcnt4):
591
      # Shared locks aren't guaranteed to be notified in order, but they'll be
592
      # first
593
      tmp = self.done.get_nowait()
594
      if tmp == "shared 1":
595
        shrcnt1 -= 1
596
      elif tmp == "shared 2":
597
        shrcnt2 -= 1
598
      elif tmp == "shared 3":
599
        shrcnt3 -= 1
600
      elif tmp == "shared 4":
601
        shrcnt4 -= 1
602
    self.assertEqual(shrcnt1, 0)
603
    self.assertEqual(shrcnt2, 0)
604
    self.assertEqual(shrcnt3, 0)
605
    self.assertEqual(shrcnt3, 0)
606

    
607
    for _ in range(3):
608
      self.assertEqual(self.done.get_nowait(), "exclusive 1")
609

    
610
    for _ in range(3):
611
      self.assertEqual(self.done.get_nowait(), "exclusive 2")
612

    
613
    self.assertRaises(Queue.Empty, self.done.get_nowait)
614

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

    
619
    # Acquire in shared mode, downgrade should be no-op
620
    self.assertTrue(self.sl.acquire(shared=1))
621
    self.assertTrue(self.sl._is_owned(shared=1))
622
    self.assertTrue(self.sl.downgrade())
623
    self.assertTrue(self.sl._is_owned(shared=1))
624
    self.sl.release()
625

    
626
  def testDowngrade(self):
627
    self.assertTrue(self.sl.acquire())
628
    self.assertTrue(self.sl._is_owned(shared=0))
629
    self.assertTrue(self.sl.downgrade())
630
    self.assertTrue(self.sl._is_owned(shared=1))
631
    self.sl.release()
632

    
633
  @_Repeat
634
  def testDowngradeJumpsAheadOfExclusive(self):
635
    def _KeepExclusive(ev_got, ev_downgrade, ev_release):
636
      self.assertTrue(self.sl.acquire())
637
      self.assertTrue(self.sl._is_owned(shared=0))
638
      ev_got.set()
639
      ev_downgrade.wait()
640
      self.assertTrue(self.sl._is_owned(shared=0))
641
      self.assertTrue(self.sl.downgrade())
642
      self.assertTrue(self.sl._is_owned(shared=1))
643
      ev_release.wait()
644
      self.assertTrue(self.sl._is_owned(shared=1))
645
      self.sl.release()
646

    
647
    def _KeepExclusive2(ev_started, ev_release):
648
      self.assertTrue(self.sl.acquire(test_notify=ev_started.set))
649
      self.assertTrue(self.sl._is_owned(shared=0))
650
      ev_release.wait()
651
      self.assertTrue(self.sl._is_owned(shared=0))
652
      self.sl.release()
653

    
654
    def _KeepShared(ev_started, ev_got, ev_release):
655
      self.assertTrue(self.sl.acquire(shared=1, test_notify=ev_started.set))
656
      self.assertTrue(self.sl._is_owned(shared=1))
657
      ev_got.set()
658
      ev_release.wait()
659
      self.assertTrue(self.sl._is_owned(shared=1))
660
      self.sl.release()
661

    
662
    # Acquire lock in exclusive mode
663
    ev_got_excl1 = threading.Event()
664
    ev_downgrade_excl1 = threading.Event()
665
    ev_release_excl1 = threading.Event()
666
    th_excl1 = self._addThread(target=_KeepExclusive,
667
                               args=(ev_got_excl1, ev_downgrade_excl1,
668
                                     ev_release_excl1))
669
    ev_got_excl1.wait()
670

    
671
    # Start a second exclusive acquire
672
    ev_started_excl2 = threading.Event()
673
    ev_release_excl2 = threading.Event()
674
    th_excl2 = self._addThread(target=_KeepExclusive2,
675
                               args=(ev_started_excl2, ev_release_excl2))
676
    ev_started_excl2.wait()
677

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

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

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

    
691
    # Check lock information
692
    self.assertEqual(self.sl.GetInfo(set([query.LQ_MODE, query.LQ_OWNER])),
693
                     (self.sl.name, "exclusive", [th_excl1.getName()], None))
694
    (_, _, _, pending) = self.sl.GetInfo(set([query.LQ_PENDING]))
695
    self.assertEqual([(pendmode, sorted(waiting))
696
                      for (pendmode, waiting) in pending],
697
                     [("exclusive", [th_excl2.getName()]),
698
                      ("shared", sorted(th.getName() for th in th_shared))])
699

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

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

    
707
    # Check lock information again
708
    self.assertEqual(self.sl.GetInfo(set([query.LQ_MODE, query.LQ_PENDING])),
709
                     (self.sl.name, "shared", None,
710
                      [("exclusive", [th_excl2.getName()])]))
711
    (_, _, owner, _) = self.sl.GetInfo(set([query.LQ_OWNER]))
712
    self.assertEqual(set(owner), set([th_excl1.getName()] +
713
                                     [th.getName() for th in th_shared]))
714

    
715
    ev_release_excl1.set()
716
    ev_release_excl2.set()
717
    ev_release_shared.set()
718

    
719
    self._waitThreads()
720

    
721
    self.assertEqual(self.sl.GetInfo(set([query.LQ_MODE, query.LQ_OWNER,
722
                                          query.LQ_PENDING])),
723
                     (self.sl.name, None, None, []))
724

    
725
  @_Repeat
726
  def testMixedAcquireTimeout(self):
727
    sync = threading.Event()
728

    
729
    def _AcquireShared(ev):
730
      if not self.sl.acquire(shared=1, timeout=None):
731
        return
732

    
733
      self.done.put("shared")
734

    
735
      # Notify main thread
736
      ev.set()
737

    
738
      # Wait for notification from main thread
739
      sync.wait()
740

    
741
      # Release lock
742
      self.sl.release()
743

    
744
    acquires = []
745
    for _ in range(3):
746
      ev = threading.Event()
747
      self._addThread(target=_AcquireShared, args=(ev, ))
748
      acquires.append(ev)
749

    
750
    # Wait for all acquires to finish
751
    for i in acquires:
752
      i.wait()
753

    
754
    self.assertEqual(self.sl._count_pending(), 0)
755

    
756
    # Try to get exclusive lock
757
    self.failIf(self.sl.acquire(shared=0, timeout=0.02))
758

    
759
    # Acquire exclusive without timeout
760
    exclsync = threading.Event()
761
    exclev = threading.Event()
762

    
763
    def _AcquireExclusive():
764
      if not self.sl.acquire(shared=0):
765
        return
766

    
767
      self.done.put("exclusive")
768

    
769
      # Notify main thread
770
      exclev.set()
771

    
772
      # Wait for notification from main thread
773
      exclsync.wait()
774

    
775
      self.sl.release()
776

    
777
    self._addThread(target=_AcquireExclusive)
778

    
779
    # Try to get exclusive lock
780
    self.failIf(self.sl.acquire(shared=0, timeout=0.02))
781

    
782
    # Make all shared holders release their locks
783
    sync.set()
784

    
785
    # Wait for exclusive acquire to succeed
786
    exclev.wait()
787

    
788
    self.assertEqual(self.sl._count_pending(), 0)
789

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

    
793
    def _AcquireSharedSimple():
794
      if self.sl.acquire(shared=1, timeout=None):
795
        self.done.put("shared2")
796
        self.sl.release()
797

    
798
    for _ in range(10):
799
      self._addThread(target=_AcquireSharedSimple)
800

    
801
    # Tell exclusive lock to release
802
    exclsync.set()
803

    
804
    # Wait for everything to finish
805
    self._waitThreads()
806

    
807
    self.assertEqual(self.sl._count_pending(), 0)
808

    
809
    # Check sequence
810
    for _ in range(3):
811
      self.assertEqual(self.done.get_nowait(), "shared")
812

    
813
    self.assertEqual(self.done.get_nowait(), "exclusive")
814

    
815
    for _ in range(10):
816
      self.assertEqual(self.done.get_nowait(), "shared2")
817

    
818
    self.assertRaises(Queue.Empty, self.done.get_nowait)
819

    
820
  def testPriority(self):
821
    # Acquire in exclusive mode
822
    self.assert_(self.sl.acquire(shared=0))
823

    
824
    # Queue acquires
825
    def _Acquire(prev, next, shared, priority, result):
826
      prev.wait()
827
      self.sl.acquire(shared=shared, priority=priority, test_notify=next.set)
828
      try:
829
        self.done.put(result)
830
      finally:
831
        self.sl.release()
832

    
833
    counter = itertools.count(0)
834
    priorities = range(-20, 30)
835
    first = threading.Event()
836
    prev = first
837

    
838
    # Data structure:
839
    # {
840
    #   priority:
841
    #     [(shared/exclusive, set(acquire names), set(pending threads)),
842
    #      (shared/exclusive, ...),
843
    #      ...,
844
    #     ],
845
    # }
846
    perprio = {}
847

    
848
    # References shared acquire per priority in L{perprio}. Data structure:
849
    # {
850
    #   priority: (shared=1, set(acquire names), set(pending threads)),
851
    # }
852
    prioshared = {}
853

    
854
    for seed in [4979, 9523, 14902, 32440]:
855
      # Use a deterministic random generator
856
      rnd = random.Random(seed)
857
      for priority in [rnd.choice(priorities) for _ in range(30)]:
858
        modes = [0, 1]
859
        rnd.shuffle(modes)
860
        for shared in modes:
861
          # Unique name
862
          acqname = "%s/shr=%s/prio=%s" % (counter.next(), shared, priority)
863

    
864
          ev = threading.Event()
865
          thread = self._addThread(target=_Acquire,
866
                                   args=(prev, ev, shared, priority, acqname))
867
          prev = ev
868

    
869
          # Record expected aqcuire, see above for structure
870
          data = (shared, set([acqname]), set([thread]))
871
          priolist = perprio.setdefault(priority, [])
872
          if shared:
873
            priosh = prioshared.get(priority, None)
874
            if priosh:
875
              # Shared acquires are merged
876
              for i, j in zip(priosh[1:], data[1:]):
877
                i.update(j)
878
              assert data[0] == priosh[0]
879
            else:
880
              prioshared[priority] = data
881
              priolist.append(data)
882
          else:
883
            priolist.append(data)
884

    
885
    # Start all acquires and wait for them
886
    first.set()
887
    prev.wait()
888

    
889
    # Check lock information
890
    self.assertEqual(self.sl.GetInfo(set()), (self.sl.name, None, None, None))
891
    self.assertEqual(self.sl.GetInfo(set([query.LQ_MODE, query.LQ_OWNER])),
892
                     (self.sl.name, "exclusive",
893
                      [threading.currentThread().getName()], None))
894

    
895
    self._VerifyPrioPending(self.sl.GetInfo(set([query.LQ_PENDING])), perprio)
896

    
897
    # Let threads acquire the lock
898
    self.sl.release()
899

    
900
    # Wait for everything to finish
901
    self._waitThreads()
902

    
903
    self.assert_(self.sl._check_empty())
904

    
905
    # Check acquires by priority
906
    for acquires in [perprio[i] for i in sorted(perprio.keys())]:
907
      for (_, names, _) in acquires:
908
        # For shared acquires, the set will contain 1..n entries. For exclusive
909
        # acquires only one.
910
        while names:
911
          names.remove(self.done.get_nowait())
912
      self.assertFalse(compat.any(names for (_, names, _) in acquires))
913

    
914
    self.assertRaises(Queue.Empty, self.done.get_nowait)
915

    
916
  def _VerifyPrioPending(self, (name, mode, owner, pending), perprio):
917
    self.assertEqual(name, self.sl.name)
918
    self.assert_(mode is None)
919
    self.assert_(owner is None)
920

    
921
    self.assertEqual([(pendmode, sorted(waiting))
922
                      for (pendmode, waiting) in pending],
923
                     [(["exclusive", "shared"][int(bool(shared))],
924
                       sorted(t.getName() for t in threads))
925
                      for acquires in [perprio[i]
926
                                       for i in sorted(perprio.keys())]
927
                      for (shared, _, threads) in acquires])
928

    
929

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

    
933
  def setUp(self):
934
    _ThreadedTestCase.setUp(self)
935
    self.sl = locking.SharedLock("TestSharedLockInCondition")
936
    self.setCondition()
937

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

    
941
  def testKeepMode(self):
942
    self.cond.acquire(shared=1)
943
    self.assert_(self.sl._is_owned(shared=1))
944
    self.cond.wait(0)
945
    self.assert_(self.sl._is_owned(shared=1))
946
    self.cond.release()
947
    self.cond.acquire(shared=0)
948
    self.assert_(self.sl._is_owned(shared=0))
949
    self.cond.wait(0)
950
    self.assert_(self.sl._is_owned(shared=0))
951
    self.cond.release()
952

    
953

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

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

    
960

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

    
964
  def setUp(self):
965
    _ThreadedTestCase.setUp(self)
966

    
967
  @locking.ssynchronized(_decoratorlock)
968
  def _doItExclusive(self):
969
    self.assert_(_decoratorlock._is_owned())
970
    self.done.put('EXC')
971

    
972
  @locking.ssynchronized(_decoratorlock, shared=1)
973
  def _doItSharer(self):
974
    self.assert_(_decoratorlock._is_owned(shared=1))
975
    self.done.put('SHR')
976

    
977
  def testDecoratedFunctions(self):
978
    self._doItExclusive()
979
    self.assertFalse(_decoratorlock._is_owned())
980
    self._doItSharer()
981
    self.assertFalse(_decoratorlock._is_owned())
982

    
983
  def testSharersCanCoexist(self):
984
    _decoratorlock.acquire(shared=1)
985
    threading.Thread(target=self._doItSharer).start()
986
    self.assert_(self.done.get(True, 1))
987
    _decoratorlock.release()
988

    
989
  @_Repeat
990
  def testExclusiveBlocksExclusive(self):
991
    _decoratorlock.acquire()
992
    self._addThread(target=self._doItExclusive)
993
    # give it a bit of time to check that it's not actually doing anything
994
    self.assertRaises(Queue.Empty, self.done.get_nowait)
995
    _decoratorlock.release()
996
    self._waitThreads()
997
    self.failUnlessEqual(self.done.get_nowait(), 'EXC')
998

    
999
  @_Repeat
1000
  def testExclusiveBlocksSharer(self):
1001
    _decoratorlock.acquire()
1002
    self._addThread(target=self._doItSharer)
1003
    self.assertRaises(Queue.Empty, self.done.get_nowait)
1004
    _decoratorlock.release()
1005
    self._waitThreads()
1006
    self.failUnlessEqual(self.done.get_nowait(), 'SHR')
1007

    
1008
  @_Repeat
1009
  def testSharerBlocksExclusive(self):
1010
    _decoratorlock.acquire(shared=1)
1011
    self._addThread(target=self._doItExclusive)
1012
    self.assertRaises(Queue.Empty, self.done.get_nowait)
1013
    _decoratorlock.release()
1014
    self._waitThreads()
1015
    self.failUnlessEqual(self.done.get_nowait(), 'EXC')
1016

    
1017

    
1018
class TestLockSet(_ThreadedTestCase):
1019
  """LockSet tests"""
1020

    
1021
  def setUp(self):
1022
    _ThreadedTestCase.setUp(self)
1023
    self._setUpLS()
1024

    
1025
  def _setUpLS(self):
1026
    """Helper to (re)initialize the lock set"""
1027
    self.resources = ['one', 'two', 'three']
1028
    self.ls = locking.LockSet(self.resources, "TestLockSet")
1029

    
1030
  def testResources(self):
1031
    self.assertEquals(self.ls._names(), set(self.resources))
1032
    newls = locking.LockSet([], "TestLockSet.testResources")
1033
    self.assertEquals(newls._names(), set())
1034

    
1035
  def testAcquireRelease(self):
1036
    self.assert_(self.ls.acquire('one'))
1037
    self.assertEquals(self.ls._list_owned(), set(['one']))
1038
    self.ls.release()
1039
    self.assertEquals(self.ls._list_owned(), set())
1040
    self.assertEquals(self.ls.acquire(['one']), set(['one']))
1041
    self.assertEquals(self.ls._list_owned(), set(['one']))
1042
    self.ls.release()
1043
    self.assertEquals(self.ls._list_owned(), set())
1044
    self.ls.acquire(['one', 'two', 'three'])
1045
    self.assertEquals(self.ls._list_owned(), set(['one', 'two', 'three']))
1046
    self.ls.release('one')
1047
    self.assertEquals(self.ls._list_owned(), set(['two', 'three']))
1048
    self.ls.release(['three'])
1049
    self.assertEquals(self.ls._list_owned(), set(['two']))
1050
    self.ls.release()
1051
    self.assertEquals(self.ls._list_owned(), set())
1052
    self.assertEquals(self.ls.acquire(['one', 'three']), set(['one', 'three']))
1053
    self.assertEquals(self.ls._list_owned(), set(['one', 'three']))
1054
    self.ls.release()
1055
    self.assertEquals(self.ls._list_owned(), set())
1056

    
1057
  def testNoDoubleAcquire(self):
1058
    self.ls.acquire('one')
1059
    self.assertRaises(AssertionError, self.ls.acquire, 'one')
1060
    self.assertRaises(AssertionError, self.ls.acquire, ['two'])
1061
    self.assertRaises(AssertionError, self.ls.acquire, ['two', 'three'])
1062
    self.ls.release()
1063
    self.ls.acquire(['one', 'three'])
1064
    self.ls.release('one')
1065
    self.assertRaises(AssertionError, self.ls.acquire, ['two'])
1066
    self.ls.release('three')
1067

    
1068
  def testNoWrongRelease(self):
1069
    self.assertRaises(AssertionError, self.ls.release)
1070
    self.ls.acquire('one')
1071
    self.assertRaises(AssertionError, self.ls.release, 'two')
1072

    
1073
  def testAddRemove(self):
1074
    self.ls.add('four')
1075
    self.assertEquals(self.ls._list_owned(), set())
1076
    self.assert_('four' in self.ls._names())
1077
    self.ls.add(['five', 'six', 'seven'], acquired=1)
1078
    self.assert_('five' in self.ls._names())
1079
    self.assert_('six' in self.ls._names())
1080
    self.assert_('seven' in self.ls._names())
1081
    self.assertEquals(self.ls._list_owned(), set(['five', 'six', 'seven']))
1082
    self.assertEquals(self.ls.remove(['five', 'six']), ['five', 'six'])
1083
    self.assert_('five' not in self.ls._names())
1084
    self.assert_('six' not in self.ls._names())
1085
    self.assertEquals(self.ls._list_owned(), set(['seven']))
1086
    self.assertRaises(AssertionError, self.ls.add, 'eight', acquired=1)
1087
    self.ls.remove('seven')
1088
    self.assert_('seven' not in self.ls._names())
1089
    self.assertEquals(self.ls._list_owned(), set([]))
1090
    self.ls.acquire(None, shared=1)
1091
    self.assertRaises(AssertionError, self.ls.add, 'eight')
1092
    self.ls.release()
1093
    self.ls.acquire(None)
1094
    self.ls.add('eight', acquired=1)
1095
    self.assert_('eight' in self.ls._names())
1096
    self.assert_('eight' in self.ls._list_owned())
1097
    self.ls.add('nine')
1098
    self.assert_('nine' in self.ls._names())
1099
    self.assert_('nine' not in self.ls._list_owned())
1100
    self.ls.release()
1101
    self.ls.remove(['two'])
1102
    self.assert_('two' not in self.ls._names())
1103
    self.ls.acquire('three')
1104
    self.assertEquals(self.ls.remove(['three']), ['three'])
1105
    self.assert_('three' not in self.ls._names())
1106
    self.assertEquals(self.ls.remove('three'), [])
1107
    self.assertEquals(self.ls.remove(['one', 'three', 'six']), ['one'])
1108
    self.assert_('one' not in self.ls._names())
1109

    
1110
  def testRemoveNonBlocking(self):
1111
    self.ls.acquire('one')
1112
    self.assertEquals(self.ls.remove('one'), ['one'])
1113
    self.ls.acquire(['two', 'three'])
1114
    self.assertEquals(self.ls.remove(['two', 'three']),
1115
                      ['two', 'three'])
1116

    
1117
  def testNoDoubleAdd(self):
1118
    self.assertRaises(errors.LockError, self.ls.add, 'two')
1119
    self.ls.add('four')
1120
    self.assertRaises(errors.LockError, self.ls.add, 'four')
1121

    
1122
  def testNoWrongRemoves(self):
1123
    self.ls.acquire(['one', 'three'], shared=1)
1124
    # Cannot remove 'two' while holding something which is not a superset
1125
    self.assertRaises(AssertionError, self.ls.remove, 'two')
1126
    # Cannot remove 'three' as we are sharing it
1127
    self.assertRaises(AssertionError, self.ls.remove, 'three')
1128

    
1129
  def testAcquireSetLock(self):
1130
    # acquire the set-lock exclusively
1131
    self.assertEquals(self.ls.acquire(None), set(['one', 'two', 'three']))
1132
    self.assertEquals(self.ls._list_owned(), set(['one', 'two', 'three']))
1133
    self.assertEquals(self.ls._is_owned(), True)
1134
    self.assertEquals(self.ls._names(), set(['one', 'two', 'three']))
1135
    # I can still add/remove elements...
1136
    self.assertEquals(self.ls.remove(['two', 'three']), ['two', 'three'])
1137
    self.assert_(self.ls.add('six'))
1138
    self.ls.release()
1139
    # share the set-lock
1140
    self.assertEquals(self.ls.acquire(None, shared=1), set(['one', 'six']))
1141
    # adding new elements is not possible
1142
    self.assertRaises(AssertionError, self.ls.add, 'five')
1143
    self.ls.release()
1144

    
1145
  def testAcquireWithRepetitions(self):
1146
    self.assertEquals(self.ls.acquire(['two', 'two', 'three'], shared=1),
1147
                      set(['two', 'two', 'three']))
1148
    self.ls.release(['two', 'two'])
1149
    self.assertEquals(self.ls._list_owned(), set(['three']))
1150

    
1151
  def testEmptyAcquire(self):
1152
    # Acquire an empty list of locks...
1153
    self.assertEquals(self.ls.acquire([]), set())
1154
    self.assertEquals(self.ls._list_owned(), set())
1155
    # New locks can still be addded
1156
    self.assert_(self.ls.add('six'))
1157
    # "re-acquiring" is not an issue, since we had really acquired nothing
1158
    self.assertEquals(self.ls.acquire([], shared=1), set())
1159
    self.assertEquals(self.ls._list_owned(), set())
1160
    # We haven't really acquired anything, so we cannot release
1161
    self.assertRaises(AssertionError, self.ls.release)
1162

    
1163
  def _doLockSet(self, names, shared):
1164
    try:
1165
      self.ls.acquire(names, shared=shared)
1166
      self.done.put('DONE')
1167
      self.ls.release()
1168
    except errors.LockError:
1169
      self.done.put('ERR')
1170

    
1171
  def _doAddSet(self, names):
1172
    try:
1173
      self.ls.add(names, acquired=1)
1174
      self.done.put('DONE')
1175
      self.ls.release()
1176
    except errors.LockError:
1177
      self.done.put('ERR')
1178

    
1179
  def _doRemoveSet(self, names):
1180
    self.done.put(self.ls.remove(names))
1181

    
1182
  @_Repeat
1183
  def testConcurrentSharedAcquire(self):
1184
    self.ls.acquire(['one', 'two'], shared=1)
1185
    self._addThread(target=self._doLockSet, args=(['one', 'two'], 1))
1186
    self._waitThreads()
1187
    self.assertEqual(self.done.get_nowait(), 'DONE')
1188
    self._addThread(target=self._doLockSet, args=(['one', 'two', 'three'], 1))
1189
    self._waitThreads()
1190
    self.assertEqual(self.done.get_nowait(), 'DONE')
1191
    self._addThread(target=self._doLockSet, args=('three', 1))
1192
    self._waitThreads()
1193
    self.assertEqual(self.done.get_nowait(), 'DONE')
1194
    self._addThread(target=self._doLockSet, args=(['one', 'two'], 0))
1195
    self._addThread(target=self._doLockSet, args=(['two', 'three'], 0))
1196
    self.assertRaises(Queue.Empty, self.done.get_nowait)
1197
    self.ls.release()
1198
    self._waitThreads()
1199
    self.assertEqual(self.done.get_nowait(), 'DONE')
1200
    self.assertEqual(self.done.get_nowait(), 'DONE')
1201

    
1202
  @_Repeat
1203
  def testConcurrentExclusiveAcquire(self):
1204
    self.ls.acquire(['one', 'two'])
1205
    self._addThread(target=self._doLockSet, args=('three', 1))
1206
    self._waitThreads()
1207
    self.assertEqual(self.done.get_nowait(), 'DONE')
1208
    self._addThread(target=self._doLockSet, args=('three', 0))
1209
    self._waitThreads()
1210
    self.assertEqual(self.done.get_nowait(), 'DONE')
1211
    self.assertRaises(Queue.Empty, self.done.get_nowait)
1212
    self._addThread(target=self._doLockSet, args=(['one', 'two'], 0))
1213
    self._addThread(target=self._doLockSet, args=(['one', 'two'], 1))
1214
    self._addThread(target=self._doLockSet, args=('one', 0))
1215
    self._addThread(target=self._doLockSet, args=('one', 1))
1216
    self._addThread(target=self._doLockSet, args=(['two', 'three'], 0))
1217
    self._addThread(target=self._doLockSet, args=(['two', 'three'], 1))
1218
    self.assertRaises(Queue.Empty, self.done.get_nowait)
1219
    self.ls.release()
1220
    self._waitThreads()
1221
    for _ in range(6):
1222
      self.failUnlessEqual(self.done.get_nowait(), 'DONE')
1223

    
1224
  @_Repeat
1225
  def testSimpleAcquireTimeoutExpiring(self):
1226
    names = sorted(self.ls._names())
1227
    self.assert_(len(names) >= 3)
1228

    
1229
    # Get name of first lock
1230
    first = names[0]
1231

    
1232
    # Get name of last lock
1233
    last = names.pop()
1234

    
1235
    checks = [
1236
      # Block first and try to lock it again
1237
      (first, first),
1238

    
1239
      # Block last and try to lock all locks
1240
      (None, first),
1241

    
1242
      # Block last and try to lock it again
1243
      (last, last),
1244
      ]
1245

    
1246
    for (wanted, block) in checks:
1247
      # Lock in exclusive mode
1248
      self.assert_(self.ls.acquire(block, shared=0))
1249

    
1250
      def _AcquireOne():
1251
        # Try to get the same lock again with a timeout (should never succeed)
1252
        acquired = self.ls.acquire(wanted, timeout=0.1, shared=0)
1253
        if acquired:
1254
          self.done.put("acquired")
1255
          self.ls.release()
1256
        else:
1257
          self.assert_(acquired is None)
1258
          self.assertFalse(self.ls._list_owned())
1259
          self.assertFalse(self.ls._is_owned())
1260
          self.done.put("not acquired")
1261

    
1262
      self._addThread(target=_AcquireOne)
1263

    
1264
      # Wait for timeout in thread to expire
1265
      self._waitThreads()
1266

    
1267
      # Release exclusive lock again
1268
      self.ls.release()
1269

    
1270
      self.assertEqual(self.done.get_nowait(), "not acquired")
1271
      self.assertRaises(Queue.Empty, self.done.get_nowait)
1272

    
1273
  @_Repeat
1274
  def testDelayedAndExpiringLockAcquire(self):
1275
    self._setUpLS()
1276
    self.ls.add(['five', 'six', 'seven', 'eight', 'nine'])
1277

    
1278
    for expire in (False, True):
1279
      names = sorted(self.ls._names())
1280
      self.assertEqual(len(names), 8)
1281

    
1282
      lock_ev = dict([(i, threading.Event()) for i in names])
1283

    
1284
      # Lock all in exclusive mode
1285
      self.assert_(self.ls.acquire(names, shared=0))
1286

    
1287
      if expire:
1288
        # We'll wait at least 300ms per lock
1289
        lockwait = len(names) * [0.3]
1290

    
1291
        # Fail if we can't acquire all locks in 400ms. There are 8 locks, so
1292
        # this gives us up to 2.4s to fail.
1293
        lockall_timeout = 0.4
1294
      else:
1295
        # This should finish rather quickly
1296
        lockwait = None
1297
        lockall_timeout = len(names) * 5.0
1298

    
1299
      def _LockAll():
1300
        def acquire_notification(name):
1301
          if not expire:
1302
            self.done.put("getting %s" % name)
1303

    
1304
          # Kick next lock
1305
          lock_ev[name].set()
1306

    
1307
        if self.ls.acquire(names, shared=0, timeout=lockall_timeout,
1308
                           test_notify=acquire_notification):
1309
          self.done.put("got all")
1310
          self.ls.release()
1311
        else:
1312
          self.done.put("timeout on all")
1313

    
1314
        # Notify all locks
1315
        for ev in lock_ev.values():
1316
          ev.set()
1317

    
1318
      t = self._addThread(target=_LockAll)
1319

    
1320
      for idx, name in enumerate(names):
1321
        # Wait for actual acquire on this lock to start
1322
        lock_ev[name].wait(10.0)
1323

    
1324
        if expire and t.isAlive():
1325
          # Wait some time after getting the notification to make sure the lock
1326
          # acquire will expire
1327
          SafeSleep(lockwait[idx])
1328

    
1329
        self.ls.release(names=name)
1330

    
1331
      self.assertFalse(self.ls._list_owned())
1332

    
1333
      self._waitThreads()
1334

    
1335
      if expire:
1336
        # Not checking which locks were actually acquired. Doing so would be
1337
        # too timing-dependant.
1338
        self.assertEqual(self.done.get_nowait(), "timeout on all")
1339
      else:
1340
        for i in names:
1341
          self.assertEqual(self.done.get_nowait(), "getting %s" % i)
1342
        self.assertEqual(self.done.get_nowait(), "got all")
1343
      self.assertRaises(Queue.Empty, self.done.get_nowait)
1344

    
1345
  @_Repeat
1346
  def testConcurrentRemove(self):
1347
    self.ls.add('four')
1348
    self.ls.acquire(['one', 'two', 'four'])
1349
    self._addThread(target=self._doLockSet, args=(['one', 'four'], 0))
1350
    self._addThread(target=self._doLockSet, args=(['one', 'four'], 1))
1351
    self._addThread(target=self._doLockSet, args=(['one', 'two'], 0))
1352
    self._addThread(target=self._doLockSet, args=(['one', 'two'], 1))
1353
    self.assertRaises(Queue.Empty, self.done.get_nowait)
1354
    self.ls.remove('one')
1355
    self.ls.release()
1356
    self._waitThreads()
1357
    for i in range(4):
1358
      self.failUnlessEqual(self.done.get_nowait(), 'ERR')
1359
    self.ls.add(['five', 'six'], acquired=1)
1360
    self._addThread(target=self._doLockSet, args=(['three', 'six'], 1))
1361
    self._addThread(target=self._doLockSet, args=(['three', 'six'], 0))
1362
    self._addThread(target=self._doLockSet, args=(['four', 'six'], 1))
1363
    self._addThread(target=self._doLockSet, args=(['four', 'six'], 0))
1364
    self.ls.remove('five')
1365
    self.ls.release()
1366
    self._waitThreads()
1367
    for i in range(4):
1368
      self.failUnlessEqual(self.done.get_nowait(), 'DONE')
1369
    self.ls.acquire(['three', 'four'])
1370
    self._addThread(target=self._doRemoveSet, args=(['four', 'six'], ))
1371
    self.assertRaises(Queue.Empty, self.done.get_nowait)
1372
    self.ls.remove('four')
1373
    self._waitThreads()
1374
    self.assertEqual(self.done.get_nowait(), ['six'])
1375
    self._addThread(target=self._doRemoveSet, args=(['two']))
1376
    self._waitThreads()
1377
    self.assertEqual(self.done.get_nowait(), ['two'])
1378
    self.ls.release()
1379
    # reset lockset
1380
    self._setUpLS()
1381

    
1382
  @_Repeat
1383
  def testConcurrentSharedSetLock(self):
1384
    # share the set-lock...
1385
    self.ls.acquire(None, shared=1)
1386
    # ...another thread can share it too
1387
    self._addThread(target=self._doLockSet, args=(None, 1))
1388
    self._waitThreads()
1389
    self.assertEqual(self.done.get_nowait(), 'DONE')
1390
    # ...or just share some elements
1391
    self._addThread(target=self._doLockSet, args=(['one', 'three'], 1))
1392
    self._waitThreads()
1393
    self.assertEqual(self.done.get_nowait(), 'DONE')
1394
    # ...but not add new ones or remove any
1395
    t = self._addThread(target=self._doAddSet, args=(['nine']))
1396
    self._addThread(target=self._doRemoveSet, args=(['two'], ))
1397
    self.assertRaises(Queue.Empty, self.done.get_nowait)
1398
    # this just releases the set-lock
1399
    self.ls.release([])
1400
    t.join(60)
1401
    self.assertEqual(self.done.get_nowait(), 'DONE')
1402
    # release the lock on the actual elements so remove() can proceed too
1403
    self.ls.release()
1404
    self._waitThreads()
1405
    self.failUnlessEqual(self.done.get_nowait(), ['two'])
1406
    # reset lockset
1407
    self._setUpLS()
1408

    
1409
  @_Repeat
1410
  def testConcurrentExclusiveSetLock(self):
1411
    # acquire the set-lock...
1412
    self.ls.acquire(None, shared=0)
1413
    # ...no one can do anything else
1414
    self._addThread(target=self._doLockSet, args=(None, 1))
1415
    self._addThread(target=self._doLockSet, args=(None, 0))
1416
    self._addThread(target=self._doLockSet, args=(['three'], 0))
1417
    self._addThread(target=self._doLockSet, args=(['two'], 1))
1418
    self._addThread(target=self._doAddSet, args=(['nine']))
1419
    self.assertRaises(Queue.Empty, self.done.get_nowait)
1420
    self.ls.release()
1421
    self._waitThreads()
1422
    for _ in range(5):
1423
      self.assertEqual(self.done.get(True, 1), 'DONE')
1424
    # cleanup
1425
    self._setUpLS()
1426

    
1427
  @_Repeat
1428
  def testConcurrentSetLockAdd(self):
1429
    self.ls.acquire('one')
1430
    # Another thread wants the whole SetLock
1431
    self._addThread(target=self._doLockSet, args=(None, 0))
1432
    self._addThread(target=self._doLockSet, args=(None, 1))
1433
    self.assertRaises(Queue.Empty, self.done.get_nowait)
1434
    self.assertRaises(AssertionError, self.ls.add, 'four')
1435
    self.ls.release()
1436
    self._waitThreads()
1437
    self.assertEqual(self.done.get_nowait(), 'DONE')
1438
    self.assertEqual(self.done.get_nowait(), 'DONE')
1439
    self.ls.acquire(None)
1440
    self._addThread(target=self._doLockSet, args=(None, 0))
1441
    self._addThread(target=self._doLockSet, args=(None, 1))
1442
    self.assertRaises(Queue.Empty, self.done.get_nowait)
1443
    self.ls.add('four')
1444
    self.ls.add('five', acquired=1)
1445
    self.ls.add('six', acquired=1, shared=1)
1446
    self.assertEquals(self.ls._list_owned(),
1447
      set(['one', 'two', 'three', 'five', 'six']))
1448
    self.assertEquals(self.ls._is_owned(), True)
1449
    self.assertEquals(self.ls._names(),
1450
      set(['one', 'two', 'three', 'four', 'five', 'six']))
1451
    self.ls.release()
1452
    self._waitThreads()
1453
    self.assertEqual(self.done.get_nowait(), 'DONE')
1454
    self.assertEqual(self.done.get_nowait(), 'DONE')
1455
    self._setUpLS()
1456

    
1457
  @_Repeat
1458
  def testEmptyLockSet(self):
1459
    # get the set-lock
1460
    self.assertEqual(self.ls.acquire(None), set(['one', 'two', 'three']))
1461
    # now empty it...
1462
    self.ls.remove(['one', 'two', 'three'])
1463
    # and adds/locks by another thread still wait
1464
    self._addThread(target=self._doAddSet, args=(['nine']))
1465
    self._addThread(target=self._doLockSet, args=(None, 1))
1466
    self._addThread(target=self._doLockSet, args=(None, 0))
1467
    self.assertRaises(Queue.Empty, self.done.get_nowait)
1468
    self.ls.release()
1469
    self._waitThreads()
1470
    for _ in range(3):
1471
      self.assertEqual(self.done.get_nowait(), 'DONE')
1472
    # empty it again...
1473
    self.assertEqual(self.ls.remove(['nine']), ['nine'])
1474
    # now share it...
1475
    self.assertEqual(self.ls.acquire(None, shared=1), set())
1476
    # other sharers can go, adds still wait
1477
    self._addThread(target=self._doLockSet, args=(None, 1))
1478
    self._waitThreads()
1479
    self.assertEqual(self.done.get_nowait(), 'DONE')
1480
    self._addThread(target=self._doAddSet, args=(['nine']))
1481
    self.assertRaises(Queue.Empty, self.done.get_nowait)
1482
    self.ls.release()
1483
    self._waitThreads()
1484
    self.assertEqual(self.done.get_nowait(), 'DONE')
1485
    self._setUpLS()
1486

    
1487
  def testAcquireWithNamesDowngrade(self):
1488
    self.assertEquals(self.ls.acquire("two", shared=0), set(["two"]))
1489
    self.assertTrue(self.ls._is_owned())
1490
    self.assertFalse(self.ls._get_lock()._is_owned())
1491
    self.ls.release()
1492
    self.assertFalse(self.ls._is_owned())
1493
    self.assertFalse(self.ls._get_lock()._is_owned())
1494
    # Can't downgrade after releasing
1495
    self.assertRaises(AssertionError, self.ls.downgrade, "two")
1496

    
1497
  def testDowngrade(self):
1498
    # Not owning anything, must raise an exception
1499
    self.assertFalse(self.ls._is_owned())
1500
    self.assertRaises(AssertionError, self.ls.downgrade)
1501

    
1502
    self.assertFalse(compat.any(i._is_owned()
1503
                                for i in self.ls._get_lockdict().values()))
1504

    
1505
    self.assertEquals(self.ls.acquire(None, shared=0),
1506
                      set(["one", "two", "three"]))
1507
    self.assertRaises(AssertionError, self.ls.downgrade, "unknown lock")
1508

    
1509
    self.assertTrue(self.ls._get_lock()._is_owned(shared=0))
1510
    self.assertTrue(compat.all(i._is_owned(shared=0)
1511
                               for i in self.ls._get_lockdict().values()))
1512

    
1513
    # Start downgrading locks
1514
    self.assertTrue(self.ls.downgrade(names=["one"]))
1515
    self.assertTrue(self.ls._get_lock()._is_owned(shared=0))
1516
    self.assertTrue(compat.all(lock._is_owned(shared=[0, 1][int(name == "one")])
1517
                               for name, lock in
1518
                                 self.ls._get_lockdict().items()))
1519

    
1520
    self.assertTrue(self.ls.downgrade(names="two"))
1521
    self.assertTrue(self.ls._get_lock()._is_owned(shared=0))
1522
    should_share = lambda name: [0, 1][int(name in ("one", "two"))]
1523
    self.assertTrue(compat.all(lock._is_owned(shared=should_share(name))
1524
                               for name, lock in
1525
                                 self.ls._get_lockdict().items()))
1526

    
1527
    # Downgrading the last exclusive lock to shared must downgrade the
1528
    # lockset-internal lock too
1529
    self.assertTrue(self.ls.downgrade(names="three"))
1530
    self.assertTrue(self.ls._get_lock()._is_owned(shared=1))
1531
    self.assertTrue(compat.all(i._is_owned(shared=1)
1532
                               for i in self.ls._get_lockdict().values()))
1533

    
1534
    # Downgrading a shared lock must be a no-op
1535
    self.assertTrue(self.ls.downgrade(names=["one", "three"]))
1536
    self.assertTrue(self.ls._get_lock()._is_owned(shared=1))
1537
    self.assertTrue(compat.all(i._is_owned(shared=1)
1538
                               for i in self.ls._get_lockdict().values()))
1539

    
1540
    self.ls.release()
1541

    
1542
  def testPriority(self):
1543
    def _Acquire(prev, next, name, priority, success_fn):
1544
      prev.wait()
1545
      self.assert_(self.ls.acquire(name, shared=0,
1546
                                   priority=priority,
1547
                                   test_notify=lambda _: next.set()))
1548
      try:
1549
        success_fn()
1550
      finally:
1551
        self.ls.release()
1552

    
1553
    # Get all in exclusive mode
1554
    self.assert_(self.ls.acquire(locking.ALL_SET, shared=0))
1555

    
1556
    done_two = Queue.Queue(0)
1557

    
1558
    first = threading.Event()
1559
    prev = first
1560

    
1561
    acquires = [("one", prio, self.done) for prio in range(1, 33)]
1562
    acquires.extend([("two", prio, done_two) for prio in range(1, 33)])
1563

    
1564
    # Use a deterministic random generator
1565
    random.Random(741).shuffle(acquires)
1566

    
1567
    for (name, prio, done) in acquires:
1568
      ev = threading.Event()
1569
      self._addThread(target=_Acquire,
1570
                      args=(prev, ev, name, prio,
1571
                            compat.partial(done.put, "Prio%s" % prio)))
1572
      prev = ev
1573

    
1574
    # Start acquires
1575
    first.set()
1576

    
1577
    # Wait for last acquire to start
1578
    prev.wait()
1579

    
1580
    # Let threads acquire locks
1581
    self.ls.release()
1582

    
1583
    # Wait for threads to finish
1584
    self._waitThreads()
1585

    
1586
    for i in range(1, 33):
1587
      self.assertEqual(self.done.get_nowait(), "Prio%s" % i)
1588
      self.assertEqual(done_two.get_nowait(), "Prio%s" % i)
1589

    
1590
    self.assertRaises(Queue.Empty, self.done.get_nowait)
1591
    self.assertRaises(Queue.Empty, done_two.get_nowait)
1592

    
1593

    
1594
class TestGanetiLockManager(_ThreadedTestCase):
1595

    
1596
  def setUp(self):
1597
    _ThreadedTestCase.setUp(self)
1598
    self.nodes=['n1', 'n2']
1599
    self.nodegroups=['g1', 'g2']
1600
    self.instances=['i1', 'i2', 'i3']
1601
    self.GL = locking.GanetiLockManager(self.nodes, self.nodegroups,
1602
                                        self.instances)
1603

    
1604
  def tearDown(self):
1605
    # Don't try this at home...
1606
    locking.GanetiLockManager._instance = None
1607

    
1608
  def testLockingConstants(self):
1609
    # The locking library internally cheats by assuming its constants have some
1610
    # relationships with each other. Check those hold true.
1611
    # This relationship is also used in the Processor to recursively acquire
1612
    # the right locks. Again, please don't break it.
1613
    for i in range(len(locking.LEVELS)):
1614
      self.assertEqual(i, locking.LEVELS[i])
1615

    
1616
  def testDoubleGLFails(self):
1617
    self.assertRaises(AssertionError, locking.GanetiLockManager, [], [], [])
1618

    
1619
  def testLockNames(self):
1620
    self.assertEqual(self.GL._names(locking.LEVEL_CLUSTER), set(['BGL']))
1621
    self.assertEqual(self.GL._names(locking.LEVEL_NODE), set(self.nodes))
1622
    self.assertEqual(self.GL._names(locking.LEVEL_NODEGROUP),
1623
                     set(self.nodegroups))
1624
    self.assertEqual(self.GL._names(locking.LEVEL_INSTANCE),
1625
                     set(self.instances))
1626

    
1627
  def testInitAndResources(self):
1628
    locking.GanetiLockManager._instance = None
1629
    self.GL = locking.GanetiLockManager([], [], [])
1630
    self.assertEqual(self.GL._names(locking.LEVEL_CLUSTER), set(['BGL']))
1631
    self.assertEqual(self.GL._names(locking.LEVEL_NODE), set())
1632
    self.assertEqual(self.GL._names(locking.LEVEL_NODEGROUP), set())
1633
    self.assertEqual(self.GL._names(locking.LEVEL_INSTANCE), set())
1634

    
1635
    locking.GanetiLockManager._instance = None
1636
    self.GL = locking.GanetiLockManager(self.nodes, self.nodegroups, [])
1637
    self.assertEqual(self.GL._names(locking.LEVEL_CLUSTER), set(['BGL']))
1638
    self.assertEqual(self.GL._names(locking.LEVEL_NODE), set(self.nodes))
1639
    self.assertEqual(self.GL._names(locking.LEVEL_NODEGROUP),
1640
                                    set(self.nodegroups))
1641
    self.assertEqual(self.GL._names(locking.LEVEL_INSTANCE), set())
1642

    
1643
    locking.GanetiLockManager._instance = None
1644
    self.GL = locking.GanetiLockManager([], [], self.instances)
1645
    self.assertEqual(self.GL._names(locking.LEVEL_CLUSTER), set(['BGL']))
1646
    self.assertEqual(self.GL._names(locking.LEVEL_NODE), set())
1647
    self.assertEqual(self.GL._names(locking.LEVEL_NODEGROUP), set())
1648
    self.assertEqual(self.GL._names(locking.LEVEL_INSTANCE),
1649
                     set(self.instances))
1650

    
1651
  def testAcquireRelease(self):
1652
    self.GL.acquire(locking.LEVEL_CLUSTER, ['BGL'], shared=1)
1653
    self.assertEquals(self.GL._list_owned(locking.LEVEL_CLUSTER), set(['BGL']))
1654
    self.GL.acquire(locking.LEVEL_INSTANCE, ['i1'])
1655
    self.GL.acquire(locking.LEVEL_NODEGROUP, ['g2'])
1656
    self.GL.acquire(locking.LEVEL_NODE, ['n1', 'n2'], shared=1)
1657
    self.GL.release(locking.LEVEL_NODE, ['n2'])
1658
    self.assertEquals(self.GL._list_owned(locking.LEVEL_NODE), set(['n1']))
1659
    self.assertEquals(self.GL._list_owned(locking.LEVEL_NODEGROUP), set(['g2']))
1660
    self.assertEquals(self.GL._list_owned(locking.LEVEL_INSTANCE), set(['i1']))
1661
    self.GL.release(locking.LEVEL_NODE)
1662
    self.assertEquals(self.GL._list_owned(locking.LEVEL_NODE), set())
1663
    self.assertEquals(self.GL._list_owned(locking.LEVEL_NODEGROUP), set(['g2']))
1664
    self.assertEquals(self.GL._list_owned(locking.LEVEL_INSTANCE), set(['i1']))
1665
    self.GL.release(locking.LEVEL_NODEGROUP)
1666
    self.GL.release(locking.LEVEL_INSTANCE)
1667
    self.assertRaises(errors.LockError, self.GL.acquire,
1668
                      locking.LEVEL_INSTANCE, ['i5'])
1669
    self.GL.acquire(locking.LEVEL_INSTANCE, ['i3'], shared=1)
1670
    self.assertEquals(self.GL._list_owned(locking.LEVEL_INSTANCE), set(['i3']))
1671

    
1672
  def testAcquireWholeSets(self):
1673
    self.GL.acquire(locking.LEVEL_CLUSTER, ['BGL'], shared=1)
1674
    self.assertEquals(self.GL.acquire(locking.LEVEL_INSTANCE, None),
1675
                      set(self.instances))
1676
    self.assertEquals(self.GL._list_owned(locking.LEVEL_INSTANCE),
1677
                      set(self.instances))
1678
    self.assertEquals(self.GL.acquire(locking.LEVEL_NODEGROUP, None),
1679
                      set(self.nodegroups))
1680
    self.assertEquals(self.GL._list_owned(locking.LEVEL_NODEGROUP),
1681
                      set(self.nodegroups))
1682
    self.assertEquals(self.GL.acquire(locking.LEVEL_NODE, None, shared=1),
1683
                      set(self.nodes))
1684
    self.assertEquals(self.GL._list_owned(locking.LEVEL_NODE),
1685
                      set(self.nodes))
1686
    self.GL.release(locking.LEVEL_NODE)
1687
    self.GL.release(locking.LEVEL_NODEGROUP)
1688
    self.GL.release(locking.LEVEL_INSTANCE)
1689
    self.GL.release(locking.LEVEL_CLUSTER)
1690

    
1691
  def testAcquireWholeAndPartial(self):
1692
    self.GL.acquire(locking.LEVEL_CLUSTER, ['BGL'], shared=1)
1693
    self.assertEquals(self.GL.acquire(locking.LEVEL_INSTANCE, None),
1694
                      set(self.instances))
1695
    self.assertEquals(self.GL._list_owned(locking.LEVEL_INSTANCE),
1696
                      set(self.instances))
1697
    self.assertEquals(self.GL.acquire(locking.LEVEL_NODE, ['n2'], shared=1),
1698
                      set(['n2']))
1699
    self.assertEquals(self.GL._list_owned(locking.LEVEL_NODE),
1700
                      set(['n2']))
1701
    self.GL.release(locking.LEVEL_NODE)
1702
    self.GL.release(locking.LEVEL_INSTANCE)
1703
    self.GL.release(locking.LEVEL_CLUSTER)
1704

    
1705
  def testBGLDependency(self):
1706
    self.assertRaises(AssertionError, self.GL.acquire,
1707
                      locking.LEVEL_NODE, ['n1', 'n2'])
1708
    self.assertRaises(AssertionError, self.GL.acquire,
1709
                      locking.LEVEL_INSTANCE, ['i3'])
1710
    self.assertRaises(AssertionError, self.GL.acquire,
1711
                      locking.LEVEL_NODEGROUP, ['g1'])
1712
    self.GL.acquire(locking.LEVEL_CLUSTER, ['BGL'], shared=1)
1713
    self.GL.acquire(locking.LEVEL_NODE, ['n1'])
1714
    self.assertRaises(AssertionError, self.GL.release,
1715
                      locking.LEVEL_CLUSTER, ['BGL'])
1716
    self.assertRaises(AssertionError, self.GL.release,
1717
                      locking.LEVEL_CLUSTER)
1718
    self.GL.release(locking.LEVEL_NODE)
1719
    self.GL.acquire(locking.LEVEL_INSTANCE, ['i1', 'i2'])
1720
    self.assertRaises(AssertionError, self.GL.release,
1721
                      locking.LEVEL_CLUSTER, ['BGL'])
1722
    self.assertRaises(AssertionError, self.GL.release,
1723
                      locking.LEVEL_CLUSTER)
1724
    self.GL.release(locking.LEVEL_INSTANCE)
1725
    self.GL.acquire(locking.LEVEL_NODEGROUP, None)
1726
    self.GL.release(locking.LEVEL_NODEGROUP, ['g1'])
1727
    self.assertRaises(AssertionError, self.GL.release,
1728
                      locking.LEVEL_CLUSTER, ['BGL'])
1729
    self.assertRaises(AssertionError, self.GL.release,
1730
                      locking.LEVEL_CLUSTER)
1731
    self.GL.release(locking.LEVEL_NODEGROUP)
1732
    self.GL.release(locking.LEVEL_CLUSTER)
1733

    
1734
  def testWrongOrder(self):
1735
    self.GL.acquire(locking.LEVEL_CLUSTER, ['BGL'], shared=1)
1736
    self.GL.acquire(locking.LEVEL_NODE, ['n2'])
1737
    self.assertRaises(AssertionError, self.GL.acquire,
1738
                      locking.LEVEL_NODE, ['n1'])
1739
    self.assertRaises(AssertionError, self.GL.acquire,
1740
                      locking.LEVEL_NODEGROUP, ['g1'])
1741
    self.assertRaises(AssertionError, self.GL.acquire,
1742
                      locking.LEVEL_INSTANCE, ['i2'])
1743

    
1744
  def testModifiableLevels(self):
1745
    self.assertRaises(AssertionError, self.GL.add, locking.LEVEL_CLUSTER,
1746
                      ['BGL2'])
1747
    self.GL.acquire(locking.LEVEL_CLUSTER, ['BGL'])
1748
    self.GL.add(locking.LEVEL_INSTANCE, ['i4'])
1749
    self.GL.remove(locking.LEVEL_INSTANCE, ['i3'])
1750
    self.GL.remove(locking.LEVEL_INSTANCE, ['i1'])
1751
    self.assertEqual(self.GL._names(locking.LEVEL_INSTANCE), set(['i2', 'i4']))
1752
    self.GL.add(locking.LEVEL_NODE, ['n3'])
1753
    self.GL.remove(locking.LEVEL_NODE, ['n1'])
1754
    self.assertEqual(self.GL._names(locking.LEVEL_NODE), set(['n2', 'n3']))
1755
    self.GL.add(locking.LEVEL_NODEGROUP, ['g3'])
1756
    self.GL.remove(locking.LEVEL_NODEGROUP, ['g2'])
1757
    self.GL.remove(locking.LEVEL_NODEGROUP, ['g1'])
1758
    self.assertEqual(self.GL._names(locking.LEVEL_NODEGROUP), set(['g3']))
1759
    self.assertRaises(AssertionError, self.GL.remove, locking.LEVEL_CLUSTER,
1760
                      ['BGL2'])
1761

    
1762
  # Helper function to run as a thread that shared the BGL and then acquires
1763
  # some locks at another level.
1764
  def _doLock(self, level, names, shared):
1765
    try:
1766
      self.GL.acquire(locking.LEVEL_CLUSTER, ['BGL'], shared=1)
1767
      self.GL.acquire(level, names, shared=shared)
1768
      self.done.put('DONE')
1769
      self.GL.release(level)
1770
      self.GL.release(locking.LEVEL_CLUSTER)
1771
    except errors.LockError:
1772
      self.done.put('ERR')
1773

    
1774
  @_Repeat
1775
  def testConcurrency(self):
1776
    self.GL.acquire(locking.LEVEL_CLUSTER, ['BGL'], shared=1)
1777
    self._addThread(target=self._doLock,
1778
                    args=(locking.LEVEL_INSTANCE, 'i1', 1))
1779
    self._waitThreads()
1780
    self.assertEqual(self.done.get_nowait(), 'DONE')
1781
    self.GL.acquire(locking.LEVEL_INSTANCE, ['i3'])
1782
    self._addThread(target=self._doLock,
1783
                    args=(locking.LEVEL_INSTANCE, 'i1', 1))
1784
    self._waitThreads()
1785
    self.assertEqual(self.done.get_nowait(), 'DONE')
1786
    self._addThread(target=self._doLock,
1787
                    args=(locking.LEVEL_INSTANCE, 'i3', 1))
1788
    self.assertRaises(Queue.Empty, self.done.get_nowait)
1789
    self.GL.release(locking.LEVEL_INSTANCE)
1790
    self._waitThreads()
1791
    self.assertEqual(self.done.get_nowait(), 'DONE')
1792
    self.GL.acquire(locking.LEVEL_INSTANCE, ['i2'], shared=1)
1793
    self._addThread(target=self._doLock,
1794
                    args=(locking.LEVEL_INSTANCE, 'i2', 1))
1795
    self._waitThreads()
1796
    self.assertEqual(self.done.get_nowait(), 'DONE')
1797
    self._addThread(target=self._doLock,
1798
                    args=(locking.LEVEL_INSTANCE, 'i2', 0))
1799
    self.assertRaises(Queue.Empty, self.done.get_nowait)
1800
    self.GL.release(locking.LEVEL_INSTANCE)
1801
    self._waitThreads()
1802
    self.assertEqual(self.done.get(True, 1), 'DONE')
1803
    self.GL.release(locking.LEVEL_CLUSTER, ['BGL'])
1804

    
1805

    
1806
class TestLockMonitor(_ThreadedTestCase):
1807
  def setUp(self):
1808
    _ThreadedTestCase.setUp(self)
1809
    self.lm = locking.LockMonitor()
1810

    
1811
  def testSingleThread(self):
1812
    locks = []
1813

    
1814
    for i in range(100):
1815
      name = "TestLock%s" % i
1816
      locks.append(locking.SharedLock(name, monitor=self.lm))
1817

    
1818
    self.assertEqual(len(self.lm._locks), len(locks))
1819
    result = objects.QueryResponse.FromDict(self.lm.QueryLocks(["name"]))
1820
    self.assertEqual(len(result.fields), 1)
1821
    self.assertEqual(len(result.data), 100)
1822

    
1823
    # Delete all locks
1824
    del locks[:]
1825

    
1826
    # The garbage collector might needs some time
1827
    def _CheckLocks():
1828
      if self.lm._locks:
1829
        raise utils.RetryAgain()
1830

    
1831
    utils.Retry(_CheckLocks, 0.1, 30.0)
1832

    
1833
    self.assertFalse(self.lm._locks)
1834

    
1835
  def testMultiThread(self):
1836
    locks = []
1837

    
1838
    def _CreateLock(prev, next, name):
1839
      prev.wait()
1840
      locks.append(locking.SharedLock(name, monitor=self.lm))
1841
      if next:
1842
        next.set()
1843

    
1844
    expnames = []
1845

    
1846
    first = threading.Event()
1847
    prev = first
1848

    
1849
    # Use a deterministic random generator
1850
    for i in random.Random(4263).sample(range(100), 33):
1851
      name = "MtTestLock%s" % i
1852
      expnames.append(name)
1853

    
1854
      ev = threading.Event()
1855
      self._addThread(target=_CreateLock, args=(prev, ev, name))
1856
      prev = ev
1857

    
1858
    # Add locks
1859
    first.set()
1860
    self._waitThreads()
1861

    
1862
    # Check order in which locks were added
1863
    self.assertEqual([i.name for i in locks], expnames)
1864

    
1865
    # Check query result
1866
    result = self.lm.QueryLocks(["name", "mode", "owner", "pending"])
1867
    self.assert_(isinstance(result, dict))
1868
    response = objects.QueryResponse.FromDict(result)
1869
    self.assertEqual(response.data,
1870
                     [[(constants.RS_NORMAL, name),
1871
                       (constants.RS_NORMAL, None),
1872
                       (constants.RS_NORMAL, None),
1873
                       (constants.RS_NORMAL, [])]
1874
                      for name in utils.NiceSort(expnames)])
1875
    self.assertEqual(len(response.fields), 4)
1876
    self.assertEqual(["name", "mode", "owner", "pending"],
1877
                     [fdef.name for fdef in response.fields])
1878

    
1879
    # Test exclusive acquire
1880
    for tlock in locks[::4]:
1881
      tlock.acquire(shared=0)
1882
      try:
1883
        def _GetExpResult(name):
1884
          if tlock.name == name:
1885
            return [(constants.RS_NORMAL, name),
1886
                    (constants.RS_NORMAL, "exclusive"),
1887
                    (constants.RS_NORMAL,
1888
                     [threading.currentThread().getName()]),
1889
                    (constants.RS_NORMAL, [])]
1890
          return [(constants.RS_NORMAL, name),
1891
                  (constants.RS_NORMAL, None),
1892
                  (constants.RS_NORMAL, None),
1893
                  (constants.RS_NORMAL, [])]
1894

    
1895
        result = self.lm.QueryLocks(["name", "mode", "owner", "pending"])
1896
        self.assertEqual(objects.QueryResponse.FromDict(result).data,
1897
                         [_GetExpResult(name)
1898
                          for name in utils.NiceSort(expnames)])
1899
      finally:
1900
        tlock.release()
1901

    
1902
    # Test shared acquire
1903
    def _Acquire(lock, shared, ev, notify):
1904
      lock.acquire(shared=shared)
1905
      try:
1906
        notify.set()
1907
        ev.wait()
1908
      finally:
1909
        lock.release()
1910

    
1911
    for tlock1 in locks[::11]:
1912
      for tlock2 in locks[::-15]:
1913
        if tlock2 == tlock1:
1914
          # Avoid deadlocks
1915
          continue
1916

    
1917
        for tlock3 in locks[::10]:
1918
          if tlock3 in (tlock2, tlock1):
1919
            # Avoid deadlocks
1920
            continue
1921

    
1922
          releaseev = threading.Event()
1923

    
1924
          # Acquire locks
1925
          acquireev = []
1926
          tthreads1 = []
1927
          for i in range(3):
1928
            ev = threading.Event()
1929
            tthreads1.append(self._addThread(target=_Acquire,
1930
                                             args=(tlock1, 1, releaseev, ev)))
1931
            acquireev.append(ev)
1932

    
1933
          ev = threading.Event()
1934
          tthread2 = self._addThread(target=_Acquire,
1935
                                     args=(tlock2, 1, releaseev, ev))
1936
          acquireev.append(ev)
1937

    
1938
          ev = threading.Event()
1939
          tthread3 = self._addThread(target=_Acquire,
1940
                                     args=(tlock3, 0, releaseev, ev))
1941
          acquireev.append(ev)
1942

    
1943
          # Wait for all locks to be acquired
1944
          for i in acquireev:
1945
            i.wait()
1946

    
1947
          # Check query result
1948
          result = self.lm.QueryLocks(["name", "mode", "owner"])
1949
          response = objects.QueryResponse.FromDict(result)
1950
          for (name, mode, owner) in response.data:
1951
            (name_status, name_value) = name
1952
            (owner_status, owner_value) = owner
1953

    
1954
            self.assertEqual(name_status, constants.RS_NORMAL)
1955
            self.assertEqual(owner_status, constants.RS_NORMAL)
1956

    
1957
            if name_value == tlock1.name:
1958
              self.assertEqual(mode, (constants.RS_NORMAL, "shared"))
1959
              self.assertEqual(set(owner_value),
1960
                               set(i.getName() for i in tthreads1))
1961
              continue
1962

    
1963
            if name_value == tlock2.name:
1964
              self.assertEqual(mode, (constants.RS_NORMAL, "shared"))
1965
              self.assertEqual(owner_value, [tthread2.getName()])
1966
              continue
1967

    
1968
            if name_value == tlock3.name:
1969
              self.assertEqual(mode, (constants.RS_NORMAL, "exclusive"))
1970
              self.assertEqual(owner_value, [tthread3.getName()])
1971
              continue
1972

    
1973
            self.assert_(name_value in expnames)
1974
            self.assertEqual(mode, (constants.RS_NORMAL, None))
1975
            self.assert_(owner_value is None)
1976

    
1977
          # Release locks again
1978
          releaseev.set()
1979

    
1980
          self._waitThreads()
1981

    
1982
          result = self.lm.QueryLocks(["name", "mode", "owner"])
1983
          self.assertEqual(objects.QueryResponse.FromDict(result).data,
1984
                           [[(constants.RS_NORMAL, name),
1985
                             (constants.RS_NORMAL, None),
1986
                             (constants.RS_NORMAL, None)]
1987
                            for name in utils.NiceSort(expnames)])
1988

    
1989
  def testDelete(self):
1990
    lock = locking.SharedLock("TestLock", monitor=self.lm)
1991

    
1992
    self.assertEqual(len(self.lm._locks), 1)
1993
    result = self.lm.QueryLocks(["name", "mode", "owner"])
1994
    self.assertEqual(objects.QueryResponse.FromDict(result).data,
1995
                     [[(constants.RS_NORMAL, lock.name),
1996
                       (constants.RS_NORMAL, None),
1997
                       (constants.RS_NORMAL, None)]])
1998

    
1999
    lock.delete()
2000

    
2001
    result = self.lm.QueryLocks(["name", "mode", "owner"])
2002
    self.assertEqual(objects.QueryResponse.FromDict(result).data,
2003
                     [[(constants.RS_NORMAL, lock.name),
2004
                       (constants.RS_NORMAL, "deleted"),
2005
                       (constants.RS_NORMAL, None)]])
2006
    self.assertEqual(len(self.lm._locks), 1)
2007

    
2008
  def testPending(self):
2009
    def _Acquire(lock, shared, prev, next):
2010
      prev.wait()
2011

    
2012
      lock.acquire(shared=shared, test_notify=next.set)
2013
      try:
2014
        pass
2015
      finally:
2016
        lock.release()
2017

    
2018
    lock = locking.SharedLock("ExcLock", monitor=self.lm)
2019

    
2020
    for shared in [0, 1]:
2021
      lock.acquire()
2022
      try:
2023
        self.assertEqual(len(self.lm._locks), 1)
2024
        result = self.lm.QueryLocks(["name", "mode", "owner"])
2025
        self.assertEqual(objects.QueryResponse.FromDict(result).data,
2026
                         [[(constants.RS_NORMAL, lock.name),
2027
                           (constants.RS_NORMAL, "exclusive"),
2028
                           (constants.RS_NORMAL,
2029
                            [threading.currentThread().getName()])]])
2030

    
2031
        threads = []
2032

    
2033
        first = threading.Event()
2034
        prev = first
2035

    
2036
        for i in range(5):
2037
          ev = threading.Event()
2038
          threads.append(self._addThread(target=_Acquire,
2039
                                          args=(lock, shared, prev, ev)))
2040
          prev = ev
2041

    
2042
        # Start acquires
2043
        first.set()
2044

    
2045
        # Wait for last acquire to start waiting
2046
        prev.wait()
2047

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

    
2053
        # All acquires are waiting now
2054
        if shared:
2055
          pending = [("shared", utils.NiceSort(t.getName() for t in threads))]
2056
        else:
2057
          pending = [("exclusive", [t.getName()]) for t in threads]
2058

    
2059
        result = self.lm.QueryLocks(["name", "mode", "owner", "pending"])
2060
        self.assertEqual(objects.QueryResponse.FromDict(result).data,
2061
                         [[(constants.RS_NORMAL, lock.name),
2062
                           (constants.RS_NORMAL, "exclusive"),
2063
                           (constants.RS_NORMAL,
2064
                            [threading.currentThread().getName()]),
2065
                           (constants.RS_NORMAL, pending)]])
2066

    
2067
        self.assertEqual(len(self.lm._locks), 1)
2068
      finally:
2069
        lock.release()
2070

    
2071
      self._waitThreads()
2072

    
2073
      # No pending acquires
2074
      result = self.lm.QueryLocks(["name", "mode", "owner", "pending"])
2075
      self.assertEqual(objects.QueryResponse.FromDict(result).data,
2076
                       [[(constants.RS_NORMAL, lock.name),
2077
                         (constants.RS_NORMAL, None),
2078
                         (constants.RS_NORMAL, None),
2079
                         (constants.RS_NORMAL, [])]])
2080

    
2081
      self.assertEqual(len(self.lm._locks), 1)
2082

    
2083
  def testDeleteAndRecreate(self):
2084
    lname = "TestLock101923193"
2085

    
2086
    # Create some locks with the same name and keep all references
2087
    locks = [locking.SharedLock(lname, monitor=self.lm)
2088
             for _ in range(5)]
2089

    
2090
    self.assertEqual(len(self.lm._locks), len(locks))
2091

    
2092
    result = self.lm.QueryLocks(["name", "mode", "owner"])
2093
    self.assertEqual(objects.QueryResponse.FromDict(result).data,
2094
                     [[(constants.RS_NORMAL, lname),
2095
                       (constants.RS_NORMAL, None),
2096
                       (constants.RS_NORMAL, None)]] * 5)
2097

    
2098
    locks[2].delete()
2099

    
2100
    # Check information order
2101
    result = self.lm.QueryLocks(["name", "mode", "owner"])
2102
    self.assertEqual(objects.QueryResponse.FromDict(result).data,
2103
                     [[(constants.RS_NORMAL, lname),
2104
                       (constants.RS_NORMAL, None),
2105
                       (constants.RS_NORMAL, None)]] * 2 +
2106
                     [[(constants.RS_NORMAL, lname),
2107
                       (constants.RS_NORMAL, "deleted"),
2108
                       (constants.RS_NORMAL, None)]] +
2109
                     [[(constants.RS_NORMAL, lname),
2110
                       (constants.RS_NORMAL, None),
2111
                       (constants.RS_NORMAL, None)]] * 2)
2112

    
2113
    locks[1].acquire(shared=0)
2114

    
2115
    last_status = [
2116
      [(constants.RS_NORMAL, lname),
2117
       (constants.RS_NORMAL, None),
2118
       (constants.RS_NORMAL, None)],
2119
      [(constants.RS_NORMAL, lname),
2120
       (constants.RS_NORMAL, "exclusive"),
2121
       (constants.RS_NORMAL, [threading.currentThread().getName()])],
2122
      [(constants.RS_NORMAL, lname),
2123
       (constants.RS_NORMAL, "deleted"),
2124
       (constants.RS_NORMAL, None)],
2125
      [(constants.RS_NORMAL, lname),
2126
       (constants.RS_NORMAL, None),
2127
       (constants.RS_NORMAL, None)],
2128
      [(constants.RS_NORMAL, lname),
2129
       (constants.RS_NORMAL, None),
2130
       (constants.RS_NORMAL, None)],
2131
      ]
2132

    
2133
    # Check information order
2134
    result = self.lm.QueryLocks(["name", "mode", "owner"])
2135
    self.assertEqual(objects.QueryResponse.FromDict(result).data, last_status)
2136

    
2137
    self.assertEqual(len(set(self.lm._locks.values())), len(locks))
2138
    self.assertEqual(len(self.lm._locks), len(locks))
2139

    
2140
    # Check lock deletion
2141
    for idx in range(len(locks)):
2142
      del locks[0]
2143
      assert gc.isenabled()
2144
      gc.collect()
2145
      self.assertEqual(len(self.lm._locks), len(locks))
2146
      result = self.lm.QueryLocks(["name", "mode", "owner"])
2147
      self.assertEqual(objects.QueryResponse.FromDict(result).data,
2148
                       last_status[idx + 1:])
2149

    
2150
    # All locks should have been deleted
2151
    assert not locks
2152
    self.assertFalse(self.lm._locks)
2153

    
2154
    result = self.lm.QueryLocks(["name", "mode", "owner"])
2155
    self.assertEqual(objects.QueryResponse.FromDict(result).data, [])
2156

    
2157

    
2158
if __name__ == '__main__':
2159
  testutils.GanetiTestProgram()