code.grnet.gr Git - ganeti-local/blob - lib/locking.py

   1 #
   2 #
   3
   4 # Copyright (C) 2006, 2007 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-msg=W0613,W0201
  24
  25 import threading
  26 # Wouldn't it be better to define LockingError in the locking module?
  27 # Well, for now that's how the rest of the code does it...
  28 from ganeti import errors
  29
  30
  31 class SharedLock:
  32   """Implements a shared lock.
  33
  34   Multiple threads can acquire the lock in a shared way, calling
  35   acquire_shared().  In order to acquire the lock in an exclusive way threads
  36   can call acquire_exclusive().
  37
  38   The lock prevents starvation but does not guarantee that threads will acquire
  39   the shared lock in the order they queued for it, just that they will
  40   eventually do so.
  41
  42   """
  43   def __init__(self):
  44     """Construct a new SharedLock"""
  45     # we have two conditions, c_shr and c_exc, sharing the same lock.
  46     self.__lock = threading.Lock()
  47     self.__turn_shr = threading.Condition(self.__lock)
  48     self.__turn_exc = threading.Condition(self.__lock)
  49
  50     # current lock holders
  51     self.__shr = set()
  52     self.__exc = None
  53
  54     # lock waiters
  55     self.__nwait_exc = 0
  56     self.__nwait_shr = 0
  57
  58     # is this lock in the deleted state?
  59     self.__deleted = False
  60
  61   def __is_sharer(self):
  62     """Is the current thread sharing the lock at this time?"""
  63     return threading.currentThread() in self.__shr
  64
  65   def __is_exclusive(self):
  66     """Is the current thread holding the lock exclusively at this time?"""
  67     return threading.currentThread() == self.__exc
  68
  69   def __is_owned(self, shared=-1):
  70     """Is the current thread somehow owning the lock at this time?
  71
  72     This is a private version of the function, which presumes you're holding
  73     the internal lock.
  74
  75     """
  76     if shared < 0:
  77       return self.__is_sharer() or self.__is_exclusive()
  78     elif shared:
  79       return self.__is_sharer()
  80     else:
  81       return self.__is_exclusive()
  82
  83   def _is_owned(self, shared=-1):
  84     """Is the current thread somehow owning the lock at this time?
  85
  86     Args:
  87       shared:
  88         < 0: check for any type of ownership (default)
  89         0: check for exclusive ownership
  90         > 0: check for shared ownership
  91
  92     """
  93     self.__lock.acquire()
  94     try:
  95       result = self.__is_owned(shared)
  96     finally:
  97       self.__lock.release()
  98
  99     return result
 100
 101   def __wait(self,c):
 102     """Wait on the given condition, and raise an exception if the current lock
 103     is declared deleted in the meantime.
 104
 105     Args:
 106       c: condition to wait on
 107
 108     """
 109     c.wait()
 110     if self.__deleted:
 111       raise errors.LockError('deleted lock')
 112
 113   def __exclusive_acquire(self):
 114     """Acquire the lock exclusively.
 115
 116     This is a private function that presumes you are already holding the
 117     internal lock. It's defined separately to avoid code duplication between
 118     acquire() and delete()
 119
 120     """
 121     self.__nwait_exc += 1
 122     try:
 123       # This is to save ourselves from a nasty race condition that could
 124       # theoretically make the sharers starve.
 125       if self.__nwait_shr > 0 or self.__nwait_exc > 1:
 126         self.__wait(self.__turn_exc)
 127
 128       while len(self.__shr) > 0 or self.__exc is not None:
 129         self.__wait(self.__turn_exc)
 130
 131       self.__exc = threading.currentThread()
 132     finally:
 133       self.__nwait_exc -= 1
 134
 135
 136   def acquire(self, blocking=1, shared=0):
 137     """Acquire a shared lock.
 138
 139     Args:
 140       shared: whether to acquire in shared mode. By default an exclusive lock
 141               will be acquired.
 142       blocking: whether to block while trying to acquire or to operate in try-lock mode.
 143                 this locking mode is not supported yet.
 144
 145     """
 146     if not blocking:
 147       # We don't have non-blocking mode for now
 148       raise NotImplementedError
 149
 150     self.__lock.acquire()
 151     try:
 152       if self.__deleted:
 153         raise errors.LockError('deleted lock')
 154
 155       # We cannot acquire the lock if we already have it
 156       assert not self.__is_owned(), "double acquire() on a non-recursive lock"
 157
 158       if shared:
 159         self.__nwait_shr += 1
 160         try:
 161           # If there is an exclusive holder waiting we have to wait.  We'll
 162           # only do this once, though, when we start waiting for the lock. Then
 163           # we'll just wait while there are no exclusive holders.
 164           if self.__nwait_exc > 0:
 165             # TODO: if !blocking...
 166             self.__wait(self.__turn_shr)
 167
 168           while self.__exc is not None:
 169             # TODO: if !blocking...
 170             self.__wait(self.__turn_shr)
 171
 172           self.__shr.add(threading.currentThread())
 173         finally:
 174           self.__nwait_shr -= 1
 175
 176       else:
 177         # TODO: if !blocking...
 178         # (or modify __exclusive_acquire for non-blocking mode)
 179         self.__exclusive_acquire()
 180
 181     finally:
 182       self.__lock.release()
 183
 184     return True
 185
 186   def release(self):
 187     """Release a Shared Lock.
 188
 189     You must have acquired the lock, either in shared or in exclusive mode,
 190     before calling this function.
 191
 192     """
 193     self.__lock.acquire()
 194     try:
 195       # Autodetect release type
 196       if self.__is_exclusive():
 197         self.__exc = None
 198
 199         # An exclusive holder has just had the lock, time to put it in shared
 200         # mode if there are shared holders waiting. Otherwise wake up the next
 201         # exclusive holder.
 202         if self.__nwait_shr > 0:
 203           self.__turn_shr.notifyAll()
 204         elif self.__nwait_exc > 0:
 205          self.__turn_exc.notify()
 206
 207       elif self.__is_sharer():
 208         self.__shr.remove(threading.currentThread())
 209
 210         # If there are shared holders waiting there *must* be an exclusive holder
 211         # waiting as well; otherwise what were they waiting for?
 212         assert (self.__nwait_shr == 0 or self.__nwait_exc > 0,
 213                 "Lock sharers waiting while no exclusive is queueing")
 214
 215         # If there are no more shared holders and some exclusive holders are
 216         # waiting let's wake one up.
 217         if len(self.__shr) == 0 and self.__nwait_exc > 0:
 218           self.__turn_exc.notify()
 219
 220       else:
 221         assert False, "Cannot release non-owned lock"
 222
 223     finally:
 224       self.__lock.release()
 225
 226   def delete(self, blocking=1):
 227     """Delete a Shared Lock.
 228
 229     This operation will declare the lock for removal. First the lock will be
 230     acquired in exclusive mode if you don't already own it, then the lock
 231     will be put in a state where any future and pending acquire() fail.
 232
 233     Args:
 234       blocking: whether to block while trying to acquire or to operate in
 235                 try-lock mode.  this locking mode is not supported yet unless
 236                 you are already holding exclusively the lock.
 237
 238     """
 239     self.__lock.acquire()
 240     try:
 241       assert not self.__is_sharer(), "cannot delete() a lock while sharing it"
 242
 243       if self.__deleted:
 244         raise errors.LockError('deleted lock')
 245
 246       if not self.__is_exclusive():
 247         if not blocking:
 248           # We don't have non-blocking mode for now
 249           raise NotImplementedError
 250         self.__exclusive_acquire()
 251
 252       self.__deleted = True
 253       self.__exc = None
 254       # Wake up everybody, they will fail acquiring the lock and
 255       # raise an exception instead.
 256       self.__turn_exc.notifyAll()
 257       self.__turn_shr.notifyAll()
 258
 259     finally:
 260       self.__lock.release()
 261