MCPcopy Create free account
hub / github.com/livekit/agents / _VirtualTimeLoop

Class _VirtualTimeLoop

tests/virtual_time.py:290–316  ·  view source on GitHub ↗

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

Source from the content-addressed store, hash-verified

288
289
290class _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
319class _VirtualTimePolicy(asyncio.DefaultEventLoopPolicy):

Callers 1

new_event_loopMethod · 0.85

Calls

no outgoing calls

Tested by

no test coverage detected