Synchronization classes using decorators (Python recipe) by Anand

ActiveState Code (http://code.activestate.com/recipes/533135/)

# Synchronization classes using decorators. Provides synchronized, semaphore
# and event classes which provide transparent decorator patterns for
# Lock, BoundedSemaphore and Event objects in Python.

from threading import Thread, Lock, BoundedSemaphore, Event, currentThread
from time import sleep
from random import random
    
class synchronized(object):
    """ Class enapsulating a lock and a function
    allowing it to be used as a synchronizing
    decorator making the wrapped function
    thread-safe """
    
    def __init__(self, *args):
        self.lock = Lock()
        
    def __call__(self, f):
        def lockedfunc(*args, **kwargs):
            try:
                self.lock.acquire()
                print 'Acquired lock=>',currentThread()
                try:
                    return f(*args, **kwargs)
                except Exception, e:
                    raise
            finally:
                self.lock.release()
                print 'Released lock=>',currentThread()

        return lockedfunc


class semaphore(object):
    """ Class encapsulating a semaphore to limit
    number of resources  """

    def __init__(self, *args):
        self.sem = BoundedSemaphore(args[0])
    
    def __call__(self, f):
        def semfunc(*args, **kwargs):
            try:
                print 'Trying to acquire sem=>',currentThread()
                self.sem.acquire()
                print 'Acquired sem=>',currentThread()
                try:
                    return f(*args, **kwargs)
                except Exception, e:
                    raise
            finally:
                self.sem.release()
                print 'Released sem=>',currentThread()

        
        return semfunc

class event(object):
    """ Class encapsulating an event object to control
    sequential access to a resource """

    def __init__(self, *args):
        self.evt = Event()
        self.evt.set()
    
    def __call__(self, f):
        def eventfunc(*args, **kwargs):
            try:
                print 'Waiting on event =>',currentThread()
                self.evt.wait()
                # First thread will clear the event and
                # make others wait, once it is done with the
                # job, it sets the event which wakes up
                # another thread, which does the same thing...
                # This provides sequential access to a
                # resource...
                self.evt.clear()
                print 'Cleared event =>',currentThread()
                try:
                    return f(*args, **kwargs)
                except Exception, e:
                    raise
            finally:
                # Wake up another thread...
                self.evt.set()
                print 'Set event=>',currentThread()

        return eventfunc

##############################################################################
# Test Code                                                                  #
##############################################################################
# Demonstrating the synchronization classes...
# Use a global list

l=range(10)

def reset():
    global l
    l = range(10)

# Not thread-safe        
def func1(begin, end):
    for x in range(begin, end):
        sleep(random()*0.5)
        l.append(x)

# Thread-safe!
@synchronized()
def func2(begin, end):
    for x in range(begin, end):
        sleep(random()*0.5)        
        l.append(x)


# Limited access, thread-safe
class DBConnection(object):
    """ A dummy db connection class """

    MAX = 5
    # We want to limit the number of DB connections to MAX
    # at a given time
    @semaphore(MAX)
    def connect(self, host):
        print "Connecting...",currentThread()
        # Sleep for some time
        sleep(3.0)
        pass

    # We want sequential access to this function
    @event()
    def connect2(self, host):
        print "Connecting...",currentThread()
        # Sleep for some time
        sleep(3.0)
        pass    
    

class PrintMsg(object):
    def startmsg(self):
        print '%s started...' % self.__class__.__name__
    def endmsg(self):
        print '%s ended...' % self.__class__.__name__        

class BaseThread(Thread, PrintMsg):
    pass

class MyThread1(BaseThread):
    def run(self):
        self.startmsg()
        func1(10, 20)
        self.endmsg()        

class MyThread2(BaseThread):
    def run(self):
        self.startmsg()        
        func1(20, 30)
        self.endmsg()

class MyThread3(BaseThread):
    def run(self):
        self.startmsg()
        func2(10, 20)
        self.endmsg()        

class MyThread4(BaseThread):
    def run(self):
        self.startmsg()        
        func2(20, 30)
        self.endmsg()                

class DBThread(BaseThread):
    def run(self):
        db = DBConnection()
        db.connect('localhost')


class DBThread2(BaseThread):
    def run(self):
        db = DBConnection()
        db.connect2('localhost')


print 'Starting the lock test...'

t1 = MyThread1()
t2 = MyThread2()

t1.start(); t2.start()
t1.join(); t2.join()

# List will not have elements in order
print l

reset()

t3 = MyThread3()
t4 = MyThread4()

t3.start(); t4.start()
t3.join(); t4.join()

# List will have elements in order
print l


sleep(3.0)

print 'Starting the sem test...'
# Sem test, init 8 threads and call connect
# on the DBConnection object...
for x in range(8):
   t = DBThread()
   t.start()#

sleep(3.0)

print 'Starting event test..'

# Event test, init 8 threads and 
# increment counter
for x in range(8):
    t = DBThread2()
    t.start()

print 'All tests completed.'
###############################################################################
#  End of test code                                                           #
###############################################################################