A `SelectorEventLoop` whose clock only advances when the loop would otherwise sleep. The selector jumps the virtual clock straight to the next scheduled callback (see :class:`_AutojumpSelector`), so timing-coupled tests run in ~0 wall time and deterministically. Same-instant timers are
| 288 | |
| 289 | |
| 290 | class _VirtualTimeLoop(asyncio.SelectorEventLoop): |
| 291 | """A `SelectorEventLoop` whose clock only advances when the loop would otherwise sleep. |
| 292 | |
| 293 | The selector jumps the virtual clock straight to the next scheduled callback (see |
| 294 | :class:`_AutojumpSelector`), so timing-coupled tests run in ~0 wall time and deterministically. |
| 295 | Same-instant timers are nudged into scheduling (FIFO) order so they fire in the causal order |
| 296 | real wall time would produce. |
| 297 | """ |
| 298 | |
| 299 | def __init__(self) -> None: |
| 300 | self._virtual_clock = _VirtualClock() |
| 301 | super().__init__(selector=_AutojumpSelector(self._virtual_clock)) |
| 302 | # Resolve finely enough that the sub-nanosecond tie-break is representable and only the |
| 303 | # exact-same-instant timer fires per jump. |
| 304 | self._clock_resolution = _FINE_RESOLUTION |
| 305 | self._tie_seq = 0 |
| 306 | |
| 307 | def time(self) -> float: |
| 308 | return self._virtual_clock.time() |
| 309 | |
| 310 | def monotonic_read(self) -> float: |
| 311 | """Strictly-increasing wall-clock read off this loop's clock (see _VirtualClock.read).""" |
| 312 | return self._virtual_clock.read() |
| 313 | |
| 314 | def call_at(self, when: float, callback: Any, *args: Any, **kwargs: Any) -> Any: |
| 315 | self._tie_seq += 1 |
| 316 | return super().call_at(when + self._tie_seq * _TIE_BREAK, callback, *args, **kwargs) |
| 317 | |
| 318 | |
| 319 | class _VirtualTimePolicy(asyncio.DefaultEventLoopPolicy): |