19

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 ...

I would prefer to use a with-statement instead of explicit "acquire" and "release", but I don't know how to get the timeout effect.

Zero Piraeus
  • 52,181
  • 26
  • 146
  • 158
Tsf
  • 1,259
  • 4
  • 16
  • 28

2 Answers2

17

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()

with acquire_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

robbles
  • 2,521
  • 1
  • 22
  • 29
  • this is working for me, thanks. But can you explain me why is it neccesary to use `yield` when using `with` in this case? –  Feb 21 '19 at 21:03
  • 1
    @monkjuice Inside a context manager, the `yield` statement is where control is passed back to the `with` block provided by the caller. So `yield result` puts the value of `result` into the `acquired` variable and runs the indented block below `with acquire_timeout...` and continues inside the context manager once it returns. – robbles Feb 23 '19 at 00:45
  • The `acquire_timeout` function in this answer is wrong; it should release the lock in a `finally` block with the `yield result` in the `try`. That way the lock will still be released when an exception is raised out of the code that runs in the context manager. – Christopher Armstrong May 17 '22 at 20:04
9

Slightly nicer version:

import threading
from contextlib import contextmanager


class TimeoutLock(object):
    def __init__(self):
        self._lock = threading.Lock()

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

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

    def release(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.

robbles
  • 2,521
  • 1
  • 22
  • 29