Python Lock With-statement And Timeout

I am using a Python 3 sequence like this: lock = threading.Lock() res = lock.acquire(timeout=10) if res: # do something .... lock.release() else: # do something else ..

Solution 1:

You can do this pretty easily with a context manager:

import threading
from contextlib import contextmanager

@contextmanager
def acquire_timeout(lock, timeout):
    result = lock.acquire(timeout=timeout)
    yield result
    if result:
        lock.release()


# Usage:lock = threading.Lock()

withacquire_timeout(lock, 2) as acquired:
    if acquired:
        print('got the lock')
        # do something ....
    else:
        print('timeout: lock not available')
        # do something else ...

*Note: This won't work in Python 2.x since there isn't a timeout argument to Lock.acquire

Solution 2:

Slightly nicer version:

import threading
from contextlib import contextmanager


classTimeoutLock(object):
    def__init__(self):
        self._lock = threading.Lock()

    defacquire(self, blocking=True, timeout=-1):
        return self._lock.acquire(blocking, timeout)

    @contextmanagerdefacquire_timeout(self, timeout):
        result = self._lock.acquire(timeout=timeout)
        yield result
        if result:
            self._lock.release()

    defrelease(self):
        self._lock.release()

# Usage:
lock = TimeoutLock()

with lock.acquire_timeout(3) as result:
    if result:
        print('got the lock')
        # do something ....else:
        print('timeout: lock not available')
        # do something else ...

It appears you can't subclass threading.Lock, so I had to make a wrapper class instead.