Function _handle_lock

certbot/src/certbot/tests/util.py:400–416

Acquire a file lock on given path, then wait to release it. This worker is coordinated using events to signal when the lock should be acquired and released. :param multiprocessing.Event event_in: event object to signal when to release the lock :param multiprocessing.Event event_out:

(event_in: synchronize.Event, event_out: synchronize.Event, path: str)

def _handle_lock(event_in: synchronize.Event, event_out: synchronize.Event, path: str) -> None:
    """
    Acquire a file lock on given path, then wait to release it. This worker is coordinated
    using events to signal when the lock should be acquired and released.
    :param multiprocessing.Event event_in: event object to signal when to release the lock
    :param multiprocessing.Event event_out: event object to signal when the lock is acquired
    :param path: the path to lock
    """
    if os.path.isdir(path):
        my_lock = lock.lock_dir(path)
    else:
        my_lock = lock.LockFile(path)
    try:
        event_out.set()
        assert event_in.wait(timeout=20), 'Timeout while waiting to release the lock.'
    finally:
        my_lock.release()


def lock_and_call(callback: Callable[[], Any], path_to_lock: str) -> None: