lock_acquire_timeout

privex.helpers.thread.lock_acquire_timeout(lock: _thread.allocate_lock, timeout: Union[int, float] = 10, fail=False, block=True)[source]

A context manager (with lock_acquire_timeout(mylock) as locked:) for acquiring thread locks and waiting for them to be released, and giving up if the lock isn’t released within timeout.

Yields a boolean in the with context which is True if the threading.Lock was acquired within timeout, or False if it wasn’t.

>>> from privex.helpers import lock_acquire_timeout
>>> from threading import Lock
>>>
>>> my_lock = Lock()
>>>
>>> def some_func():
...     print("attempting to acquire a lock on 'my_lock'... will wait up to 30 secs...")
...     with lock_acquire_timeout(my_lock, timeout=30) as locked:
...         if not locked:
...             raise Exception("Failed to acquire 'my_lock' after waiting 30 seconds!")
...         print("successfully acquired a lock on 'my_lock'")
...     print("finished. my_lock should have been automatically released.")

Original written by “robbles” on StackOverflow: https://stackoverflow.com/a/16782391/2648583

Parameters

  • lock (Lock) – The threading.Lock object to attempt to acquire a lock on

  • timeout (int|float) – The amount of seconds to wait for lock to be released if it’s already locked

  • fail (bool) – (Default: False) If this is True, will raise LockWaitTimeout if we fail to acquire the lock lock within timeout seconds.

  • block (bool) – If this is set to False, timeout will be nulled and a non-blocking acquire will be done.

Raises

LockWaitTimeout – When fail is True and we fail to acquire lock within timeout seconds.