easy_lock.py

# Using an RLock so that one decorated function may call
# another decorated function without blocking
from threading import RLock

class EasyLock:
    # Provides a decorator that wraps one or more functions
    # in an threading.RLock.
    #
    # When two or more functions are decorated using the same *name*,
    # the lock applies to all those functions collectively.
    # That is, decorated functions with different *name* can overlap in time,
    # but decorated functions sharing the same *name* will not.


    LOCKS = {}

    @classmethod
    def locked(cls, name=None):
        if not name in cls.LOCKS:
            cls.LOCKS[name] = RLock()

        def generated_decorator(func):
            def inner(*args, **kwargs):
                with cls.LOCKS[name]:
                    return func(*args, **kwargs)
            return inner

        return generated_decorator



if __name__ == '__main__':
    # The following example defines four functions.
    # Each function is the same except for the content of its print() statement.
    # From the printed output you can verify that f_a1 and f_a2 share the same lock,
    # whereas f_b1 and f_b2 each have their own lock.

    from time import sleep
    from threading import Thread

    locked = EasyLock.locked

    @locked('a')
    def f_a1(words):
        for word in words:
            print(f'f_a1 (lock a): {word}')
            sleep(1)

    @locked('a')
    def f_a2(words):
        for word in words:
            print(f'f_a2 (lock a): {word}')
            sleep(1)


    @locked('b')
    def f_b1(words):
        for word in words:
            print(f'f_b1 (lock b): {word}')
            sleep(1)

    @locked()
    def f_b2(words):
        for word in words:
            print(f'f_b2 (lock None): {word}')
            sleep(1)


    t1 = Thread(target=f_a1, args=(['Two',     'is',     'a',   'tiny',    'party.'],))
    t2 = Thread(target=f_a2, args=(['Buster',  'Rabbit', 'is',  'long',    'gone.'],))
    t3 = Thread(target=f_b1, args=(['Orange',  'and',    'red', 'make',    'pizza.'],))
    t4 = Thread(target=f_b2, args=(["Robin's", 'song',   'is',  'getting', 'monotonous.'],))

    threads = [t1, t2, t3, t4]
    for t in threads:
        t.start()

    for t in threads:
        t.join()