SyncScheduler

__init__() -> None

def __init__(self) -> None:
    super().__init__()
    self._last_activity = time.monotonic()
    self._ready: deque[Task] = deque()
    self._timers: list[tuple[float, int, _Timer]] = []
    self._seq = itertools.count()
    self._cond = threading.Condition()

schedule(task) -> None

def schedule(self, task: Task) -> None:
    task = self._tracked(task)
    with self._cond:
        self._ready.append(task)
        self._last_activity = time.monotonic()
        self._cond.notify()

call_later(delay, task, *, daemon=False) -> Cancel

def call_later(self, delay: Seconds, task: Task, *, daemon: bool = False) -> Cancel:
    timer = _Timer(time.monotonic() + max(delay, 0.0), task, daemon=daemon)
    with self._cond:
        heapq.heappush(self._timers, (timer.when, next(self._seq), timer))
        self._cond.notify()
    return timer.cancel

run_until(done, *, stall_timeout=None) -> None

def run_until(self, done: Callable[[], bool], *, stall_timeout: float | None = None) -> None:
    """Run tasks and timers until ``done()`` is true.

    Blocks while nothing is ready. With ``stall_timeout``, raises
    ``StalledError`` once nothing has been ready, no timer but daemon
    timers has been pending, and no other thread has scheduled anything
    for that long.

    A ``KeyboardInterrupt`` raised while blocked propagates; the caller
    decides how to unwind.
    """
    self._last_activity = time.monotonic()
    while not done():
        with self._cond:
            self._move_due_timers()
            if not self._ready:
                self._cond.wait(self._wait_time(stall_timeout))
                continue
            batch, self._ready = self._ready, deque()
        for task in batch:
            task()

track(tracker, f) -> None

def track(self, tracker: WakeTracker, f: Callable[[], None]) -> None:
    """Run ``f``, counting what it schedules against ``tracker``."""
    previous = self._tracking.tracker
    self._tracking.tracker = tracker
    try:
        f()
    finally:
        self._tracking.tracker = previous