(lock_path: Path)
| 90 | |
| 91 | |
| 92 | def _local_lock(lock_path: Path) -> _LocalRwLock: |
| 93 | resolved = lock_path.resolve() |
| 94 | with _LOCKS_GUARD: |
| 95 | lock = _LOCAL_LOCKS.get(resolved) |
| 96 | if lock is None: |
| 97 | lock = _LocalRwLock() |
| 98 | _LOCAL_LOCKS[resolved] = lock |
| 99 | return lock |
| 100 | |
| 101 | |
| 102 | def _drain_pending_journals(openkb_dir: Path) -> None: |
no test coverage detected