Design a task scheduler (cron, LLD)¶
TL;DR¶
- You build an in-process scheduler: a min-heap ordered by due time, one timer thread that waits on a
Condition, and a bounded pool of workers that run the tasks. - Three decisions carry the interview:
Condition.wait(timeout)instead of a poll loop, so scheduling something sooner wakes the timer immediately; cancellation tombstones the heap entry instead of removing it; and a run that outlasts its own period needs a stated policy, not silent pile-up. - Patterns that earn their place: Command (
Task), Strategy (Schedule), Observer (TaskListener), Producer-Consumer (heap plus pool), Dependency Injection (Clock).
Problem statement¶
"Design an in-process task scheduler. Callers register work to run once, at a fixed rate, after a fixed delay, or on a cron expression. Tasks have priorities, can be cancelled, paused and resumed, and are retried with backoff when they fail. Several tasks may be due at the same instant, so there is a pool of worker threads. It must shut down without losing work. Show me the data structure the timer uses, and be precise about what the timer thread does while it has nothing to do."
Requirements¶
Functional
- Schedules: one-off, fixed rate, fixed delay, and cron to the minute.
- Priorities, used to break a tie between tasks due at the same instant.
- A worker pool, so a slow task does not delay the next due task.
- Cancel, pause and resume for a registered task.
- Retries with exponential backoff and a cap; exhausted tasks go to a dead-letter list.
- Per-task timeouts, and a stated answer for what a timeout can and cannot do in Python.
- Status and per-run history: attempt number, start, finish, outcome, error.
- A persistence interface every status change is written through.
- Graceful shutdown that stops timing first and then drains what is in flight.
Non-functional and constraints
- No polling. An idle scheduler must consume no CPU, and the timer must react immediately when work is scheduled earlier than the current head.
- Deterministic and testable: the clock is injected, so a test proves a twelve-hour delay in milliseconds.
- A failing task must never kill its worker thread, and a slow listener must never block the scheduler.
Out of scope: distribution and leader election, task dependencies as a DAG, and any form of preemption.
Clarifying questions and assumptions¶
| Question to ask | Assumption taken here |
|---|---|
| One process or a cluster? | One. The TaskStore protocol is the seam to a shared store, and distribution is the HLD case study. |
| Is fixed rate measured from the start or the finish of a run? | From the start, which is what makes it rate. FixedDelay measures from the finish, and that is the safer default. |
| What if a run takes longer than its period? | An explicit OverrunPolicy: skip to the next slot in the future, or catch up. Never "keep queueing". |
| Can I stop a task that is already running? | No — Python cannot preempt a thread. cancel stops future runs; real cancellation is cooperative. |
| Does the cron need seconds, or time zones? | Minutes, in UTC. Both are extensions and neither changes the design. |
| Is a retry a new run or the same run? | The same task, attempt n + 1, re-queued at finished_at + backoff. The schedule is not consulted. |
| What does the queue do while it waits? | Blocks on a Condition with the exact remaining delay. This is the answer they are probing for. |
Core entities and relationships¶
Task— the Command: an id, a name, the callable, a priority, a retry policy and a timeout. It knows nothing about when.ScheduledTask— the task plus its schedule, status, next run, attempt counter,generationand history.Schedule(Protocol) withOneTime,FixedDelay,FixedRate,CronSchedule— two questions each: when first, and when again after a given run.QueueEntry— what the heap orders: due time, then priority, then arrival sequence, plus the task id and generation.TaskQueue— the min-heap and theCondition.WorkerPool— fixed threads draining a bounded queue.Scheduler— the registry, the timer thread, the pool, the dead-letter list and the lock over all of it.ExecutionRecord— one attempt, kept for history and used by the schedule to compute the next run.TaskStore(Protocol) withInMemoryTaskStore— the persistence seam.TaskListenerwithEventLog— the Observer seam, carrying an immutableTaskEvent.
Multiplicities: scheduler 1 → * scheduled tasks, scheduled task 1 → 1 task and 1 → 1 schedule, scheduled task 1 → * execution records, queue 1 → * entries, scheduler 1 → * listeners.
Class diagram¶
The machine: registry, heap, pool, and the two seams that hang off it.
classDiagram
class Scheduler {
-_lock: Lock
-_tasks: Dict~str,ScheduledTask~
+schedule(name, action, schedule) ScheduledTask
+cancel(task_id) bool
+pause(task_id) bool
+resume(task_id) bool
+start()
+shutdown(drain, timeout)
}
class TaskQueue {
-_condition: Condition
-_heap: List~QueueEntry~
+push(entry)
+pop_due() QueueEntry
+wake()
+close()
}
class WorkerPool {
-_queue: Queue
+start()
+submit(job)
+shutdown(drain, timeout)
}
class QueueEntry {
+due_at: float
+priority: Priority
+sequence: int
+task_id: str
+generation: int
}
class ScheduledTask {
+status: TaskStatus
+next_run_at: float
+attempt: int
+generation: int
+history: List~ExecutionRecord~
}
class Task {
+id: str
+name: str
+action: Callable
+priority: Priority
+retry: RetryPolicy
+timeout: float
}
class RetryPolicy {
+max_attempts: int
+should_retry(attempt) bool
+backoff(attempt) float
}
class ExecutionRecord {
+attempt: int
+started_at: float
+finished_at: float
+ok: bool
+timed_out: bool
}
class TaskStore {
<<interface>>
+save(task)
+delete(task_id)
}
class InMemoryTaskStore
class TaskListener {
<<interface>>
+on_task_event(event)
}
class EventLog {
+wait_for(predicate, timeout) bool
+runs(name) int
}
class TaskEvent {
+task_id: str
+status: TaskStatus
+record: ExecutionRecord
}
Scheduler "1" *-- "1" TaskQueue
Scheduler "1" *-- "1" WorkerPool
Scheduler "1" o-- "many" ScheduledTask
Scheduler --> TaskStore
Scheduler --> "many" TaskListener
TaskQueue "1" o-- "many" QueueEntry
ScheduledTask "1" *-- "1" Task
ScheduledTask "1" o-- "many" ExecutionRecord
Task --> RetryPolicy
TaskStore <|.. InMemoryTaskStore
TaskListener <|.. EventLog
Scheduler ..> TaskEvent : publishesThe schedules: one Protocol, four answers to "when again?".
classDiagram
class Schedule {
<<interface>>
+first_run(now) float
+next_run(record, now) float
}
class OneTime {
+at: float
}
class FixedDelay {
+delay: float
+start_delay: float
}
class FixedRate {
+period: float
+on_overrun: OverrunPolicy
}
class CronSchedule {
+expression: str
+next_after(timestamp) float
}
class OverrunPolicy {
<<enumeration>>
SKIP
CATCH_UP
}
class ExecutionRecord
class ScheduledTask
Schedule <|.. OneTime
Schedule <|.. FixedDelay
Schedule <|.. FixedRate
Schedule <|.. CronSchedule
FixedRate --> OverrunPolicy
Schedule ..> ExecutionRecord : reads
ScheduledTask --> ScheduleDesign patterns applied¶
| Pattern | Where | Why it earns its place |
|---|---|---|
| Command | Task |
The work is an object with its own retry policy and timeout, so the queue, the pool and the store move it around without knowing what it does. It is also what makes a history of attempts meaningful. |
| Strategy | Schedule with four implementations |
"Now support cron" is a new class and nothing else. Each schedule answers two questions and holds no state, so one instance is safely shared by many tasks. |
| Producer-Consumer | TaskQueue (heap plus Condition) feeding WorkerPool |
The timer produces due tasks, the pool consumes them, and the bounded queue between them is backpressure rather than an unbounded backlog. |
| Observer | TaskListener, EventLog |
Status changes are published to listeners outside the scheduler lock, so a slow metrics sink cannot stall the timer. |
| Repository | TaskStore |
Every transition is written through one interface, so adding durability is one class, not an edit to _settle. |
| Dependency Injection | Clock, IdGenerator, TaskStore, listeners |
A twelve-hour delay is tested in milliseconds and task ids are t-1, t-2. |
What was deliberately not used: Singleton for the scheduler, even though the brief suggests it. One instance built in main and injected is strictly better here — the tests build one scheduler per case and shut it down in a fixture, which a module-level singleton makes impossible without a reset() back door. Say that out loud. State classes per TaskStatus are also skipped: seven statuses with guarded transitions in three methods is an enum's job, and a class per status would be ceremony around two if statements.
Key flows¶
One tick: the timer waits exactly as long as it must, then hands work to a pool.
sequenceDiagram
autonumber
participant T as TimerThread
participant Q as TaskQueue
participant P as WorkerPool
participant W as Worker
participant S as Scheduler
T->>Q: pop_due()
Q->>Q: delay = head.due_at - clock.now()
alt nothing due yet
Q->>Q: condition.wait(delay)
Note over Q: a push with an earlier due time notifies and re-arms
end
Q-->>T: QueueEntry
T->>P: submit(execute(entry))
P->>W: dequeue the job
W->>S: check generation, mark RUNNING, run the action
alt run succeeded
W->>S: schedule.next_run(record, now)
S->>Q: push the next entry
else run failed and retries remain
W->>S: retry.backoff(attempt)
S->>Q: push at finished_at + backoff
else retries exhausted
W->>S: status FAILED, id to the dead letter
endThe lifecycle. RETRYING is a separate state from SCHEDULED on purpose: a retry keeps the attempt counter, a normal reschedule resets it.
stateDiagram-v2
[*] --> SCHEDULED : schedule(task, rule)
SCHEDULED --> RUNNING : due, worker picks it up
RUNNING --> SUCCEEDED : action returned and the rule has no next run
RUNNING --> SCHEDULED : action returned and the rule has a next run
RUNNING --> RETRYING : raised or timed out, attempts remain
RETRYING --> RUNNING : backoff elapsed
RUNNING --> FAILED : retries exhausted, id to the dead letter
SCHEDULED --> PAUSED : pause
PAUSED --> SCHEDULED : resume
SCHEDULED --> CANCELLED : cancel
PAUSED --> CANCELLED : cancel
SUCCEEDED --> [*]
FAILED --> [*]
CANCELLED --> [*]Implementation¶
Write the vocabulary first, then the schedules, then the queue — because the queue is what the interview is really about — and only then the scheduler that ties them together.
Seven statuses, three priorities that sort correctly by being an IntEnum, and an overrun policy that forces the awkward question to be answered at configuration time:
code/lld/task_scheduler/models.py — statuses, priorities and errors
class TaskStatus(StrEnum):
SCHEDULED = "scheduled" # sitting in the heap with a due time
RUNNING = "running" # a worker is executing it now
SUCCEEDED = "succeeded" # last run passed (a recurring task returns to SCHEDULED)
RETRYING = "retrying" # last run failed and a retry is queued
FAILED = "failed" # retries exhausted; the id is on the dead-letter list
PAUSED = "paused" # stays registered, never dispatched
CANCELLED = "cancelled" # terminal, removed from consideration
class Priority(IntEnum):
"""Lower sorts first, so the heap orders by priority without a custom comparator."""
HIGH = 0
NORMAL = 1
LOW = 2
class OverrunPolicy(StrEnum):
"""What a fixed-rate schedule does when a run outlasts its own period."""
SKIP = "skip" # jump forward to the next slot in the future: no pile-up
CATCH_UP = "catch_up" # run again immediately, once per missed slot
class TaskNotFoundError(NotFoundError):
"""No task with that id is registered."""
class SchedulerStateError(InvalidStateError):
"""The scheduler is not running, or is already shutting down."""
class ScheduleError(ValidationError):
"""A schedule cannot be built or can never fire again."""
Task is the Command and RetryPolicy is frozen so one policy can be shared by every task that wants it. ExecutionRecord is not only history: the schedule reads it to work out the next run.
code/lld/task_scheduler/models.py — the command and its record
@dataclass(frozen=True, slots=True)
class RetryPolicy:
"""Exponential backoff with a ceiling. Frozen, so one policy is safely shared."""
max_attempts: int = 3
initial_backoff: float = 1.0
multiplier: float = 2.0
max_backoff: float = 60.0
def __post_init__(self) -> None:
if self.max_attempts < 1 or self.initial_backoff <= 0 or self.multiplier < 1:
raise ValidationError("retry policy must allow at least one attempt with positive backoff")
def should_retry(self, attempt: int) -> bool:
return attempt < self.max_attempts
def backoff(self, attempt: int) -> float:
"""Delay before attempt ``attempt + 1``: 1 s, 2 s, 4 s, ... capped."""
return min(self.max_backoff, self.initial_backoff * self.multiplier ** (attempt - 1))
DEFAULT_RETRY = RetryPolicy()
NO_RETRY = RetryPolicy(max_attempts=1)
@dataclass(frozen=True, slots=True)
class Task:
"""The Command: the work and how to treat it, with nothing about *when*.
Separating this from ``ScheduledTask`` is what lets the same callable be
registered twice under two schedules, and it is what a persistence layer
stores as the definition rather than as the running state.
"""
id: str
name: str
action: Callable[[], object]
priority: Priority = Priority.NORMAL
retry: RetryPolicy = DEFAULT_RETRY
timeout: float | None = None # seconds. See the note on cooperative cancellation.
@dataclass(frozen=True, slots=True)
class ExecutionRecord:
"""One attempt, kept for history and for the schedule's next-run arithmetic."""
attempt: int
started_at: float
finished_at: float
ok: bool
error: str | None = None
timed_out: bool = False
@property
def duration(self) -> float:
return self.finished_at - self.started_at
ScheduledTask carries the running state, and QueueEntry is what the heap actually orders. The sequence field is the detail worth pointing at: without it, two entries that tie on time and priority would fall back to comparing task ids.
code/lld/task_scheduler/models.py — running state and the heap entry
@dataclass(slots=True)
class ScheduledTask:
"""A task plus when it next runs, what state it is in, and what happened before.
``generation`` is the tombstone trick: removing an entry from the middle of a
heap is O(n), so cancel, pause and reschedule bump this counter instead and a
popped entry whose generation no longer matches is simply dropped.
"""
task: Task
schedule: Schedule
status: TaskStatus = TaskStatus.SCHEDULED
next_run_at: float | None = None
attempt: int = 0
generation: int = 0
history: list[ExecutionRecord] = field(default_factory=list)
@property
def id(self) -> str:
return self.task.id
@property
def name(self) -> str:
return self.task.name
@property
def is_active(self) -> bool:
return self.status not in (TaskStatus.CANCELLED, TaskStatus.PAUSED, TaskStatus.FAILED)
def last_run(self) -> ExecutionRecord | None:
return self.history[-1] if self.history else None
def summary(self) -> str:
runs = len(self.history)
ok = sum(1 for r in self.history if r.ok)
return f"{self.name}: {self.status} after {runs} run(s), {ok} ok"
@dataclass(frozen=True, slots=True)
class TaskEvent:
"""What a listener receives. Immutable, so it can leave the lock safely."""
task_id: str
name: str
status: TaskStatus
record: ExecutionRecord | None = None
@dataclass(frozen=True, slots=True, order=True)
class QueueEntry:
"""What the heap orders: due time first, then priority, then arrival.
``sequence`` is what makes the ordering total. Without it, two entries with
the same due time and priority would compare on ``task_id`` - alphabetical
order masquerading as fairness.
"""
due_at: float
priority: Priority
sequence: int
task_id: str
generation: int
The three simple schedules fit on one screen, and the difference between fixed rate and fixed delay — start versus finish — is one attribute access:
code/lld/task_scheduler/schedules.py — the Schedule protocol and three rules
class Schedule(Protocol):
"""A recurrence rule. Pure and stateless, so one instance is safely shared."""
def first_run(self, now: float) -> float | None:
"""When this task should run for the first time."""
def next_run(self, record: ExecutionRecord, now: float) -> float | None:
"""When to run again after ``record``. ``None`` retires the task."""
@dataclass(frozen=True, slots=True)
class OneTime:
"""Run once, at ``at`` - or immediately if that instant has already passed."""
at: float
def first_run(self, now: float) -> float | None:
return max(self.at, now)
def next_run(self, record: ExecutionRecord, now: float) -> float | None:
return None
@dataclass(frozen=True, slots=True)
class FixedDelay:
"""Wait ``delay`` seconds *after the previous run finished*.
Two runs can therefore never overlap, whatever a run costs. This is the right
default for jobs whose duration varies - a backup, a sync, an import.
"""
delay: float
start_delay: float = 0.0
def first_run(self, now: float) -> float | None:
return now + self.start_delay
def next_run(self, record: ExecutionRecord, now: float) -> float | None:
return record.finished_at + self.delay
@dataclass(frozen=True, slots=True)
class FixedRate:
"""Fire every ``period`` seconds measured from when a run *started*.
The interesting case is a run that outlasts its own period. ``SKIP`` jumps to
the next slot in the future, so a job that hangs for an hour does not come
back to a queue of sixty missed minutes. ``CATCH_UP`` runs the backlog. Pick
one deliberately: the default of "just keep adding" is how schedulers melt.
"""
period: float
on_overrun: OverrunPolicy = OverrunPolicy.SKIP
start_delay: float = 0.0
def first_run(self, now: float) -> float | None:
return now + self.start_delay
def next_run(self, record: ExecutionRecord, now: float) -> float | None:
due = record.started_at + self.period
if due > now or self.on_overrun is OverrunPolicy.CATCH_UP:
return due
missed = int((now - due) // self.period) + 1
return due + missed * self.period
Cron is where candidates lose time. Parse into five sets of allowed values, then walk forward from the next minute, skipping a whole day the moment the date fields do not match:
code/lld/task_scheduler/schedules.py — cron
@dataclass(frozen=True, slots=True)
class CronSchedule:
"""A five-field cron expression evaluated in UTC, to the minute.
``*/15 * * * *`` is every quarter hour, ``0 3 * * 1-5`` is 03:00 on weekdays.
The search steps minute by minute but skips a whole day as soon as the date
fields do not match, so finding the next fire time never walks a year of
minutes. When both day-of-month and day-of-week are restricted, cron's
historical rule applies: either one matching is enough.
"""
expression: str
def __post_init__(self) -> None:
self._parsed() # fail at construction, not on the first tick
def first_run(self, now: float) -> float | None:
return self.next_after(now)
def next_run(self, record: ExecutionRecord, now: float) -> float | None:
return self.next_after(max(now, record.finished_at))
def next_after(self, timestamp: float) -> float:
minutes, hours, days, months, weekdays = self._parsed()
moment = datetime.fromtimestamp(timestamp, UTC).replace(second=0, microsecond=0)
moment += timedelta(minutes=1)
for _ in range(MAX_CRON_STEPS):
if not _date_matches(moment, days, months, weekdays):
moment = (moment + timedelta(days=1)).replace(hour=0, minute=0)
continue
if moment.hour in hours and moment.minute in minutes:
return moment.timestamp()
moment += timedelta(minutes=1)
raise ScheduleError(f"cron expression {self.expression!r} has no run time in range")
def _parsed(self) -> tuple[frozenset[int], ...]:
parts = self.expression.split()
if len(parts) != len(CRON_FIELDS):
raise ScheduleError(f"cron needs {len(CRON_FIELDS)} fields, got {len(parts)}: {self.expression!r}")
return tuple(
_parse_field(part, low, high, name)
for part, (low, high), name in zip(parts, CRON_RANGES, CRON_FIELDS, strict=True)
)
def _date_matches(
moment: datetime, days: frozenset[int], months: frozenset[int], weekdays: frozenset[int]
) -> bool:
if moment.month not in months:
return False
# cron weekdays are Sunday=0; Python's weekday() is Monday=0
day_ok = moment.day in days
weekday_ok = (moment.weekday() + 1) % 7 in weekdays
restricted_day = len(days) < 31
restricted_weekday = len(weekdays) < 7
if restricted_day and restricted_weekday:
return day_ok or weekday_ok
return day_ok and weekday_ok
def _parse_field(spec: str, low: int, high: int, name: str) -> frozenset[int]:
"""``*``, ``5``, ``1-5``, ``*/15`` and comma-separated lists of those."""
values: set[int] = set()
for part in spec.split(","):
body, _, step_text = part.partition("/")
try:
step = int(step_text) if step_text else 1
except ValueError as exc:
raise ScheduleError(f"cron {name} field has a bad step: {part!r}") from exc
if step < 1:
raise ScheduleError(f"cron {name} field step must be positive: {part!r}")
start, end = _bounds(body, low, high, name)
values.update(range(start, end + 1, step))
if not values:
raise ScheduleError(f"cron {name} field matches nothing: {spec!r}")
return frozenset(values)
def _bounds(body: str, low: int, high: int, name: str) -> tuple[int, int]:
if body in ("*", ""):
return low, high
first, sep, last = body.partition("-")
try:
start = int(first)
end = int(last) if sep else start
except ValueError as exc:
raise ScheduleError(f"cron {name} field is not a number: {body!r}") from exc
if not (low <= start <= high and low <= end <= high and start <= end):
raise ScheduleError(f"cron {name} field {body!r} is outside {low}-{high}")
return start, end
Now the heart of it. pop_due is twenty lines and it is what the interviewer is grading:
code/lld/task_scheduler/queues.py — the heap and the Condition
class TaskQueue:
"""A min-heap of due times behind one ``Condition``: the timer half of the scheduler.
``pop_due`` is the method the whole design turns on. It waits for exactly as
long as the earliest entry is not yet due - never a poll loop - and ``push``
notifies, so scheduling something *sooner* than the current head wakes the
timer immediately instead of leaving it parked for the old delay.
``_condition`` guards ``_heap`` and ``_closed``. Nothing else touches them.
"""
def __init__(self, clock: Clock) -> None:
self._clock = clock
self._condition = threading.Condition()
self._heap: list[QueueEntry] = []
self._closed = False
def push(self, entry: QueueEntry) -> None:
with self._condition:
heapq.heappush(self._heap, entry)
self._condition.notify_all() # an earlier due time must re-arm the timer
def pop_due(self) -> QueueEntry | None:
"""Block until the earliest entry is due. ``None`` means the queue is closed."""
with self._condition:
while not self._closed:
if not self._heap:
self._condition.wait() # nothing to time: sleep until a push
continue
delay = self._heap[0].due_at - self._clock.now()
if delay <= 0:
return heapq.heappop(self._heap)
self._condition.wait(timeout=delay)
return None
def wake(self) -> None:
"""Force a re-evaluation. Production never needs it; a test with a fake clock does."""
with self._condition:
self._condition.notify_all()
def close(self) -> None:
with self._condition:
self._closed = True
self._condition.notify_all()
def pending(self) -> list[QueueEntry]:
with self._condition:
return sorted(self._heap)
def __len__(self) -> int:
with self._condition:
return len(self._heap)
The pool is the consumer half. Note the bounded queue and the sentinel-per-worker shutdown:
code/lld/task_scheduler/queues.py — the worker pool
class WorkerPool:
"""Fixed threads draining a bounded queue - the consumer half of producer-consumer.
The bound is deliberate. If work arrives faster than it finishes, ``submit``
blocks the timer thread, and that backpressure is the honest behaviour: an
unbounded queue would absorb the backlog silently until memory ran out.
Shutdown pushes one sentinel per worker, so every thread leaves its blocking
``get`` without a flag anyone has to poll.
"""
def __init__(self, size: int = DEFAULT_WORKERS, name: str = "worker", queue_size: int = 0) -> None:
if size < 1:
raise ValidationError("worker pool needs at least one thread")
self._queue: queue.Queue[Callable[[], None] | None] = queue.Queue(maxsize=queue_size or size * 4)
self._threads = [
threading.Thread(target=self._run, name=f"{name}-{i}", daemon=True) for i in range(size)
]
self._started = False
self._closed = False
@property
def size(self) -> int:
return len(self._threads)
def start(self) -> None:
self._started = True
for thread in self._threads:
thread.start()
def submit(self, job: Callable[[], None]) -> None:
if self._closed:
raise SchedulerStateError("worker pool is shut down")
self._queue.put(job)
def shutdown(self, drain: bool = True, timeout: float = DEFAULT_SHUTDOWN_TIMEOUT) -> None:
"""``drain=True`` finishes what is queued. ``drain=False`` discards it first."""
self._closed = True
if not drain:
self._discard()
for _ in self._threads:
self._queue.put(None)
if self._started:
for thread in self._threads:
thread.join(timeout)
def _discard(self) -> None:
while True:
try:
self._queue.get_nowait()
except queue.Empty:
return
self._queue.task_done()
def _run(self) -> None:
while True:
job = self._queue.get()
try:
if job is None:
return
job()
finally:
self._queue.task_done()
The scheduler is then bookkeeping: one lock over the registry, and the three paths a finished run can take.
code/lld/task_scheduler/services.py — the scheduler
class Scheduler:
"""Registry plus timer thread plus worker pool.
``_lock`` guards ``_tasks`` and every field of every task in it - status,
attempt, generation, next run and history. The queue owns its own condition,
and the two are never held at the same time except in the one direction
``_lock`` then ``queue.push``, which is why there is no lock-order cycle.
"""
def __init__(
self,
workers: int = DEFAULT_WORKERS,
clock: Clock | None = None,
ids: IdGenerator | None = None,
store: TaskStore | None = None,
listeners: Iterable[TaskListener] = (),
) -> None:
self._clock = clock or SystemClock()
self._ids = ids or SequentialIdGenerator("task")
self._store = store or InMemoryTaskStore()
self._listeners = list(listeners)
self._lock = threading.Lock()
self._tasks: dict[str, ScheduledTask] = {}
self._dead_letter: list[str] = []
self._sequence = itertools.count()
self._queue = TaskQueue(self._clock)
self._pool = WorkerPool(workers, name="scheduler-worker")
self._timer = threading.Thread(target=self._timer_loop, name="scheduler-timer", daemon=True)
self._started = False
self._closed = False
# -- lifecycle ---------------------------------------------------------------
def start(self) -> None:
if self._started or self._closed:
raise SchedulerStateError("a scheduler starts once and never restarts after shutdown")
self._started = True
self._pool.start()
self._timer.start()
def shutdown(self, drain: bool = True, timeout: float = DEFAULT_SHUTDOWN_TIMEOUT) -> None:
"""Stop timing first, then let the workers finish (or drop) what is in flight.
Idempotent, so a ``finally`` block and a context manager can both call it.
"""
if self._closed:
return
self._closed = True
self._queue.close()
if self._timer.is_alive():
self._timer.join(timeout)
self._pool.shutdown(drain=drain, timeout=timeout)
def wake(self) -> None:
self._queue.wake()
# -- registration ------------------------------------------------------------
def schedule(
self,
name: str,
action: Callable[[], object],
schedule: Schedule,
priority: Priority = Priority.NORMAL,
retry: RetryPolicy = DEFAULT_RETRY,
timeout: float | None = None,
) -> ScheduledTask:
if self._closed:
raise SchedulerStateError("the scheduler is shut down and accepts no new tasks")
due = schedule.first_run(self._clock.now())
if due is None:
raise ScheduleError(f"schedule for {name!r} would never fire")
task = Task(self._ids.next_id(), name, action, priority=priority, retry=retry, timeout=timeout)
scheduled = ScheduledTask(task=task, schedule=schedule)
with self._lock:
self._tasks[task.id] = scheduled
self._enqueue(scheduled, due, TaskStatus.SCHEDULED)
return scheduled
def cancel(self, task_id: str) -> bool:
"""Tombstone the queued entry. A run already in flight finishes on its own."""
return self._transition(task_id, TaskStatus.CANCELLED, from_active=True)
def pause(self, task_id: str) -> bool:
return self._transition(task_id, TaskStatus.PAUSED, from_active=True)
def resume(self, task_id: str) -> bool:
with self._lock:
task = self._require(task_id)
if task.status is not TaskStatus.PAUSED:
return False
due = max(task.next_run_at or self._clock.now(), self._clock.now())
self._enqueue(task, due, TaskStatus.SCHEDULED)
return True
def task(self, task_id: str) -> ScheduledTask:
with self._lock:
return self._require(task_id)
def tasks(self) -> list[ScheduledTask]:
with self._lock:
return list(self._tasks.values())
def dead_letter(self) -> list[str]:
with self._lock:
return list(self._dead_letter)
def pending(self) -> int:
return len(self._queue)
# -- the loop ----------------------------------------------------------------
def _timer_loop(self) -> None:
while True:
entry = self._queue.pop_due()
if entry is None:
return
try:
self._pool.submit(partial(self._execute, entry))
except SchedulerStateError:
return
def _execute(self, entry: QueueEntry) -> None:
with self._lock:
task = self._tasks.get(entry.task_id)
if task is None or entry.generation != task.generation:
return # cancelled, paused or superseded: the entry is a tombstone
if task.status not in (TaskStatus.SCHEDULED, TaskStatus.RETRYING):
return
task.status = TaskStatus.RUNNING
task.attempt += 1
attempt, action = task.attempt, task.task.action
limit, retry = task.task.timeout, task.task.retry
record = self._run_once(attempt, action, limit)
self._settle(entry.task_id, record, retry)
def _run_once(self, attempt: int, action: Callable[[], object], limit: float | None) -> ExecutionRecord:
started = self._clock.now()
error: str | None = None
try:
action()
except Exception as exc: # a failing task must never kill its worker thread
error = f"{type(exc).__name__}: {exc}"
finished = self._clock.now()
# Python cannot preempt a thread, so a timeout is *detected*, not enforced.
# Real cancellation is cooperative: hand the task an Event it checks.
timed_out = limit is not None and finished - started > limit
if timed_out and error is None:
error = f"exceeded its {limit} s budget"
return ExecutionRecord(attempt, started, finished, ok=error is None, error=error, timed_out=timed_out)
def _settle(self, task_id: str, record: ExecutionRecord, retry: RetryPolicy) -> None:
with self._lock:
task = self._tasks.get(task_id)
if task is None:
return
task.history.append(record)
if task.status is not TaskStatus.RUNNING:
pass # cancelled or paused mid-run: keep the record, schedule nothing
elif record.ok:
task.attempt = 0
self._reschedule(task, record)
elif retry.should_retry(record.attempt):
self._enqueue(task, record.finished_at + retry.backoff(record.attempt), TaskStatus.RETRYING)
else:
task.status = TaskStatus.FAILED
task.next_run_at = None
self._dead_letter.append(task.id)
self._store.save(task)
event = TaskEvent(task.id, task.name, task.status, record)
self._notify(event)
def _reschedule(self, task: ScheduledTask, record: ExecutionRecord) -> None:
"""Caller holds the lock. A recurring task goes straight back into the heap."""
due = task.schedule.next_run(record, self._clock.now())
if due is None:
task.status = TaskStatus.SUCCEEDED
task.next_run_at = None
self._store.save(task)
return
self._enqueue(task, due, TaskStatus.SCHEDULED)
def _enqueue(self, task: ScheduledTask, due_at: float, status: TaskStatus) -> None:
"""Caller holds the lock. Bumping the generation retires any older entry."""
task.generation += 1
task.status = status
task.next_run_at = due_at
self._store.save(task)
self._queue.push(QueueEntry(due_at, task.task.priority, next(self._sequence), task.id, task.generation))
def _transition(self, task_id: str, status: TaskStatus, from_active: bool) -> bool:
with self._lock:
task = self._require(task_id)
if from_active and not task.is_active:
return False
task.generation += 1 # every queued entry for this task is now stale
task.status = status
self._store.save(task)
event = TaskEvent(task.id, task.name, status)
self._notify(event)
return True
def _require(self, task_id: str) -> ScheduledTask:
task = self._tasks.get(task_id)
if task is None:
raise TaskNotFoundError(f"unknown task {task_id}")
return task
def _notify(self, event: TaskEvent) -> None:
# Outside the lock: a slow listener must never block the scheduler.
for listener in self._listeners:
listener.on_task_event(event)
Running python -m lld.task_scheduler.demo shows the cron arithmetic, a recurring task, a retry sequence, the early wake-up, pause and resume, and shutdown:
cron '*/15 * * * *' from 09:07 UTC fires at 09:15, 09:30, 09:45
fixed rate 60 s after a run of 150 s: skip -> t+180 s, catch-up -> t+60 s
heartbeat every 30 s: 3 runs in the first minute
import: succeeded on attempt 3 after retries at +1 s and +2 s
cleanup: failed after 2 attempts, last error 'RuntimeError: disk is full'
dead letter: ['task-3']
timer parked 12 h out, then an insertion due now: urgent-alert ran without a nudge = True
digest paused through its slot: 0 runs, status paused
digest resumed: 1 run, next due in 3600 s
cancelled digest and heartbeat: heap holds 3 entries, 2 tombstoned
shutdown: 6 tasks registered, 2 succeeded, 1 dead
Concurrency and edge cases¶
Which lock protects what. Two, and they are acquired in one direction only.
TaskQueue._conditionguards the heap and the closed flag. It is aConditionrather than aLockbecause the timer has to wait for a time that can change:wait(delay)parks with no CPU cost, andpushnotifies, so a task scheduled sooner than the current head re-arms the timer instead of leaving it asleep until the old deadline.Scheduler._lockguards the task registry and every mutable field on a task: status, attempt, generation, next run and history. The race it prevents is two workers settling the same task at once and both re-queueing it.
The order is always Scheduler._lock then the queue's condition — _enqueue holds the registry lock and calls push, which never blocks. The timer thread does the reverse in time but never holds both, so there is no cycle to deadlock on.
Why not poll. The obvious first answer is a loop that sleeps a second and scans for due tasks. It costs 86,400 scans a day per scheduler, it adds up to a second of latency to every task, and it still gets the interesting case wrong: a task scheduled to run now while the loop is mid-sleep waits anyway. Condition.wait(delay) is both cheaper and more accurate, and an uncontended acquire around it costs about 17 ns.
Cancelling a queued entry. Removing an arbitrary element from a binary heap is O(n) plus a re-heapify, so you do not: cancel, pause and every re-queue bump the task's generation, and a popped entry whose generation no longer matches the task is dropped. The heap therefore holds tombstones — the demo ends with two of them beside one live entry — which is the accepted trade. If tombstones ever outgrew live entries you would rebuild the heap during a quiet tick, exactly as a log-structured store compacts.
Cancelling a running task. You cannot. Python has no safe way to stop a thread from outside, so cancel marks the task and _settle refuses to reschedule it; the in-flight run finishes. Timeouts are the same story: the record is marked timed_out after the fact, and the honest answer for real enforcement is cooperative cancellation — hand the task an Event it checks — or a subprocess you can actually kill. Say this rather than pretending a decorator can interrupt a thread.
Shutdown order. Stop timing first (queue.close() wakes the timer and it returns), then drain the pool. Doing it the other way round lets the timer submit into a closed pool. drain=False discards what is queued but still lets in-flight runs finish, because there is no way to stop them.
Edge cases handled: a failing task never kills its worker, because _run_once catches every exception and turns it into a record; a task whose schedule returns None retires instead of spinning; the attempt counter resets after a success so a recurring task retries afresh next time; resume runs a task that slept through its slot rather than waiting a full period; scheduling after shutdown raises instead of silently never running; shutdown is idempotent so a finally block can call it; and a cron expression is validated at construction, not on the first tick at 03:00.
Common mistake
Calling notify() on a push and thinking it is enough. It wakes one waiter — fine with a single timer thread, and quietly wrong the moment you add a second one or a shutdown waiter, because the notification can go to the thread that does not care. Use notify_all unless you can name every waiter, and always re-test the predicate in a while loop after waking: wait can return because of a notification, a timeout, or neither.
Extensibility and follow-ups¶
- Task dependencies (DAGs): give
Taska set of upstream ids and hold a task in aBLOCKEDstate until each has a successful record. The queue does not change; a readiness check beforesubmitdoes. That is the shape Airflow grew from. - Misfire policies: a scheduler that was down for an hour has to decide whether to run what it missed.
OverrunPolicyis the same question for one task; a misfire policy is it for the whole process at startup, and both belong on the schedule, not in the loop. - Persistence and recovery:
TaskStore.load_allalready exists as the seam. Recovery means rebuilding the heap from storednext_run_atvalues and deciding, per task, what to do about slots that passed while you were down. - Distribution: many schedulers, one set of jobs. That needs leases with fencing tokens (so a paused process cannot run a job someone else already took) or partitioned schedulers, plus idempotent tasks, because a lease can expire mid-run. It is the HLD case study.
- A time wheel instead of a heap: at hundreds of thousands of timers, a hierarchical timing wheel gives O(1) insertion against the heap's O(log n) — about 17 comparisons per push at 100k entries. It is the right answer for a network stack and the wrong one for a job scheduler, where the entry count is small and the arithmetic is not the bottleneck.
- Cron extensions: seconds, time zones and
@dailyaliases are all parsing changes insideCronSchedule. Time zones are the one that is genuinely hard, because a daily job at 02:30 does not exist on one day of the year and happens twice on another.
Interview tip
When you draw the queue, say the sentence "the timer waits on a condition with a timeout equal to the delay until the head is due" before you write a line. Then follow it with "and push notifies, so scheduling something sooner wakes it early." Most candidates write a poll loop and get corrected; saying the mechanism first tells the interviewer you have built one.
Tests¶
tests/test_task_scheduler.py has 24 cases. Two carry the design: the wake-up test and the fifty-tasks test.
code/lld/task_scheduler/tests/test_task_scheduler.py — the early wake-up
def test_an_earlier_insertion_wakes_the_parked_timer(clock: FakeClock, make_scheduler: Factory) -> None:
"""The reason the queue is a Condition and not a sleep.
The timer parks with a twelve-hour timeout. Pushing a task due *now* has to
interrupt that wait, or the urgent task waits half a day. Nothing here
advances the clock and nothing calls `wake()`: the notify inside `push` is
what re-arms the timer.
"""
scheduler, log = make_scheduler(workers=1)
scheduler.schedule("nightly", lambda: None, OneTime(clock.now() + 12 * 3600))
scheduler.start()
assert not log.wait_for(lambda _: log.runs("nightly") >= 1, NEVER)
scheduler.schedule("urgent", lambda: None, OneTime(clock.now()), priority=Priority.HIGH)
assert log.wait_for(lambda _: log.runs("urgent") >= 1, WAIT)
assert log.runs("nightly") == 0 # the twelve-hour task is still waiting its turn
It proves the mechanism rather than the behaviour: nothing advances the clock, so the only thing that can have released the timer from a twelve-hour wait is the notify inside push. The negative assertion uses a 50 ms bounded wait, which is how you test that something does not happen without sleeping.
code/lld/task_scheduler/tests/test_task_scheduler.py — fifty tasks, four workers
def test_fifty_tasks_due_at_once_each_run_exactly_once(clock: FakeClock, make_scheduler: Factory) -> None:
"""Four workers, fifty tasks, one instant: no task is skipped and none runs twice.
The invariant is what a broken heap or an unguarded status transition breaks:
a task popped twice would appear twice in `seen`, and one dropped by a lost
notify would never appear at all.
"""
scheduler, log = make_scheduler(workers=4)
seen: list[int] = []
for i in range(50):
scheduler.schedule(f"job-{i}", partial(seen.append, i), OneTime(clock.now()))
scheduler.start()
assert log.wait_for(lambda events: sum(1 for e in events if e.record) >= 50, 2.0)
assert sorted(seen) == list(range(50))
assert scheduler.pending() == 0
assert all(task.status is TaskStatus.SUCCEEDED for task in scheduler.tasks())
Every test waits on EventLog.wait_for, which is a Condition.wait_for under the hood — the worker notifies and the test wakes, so the suite finishes in a fraction of a second and never sleeps for a fixed interval.
The rest cover: a one-off task retiring; both overrun policies; fixed delay measured from the finish; five cron expressions including a weekday range and a month rollover; five malformed cron expressions; retry backoff of 1 s then 2 s ending in success; retries exhausted reaching the dead letter with the store recording the transition; cancel tombstoning a queued entry; pause holding a slot and resume running it; priority breaking a tie between three tasks due together; a task over its budget recorded as timed out; shutdown being final and idempotent; and the worker pool draining then refusing new work. Run them with uv run pytest code/lld/task_scheduler -q.
45-minute pacing¶
| Minutes | What to do | What to say or write |
|---|---|---|
| 0–5 | Clarify | One process or many? Rate or delay? What happens when a run overruns? Can a running task be cancelled? |
| 5–10 | Entities | Task, Schedule, ScheduledTask, QueueEntry on the board. Name the split: the task is what, the schedule is when. |
| 10–14 | The data structure | Min-heap keyed on due time, priority, sequence. Say why the sequence field exists before anyone asks. |
| 14–24 | The timer loop | Write pop_due. Say "wait with a timeout, push notifies" while writing it, then the worker pool behind it. |
| 24–32 | Lifecycle | Statuses, retry with backoff, dead letter, and the reschedule path for a recurring task. |
| 32–38 | Cancellation and shutdown | Tombstones with a generation counter, why you cannot stop a running thread, shutdown order. |
| 38–45 | Extensions | Cron parsing sketch, misfire policies, DAG dependencies, and the hand-off to a distributed scheduler with leases. |
Related¶
- Design a distributed job scheduler — the same problem once one process is not enough
- Command — the pattern behind
Task - Strategy — the four schedule rules behind one Protocol
- Observer — publishing status changes outside the lock
- Repository — the shape of the
TaskStoreseam - Concurrency for LLD in Python —
Condition, producer-consumer and the wait-in-a-while-loop rule - Primary sources: Varghese and Lauck, "Hashed and Hierarchical Timing Wheels" (SOSP 1987); the POSIX
crontab(5)specification