| """Synchronization metaclass. | |
| This metaclass makes it possible to declare synchronized methods. | |
| """ | |
| import thread | |
| # First we need to define a reentrant lock. | |
| # This is generally useful and should probably be in a standard Python | |
| # library module. For now, we in-line it. | |
| class Lock: | |
| """Reentrant lock. | |
| This is a mutex-like object which can be acquired by the same | |
| thread more than once. It keeps a reference count of the number | |
| of times it has been acquired by the same thread. Each acquire() | |
| call must be matched by a release() call and only the last | |
| release() call actually releases the lock for acquisition by | |
| another thread. | |
| The implementation uses two locks internally: | |
| __mutex is a short term lock used to protect the instance variables | |
| __wait is the lock for which other threads wait | |
| A thread intending to acquire both locks should acquire __wait | |
| first. | |
| The implementation uses two other instance variables, protected by | |
| locking __mutex: | |
| __tid is the thread ID of the thread that currently has the lock | |
| __count is the number of times the current thread has acquired it | |
| When the lock is released, __tid is None and __count is zero. | |
| """ | |
| def __init__(self): | |
| """Constructor. Initialize all instance variables.""" | |
| self.__mutex = thread.allocate_lock() | |
| self.__wait = thread.allocate_lock() | |
| self.__tid = None | |
| self.__count = 0 | |
| def acquire(self, flag=1): | |
| """Acquire the lock. | |
| If the optional flag argument is false, returns immediately | |
| when it cannot acquire the __wait lock without blocking (it | |
| may still block for a little while in order to acquire the | |
| __mutex lock). | |
| The return value is only relevant when the flag argument is | |
| false; it is 1 if the lock is acquired, 0 if not. | |
| """ | |
| self.__mutex.acquire() | |
| try: | |
| if self.__tid == thread.get_ident(): | |
| self.__count = self.__count + 1 | |
| return 1 | |
| finally: | |
| self.__mutex.release() | |
| locked = self.__wait.acquire(flag) | |
| if not flag and not locked: | |
| return 0 | |
| try: | |
| self.__mutex.acquire() | |
| assert self.__tid == None | |
| assert self.__count == 0 | |
| self.__tid = thread.get_ident() | |
| self.__count = 1 | |
| return 1 | |
| finally: | |
| self.__mutex.release() | |
| def release(self): | |
| """Release the lock. | |
| If this thread doesn't currently have the lock, an assertion | |
| error is raised. | |
| Only allow another thread to acquire the lock when the count | |
| reaches zero after decrementing it. | |
| """ | |
| self.__mutex.acquire() | |
| try: | |
| assert self.__tid == thread.get_ident() | |
| assert self.__count > 0 | |
| self.__count = self.__count - 1 | |
| if self.__count == 0: | |
| self.__tid = None | |
| self.__wait.release() | |
| finally: | |
| self.__mutex.release() | |
| def _testLock(): | |
| done = [] | |
| def f2(lock, done=done): | |
| lock.acquire() | |
| print "f2 running in thread %d\n" % thread.get_ident(), | |
| lock.release() | |
| done.append(1) | |
| def f1(lock, f2=f2, done=done): | |
| lock.acquire() | |
| print "f1 running in thread %d\n" % thread.get_ident(), | |
| try: | |
| f2(lock) | |
| finally: | |
| lock.release() | |
| done.append(1) | |
| lock = Lock() | |
| lock.acquire() | |
| f1(lock) # Adds 2 to done | |
| lock.release() | |
| lock.acquire() | |
| thread.start_new_thread(f1, (lock,)) # Adds 2 | |
| thread.start_new_thread(f1, (lock, f1)) # Adds 3 | |
| thread.start_new_thread(f2, (lock,)) # Adds 1 | |
| thread.start_new_thread(f2, (lock,)) # Adds 1 | |
| lock.release() | |
| import time | |
| while len(done) < 9: | |
| print len(done) | |
| time.sleep(0.001) | |
| print len(done) | |
| # Now, the Locking metaclass is a piece of cake. | |
| # As an example feature, methods whose name begins with exactly one | |
| # underscore are not synchronized. | |
| from Meta import MetaClass, MetaHelper, MetaMethodWrapper | |
| class LockingMethodWrapper(MetaMethodWrapper): | |
| def __call__(self, *args, **kw): | |
| if self.__name__[:1] == '_' and self.__name__[1:] != '_': | |
| return apply(self.func, (self.inst,) + args, kw) | |
| self.inst.__lock__.acquire() | |
| try: | |
| return apply(self.func, (self.inst,) + args, kw) | |
| finally: | |
| self.inst.__lock__.release() | |
| class LockingHelper(MetaHelper): | |
| __methodwrapper__ = LockingMethodWrapper | |
| def __helperinit__(self, formalclass): | |
| MetaHelper.__helperinit__(self, formalclass) | |
| self.__lock__ = Lock() | |
| class LockingMetaClass(MetaClass): | |
| __helper__ = LockingHelper | |
| Locking = LockingMetaClass('Locking', (), {}) | |
| def _test(): | |
| # For kicks, take away the Locking base class and see it die | |
| class Buffer(Locking): | |
| def __init__(self, initialsize): | |
| assert initialsize > 0 | |
| self.size = initialsize | |
| self.buffer = [None]*self.size | |
| self.first = self.last = 0 | |
| def put(self, item): | |
| # Do we need to grow the buffer? | |
| if (self.last+1) % self.size != self.first: | |
| # Insert the new item | |
| self.buffer[self.last] = item | |
| self.last = (self.last+1) % self.size | |
| return | |
| # Double the buffer size | |
| # First normalize it so that first==0 and last==size-1 | |
| print "buffer =", self.buffer | |
| print "first = %d, last = %d, size = %d" % ( | |
| self.first, self.last, self.size) | |
| if self.first <= self.last: | |
| temp = self.buffer[self.first:self.last] | |
| else: | |
| temp = self.buffer[self.first:] + self.buffer[:self.last] | |
| print "temp =", temp | |
| self.buffer = temp + [None]*(self.size+1) | |
| self.first = 0 | |
| self.last = self.size-1 | |
| self.size = self.size*2 | |
| print "Buffer size doubled to", self.size | |
| print "new buffer =", self.buffer | |
| print "first = %d, last = %d, size = %d" % ( | |
| self.first, self.last, self.size) | |
| self.put(item) # Recursive call to test the locking | |
| def get(self): | |
| # Is the buffer empty? | |
| if self.first == self.last: | |
| raise EOFError # Avoid defining a new exception | |
| item = self.buffer[self.first] | |
| self.first = (self.first+1) % self.size | |
| return item | |
| def producer(buffer, wait, n=1000): | |
| import time | |
| i = 0 | |
| while i < n: | |
| print "put", i | |
| buffer.put(i) | |
| i = i+1 | |
| print "Producer: done producing", n, "items" | |
| wait.release() | |
| def consumer(buffer, wait, n=1000): | |
| import time | |
| i = 0 | |
| tout = 0.001 | |
| while i < n: | |
| try: | |
| x = buffer.get() | |
| if x != i: | |
| raise AssertionError, \ | |
| "get() returned %s, expected %s" % (x, i) | |
| print "got", i | |
| i = i+1 | |
| tout = 0.001 | |
| except EOFError: | |
| time.sleep(tout) | |
| tout = tout*2 | |
| print "Consumer: done consuming", n, "items" | |
| wait.release() | |
| pwait = thread.allocate_lock() | |
| pwait.acquire() | |
| cwait = thread.allocate_lock() | |
| cwait.acquire() | |
| buffer = Buffer(1) | |
| n = 1000 | |
| thread.start_new_thread(consumer, (buffer, cwait, n)) | |
| thread.start_new_thread(producer, (buffer, pwait, n)) | |
| pwait.acquire() | |
| print "Producer done" | |
| cwait.acquire() | |
| print "All done" | |
| print "buffer size ==", len(buffer.buffer) | |
| if __name__ == '__main__': | |
| _testLock() | |
| _test() |