Hold a KB-level advisory lock.
(openkb_dir: Path, *, exclusive: bool)
| 127 | |
| 128 | @contextlib.contextmanager |
| 129 | def kb_lock(openkb_dir: Path, *, exclusive: bool) -> Iterator[None]: |
| 130 | """Hold a KB-level advisory lock.""" |
| 131 | lock_path = openkb_dir / "ingest.lock" |
| 132 | lock_path.parent.mkdir(parents=True, exist_ok=True) |
| 133 | resolved = lock_path.resolve() |
| 134 | held = _held_locks() |
| 135 | exclusive_depth, shared_depth = held.get(resolved, (0, 0)) |
| 136 | |
| 137 | if exclusive_depth or shared_depth: |
| 138 | if exclusive and not exclusive_depth: |
| 139 | raise RuntimeError("Cannot upgrade an existing KB read lock to a write lock") |
| 140 | held[resolved] = ( |
| 141 | exclusive_depth + (1 if exclusive else 0), |
| 142 | shared_depth + (0 if exclusive else 1), |
| 143 | ) |
| 144 | try: |
| 145 | yield |
| 146 | finally: |
| 147 | current_exclusive, current_shared = held[resolved] |
| 148 | next_counts = ( |
| 149 | current_exclusive - (1 if exclusive else 0), |
| 150 | current_shared - (0 if exclusive else 1), |
| 151 | ) |
| 152 | if next_counts == (0, 0): |
| 153 | del held[resolved] |
| 154 | else: |
| 155 | held[resolved] = next_counts |
| 156 | return |
| 157 | |
| 158 | local_lock = _local_lock(lock_path) |
| 159 | local_context = local_lock.write() if exclusive else local_lock.read() |
| 160 | with local_context: |
| 161 | with lock_path.open("a+", encoding="utf-8") as fh: |
| 162 | flock(fh, exclusive=exclusive) |
| 163 | held[resolved] = (1, 0) if exclusive else (0, 1) |
| 164 | try: |
| 165 | if exclusive: |
| 166 | _drain_pending_journals(openkb_dir) |
| 167 | yield |
| 168 | finally: |
| 169 | held.pop(resolved, None) |
| 170 | funlock(fh) |
| 171 | |
| 172 | |
| 173 | def kb_ingest_lock(openkb_dir: Path): |
no test coverage detected