Statistics
| Branch: | Tag: | Revision:

root / lib / locking.py @ 8716b1db

History | View | Annotate | Download (57.3 kB)

1
#
2
#
3

    
4
# Copyright (C) 2006, 2007, 2008, 2009, 2010, 2011, 2012 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
"""Module implementing the Ganeti locking code."""
22

    
23
# pylint: disable=W0212
24

    
25
# W0212 since e.g. LockSet methods use (a lot) the internals of
26
# SharedLock
27

    
28
import os
29
import select
30
import threading
31
import errno
32
import weakref
33
import logging
34
import heapq
35
import itertools
36
import time
37

    
38
from ganeti import errors
39
from ganeti import utils
40
from ganeti import compat
41
from ganeti import query
42

    
43

    
44
_EXCLUSIVE_TEXT = "exclusive"
45
_SHARED_TEXT = "shared"
46
_DELETED_TEXT = "deleted"
47

    
48
_DEFAULT_PRIORITY = 0
49

    
50
#: Minimum timeout required to consider scheduling a pending acquisition
51
#: (seconds)
52
_LOCK_ACQUIRE_MIN_TIMEOUT = (1.0 / 1000)
53

    
54
# Internal lock acquisition modes for L{LockSet}
55
(_LS_ACQUIRE_EXACT,
56
 _LS_ACQUIRE_ALL) = range(1, 3)
57

    
58

    
59
def ssynchronized(mylock, shared=0):
60
  """Shared Synchronization decorator.
61

62
  Calls the function holding the given lock, either in exclusive or shared
63
  mode. It requires the passed lock to be a SharedLock (or support its
64
  semantics).
65

66
  @type mylock: lockable object or string
67
  @param mylock: lock to acquire or class member name of the lock to acquire
68

69
  """
70
  def wrap(fn):
71
    def sync_function(*args, **kwargs):
72
      if isinstance(mylock, basestring):
73
        assert args, "cannot ssynchronize on non-class method: self not found"
74
        # args[0] is "self"
75
        lock = getattr(args[0], mylock)
76
      else:
77
        lock = mylock
78
      lock.acquire(shared=shared)
79
      try:
80
        return fn(*args, **kwargs)
81
      finally:
82
        lock.release()
83
    return sync_function
84
  return wrap
85

    
86

    
87
class _SingleNotifyPipeConditionWaiter(object):
88
  """Helper class for SingleNotifyPipeCondition
89

90
  """
91
  __slots__ = [
92
    "_fd",
93
    "_poller",
94
    ]
95

    
96
  def __init__(self, poller, fd):
97
    """Constructor for _SingleNotifyPipeConditionWaiter
98

99
    @type poller: select.poll
100
    @param poller: Poller object
101
    @type fd: int
102
    @param fd: File descriptor to wait for
103

104
    """
105
    object.__init__(self)
106
    self._poller = poller
107
    self._fd = fd
108

    
109
  def __call__(self, timeout):
110
    """Wait for something to happen on the pipe.
111

112
    @type timeout: float or None
113
    @param timeout: Timeout for waiting (can be None)
114

115
    """
116
    running_timeout = utils.RunningTimeout(timeout, True)
117

    
118
    while True:
119
      remaining_time = running_timeout.Remaining()
120

    
121
      if remaining_time is not None:
122
        if remaining_time < 0.0:
123
          break
124

    
125
        # Our calculation uses seconds, poll() wants milliseconds
126
        remaining_time *= 1000
127

    
128
      try:
129
        result = self._poller.poll(remaining_time)
130
      except EnvironmentError, err:
131
        if err.errno != errno.EINTR:
132
          raise
133
        result = None
134

    
135
      # Check whether we were notified
136
      if result and result[0][0] == self._fd:
137
        break
138

    
139

    
140
class _BaseCondition(object):
141
  """Base class containing common code for conditions.
142

143
  Some of this code is taken from python's threading module.
144

145
  """
146
  __slots__ = [
147
    "_lock",
148
    "acquire",
149
    "release",
150
    "_is_owned",
151
    "_acquire_restore",
152
    "_release_save",
153
    ]
154

    
155
  def __init__(self, lock):
156
    """Constructor for _BaseCondition.
157

158
    @type lock: threading.Lock
159
    @param lock: condition base lock
160

161
    """
162
    object.__init__(self)
163

    
164
    try:
165
      self._release_save = lock._release_save
166
    except AttributeError:
167
      self._release_save = self._base_release_save
168
    try:
169
      self._acquire_restore = lock._acquire_restore
170
    except AttributeError:
171
      self._acquire_restore = self._base_acquire_restore
172
    try:
173
      self._is_owned = lock.is_owned
174
    except AttributeError:
175
      self._is_owned = self._base_is_owned
176

    
177
    self._lock = lock
178

    
179
    # Export the lock's acquire() and release() methods
180
    self.acquire = lock.acquire
181
    self.release = lock.release
182

    
183
  def _base_is_owned(self):
184
    """Check whether lock is owned by current thread.
185

186
    """
187
    if self._lock.acquire(0):
188
      self._lock.release()
189
      return False
190
    return True
191

    
192
  def _base_release_save(self):
193
    self._lock.release()
194

    
195
  def _base_acquire_restore(self, _):
196
    self._lock.acquire()
197

    
198
  def _check_owned(self):
199
    """Raise an exception if the current thread doesn't own the lock.
200

201
    """
202
    if not self._is_owned():
203
      raise RuntimeError("cannot work with un-aquired lock")
204

    
205

    
206
class SingleNotifyPipeCondition(_BaseCondition):
207
  """Condition which can only be notified once.
208

209
  This condition class uses pipes and poll, internally, to be able to wait for
210
  notification with a timeout, without resorting to polling. It is almost
211
  compatible with Python's threading.Condition, with the following differences:
212
    - notifyAll can only be called once, and no wait can happen after that
213
    - notify is not supported, only notifyAll
214

215
  """
216

    
217
  __slots__ = [
218
    "_poller",
219
    "_read_fd",
220
    "_write_fd",
221
    "_nwaiters",
222
    "_notified",
223
    ]
224

    
225
  _waiter_class = _SingleNotifyPipeConditionWaiter
226

    
227
  def __init__(self, lock):
228
    """Constructor for SingleNotifyPipeCondition
229

230
    """
231
    _BaseCondition.__init__(self, lock)
232
    self._nwaiters = 0
233
    self._notified = False
234
    self._read_fd = None
235
    self._write_fd = None
236
    self._poller = None
237

    
238
  def _check_unnotified(self):
239
    """Throws an exception if already notified.
240

241
    """
242
    if self._notified:
243
      raise RuntimeError("cannot use already notified condition")
244

    
245
  def _Cleanup(self):
246
    """Cleanup open file descriptors, if any.
247

248
    """
249
    if self._read_fd is not None:
250
      os.close(self._read_fd)
251
      self._read_fd = None
252

    
253
    if self._write_fd is not None:
254
      os.close(self._write_fd)
255
      self._write_fd = None
256
    self._poller = None
257

    
258
  def wait(self, timeout):
259
    """Wait for a notification.
260

261
    @type timeout: float or None
262
    @param timeout: Waiting timeout (can be None)
263

264
    """
265
    self._check_owned()
266
    self._check_unnotified()
267

    
268
    self._nwaiters += 1
269
    try:
270
      if self._poller is None:
271
        (self._read_fd, self._write_fd) = os.pipe()
272
        self._poller = select.poll()
273
        self._poller.register(self._read_fd, select.POLLHUP)
274

    
275
      wait_fn = self._waiter_class(self._poller, self._read_fd)
276
      state = self._release_save()
277
      try:
278
        # Wait for notification
279
        wait_fn(timeout)
280
      finally:
281
        # Re-acquire lock
282
        self._acquire_restore(state)
283
    finally:
284
      self._nwaiters -= 1
285
      if self._nwaiters == 0:
286
        self._Cleanup()
287

    
288
  def notifyAll(self): # pylint: disable=C0103
289
    """Close the writing side of the pipe to notify all waiters.
290

291
    """
292
    self._check_owned()
293
    self._check_unnotified()
294
    self._notified = True
295
    if self._write_fd is not None:
296
      os.close(self._write_fd)
297
      self._write_fd = None
298

    
299

    
300
class PipeCondition(_BaseCondition):
301
  """Group-only non-polling condition with counters.
302

303
  This condition class uses pipes and poll, internally, to be able to wait for
304
  notification with a timeout, without resorting to polling. It is almost
305
  compatible with Python's threading.Condition, but only supports notifyAll and
306
  non-recursive locks. As an additional features it's able to report whether
307
  there are any waiting threads.
308

309
  """
310
  __slots__ = [
311
    "_waiters",
312
    "_single_condition",
313
    ]
314

    
315
  _single_condition_class = SingleNotifyPipeCondition
316

    
317
  def __init__(self, lock):
318
    """Initializes this class.
319

320
    """
321
    _BaseCondition.__init__(self, lock)
322
    self._waiters = set()
323
    self._single_condition = self._single_condition_class(self._lock)
324

    
325
  def wait(self, timeout):
326
    """Wait for a notification.
327

328
    @type timeout: float or None
329
    @param timeout: Waiting timeout (can be None)
330

331
    """
332
    self._check_owned()
333

    
334
    # Keep local reference to the pipe. It could be replaced by another thread
335
    # notifying while we're waiting.
336
    cond = self._single_condition
337

    
338
    self._waiters.add(threading.currentThread())
339
    try:
340
      cond.wait(timeout)
341
    finally:
342
      self._check_owned()
343
      self._waiters.remove(threading.currentThread())
344

    
345
  def notifyAll(self): # pylint: disable=C0103
346
    """Notify all currently waiting threads.
347

348
    """
349
    self._check_owned()
350
    self._single_condition.notifyAll()
351
    self._single_condition = self._single_condition_class(self._lock)
352

    
353
  def get_waiting(self):
354
    """Returns a list of all waiting threads.
355

356
    """
357
    self._check_owned()
358

    
359
    return self._waiters
360

    
361
  def has_waiting(self):
362
    """Returns whether there are active waiters.
363

364
    """
365
    self._check_owned()
366

    
367
    return bool(self._waiters)
368

    
369
  def __repr__(self):
370
    return ("<%s.%s waiters=%s at %#x>" %
371
            (self.__class__.__module__, self.__class__.__name__,
372
             self._waiters, id(self)))
373

    
374

    
375
class _PipeConditionWithMode(PipeCondition):
376
  __slots__ = [
377
    "shared",
378
    ]
379

    
380
  def __init__(self, lock, shared):
381
    """Initializes this class.
382

383
    """
384
    self.shared = shared
385
    PipeCondition.__init__(self, lock)
386

    
387

    
388
class SharedLock(object):
389
  """Implements a shared lock.
390

391
  Multiple threads can acquire the lock in a shared way by calling
392
  C{acquire(shared=1)}. In order to acquire the lock in an exclusive way
393
  threads can call C{acquire(shared=0)}.
394

395
  Notes on data structures: C{__pending} contains a priority queue (heapq) of
396
  all pending acquires: C{[(priority1: prioqueue1), (priority2: prioqueue2),
397
  ...]}. Each per-priority queue contains a normal in-order list of conditions
398
  to be notified when the lock can be acquired. Shared locks are grouped
399
  together by priority and the condition for them is stored in
400
  C{__pending_shared} if it already exists. C{__pending_by_prio} keeps
401
  references for the per-priority queues indexed by priority for faster access.
402

403
  @type name: string
404
  @ivar name: the name of the lock
405

406
  """
407
  __slots__ = [
408
    "__weakref__",
409
    "__deleted",
410
    "__exc",
411
    "__lock",
412
    "__pending",
413
    "__pending_by_prio",
414
    "__pending_shared",
415
    "__shr",
416
    "__time_fn",
417
    "name",
418
    ]
419

    
420
  __condition_class = _PipeConditionWithMode
421

    
422
  def __init__(self, name, monitor=None, _time_fn=time.time):
423
    """Construct a new SharedLock.
424

425
    @param name: the name of the lock
426
    @type monitor: L{LockMonitor}
427
    @param monitor: Lock monitor with which to register
428

429
    """
430
    object.__init__(self)
431

    
432
    self.name = name
433

    
434
    # Used for unittesting
435
    self.__time_fn = _time_fn
436

    
437
    # Internal lock
438
    self.__lock = threading.Lock()
439

    
440
    # Queue containing waiting acquires
441
    self.__pending = []
442
    self.__pending_by_prio = {}
443
    self.__pending_shared = {}
444

    
445
    # Current lock holders
446
    self.__shr = set()
447
    self.__exc = None
448

    
449
    # is this lock in the deleted state?
450
    self.__deleted = False
451

    
452
    # Register with lock monitor
453
    if monitor:
454
      logging.debug("Adding lock %s to monitor", name)
455
      monitor.RegisterLock(self)
456

    
457
  def __repr__(self):
458
    return ("<%s.%s name=%s at %#x>" %
459
            (self.__class__.__module__, self.__class__.__name__,
460
             self.name, id(self)))
461

    
462
  def GetLockInfo(self, requested):
463
    """Retrieves information for querying locks.
464

465
    @type requested: set
466
    @param requested: Requested information, see C{query.LQ_*}
467

468
    """
469
    self.__lock.acquire()
470
    try:
471
      # Note: to avoid unintentional race conditions, no references to
472
      # modifiable objects should be returned unless they were created in this
473
      # function.
474
      mode = None
475
      owner_names = None
476

    
477
      if query.LQ_MODE in requested:
478
        if self.__deleted:
479
          mode = _DELETED_TEXT
480
          assert not (self.__exc or self.__shr)
481
        elif self.__exc:
482
          mode = _EXCLUSIVE_TEXT
483
        elif self.__shr:
484
          mode = _SHARED_TEXT
485

    
486
      # Current owner(s) are wanted
487
      if query.LQ_OWNER in requested:
488
        if self.__exc:
489
          owner = [self.__exc]
490
        else:
491
          owner = self.__shr
492

    
493
        if owner:
494
          assert not self.__deleted
495
          owner_names = [i.getName() for i in owner]
496

    
497
      # Pending acquires are wanted
498
      if query.LQ_PENDING in requested:
499
        pending = []
500

    
501
        # Sorting instead of copying and using heaq functions for simplicity
502
        for (_, prioqueue) in sorted(self.__pending):
503
          for cond in prioqueue:
504
            if cond.shared:
505
              pendmode = _SHARED_TEXT
506
            else:
507
              pendmode = _EXCLUSIVE_TEXT
508

    
509
            # List of names will be sorted in L{query._GetLockPending}
510
            pending.append((pendmode, [i.getName()
511
                                       for i in cond.get_waiting()]))
512
      else:
513
        pending = None
514

    
515
      return [(self.name, mode, owner_names, pending)]
516
    finally:
517
      self.__lock.release()
518

    
519
  def __check_deleted(self):
520
    """Raises an exception if the lock has been deleted.
521

522
    """
523
    if self.__deleted:
524
      raise errors.LockError("Deleted lock %s" % self.name)
525

    
526
  def __is_sharer(self):
527
    """Is the current thread sharing the lock at this time?
528

529
    """
530
    return threading.currentThread() in self.__shr
531

    
532
  def __is_exclusive(self):
533
    """Is the current thread holding the lock exclusively at this time?
534

535
    """
536
    return threading.currentThread() == self.__exc
537

    
538
  def __is_owned(self, shared=-1):
539
    """Is the current thread somehow owning the lock at this time?
540

541
    This is a private version of the function, which presumes you're holding
542
    the internal lock.
543

544
    """
545
    if shared < 0:
546
      return self.__is_sharer() or self.__is_exclusive()
547
    elif shared:
548
      return self.__is_sharer()
549
    else:
550
      return self.__is_exclusive()
551

    
552
  def is_owned(self, shared=-1):
553
    """Is the current thread somehow owning the lock at this time?
554

555
    @param shared:
556
        - < 0: check for any type of ownership (default)
557
        - 0: check for exclusive ownership
558
        - > 0: check for shared ownership
559

560
    """
561
    self.__lock.acquire()
562
    try:
563
      return self.__is_owned(shared=shared)
564
    finally:
565
      self.__lock.release()
566

    
567
  #: Necessary to remain compatible with threading.Condition, which tries to
568
  #: retrieve a locks' "_is_owned" attribute
569
  _is_owned = is_owned
570

    
571
  def _count_pending(self):
572
    """Returns the number of pending acquires.
573

574
    @rtype: int
575

576
    """
577
    self.__lock.acquire()
578
    try:
579
      return sum(len(prioqueue) for (_, prioqueue) in self.__pending)
580
    finally:
581
      self.__lock.release()
582

    
583
  def _check_empty(self):
584
    """Checks whether there are any pending acquires.
585

586
    @rtype: bool
587

588
    """
589
    self.__lock.acquire()
590
    try:
591
      # Order is important: __find_first_pending_queue modifies __pending
592
      (_, prioqueue) = self.__find_first_pending_queue()
593

    
594
      return not (prioqueue or
595
                  self.__pending or
596
                  self.__pending_by_prio or
597
                  self.__pending_shared)
598
    finally:
599
      self.__lock.release()
600

    
601
  def __do_acquire(self, shared):
602
    """Actually acquire the lock.
603

604
    """
605
    if shared:
606
      self.__shr.add(threading.currentThread())
607
    else:
608
      self.__exc = threading.currentThread()
609

    
610
  def __can_acquire(self, shared):
611
    """Determine whether lock can be acquired.
612

613
    """
614
    if shared:
615
      return self.__exc is None
616
    else:
617
      return len(self.__shr) == 0 and self.__exc is None
618

    
619
  def __find_first_pending_queue(self):
620
    """Tries to find the topmost queued entry with pending acquires.
621

622
    Removes empty entries while going through the list.
623

624
    """
625
    while self.__pending:
626
      (priority, prioqueue) = self.__pending[0]
627

    
628
      if prioqueue:
629
        return (priority, prioqueue)
630

    
631
      # Remove empty queue
632
      heapq.heappop(self.__pending)
633
      del self.__pending_by_prio[priority]
634
      assert priority not in self.__pending_shared
635

    
636
    return (None, None)
637

    
638
  def __is_on_top(self, cond):
639
    """Checks whether the passed condition is on top of the queue.
640

641
    The caller must make sure the queue isn't empty.
642

643
    """
644
    (_, prioqueue) = self.__find_first_pending_queue()
645

    
646
    return cond == prioqueue[0]
647

    
648
  def __acquire_unlocked(self, shared, timeout, priority):
649
    """Acquire a shared lock.
650

651
    @param shared: whether to acquire in shared mode; by default an
652
        exclusive lock will be acquired
653
    @param timeout: maximum waiting time before giving up
654
    @type priority: integer
655
    @param priority: Priority for acquiring lock
656

657
    """
658
    self.__check_deleted()
659

    
660
    # We cannot acquire the lock if we already have it
661
    assert not self.__is_owned(), ("double acquire() on a non-recursive lock"
662
                                   " %s" % self.name)
663

    
664
    # Remove empty entries from queue
665
    self.__find_first_pending_queue()
666

    
667
    # Check whether someone else holds the lock or there are pending acquires.
668
    if not self.__pending and self.__can_acquire(shared):
669
      # Apparently not, can acquire lock directly.
670
      self.__do_acquire(shared)
671
      return True
672

    
673
    # The lock couldn't be acquired right away, so if a timeout is given and is
674
    # considered too short, return right away as scheduling a pending
675
    # acquisition is quite expensive
676
    if timeout is not None and timeout < _LOCK_ACQUIRE_MIN_TIMEOUT:
677
      return False
678

    
679
    prioqueue = self.__pending_by_prio.get(priority, None)
680

    
681
    if shared:
682
      # Try to re-use condition for shared acquire
683
      wait_condition = self.__pending_shared.get(priority, None)
684
      assert (wait_condition is None or
685
              (wait_condition.shared and wait_condition in prioqueue))
686
    else:
687
      wait_condition = None
688

    
689
    if wait_condition is None:
690
      if prioqueue is None:
691
        assert priority not in self.__pending_by_prio
692

    
693
        prioqueue = []
694
        heapq.heappush(self.__pending, (priority, prioqueue))
695
        self.__pending_by_prio[priority] = prioqueue
696

    
697
      wait_condition = self.__condition_class(self.__lock, shared)
698
      prioqueue.append(wait_condition)
699

    
700
      if shared:
701
        # Keep reference for further shared acquires on same priority. This is
702
        # better than trying to find it in the list of pending acquires.
703
        assert priority not in self.__pending_shared
704
        self.__pending_shared[priority] = wait_condition
705

    
706
    wait_start = self.__time_fn()
707
    acquired = False
708

    
709
    try:
710
      # Wait until we become the topmost acquire in the queue or the timeout
711
      # expires.
712
      while True:
713
        if self.__is_on_top(wait_condition) and self.__can_acquire(shared):
714
          self.__do_acquire(shared)
715
          acquired = True
716
          break
717

    
718
        # A lot of code assumes blocking acquires always succeed, therefore we
719
        # can never return False for a blocking acquire
720
        if (timeout is not None and
721
            utils.TimeoutExpired(wait_start, timeout, _time_fn=self.__time_fn)):
722
          break
723

    
724
        # Wait for notification
725
        wait_condition.wait(timeout)
726
        self.__check_deleted()
727
    finally:
728
      # Remove condition from queue if there are no more waiters
729
      if not wait_condition.has_waiting():
730
        prioqueue.remove(wait_condition)
731
        if wait_condition.shared:
732
          # Remove from list of shared acquires if it wasn't while releasing
733
          # (e.g. on lock deletion)
734
          self.__pending_shared.pop(priority, None)
735

    
736
    return acquired
737

    
738
  def acquire(self, shared=0, timeout=None, priority=None,
739
              test_notify=None):
740
    """Acquire a shared lock.
741

742
    @type shared: integer (0/1) used as a boolean
743
    @param shared: whether to acquire in shared mode; by default an
744
        exclusive lock will be acquired
745
    @type timeout: float
746
    @param timeout: maximum waiting time before giving up
747
    @type priority: integer
748
    @param priority: Priority for acquiring lock
749
    @type test_notify: callable or None
750
    @param test_notify: Special callback function for unittesting
751

752
    """
753
    if priority is None:
754
      priority = _DEFAULT_PRIORITY
755

    
756
    self.__lock.acquire()
757
    try:
758
      # We already got the lock, notify now
759
      if __debug__ and callable(test_notify):
760
        test_notify()
761

    
762
      return self.__acquire_unlocked(shared, timeout, priority)
763
    finally:
764
      self.__lock.release()
765

    
766
  def downgrade(self):
767
    """Changes the lock mode from exclusive to shared.
768

769
    Pending acquires in shared mode on the same priority will go ahead.
770

771
    """
772
    self.__lock.acquire()
773
    try:
774
      assert self.__is_owned(), "Lock must be owned"
775

    
776
      if self.__is_exclusive():
777
        # Do nothing if the lock is already acquired in shared mode
778
        self.__exc = None
779
        self.__do_acquire(1)
780

    
781
        # Important: pending shared acquires should only jump ahead if there
782
        # was a transition from exclusive to shared, otherwise an owner of a
783
        # shared lock can keep calling this function to push incoming shared
784
        # acquires
785
        (priority, prioqueue) = self.__find_first_pending_queue()
786
        if prioqueue:
787
          # Is there a pending shared acquire on this priority?
788
          cond = self.__pending_shared.pop(priority, None)
789
          if cond:
790
            assert cond.shared
791
            assert cond in prioqueue
792

    
793
            # Ensure shared acquire is on top of queue
794
            if len(prioqueue) > 1:
795
              prioqueue.remove(cond)
796
              prioqueue.insert(0, cond)
797

    
798
            # Notify
799
            cond.notifyAll()
800

    
801
      assert not self.__is_exclusive()
802
      assert self.__is_sharer()
803

    
804
      return True
805
    finally:
806
      self.__lock.release()
807

    
808
  def release(self):
809
    """Release a Shared Lock.
810

811
    You must have acquired the lock, either in shared or in exclusive mode,
812
    before calling this function.
813

814
    """
815
    self.__lock.acquire()
816
    try:
817
      assert self.__is_exclusive() or self.__is_sharer(), \
818
        "Cannot release non-owned lock"
819

    
820
      # Autodetect release type
821
      if self.__is_exclusive():
822
        self.__exc = None
823
        notify = True
824
      else:
825
        self.__shr.remove(threading.currentThread())
826
        notify = not self.__shr
827

    
828
      # Notify topmost condition in queue if there are no owners left (for
829
      # shared locks)
830
      if notify:
831
        self.__notify_topmost()
832
    finally:
833
      self.__lock.release()
834

    
835
  def __notify_topmost(self):
836
    """Notifies topmost condition in queue of pending acquires.
837

838
    """
839
    (priority, prioqueue) = self.__find_first_pending_queue()
840
    if prioqueue:
841
      cond = prioqueue[0]
842
      cond.notifyAll()
843
      if cond.shared:
844
        # Prevent further shared acquires from sneaking in while waiters are
845
        # notified
846
        self.__pending_shared.pop(priority, None)
847

    
848
  def _notify_topmost(self):
849
    """Exported version of L{__notify_topmost}.
850

851
    """
852
    self.__lock.acquire()
853
    try:
854
      return self.__notify_topmost()
855
    finally:
856
      self.__lock.release()
857

    
858
  def delete(self, timeout=None, priority=None):
859
    """Delete a Shared Lock.
860

861
    This operation will declare the lock for removal. First the lock will be
862
    acquired in exclusive mode if you don't already own it, then the lock
863
    will be put in a state where any future and pending acquire() fail.
864

865
    @type timeout: float
866
    @param timeout: maximum waiting time before giving up
867
    @type priority: integer
868
    @param priority: Priority for acquiring lock
869

870
    """
871
    if priority is None:
872
      priority = _DEFAULT_PRIORITY
873

    
874
    self.__lock.acquire()
875
    try:
876
      assert not self.__is_sharer(), "Cannot delete() a lock while sharing it"
877

    
878
      self.__check_deleted()
879

    
880
      # The caller is allowed to hold the lock exclusively already.
881
      acquired = self.__is_exclusive()
882

    
883
      if not acquired:
884
        acquired = self.__acquire_unlocked(0, timeout, priority)
885

    
886
      if acquired:
887
        assert self.__is_exclusive() and not self.__is_sharer(), \
888
          "Lock wasn't acquired in exclusive mode"
889

    
890
        self.__deleted = True
891
        self.__exc = None
892

    
893
        assert not (self.__exc or self.__shr), "Found owner during deletion"
894

    
895
        # Notify all acquires. They'll throw an error.
896
        for (_, prioqueue) in self.__pending:
897
          for cond in prioqueue:
898
            cond.notifyAll()
899

    
900
        assert self.__deleted
901

    
902
      return acquired
903
    finally:
904
      self.__lock.release()
905

    
906
  def _release_save(self):
907
    shared = self.__is_sharer()
908
    self.release()
909
    return shared
910

    
911
  def _acquire_restore(self, shared):
912
    self.acquire(shared=shared)
913

    
914

    
915
# Whenever we want to acquire a full LockSet we pass None as the value
916
# to acquire.  Hide this behind this nicely named constant.
917
ALL_SET = None
918

    
919

    
920
class _AcquireTimeout(Exception):
921
  """Internal exception to abort an acquire on a timeout.
922

923
  """
924

    
925

    
926
class LockSet:
927
  """Implements a set of locks.
928

929
  This abstraction implements a set of shared locks for the same resource type,
930
  distinguished by name. The user can lock a subset of the resources and the
931
  LockSet will take care of acquiring the locks always in the same order, thus
932
  preventing deadlock.
933

934
  All the locks needed in the same set must be acquired together, though.
935

936
  @type name: string
937
  @ivar name: the name of the lockset
938

939
  """
940
  def __init__(self, members, name, monitor=None):
941
    """Constructs a new LockSet.
942

943
    @type members: list of strings
944
    @param members: initial members of the set
945
    @type monitor: L{LockMonitor}
946
    @param monitor: Lock monitor with which to register member locks
947

948
    """
949
    assert members is not None, "members parameter is not a list"
950
    self.name = name
951

    
952
    # Lock monitor
953
    self.__monitor = monitor
954

    
955
    # Used internally to guarantee coherency
956
    self.__lock = SharedLock(self._GetLockName("[lockset]"), monitor=monitor)
957

    
958
    # The lockdict indexes the relationship name -> lock
959
    # The order-of-locking is implied by the alphabetical order of names
960
    self.__lockdict = {}
961

    
962
    for mname in members:
963
      self.__lockdict[mname] = SharedLock(self._GetLockName(mname),
964
                                          monitor=monitor)
965

    
966
    # The owner dict contains the set of locks each thread owns. For
967
    # performance each thread can access its own key without a global lock on
968
    # this structure. It is paramount though that *no* other type of access is
969
    # done to this structure (eg. no looping over its keys). *_owner helper
970
    # function are defined to guarantee access is correct, but in general never
971
    # do anything different than __owners[threading.currentThread()], or there
972
    # will be trouble.
973
    self.__owners = {}
974

    
975
  def _GetLockName(self, mname):
976
    """Returns the name for a member lock.
977

978
    """
979
    return "%s/%s" % (self.name, mname)
980

    
981
  def _get_lock(self):
982
    """Returns the lockset-internal lock.
983

984
    """
985
    return self.__lock
986

    
987
  def _get_lockdict(self):
988
    """Returns the lockset-internal lock dictionary.
989

990
    Accessing this structure is only safe in single-thread usage or when the
991
    lockset-internal lock is held.
992

993
    """
994
    return self.__lockdict
995

    
996
  def is_owned(self):
997
    """Is the current thread a current level owner?
998

999
    @note: Use L{check_owned} to check if a specific lock is held
1000

1001
    """
1002
    return threading.currentThread() in self.__owners
1003

    
1004
  def check_owned(self, names, shared=-1):
1005
    """Check if locks are owned in a specific mode.
1006

1007
    @type names: sequence or string
1008
    @param names: Lock names (or a single lock name)
1009
    @param shared: See L{SharedLock.is_owned}
1010
    @rtype: bool
1011
    @note: Use L{is_owned} to check if the current thread holds I{any} lock and
1012
      L{list_owned} to get the names of all owned locks
1013

1014
    """
1015
    if isinstance(names, basestring):
1016
      names = [names]
1017

    
1018
    # Avoid check if no locks are owned anyway
1019
    if names and self.is_owned():
1020
      candidates = []
1021

    
1022
      # Gather references to all locks (in case they're deleted in the meantime)
1023
      for lname in names:
1024
        try:
1025
          lock = self.__lockdict[lname]
1026
        except KeyError:
1027
          raise errors.LockError("Non-existing lock '%s' in set '%s' (it may"
1028
                                 " have been removed)" % (lname, self.name))
1029
        else:
1030
          candidates.append(lock)
1031

    
1032
      return compat.all(lock.is_owned(shared=shared) for lock in candidates)
1033
    else:
1034
      return False
1035

    
1036
  def _add_owned(self, name=None):
1037
    """Note the current thread owns the given lock"""
1038
    if name is None:
1039
      if not self.is_owned():
1040
        self.__owners[threading.currentThread()] = set()
1041
    else:
1042
      if self.is_owned():
1043
        self.__owners[threading.currentThread()].add(name)
1044
      else:
1045
        self.__owners[threading.currentThread()] = set([name])
1046

    
1047
  def _del_owned(self, name=None):
1048
    """Note the current thread owns the given lock"""
1049

    
1050
    assert not (name is None and self.__lock.is_owned()), \
1051
           "Cannot hold internal lock when deleting owner status"
1052

    
1053
    if name is not None:
1054
      self.__owners[threading.currentThread()].remove(name)
1055

    
1056
    # Only remove the key if we don't hold the set-lock as well
1057
    if not (self.__lock.is_owned() or
1058
            self.__owners[threading.currentThread()]):
1059
      del self.__owners[threading.currentThread()]
1060

    
1061
  def list_owned(self):
1062
    """Get the set of resource names owned by the current thread"""
1063
    if self.is_owned():
1064
      return self.__owners[threading.currentThread()].copy()
1065
    else:
1066
      return set()
1067

    
1068
  def _release_and_delete_owned(self):
1069
    """Release and delete all resources owned by the current thread"""
1070
    for lname in self.list_owned():
1071
      lock = self.__lockdict[lname]
1072
      if lock.is_owned():
1073
        lock.release()
1074
      self._del_owned(name=lname)
1075

    
1076
  def __names(self):
1077
    """Return the current set of names.
1078

1079
    Only call this function while holding __lock and don't iterate on the
1080
    result after releasing the lock.
1081

1082
    """
1083
    return self.__lockdict.keys()
1084

    
1085
  def _names(self):
1086
    """Return a copy of the current set of elements.
1087

1088
    Used only for debugging purposes.
1089

1090
    """
1091
    # If we don't already own the set-level lock acquired
1092
    # we'll get it and note we need to release it later.
1093
    release_lock = False
1094
    if not self.__lock.is_owned():
1095
      release_lock = True
1096
      self.__lock.acquire(shared=1)
1097
    try:
1098
      result = self.__names()
1099
    finally:
1100
      if release_lock:
1101
        self.__lock.release()
1102
    return set(result)
1103

    
1104
  def acquire(self, names, timeout=None, shared=0, priority=None,
1105
              test_notify=None):
1106
    """Acquire a set of resource locks.
1107

1108
    @type names: list of strings (or string)
1109
    @param names: the names of the locks which shall be acquired
1110
        (special lock names, or instance/node names)
1111
    @type shared: integer (0/1) used as a boolean
1112
    @param shared: whether to acquire in shared mode; by default an
1113
        exclusive lock will be acquired
1114
    @type timeout: float or None
1115
    @param timeout: Maximum time to acquire all locks
1116
    @type priority: integer
1117
    @param priority: Priority for acquiring locks
1118
    @type test_notify: callable or None
1119
    @param test_notify: Special callback function for unittesting
1120

1121
    @return: Set of all locks successfully acquired or None in case of timeout
1122

1123
    @raise errors.LockError: when any lock we try to acquire has
1124
        been deleted before we succeed. In this case none of the
1125
        locks requested will be acquired.
1126

1127
    """
1128
    assert timeout is None or timeout >= 0.0
1129

    
1130
    # Check we don't already own locks at this level
1131
    assert not self.is_owned(), ("Cannot acquire locks in the same set twice"
1132
                                 " (lockset %s)" % self.name)
1133

    
1134
    if priority is None:
1135
      priority = _DEFAULT_PRIORITY
1136

    
1137
    # We need to keep track of how long we spent waiting for a lock. The
1138
    # timeout passed to this function is over all lock acquires.
1139
    running_timeout = utils.RunningTimeout(timeout, False)
1140

    
1141
    try:
1142
      if names is not None:
1143
        # Support passing in a single resource to acquire rather than many
1144
        if isinstance(names, basestring):
1145
          names = [names]
1146

    
1147
        return self.__acquire_inner(names, _LS_ACQUIRE_EXACT, shared, priority,
1148
                                    running_timeout.Remaining, test_notify)
1149

    
1150
      else:
1151
        # If no names are given acquire the whole set by not letting new names
1152
        # being added before we release, and getting the current list of names.
1153
        # Some of them may then be deleted later, but we'll cope with this.
1154
        #
1155
        # We'd like to acquire this lock in a shared way, as it's nice if
1156
        # everybody else can use the instances at the same time. If we are
1157
        # acquiring them exclusively though they won't be able to do this
1158
        # anyway, though, so we'll get the list lock exclusively as well in
1159
        # order to be able to do add() on the set while owning it.
1160
        if not self.__lock.acquire(shared=shared, priority=priority,
1161
                                   timeout=running_timeout.Remaining()):
1162
          raise _AcquireTimeout()
1163
        try:
1164
          # note we own the set-lock
1165
          self._add_owned()
1166

    
1167
          return self.__acquire_inner(self.__names(), _LS_ACQUIRE_ALL, shared,
1168
                                      priority, running_timeout.Remaining,
1169
                                      test_notify)
1170
        except:
1171
          # We shouldn't have problems adding the lock to the owners list, but
1172
          # if we did we'll try to release this lock and re-raise exception.
1173
          # Of course something is going to be really wrong, after this.
1174
          self.__lock.release()
1175
          self._del_owned()
1176
          raise
1177

    
1178
    except _AcquireTimeout:
1179
      return None
1180

    
1181
  def __acquire_inner(self, names, mode, shared, priority,
1182
                      timeout_fn, test_notify):
1183
    """Inner logic for acquiring a number of locks.
1184

1185
    @param names: Names of the locks to be acquired
1186
    @param mode: Lock acquisition mode
1187
    @param shared: Whether to acquire in shared mode
1188
    @param timeout_fn: Function returning remaining timeout
1189
    @param priority: Priority for acquiring locks
1190
    @param test_notify: Special callback function for unittesting
1191

1192
    """
1193
    assert mode in (_LS_ACQUIRE_EXACT, _LS_ACQUIRE_ALL)
1194

    
1195
    acquire_list = []
1196

    
1197
    # First we look the locks up on __lockdict. We have no way of being sure
1198
    # they will still be there after, but this makes it a lot faster should
1199
    # just one of them be the already wrong. Using a sorted sequence to prevent
1200
    # deadlocks.
1201
    for lname in sorted(frozenset(names)):
1202
      try:
1203
        lock = self.__lockdict[lname] # raises KeyError if lock is not there
1204
      except KeyError:
1205
        # We are acquiring the whole set, it doesn't matter if this particular
1206
        # element is not there anymore. If, however, only certain names should
1207
        # be acquired, not finding a lock is an error.
1208
        if mode == _LS_ACQUIRE_EXACT:
1209
          raise errors.LockError("Lock '%s' not found in set '%s' (it may have"
1210
                                 " been removed)" % (lname, self.name))
1211
      else:
1212
        acquire_list.append((lname, lock))
1213

    
1214
    # This will hold the locknames we effectively acquired.
1215
    acquired = set()
1216

    
1217
    try:
1218
      # Now acquire_list contains a sorted list of resources and locks we
1219
      # want.  In order to get them we loop on this (private) list and
1220
      # acquire() them.  We gave no real guarantee they will still exist till
1221
      # this is done but .acquire() itself is safe and will alert us if the
1222
      # lock gets deleted.
1223
      for (lname, lock) in acquire_list:
1224
        if __debug__ and callable(test_notify):
1225
          test_notify_fn = lambda: test_notify(lname)
1226
        else:
1227
          test_notify_fn = None
1228

    
1229
        timeout = timeout_fn()
1230

    
1231
        try:
1232
          # raises LockError if the lock was deleted
1233
          acq_success = lock.acquire(shared=shared, timeout=timeout,
1234
                                     priority=priority,
1235
                                     test_notify=test_notify_fn)
1236
        except errors.LockError:
1237
          if mode == _LS_ACQUIRE_ALL:
1238
            # We are acquiring the whole set, it doesn't matter if this
1239
            # particular element is not there anymore.
1240
            continue
1241

    
1242
          raise errors.LockError("Lock '%s' not found in set '%s' (it may have"
1243
                                 " been removed)" % (lname, self.name))
1244

    
1245
        if not acq_success:
1246
          # Couldn't get lock or timeout occurred
1247
          if timeout is None:
1248
            # This shouldn't happen as SharedLock.acquire(timeout=None) is
1249
            # blocking.
1250
            raise errors.LockError("Failed to get lock %s (set %s)" %
1251
                                   (lname, self.name))
1252

    
1253
          raise _AcquireTimeout()
1254

    
1255
        try:
1256
          # now the lock cannot be deleted, we have it!
1257
          self._add_owned(name=lname)
1258
          acquired.add(lname)
1259

    
1260
        except:
1261
          # We shouldn't have problems adding the lock to the owners list, but
1262
          # if we did we'll try to release this lock and re-raise exception.
1263
          # Of course something is going to be really wrong after this.
1264
          if lock.is_owned():
1265
            lock.release()
1266
          raise
1267

    
1268
    except:
1269
      # Release all owned locks
1270
      self._release_and_delete_owned()
1271
      raise
1272

    
1273
    return acquired
1274

    
1275
  def downgrade(self, names=None):
1276
    """Downgrade a set of resource locks from exclusive to shared mode.
1277

1278
    The locks must have been acquired in exclusive mode.
1279

1280
    """
1281
    assert self.is_owned(), ("downgrade on lockset %s while not owning any"
1282
                             " lock" % self.name)
1283

    
1284
    # Support passing in a single resource to downgrade rather than many
1285
    if isinstance(names, basestring):
1286
      names = [names]
1287

    
1288
    owned = self.list_owned()
1289

    
1290
    if names is None:
1291
      names = owned
1292
    else:
1293
      names = set(names)
1294
      assert owned.issuperset(names), \
1295
        ("downgrade() on unheld resources %s (set %s)" %
1296
         (names.difference(owned), self.name))
1297

    
1298
    for lockname in names:
1299
      self.__lockdict[lockname].downgrade()
1300

    
1301
    # Do we own the lockset in exclusive mode?
1302
    if self.__lock.is_owned(shared=0):
1303
      # Have all locks been downgraded?
1304
      if not compat.any(lock.is_owned(shared=0)
1305
                        for lock in self.__lockdict.values()):
1306
        self.__lock.downgrade()
1307
        assert self.__lock.is_owned(shared=1)
1308

    
1309
    return True
1310

    
1311
  def release(self, names=None):
1312
    """Release a set of resource locks, at the same level.
1313

1314
    You must have acquired the locks, either in shared or in exclusive mode,
1315
    before releasing them.
1316

1317
    @type names: list of strings, or None
1318
    @param names: the names of the locks which shall be released
1319
        (defaults to all the locks acquired at that level).
1320

1321
    """
1322
    assert self.is_owned(), ("release() on lock set %s while not owner" %
1323
                             self.name)
1324

    
1325
    # Support passing in a single resource to release rather than many
1326
    if isinstance(names, basestring):
1327
      names = [names]
1328

    
1329
    if names is None:
1330
      names = self.list_owned()
1331
    else:
1332
      names = set(names)
1333
      assert self.list_owned().issuperset(names), (
1334
               "release() on unheld resources %s (set %s)" %
1335
               (names.difference(self.list_owned()), self.name))
1336

    
1337
    # First of all let's release the "all elements" lock, if set.
1338
    # After this 'add' can work again
1339
    if self.__lock.is_owned():
1340
      self.__lock.release()
1341
      self._del_owned()
1342

    
1343
    for lockname in names:
1344
      # If we are sure the lock doesn't leave __lockdict without being
1345
      # exclusively held we can do this...
1346
      self.__lockdict[lockname].release()
1347
      self._del_owned(name=lockname)
1348

    
1349
  def add(self, names, acquired=0, shared=0):
1350
    """Add a new set of elements to the set
1351

1352
    @type names: list of strings
1353
    @param names: names of the new elements to add
1354
    @type acquired: integer (0/1) used as a boolean
1355
    @param acquired: pre-acquire the new resource?
1356
    @type shared: integer (0/1) used as a boolean
1357
    @param shared: is the pre-acquisition shared?
1358

1359
    """
1360
    # Check we don't already own locks at this level
1361
    assert not self.is_owned() or self.__lock.is_owned(shared=0), \
1362
      ("Cannot add locks if the set %s is only partially owned, or shared" %
1363
       self.name)
1364

    
1365
    # Support passing in a single resource to add rather than many
1366
    if isinstance(names, basestring):
1367
      names = [names]
1368

    
1369
    # If we don't already own the set-level lock acquired in an exclusive way
1370
    # we'll get it and note we need to release it later.
1371
    release_lock = False
1372
    if not self.__lock.is_owned():
1373
      release_lock = True
1374
      self.__lock.acquire()
1375

    
1376
    try:
1377
      invalid_names = set(self.__names()).intersection(names)
1378
      if invalid_names:
1379
        # This must be an explicit raise, not an assert, because assert is
1380
        # turned off when using optimization, and this can happen because of
1381
        # concurrency even if the user doesn't want it.
1382
        raise errors.LockError("duplicate add(%s) on lockset %s" %
1383
                               (invalid_names, self.name))
1384

    
1385
      for lockname in names:
1386
        lock = SharedLock(self._GetLockName(lockname), monitor=self.__monitor)
1387

    
1388
        if acquired:
1389
          # No need for priority or timeout here as this lock has just been
1390
          # created
1391
          lock.acquire(shared=shared)
1392
          # now the lock cannot be deleted, we have it!
1393
          try:
1394
            self._add_owned(name=lockname)
1395
          except:
1396
            # We shouldn't have problems adding the lock to the owners list,
1397
            # but if we did we'll try to release this lock and re-raise
1398
            # exception.  Of course something is going to be really wrong,
1399
            # after this.  On the other hand the lock hasn't been added to the
1400
            # __lockdict yet so no other threads should be pending on it. This
1401
            # release is just a safety measure.
1402
            lock.release()
1403
            raise
1404

    
1405
        self.__lockdict[lockname] = lock
1406

    
1407
    finally:
1408
      # Only release __lock if we were not holding it previously.
1409
      if release_lock:
1410
        self.__lock.release()
1411

    
1412
    return True
1413

    
1414
  def remove(self, names):
1415
    """Remove elements from the lock set.
1416

1417
    You can either not hold anything in the lockset or already hold a superset
1418
    of the elements you want to delete, exclusively.
1419

1420
    @type names: list of strings
1421
    @param names: names of the resource to remove.
1422

1423
    @return: a list of locks which we removed; the list is always
1424
        equal to the names list if we were holding all the locks
1425
        exclusively
1426

1427
    """
1428
    # Support passing in a single resource to remove rather than many
1429
    if isinstance(names, basestring):
1430
      names = [names]
1431

    
1432
    # If we own any subset of this lock it must be a superset of what we want
1433
    # to delete. The ownership must also be exclusive, but that will be checked
1434
    # by the lock itself.
1435
    assert not self.is_owned() or self.list_owned().issuperset(names), (
1436
      "remove() on acquired lockset %s while not owning all elements" %
1437
      self.name)
1438

    
1439
    removed = []
1440

    
1441
    for lname in names:
1442
      # Calling delete() acquires the lock exclusively if we don't already own
1443
      # it, and causes all pending and subsequent lock acquires to fail. It's
1444
      # fine to call it out of order because delete() also implies release(),
1445
      # and the assertion above guarantees that if we either already hold
1446
      # everything we want to delete, or we hold none.
1447
      try:
1448
        self.__lockdict[lname].delete()
1449
        removed.append(lname)
1450
      except (KeyError, errors.LockError):
1451
        # This cannot happen if we were already holding it, verify:
1452
        assert not self.is_owned(), ("remove failed while holding lockset %s" %
1453
                                     self.name)
1454
      else:
1455
        # If no LockError was raised we are the ones who deleted the lock.
1456
        # This means we can safely remove it from lockdict, as any further or
1457
        # pending delete() or acquire() will fail (and nobody can have the lock
1458
        # since before our call to delete()).
1459
        #
1460
        # This is done in an else clause because if the exception was thrown
1461
        # it's the job of the one who actually deleted it.
1462
        del self.__lockdict[lname]
1463
        # And let's remove it from our private list if we owned it.
1464
        if self.is_owned():
1465
          self._del_owned(name=lname)
1466

    
1467
    return removed
1468

    
1469

    
1470
# Locking levels, must be acquired in increasing order. Current rules are:
1471
# - At level LEVEL_CLUSTER resides the Big Ganeti Lock (BGL) which must be
1472
#   acquired before performing any operation, either in shared or exclusive
1473
#   mode. Acquiring the BGL in exclusive mode is discouraged and should be
1474
#   avoided..
1475
# - At levels LEVEL_NODE and LEVEL_INSTANCE reside node and instance locks. If
1476
#   you need more than one node, or more than one instance, acquire them at the
1477
#   same time.
1478
# - LEVEL_NODE_RES is for node resources and should be used by operations with
1479
#   possibly high impact on the node's disks.
1480
# - LEVEL_NODE_ALLOC blocks instance allocations for the whole cluster
1481
#   ("NAL" is the only lock at this level). It should be acquired in shared
1482
#   mode when an opcode blocks all or a significant amount of a cluster's
1483
#   locks. Opcodes doing instance allocations should acquire in exclusive mode.
1484
#   Once the set of acquired locks for an opcode has been reduced to the working
1485
#   set, the NAL should be released as well to allow allocations to proceed.
1486
(LEVEL_CLUSTER,
1487
 LEVEL_NODE_ALLOC,
1488
 LEVEL_INSTANCE,
1489
 LEVEL_NODEGROUP,
1490
 LEVEL_NODE,
1491
 LEVEL_NODE_RES,
1492
 LEVEL_NETWORK) = range(0, 7)
1493

    
1494
LEVELS = [
1495
  LEVEL_CLUSTER,
1496
  LEVEL_NODE_ALLOC,
1497
  LEVEL_INSTANCE,
1498
  LEVEL_NODEGROUP,
1499
  LEVEL_NODE,
1500
  LEVEL_NODE_RES,
1501
  LEVEL_NETWORK,
1502
  ]
1503

    
1504
# Lock levels which are modifiable
1505
LEVELS_MOD = frozenset([
1506
  LEVEL_NODE_RES,
1507
  LEVEL_NODE,
1508
  LEVEL_NODEGROUP,
1509
  LEVEL_INSTANCE,
1510
  LEVEL_NETWORK,
1511
  ])
1512

    
1513
#: Lock level names (make sure to use singular form)
1514
LEVEL_NAMES = {
1515
  LEVEL_CLUSTER: "cluster",
1516
  LEVEL_INSTANCE: "instance",
1517
  LEVEL_NODEGROUP: "nodegroup",
1518
  LEVEL_NODE: "node",
1519
  LEVEL_NODE_RES: "node-res",
1520
  LEVEL_NETWORK: "network",
1521
  LEVEL_NODE_ALLOC: "node-alloc",
1522
  }
1523

    
1524
# Constant for the big ganeti lock
1525
BGL = "BGL"
1526

    
1527
#: Node allocation lock
1528
NAL = "NAL"
1529

    
1530

    
1531
class GanetiLockManager:
1532
  """The Ganeti Locking Library
1533

1534
  The purpose of this small library is to manage locking for ganeti clusters
1535
  in a central place, while at the same time doing dynamic checks against
1536
  possible deadlocks. It will also make it easier to transition to a different
1537
  lock type should we migrate away from python threads.
1538

1539
  """
1540
  _instance = None
1541

    
1542
  def __init__(self, nodes, nodegroups, instances, networks):
1543
    """Constructs a new GanetiLockManager object.
1544

1545
    There should be only a GanetiLockManager object at any time, so this
1546
    function raises an error if this is not the case.
1547

1548
    @param nodes: list of node names
1549
    @param nodegroups: list of nodegroup uuids
1550
    @param instances: list of instance names
1551

1552
    """
1553
    assert self.__class__._instance is None, \
1554
           "double GanetiLockManager instance"
1555

    
1556
    self.__class__._instance = self
1557

    
1558
    self._monitor = LockMonitor()
1559

    
1560
    # The keyring contains all the locks, at their level and in the correct
1561
    # locking order.
1562
    self.__keyring = {
1563
      LEVEL_CLUSTER: LockSet([BGL], "cluster", monitor=self._monitor),
1564
      LEVEL_NODE: LockSet(nodes, "node", monitor=self._monitor),
1565
      LEVEL_NODE_RES: LockSet(nodes, "node-res", monitor=self._monitor),
1566
      LEVEL_NODEGROUP: LockSet(nodegroups, "nodegroup", monitor=self._monitor),
1567
      LEVEL_INSTANCE: LockSet(instances, "instance", monitor=self._monitor),
1568
      LEVEL_NETWORK: LockSet(networks, "network", monitor=self._monitor),
1569
      LEVEL_NODE_ALLOC: LockSet([NAL], "node-alloc", monitor=self._monitor),
1570
      }
1571

    
1572
    assert compat.all(ls.name == LEVEL_NAMES[level]
1573
                      for (level, ls) in self.__keyring.items()), \
1574
      "Keyring name mismatch"
1575

    
1576
  def AddToLockMonitor(self, provider):
1577
    """Registers a new lock with the monitor.
1578

1579
    See L{LockMonitor.RegisterLock}.
1580

1581
    """
1582
    return self._monitor.RegisterLock(provider)
1583

    
1584
  def QueryLocks(self, fields):
1585
    """Queries information from all locks.
1586

1587
    See L{LockMonitor.QueryLocks}.
1588

1589
    """
1590
    return self._monitor.QueryLocks(fields)
1591

    
1592
  def _names(self, level):
1593
    """List the lock names at the given level.
1594

1595
    This can be used for debugging/testing purposes.
1596

1597
    @param level: the level whose list of locks to get
1598

1599
    """
1600
    assert level in LEVELS, "Invalid locking level %s" % level
1601
    return self.__keyring[level]._names()
1602

    
1603
  def is_owned(self, level):
1604
    """Check whether we are owning locks at the given level
1605

1606
    """
1607
    return self.__keyring[level].is_owned()
1608

    
1609
  def list_owned(self, level):
1610
    """Get the set of owned locks at the given level
1611

1612
    """
1613
    return self.__keyring[level].list_owned()
1614

    
1615
  def check_owned(self, level, names, shared=-1):
1616
    """Check if locks at a certain level are owned in a specific mode.
1617

1618
    @see: L{LockSet.check_owned}
1619

1620
    """
1621
    return self.__keyring[level].check_owned(names, shared=shared)
1622

    
1623
  def _upper_owned(self, level):
1624
    """Check that we don't own any lock at a level greater than the given one.
1625

1626
    """
1627
    # This way of checking only works if LEVELS[i] = i, which we check for in
1628
    # the test cases.
1629
    return compat.any((self.is_owned(l) for l in LEVELS[level + 1:]))
1630

    
1631
  def _BGL_owned(self): # pylint: disable=C0103
1632
    """Check if the current thread owns the BGL.
1633

1634
    Both an exclusive or a shared acquisition work.
1635

1636
    """
1637
    return BGL in self.__keyring[LEVEL_CLUSTER].list_owned()
1638

    
1639
  @staticmethod
1640
  def _contains_BGL(level, names): # pylint: disable=C0103
1641
    """Check if the level contains the BGL.
1642

1643
    Check if acting on the given level and set of names will change
1644
    the status of the Big Ganeti Lock.
1645

1646
    """
1647
    return level == LEVEL_CLUSTER and (names is None or BGL in names)
1648

    
1649
  def acquire(self, level, names, timeout=None, shared=0, priority=None):
1650
    """Acquire a set of resource locks, at the same level.
1651

1652
    @type level: member of locking.LEVELS
1653
    @param level: the level at which the locks shall be acquired
1654
    @type names: list of strings (or string)
1655
    @param names: the names of the locks which shall be acquired
1656
        (special lock names, or instance/node names)
1657
    @type shared: integer (0/1) used as a boolean
1658
    @param shared: whether to acquire in shared mode; by default
1659
        an exclusive lock will be acquired
1660
    @type timeout: float
1661
    @param timeout: Maximum time to acquire all locks
1662
    @type priority: integer
1663
    @param priority: Priority for acquiring lock
1664

1665
    """
1666
    assert level in LEVELS, "Invalid locking level %s" % level
1667

    
1668
    # Check that we are either acquiring the Big Ganeti Lock or we already own
1669
    # it. Some "legacy" opcodes need to be sure they are run non-concurrently
1670
    # so even if we've migrated we need to at least share the BGL to be
1671
    # compatible with them. Of course if we own the BGL exclusively there's no
1672
    # point in acquiring any other lock, unless perhaps we are half way through
1673
    # the migration of the current opcode.
1674
    assert (self._contains_BGL(level, names) or self._BGL_owned()), (
1675
      "You must own the Big Ganeti Lock before acquiring any other")
1676

    
1677
    # Check we don't own locks at the same or upper levels.
1678
    assert not self._upper_owned(level), ("Cannot acquire locks at a level"
1679
                                          " while owning some at a greater one")
1680

    
1681
    # Acquire the locks in the set.
1682
    return self.__keyring[level].acquire(names, shared=shared, timeout=timeout,
1683
                                         priority=priority)
1684

    
1685
  def downgrade(self, level, names=None):
1686
    """Downgrade a set of resource locks from exclusive to shared mode.
1687

1688
    You must have acquired the locks in exclusive mode.
1689

1690
    @type level: member of locking.LEVELS
1691
    @param level: the level at which the locks shall be downgraded
1692
    @type names: list of strings, or None
1693
    @param names: the names of the locks which shall be downgraded
1694
        (defaults to all the locks acquired at the level)
1695

1696
    """
1697
    assert level in LEVELS, "Invalid locking level %s" % level
1698

    
1699
    return self.__keyring[level].downgrade(names=names)
1700

    
1701
  def release(self, level, names=None):
1702
    """Release a set of resource locks, at the same level.
1703

1704
    You must have acquired the locks, either in shared or in exclusive
1705
    mode, before releasing them.
1706

1707
    @type level: member of locking.LEVELS
1708
    @param level: the level at which the locks shall be released
1709
    @type names: list of strings, or None
1710
    @param names: the names of the locks which shall be released
1711
        (defaults to all the locks acquired at that level)
1712

1713
    """
1714
    assert level in LEVELS, "Invalid locking level %s" % level
1715
    assert (not self._contains_BGL(level, names) or
1716
            not self._upper_owned(LEVEL_CLUSTER)), (
1717
              "Cannot release the Big Ganeti Lock while holding something"
1718
              " at upper levels (%r)" %
1719
              (utils.CommaJoin(["%s=%r" % (LEVEL_NAMES[i], self.list_owned(i))
1720
                                for i in self.__keyring.keys()]), ))
1721

    
1722
    # Release will complain if we don't own the locks already
1723
    return self.__keyring[level].release(names)
1724

    
1725
  def add(self, level, names, acquired=0, shared=0):
1726
    """Add locks at the specified level.
1727

1728
    @type level: member of locking.LEVELS_MOD
1729
    @param level: the level at which the locks shall be added
1730
    @type names: list of strings
1731
    @param names: names of the locks to acquire
1732
    @type acquired: integer (0/1) used as a boolean
1733
    @param acquired: whether to acquire the newly added locks
1734
    @type shared: integer (0/1) used as a boolean
1735
    @param shared: whether the acquisition will be shared
1736

1737
    """
1738
    assert level in LEVELS_MOD, "Invalid or immutable level %s" % level
1739
    assert self._BGL_owned(), ("You must own the BGL before performing other"
1740
                               " operations")
1741
    assert not self._upper_owned(level), ("Cannot add locks at a level"
1742
                                          " while owning some at a greater one")
1743
    return self.__keyring[level].add(names, acquired=acquired, shared=shared)
1744

    
1745
  def remove(self, level, names):
1746
    """Remove locks from the specified level.
1747

1748
    You must either already own the locks you are trying to remove
1749
    exclusively or not own any lock at an upper level.
1750

1751
    @type level: member of locking.LEVELS_MOD
1752
    @param level: the level at which the locks shall be removed
1753
    @type names: list of strings
1754
    @param names: the names of the locks which shall be removed
1755
        (special lock names, or instance/node names)
1756

1757
    """
1758
    assert level in LEVELS_MOD, "Invalid or immutable level %s" % level
1759
    assert self._BGL_owned(), ("You must own the BGL before performing other"
1760
                               " operations")
1761
    # Check we either own the level or don't own anything from here
1762
    # up. LockSet.remove() will check the case in which we don't own
1763
    # all the needed resources, or we have a shared ownership.
1764
    assert self.is_owned(level) or not self._upper_owned(level), (
1765
           "Cannot remove locks at a level while not owning it or"
1766
           " owning some at a greater one")
1767
    return self.__keyring[level].remove(names)
1768

    
1769

    
1770
def _MonitorSortKey((item, idx, num)):
1771
  """Sorting key function.
1772

1773
  Sort by name, registration order and then order of information. This provides
1774
  a stable sort order over different providers, even if they return the same
1775
  name.
1776

1777
  """
1778
  (name, _, _, _) = item
1779

    
1780
  return (utils.NiceSortKey(name), num, idx)
1781

    
1782

    
1783
class LockMonitor(object):
1784
  _LOCK_ATTR = "_lock"
1785

    
1786
  def __init__(self):
1787
    """Initializes this class.
1788

1789
    """
1790
    self._lock = SharedLock("LockMonitor")
1791

    
1792
    # Counter for stable sorting
1793
    self._counter = itertools.count(0)
1794

    
1795
    # Tracked locks. Weak references are used to avoid issues with circular
1796
    # references and deletion.
1797
    self._locks = weakref.WeakKeyDictionary()
1798

    
1799
  @ssynchronized(_LOCK_ATTR)
1800
  def RegisterLock(self, provider):
1801
    """Registers a new lock.
1802

1803
    @param provider: Object with a callable method named C{GetLockInfo}, taking
1804
      a single C{set} containing the requested information items
1805
    @note: It would be nicer to only receive the function generating the
1806
      requested information but, as it turns out, weak references to bound
1807
      methods (e.g. C{self.GetLockInfo}) are tricky; there are several
1808
      workarounds, but none of the ones I found works properly in combination
1809
      with a standard C{WeakKeyDictionary}
1810

1811
    """
1812
    assert provider not in self._locks, "Duplicate registration"
1813

    
1814
    # There used to be a check for duplicate names here. As it turned out, when
1815
    # a lock is re-created with the same name in a very short timeframe, the
1816
    # previous instance might not yet be removed from the weakref dictionary.
1817
    # By keeping track of the order of incoming registrations, a stable sort
1818
    # ordering can still be guaranteed.
1819

    
1820
    self._locks[provider] = self._counter.next()
1821

    
1822
  def _GetLockInfo(self, requested):
1823
    """Get information from all locks.
1824

1825
    """
1826
    # Must hold lock while getting consistent list of tracked items
1827
    self._lock.acquire(shared=1)
1828
    try:
1829
      items = self._locks.items()
1830
    finally:
1831
      self._lock.release()
1832

    
1833
    return [(info, idx, num)
1834
            for (provider, num) in items
1835
            for (idx, info) in enumerate(provider.GetLockInfo(requested))]
1836

    
1837
  def _Query(self, fields):
1838
    """Queries information from all locks.
1839

1840
    @type fields: list of strings
1841
    @param fields: List of fields to return
1842

1843
    """
1844
    qobj = query.Query(query.LOCK_FIELDS, fields)
1845

    
1846
    # Get all data with internal lock held and then sort by name and incoming
1847
    # order
1848
    lockinfo = sorted(self._GetLockInfo(qobj.RequestedData()),
1849
                      key=_MonitorSortKey)
1850

    
1851
    # Extract lock information and build query data
1852
    return (qobj, query.LockQueryData(map(compat.fst, lockinfo)))
1853

    
1854
  def QueryLocks(self, fields):
1855
    """Queries information from all locks.
1856

1857
    @type fields: list of strings
1858
    @param fields: List of fields to return
1859

1860
    """
1861
    (qobj, ctx) = self._Query(fields)
1862

    
1863
    # Prepare query response
1864
    return query.GetQueryResponse(qobj, ctx)