Saturday, March 28, 2020

Custom Pytyhon Lock with statement

class TimeoutLock(object):
  """Customized threading.Lock with timeout functionality for with statement."""

  def __init__(self):
    self.lock = threading.Lock()

  @contextlib2.contextmanager
  def acquire(self, timeout_ms: int):
    """Tries to acquires lock within specified timeout.

    Args:
      timeout_ms (int): timeout for acquiring lock.

    Yields:
      The lock acquiring result.
    """
    ok = self.lock.acquire(blocking=True, timeout=timeout_ms/1000)
    logging.info('****** LOCK acquired ********')

    yield ok
    if ok:
      self.lock.release()
      logging.info('********* LOCK released ************')
    else:
      logging.warning('couldn\'t acquire lock, likely timed out.')

  def release(self):
    """Wrapper for threading lock."""
    if self.lock.locked():
      self.lock.release()