(self)
| 69 | |
| 70 | @contextlib.contextmanager |
| 71 | def write(self) -> Iterator[None]: |
| 72 | with self._condition: |
| 73 | while self._writer or self._readers: |
| 74 | self._condition.wait() |
| 75 | self._writer = True |
| 76 | try: |
| 77 | yield |
| 78 | finally: |
| 79 | with self._condition: |
| 80 | self._writer = False |
| 81 | self._condition.notify_all() |
| 82 | |
| 83 | |
| 84 | def _held_locks() -> dict[Path, tuple[int, int]]: |
no outgoing calls
no test coverage detected