| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381 |
- """Synchronization functions.
- File- and mutex-based mutual exclusion synchronizers are provided,
- as well as a name-based mutex which locks within an application
- based on a string name.
- """
- import os
- import sys
- import tempfile
- try:
- import threading as _threading
- except ImportError:
- import dummy_threading as _threading
- # check for fcntl module
- try:
- sys.getwindowsversion()
- has_flock = False
- except:
- try:
- import fcntl
- has_flock = True
- except ImportError:
- has_flock = False
- from beaker import util
- from beaker.exceptions import LockError
- __all__ = ["file_synchronizer", "mutex_synchronizer", "null_synchronizer",
- "NameLock", "_threading"]
- class NameLock(object):
- """a proxy for an RLock object that is stored in a name based
- registry.
-
- Multiple threads can get a reference to the same RLock based on the
- name alone, and synchronize operations related to that name.
- """
- locks = util.WeakValuedRegistry()
- class NLContainer(object):
- def __init__(self, reentrant):
- if reentrant:
- self.lock = _threading.RLock()
- else:
- self.lock = _threading.Lock()
- def __call__(self):
- return self.lock
- def __init__(self, identifier = None, reentrant = False):
- if identifier is None:
- self._lock = NameLock.NLContainer(reentrant)
- else:
- self._lock = NameLock.locks.get(identifier, NameLock.NLContainer,
- reentrant)
- def acquire(self, wait = True):
- return self._lock().acquire(wait)
- def release(self):
- self._lock().release()
- _synchronizers = util.WeakValuedRegistry()
- def _synchronizer(identifier, cls, **kwargs):
- return _synchronizers.sync_get((identifier, cls), cls, identifier, **kwargs)
- def file_synchronizer(identifier, **kwargs):
- if not has_flock or 'lock_dir' not in kwargs:
- return mutex_synchronizer(identifier)
- else:
- return _synchronizer(identifier, FileSynchronizer, **kwargs)
- def mutex_synchronizer(identifier, **kwargs):
- return _synchronizer(identifier, ConditionSynchronizer, **kwargs)
- class null_synchronizer(object):
- def acquire_write_lock(self, wait=True):
- return True
- def acquire_read_lock(self):
- pass
- def release_write_lock(self):
- pass
- def release_read_lock(self):
- pass
- acquire = acquire_write_lock
- release = release_write_lock
- class SynchronizerImpl(object):
- def __init__(self):
- self._state = util.ThreadLocal()
- class SyncState(object):
- __slots__ = 'reentrantcount', 'writing', 'reading'
- def __init__(self):
- self.reentrantcount = 0
- self.writing = False
- self.reading = False
- def state(self):
- if not self._state.has():
- state = SynchronizerImpl.SyncState()
- self._state.put(state)
- return state
- else:
- return self._state.get()
- state = property(state)
-
- def release_read_lock(self):
- state = self.state
- if state.writing:
- raise LockError("lock is in writing state")
- if not state.reading:
- raise LockError("lock is not in reading state")
-
- if state.reentrantcount == 1:
- self.do_release_read_lock()
- state.reading = False
- state.reentrantcount -= 1
-
- def acquire_read_lock(self, wait = True):
- state = self.state
- if state.writing:
- raise LockError("lock is in writing state")
-
- if state.reentrantcount == 0:
- x = self.do_acquire_read_lock(wait)
- if (wait or x):
- state.reentrantcount += 1
- state.reading = True
- return x
- elif state.reading:
- state.reentrantcount += 1
- return True
-
- def release_write_lock(self):
- state = self.state
- if state.reading:
- raise LockError("lock is in reading state")
- if not state.writing:
- raise LockError("lock is not in writing state")
- if state.reentrantcount == 1:
- self.do_release_write_lock()
- state.writing = False
- state.reentrantcount -= 1
-
- release = release_write_lock
-
- def acquire_write_lock(self, wait = True):
- state = self.state
- if state.reading:
- raise LockError("lock is in reading state")
-
- if state.reentrantcount == 0:
- x = self.do_acquire_write_lock(wait)
- if (wait or x):
- state.reentrantcount += 1
- state.writing = True
- return x
- elif state.writing:
- state.reentrantcount += 1
- return True
- acquire = acquire_write_lock
- def do_release_read_lock(self):
- raise NotImplementedError()
-
- def do_acquire_read_lock(self):
- raise NotImplementedError()
-
- def do_release_write_lock(self):
- raise NotImplementedError()
-
- def do_acquire_write_lock(self):
- raise NotImplementedError()
- class FileSynchronizer(SynchronizerImpl):
- """a synchronizer which locks using flock().
- Adapted for Python/multithreads from Apache::Session::Lock::File,
- http://search.cpan.org/src/CWEST/Apache-Session-1.81/Session/Lock/File.pm
-
- This module does not unlink temporary files,
- because it interferes with proper locking. This can cause
- problems on certain systems (Linux) whose file systems (ext2) do not
- perform well with lots of files in one directory. To prevent this
- you should use a script to clean out old files from your lock directory.
-
- """
- def __init__(self, identifier, lock_dir):
- super(FileSynchronizer, self).__init__()
- self._filedescriptor = util.ThreadLocal()
-
- if lock_dir is None:
- lock_dir = tempfile.gettempdir()
- else:
- lock_dir = lock_dir
- self.filename = util.encoded_path(
- lock_dir,
- [identifier],
- extension='.lock'
- )
- def _filedesc(self):
- return self._filedescriptor.get()
- _filedesc = property(_filedesc)
-
- def _open(self, mode):
- filedescriptor = self._filedesc
- if filedescriptor is None:
- filedescriptor = os.open(self.filename, mode)
- self._filedescriptor.put(filedescriptor)
- return filedescriptor
-
- def do_acquire_read_lock(self, wait):
- filedescriptor = self._open(os.O_CREAT | os.O_RDONLY)
- if not wait:
- try:
- fcntl.flock(filedescriptor, fcntl.LOCK_SH | fcntl.LOCK_NB)
- return True
- except IOError:
- os.close(filedescriptor)
- self._filedescriptor.remove()
- return False
- else:
- fcntl.flock(filedescriptor, fcntl.LOCK_SH)
- return True
- def do_acquire_write_lock(self, wait):
- filedescriptor = self._open(os.O_CREAT | os.O_WRONLY)
- if not wait:
- try:
- fcntl.flock(filedescriptor, fcntl.LOCK_EX | fcntl.LOCK_NB)
- return True
- except IOError:
- os.close(filedescriptor)
- self._filedescriptor.remove()
- return False
- else:
- fcntl.flock(filedescriptor, fcntl.LOCK_EX)
- return True
-
- def do_release_read_lock(self):
- self._release_all_locks()
-
- def do_release_write_lock(self):
- self._release_all_locks()
-
- def _release_all_locks(self):
- filedescriptor = self._filedesc
- if filedescriptor is not None:
- fcntl.flock(filedescriptor, fcntl.LOCK_UN)
- os.close(filedescriptor)
- self._filedescriptor.remove()
- class ConditionSynchronizer(SynchronizerImpl):
- """a synchronizer using a Condition."""
-
- def __init__(self, identifier):
- super(ConditionSynchronizer, self).__init__()
- # counts how many asynchronous methods are executing
- self.async = 0
- # pointer to thread that is the current sync operation
- self.current_sync_operation = None
- # condition object to lock on
- self.condition = _threading.Condition(_threading.Lock())
- def do_acquire_read_lock(self, wait = True):
- self.condition.acquire()
- try:
- # see if a synchronous operation is waiting to start
- # or is already running, in which case we wait (or just
- # give up and return)
- if wait:
- while self.current_sync_operation is not None:
- self.condition.wait()
- else:
- if self.current_sync_operation is not None:
- return False
- self.async += 1
- finally:
- self.condition.release()
- if not wait:
- return True
-
- def do_release_read_lock(self):
- self.condition.acquire()
- try:
- self.async -= 1
-
- # check if we are the last asynchronous reader thread
- # out the door.
- if self.async == 0:
- # yes. so if a sync operation is waiting, notifyAll to wake
- # it up
- if self.current_sync_operation is not None:
- self.condition.notifyAll()
- elif self.async < 0:
- raise LockError("Synchronizer error - too many "
- "release_read_locks called")
- finally:
- self.condition.release()
-
- def do_acquire_write_lock(self, wait = True):
- self.condition.acquire()
- try:
- # here, we are not a synchronous reader, and after returning,
- # assuming waiting or immediate availability, we will be.
-
- if wait:
- # if another sync is working, wait
- while self.current_sync_operation is not None:
- self.condition.wait()
- else:
- # if another sync is working,
- # we dont want to wait, so forget it
- if self.current_sync_operation is not None:
- return False
-
- # establish ourselves as the current sync
- # this indicates to other read/write operations
- # that they should wait until this is None again
- self.current_sync_operation = _threading.currentThread()
- # now wait again for asyncs to finish
- if self.async > 0:
- if wait:
- # wait
- self.condition.wait()
- else:
- # we dont want to wait, so forget it
- self.current_sync_operation = None
- return False
- finally:
- self.condition.release()
-
- if not wait:
- return True
- def do_release_write_lock(self):
- self.condition.acquire()
- try:
- if self.current_sync_operation is not _threading.currentThread():
- raise LockError("Synchronizer error - current thread doesnt "
- "have the write lock")
- # reset the current sync operation so
- # another can get it
- self.current_sync_operation = None
- # tell everyone to get ready
- self.condition.notifyAll()
- finally:
- # everyone go !!
- self.condition.release()
|