(prober: ReadWriteLock, deadline: float)
| 550 | |
| 551 | |
| 552 | def _write_intent_pending(prober: ReadWriteLock, deadline: float) -> bool: |
| 553 | try: |
| 554 | prober.acquire_read(blocking=False) |
| 555 | except Timeout: |
| 556 | return False |
| 557 | prober.release() |
| 558 | assert time.monotonic() < deadline, "Writer never registered its write intent" |
| 559 | time.sleep(0.05) |
| 560 | return True |
| 561 | |
| 562 | |
| 563 | def chain_reader( |
no test coverage detected