Function acquire

tests/test_fork_coordination.py:1113–1117

()

        _release = _acquire

    def acquire() -> None:
        try:
            BlockingTransitionLock(str(tmp_path / "transition.lock"), is_singleton=False).acquire(timeout=0)
        except Timeout:
            finished.set()

    worker = threading.Thread(target=acquire)
    worker.start()