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

Class _VirtualClock

tests/virtual_time.py:210–236  ·  view source on GitHub ↗

The loop's virtual clock. `time()` only moves when `advance()` is called (by the selector), and is what asyncio reads for scheduling -- it must stay raw so the sub-nanosecond timer tie-break (see _VirtualTimeLoop) is not swamped. `read()` is the wall-clock view: the same virtual time, b

Source from the content-addressed store, hash-verified

208
209
210class _VirtualClock:
211 """The loop's virtual clock.
212
213 `time()` only moves when `advance()` is called (by the selector), and is what asyncio reads for
214 scheduling -- it must stay raw so the sub-nanosecond timer tie-break (see _VirtualTimeLoop) is
215 not swamped. `read()` is the wall-clock view: the same virtual time, but nudged to stay
216 strictly increasing within a tick so synchronously-created items get distinct, ordered
217 `created_at`. The high-water mark lives here (per loop), so it resets naturally with each test's
218 fresh loop -- no global state.
219 """
220
221 __slots__ = ("_time", "_read")
222
223 def __init__(self) -> None:
224 self._time = 0.0
225 self._read = 0.0
226
227 def time(self) -> float:
228 return self._time
229
230 def advance(self, seconds: float) -> None:
231 if seconds > 0:
232 self._time += seconds
233
234 def read(self) -> float:
235 self._read = self._time if self._time > self._read else self._read + _TICK_EPSILON
236 return self._read
237
238
239class _AutojumpSelector(selectors.BaseSelector):

Callers 1

__init__Method · 0.85

Calls

no outgoing calls

Tested by

no test coverage detected