Courses › Low-Level Design › Problem Walkthroughs

Design a Task Scheduler

A run-time heap, a dispatcher that sleeps until due, worker pools, cancels and recurrence.

"Design a task scheduler" (run jobs at a given time, after a delay, or on a repeating schedule) is a concurrency-heavy LLD question. It's built from three pieces: a priority queue ordered by run time, a dispatcher thread that sleeps until the next task is due, and a worker pool that runs tasks without blocking the dispatcher. The follow-ups cover cancellation, retries and recurring jobs.

Requirements

  • schedule(task, run_at) and schedule_after(task, delay) return a task ID.
  • schedule_every(task, interval) for recurring tasks.
  • cancel(task_id).
  • Tasks run on a pool of N workers. A slow task must not delay other tasks that are due.
  • A failing task doesn't crash the scheduler. Optionally, retry with backoff.
  • shutdown(wait=True).

Model

ScheduledTask   id, fn, run_at, interval | None, retries_left, cancelled, seq
Scheduler       heap of (run_at, seq, task); condition variable; dispatcher thread;
                worker pool (ThreadPoolExecutor); tasks by id
                schedule(fn, run_at), schedule_after, schedule_every, cancel, shutdown
Clock           injected: now() and a way to wait (so tests can control time)

The dispatcher loop

[wait until heap non-empty] -> [peek earliest] -> [due?] -> [pop + hand to worker pool] -> [wait until heap non-empty]
[due?] -> [no: wait(timeout = run_at − now), wake early on new task] -> [peek earliest]

The detail that matters most: when a new task is scheduled earlier than the one the dispatcher is waiting for, the dispatcher has to wake up and re-check. A condition variable's notify handles this.

Implementation

import heapq
import itertools
import threading
import time
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, field


@dataclass(order=True)
class ScheduledTask:
    run_at: float
    seq: int
    id: str = field(compare=False)
    fn: object = field(compare=False)
    interval: float | None = field(default=None, compare=False)
    retries_left: int = field(default=0, compare=False)
    cancelled: bool = field(default=False, compare=False)


class Scheduler:
    def __init__(self, workers=4, clock=time.monotonic):
        self._heap = []
        self._tasks = {}
        self._cv = threading.Condition()
        self._seq = itertools.count()
        self._clock = clock
        self._pool = ThreadPoolExecutor(max_workers=workers)
        self._running = True
        self._dispatcher = threading.Thread(target=self._dispatch, daemon=True)
        self._dispatcher.start()

    # ── public API ──
    def schedule(self, fn, run_at, interval=None, retries=0):
        with self._cv:
            if not self._running:
                raise RuntimeError("scheduler is shut down")
            seq = next(self._seq)
            task = ScheduledTask(run_at, seq, f"t{seq}", fn, interval, retries)
            self._tasks[task.id] = task
            heapq.heappush(self._heap, task)
            self._cv.notify()                 # may be earlier than what we're waiting for
            return task.id

    def schedule_after(self, fn, delay, **kw):
        return self.schedule(fn, self._clock() + delay, **kw)

    def schedule_every(self, fn, interval, **kw):
        return self.schedule(fn, self._clock() + interval, interval=interval, **kw)

    def cancel(self, task_id):
        with self._cv:
            task = self._tasks.pop(task_id, None)
            if task:
                task.cancelled = True         # lazy: skipped when it reaches the top
            return task is not None

    def shutdown(self, wait=True):
        with self._cv:
            self._running = False
            self._cv.notify()
        self._dispatcher.join()
        self._pool.shutdown(wait=wait)

    # ── internals ──
    def _dispatch(self):
        while True:
            with self._cv:
                while self._running and (not self._heap or self._heap[0].run_at > self._clock()):
                    timeout = self._heap[0].run_at - self._clock() if self._heap else None
                    self._cv.wait(timeout)
                if not self._running:
                    return
                task = heapq.heappop(self._heap)
                if task.cancelled:
                    continue
            self._pool.submit(self._run, task)

    def _run(self, task):
        try:
            task.fn()
        except Exception:
            if task.retries_left > 0:
                task.retries_left -= 1
                self._requeue(task, self._clock() + 1.0)   # simple fixed backoff
                return
        if task.interval is not None:
            self._requeue(task, task.run_at + task.interval)  # fixed rate: no drift
        else:
            with self._cv:
                self._tasks.pop(task.id, None)

    def _requeue(self, task, run_at):
        with self._cv:
            if task.cancelled or not self._running:
                return
            task.run_at, task.seq = run_at, next(self._seq)
            heapq.heappush(self._heap, task)
            self._cv.notify()
ran = []
s = Scheduler(workers=2)
s.schedule_after(lambda: ran.append("b"), 0.10)
s.schedule_after(lambda: ran.append("a"), 0.05)
dropped = s.schedule_after(lambda: ran.append("x"), 0.08)
s.cancel(dropped)
time.sleep(0.25)
s.shutdown()
assert ran == ["a", "b"]

Design decisions to explain

  • Heap keyed by (run_at, seq). The sequence number breaks ties in scheduling order and avoids comparing the functions themselves.
  • The dispatcher never runs task code. It only hands tasks to the pool. A slow task occupies one worker, not the scheduler.
  • Wait with a timeout, and notify on schedule. The dispatcher sleeps exactly until the next due time, and wakes early when an earlier task arrives. No busy polling.
  • Lazy cancellation: mark the task and skip it when it's popped. Removing it from the middle of a heap is O(n).
  • Fixed-rate vs. fixed-delay recurrence: run_at + interval keeps a steady cadence, though runs can pile up if a task is slower than its interval. finish_time + interval never overlaps but drifts over time. Name the trade-off.

[!WARNING] Wall-clock time vs. monotonic time Use a monotonic clock for delays and intervals, so a system clock change can't fire everything at once or stall the scheduler. Use wall-clock time (with time zones) only for "run at 9:00 every day" schedules, and convert that to the monotonic clock when queueing.

Follow-ups

  • Overlapping recurring runs: skip a run if the previous one is still going, or allow at most one concurrent run per task with a per-task lock.
  • Priorities: add a priority to the heap key after run_at, or keep separate pools so urgent tasks never wait behind batch work.
  • Persistence: store tasks in a database so they survive restarts. At startup, decide what to do with runs that were missed: fire them immediately, skip them, or fire once.
  • Several machines: the in-memory heap becomes a table polled with leases. See Scheduling Future Work.
  • Testing: inject a fake clock and a synchronous executor so tests are deterministic.

[!TIP] Practice Design a Task Scheduler · Bounded Blocking Queue · Concurrency & Thread Safety