Seven build steps, one of them flaky. Deploy waits for package and lint, package waits for test, test fails twice and passes on the third try, and docs, which nobody depends on, runs in the gap while test backs off.
Build systems, data pipelines, CI runners and workflow engines are all this program underneath: a graph of tasks, a queue of what's ready, workers, and rules for failure. The round asks for it because it touches everything at once: graph algorithms (cycles, topological order), heaps (for priority and for time), concurrency (a worker pool, one condition variable) and state machines.
This lesson builds it in one process: Kahn's algorithm at submit time, counts that release tasks as their dependencies succeed, a ready heap and a delayed heap, retries, cascading cancellation, and fixed-rate recurring tasks.
How the round runs#
"Build a task scheduler. A task is a function plus the ids of tasks it depends on. Run tasks in parallel on a fixed number of workers, never before their dependencies succeed. Tasks have priorities. Failed tasks should retry with backoff. I need to cancel tasks, schedule a task for later, and run some tasks periodically. A distributed version is out of scope; one process, real threads."
- 0–10 min — clarify: what a failed dependency means, retry rules, cancelling a running task, recurring semantics
- 10–20 min — task states, then
Task,RetryPolicyand the scheduler's structures - 20–40 min —
submit: validation, Kahn's algorithm, counting unmet dependencies - 40–65 min — workers: the ready heap, the delayed heap, the condition variable, finishing a task
- 65–80 min — retries, cancellation, recurring tasks, then the demo
- 80–90 min — follow-ups: persistence, many machines, cron expressions, resource limits
Requirements and clarifying questions#
| You ask | They answer | What it changes |
|---|---|---|
| What if a task depends on itself, indirectly? | Reject the whole batch. | Cycle detection before anything is stored |
| If a dependency fails for good, what happens downstream? | Don't run it; mark it skipped. | A SKIPPED state that cascades |
| Priority among ready tasks? | Lower number first; ties by submission order. | A heap of (priority, seq, id) |
| Retry rules? | Per task: up to N attempts, delay doubling from a base, capped. | RetryPolicy; a failed attempt goes to the delayed heap |
| Cancel a running task? | Ask it to stop; we can't kill a thread. | A cancel flag passed to every task |
| Delayed and periodic tasks? | "Not before T", and "every N seconds, K times". | A delayed heap, and fixed-rate rescheduling |
| Can new tasks depend on old ones? | Yes. Old ones can't depend on new ones. | A cycle can only be inside the new batch |
Why count unmet dependencies instead of checking them?
A worker finishes build. How do you find which tasks can now start, without scanning every task and every dependency?
One strong answer
Keep two maps. children: for each task, the tasks that depend on it. unmet: for each waiting task, how many of its dependencies haven't succeeded yet. When build succeeds, walk only its children and decrement their counts; any child that reaches 0 goes into the ready heap. That's Kahn's topological sort, run live: each edge is touched once when its source succeeds, so releasing tasks costs O(out-degree), not O(all tasks). The same algorithm, run once on a copy at submit time, detects cycles: if it can't consume every node, some are stuck in a loop.
%% title: Task scheduler: a task's states stateDiagram-v2 [*] --> Pending Pending --> Ready: dependencies met, start time reached Ready --> Running: a worker picks it Running --> Succeeded: returned Running --> Pending: failed, retries left / recurring, more runs Running --> Failed: failed, no retries left Pending --> Cancelled: cancel Ready --> Cancelled: cancel Running --> Cancelled: cancel flag set, task returned Pending --> Skipped: a dependency failed or was cancelled Succeeded --> [*] Failed --> [*] Cancelled --> [*] Skipped --> [*]
The classes#
%% title: Task scheduler: the classes
classDiagram
class Task {
+task_id str
+fn callable
+deps frozenset
+priority int
+retry RetryPolicy
+delay float
+every float
+times int
+status Status
+attempts int
}
class RetryPolicy {
+max_attempts int
+base_delay float
+max_delay float
+delay(failures) float
}
class Status {
<<enumeration>>
PENDING
READY
RUNNING
SUCCEEDED
FAILED
CANCELLED
SKIPPED
}
class Scheduler {
-children dict
-unmet dict
-ready heap
-delayed heap
-cv Condition
+submit(tasks)
+cancel(task_id) bool
+wait(timeout) bool
+shutdown()
}
Task --> RetryPolicy
Task --> Status
Scheduler "1" o-- "many" Task
Scheduler ..> CycleErrorEverything the scheduler knows lives behind one condition variable. The state changes are tiny and touch several structures at once, so one lock is simpler and correct; the expensive part, running the task, happens outside it.
| Structure | Holds | Cost |
|---|---|---|
children: dict[id, list[id]] |
the graph, as out-edges | O(out-degree) to release children |
unmet: dict[id, int] |
dependencies not yet succeeded | O(1) per decrement |
ready heap (priority, seq, id) |
tasks that can run now | O(log n) push and pop |
delayed heap (run_at, seq, id) |
tasks waiting for a time: delayed starts, retries, recurring runs | O(log n) push and pop |
The interfaces#
# scheduler.py, part 1: task states, the retry policy, and the task
from __future__ import annotations
import heapq
import itertools
import threading
import time
from collections.abc import Callable
from dataclasses import dataclass
from enum import Enum
class Status(Enum):
PENDING = "pending" # waiting for dependencies or for its start time
READY = "ready" # in the ready heap
RUNNING = "running"
SUCCEEDED = "succeeded"
FAILED = "failed" # out of retries
CANCELLED = "cancelled"
SKIPPED = "skipped" # a dependency failed or was cancelled
DONE = frozenset({Status.SUCCEEDED, Status.FAILED, Status.CANCELLED, Status.SKIPPED})
@dataclass(frozen=True)
class RetryPolicy:
max_attempts: int = 1
base_delay: float = 0.05 # seconds before the first retry; doubles after each failure
max_delay: float = 30.0
def delay(self, failures: int) -> float:
return min(self.max_delay, self.base_delay * 2 ** (failures - 1))
@dataclass
class Task:
task_id: str
fn: Callable[[threading.Event], object] # receives a cancel flag it may check
deps: frozenset[str] = frozenset()
priority: int = 0 # lower runs first among ready tasks
retry: RetryPolicy = RetryPolicy()
delay: float = 0.0 # don't start before submit time + delay
every: float | None = None # recurring: next run planned this long after the last planned start
times: int = 1 # total runs, for a recurring task
status: Status = Status.PENDING
attempts: int = 0 # attempts at the current run
runs: int = 0
run_at: float = 0.0 # set by the scheduler
error: str | None = None
def __post_init__(self) -> None:
self.deps = frozenset(self.deps)The patterns, and why#
| Pattern | Where | Why it earns its place |
|---|---|---|
| Topological sort (Kahn) | _check_acyclic, and unmet at run time |
One algorithm both rejects cycles up front and decides, live, what can start next. |
| Producer–consumer with a condition variable | _cv, _next, _work |
Workers sleep until there's work or the next delayed task is due; nothing polls. |
| Strategy (as a value) | RetryPolicy per task |
A flaky network call and a deterministic bug need different retry rules. |
| Cooperative cancellation | the threading.Event passed to fn |
Python can't kill a thread safely. The task checks the flag at safe points and returns early. |
| Lazy deletion | cancelled tasks left in the heaps | Cancel is O(1) plus the cascade; the stale heap entry is dropped when popped. |
Concurrency and edge cases#
Three moments need care. Releasing children: two workers finish package's two dependencies at once. Without the lock, both could read unmet as 2 and write 1, and package never starts. Every state change is under _cv. Sleeping: an idle worker waits on the condition with a timeout equal to the time until the next delayed task; submit, finish and cancel call notify_all, so it wakes early for new work. Running: fn is called outside the lock, so four workers run four tasks at once.
- A dependency that already failed when the child is submitted: skip the child at once instead of waiting forever
- Cancel a running task: set its flag; it ends as
CANCELLEDwhen it returns, even if it succeeded, and its children are skipped - Cancel a task twice, or a finished one: return
False - A recurring run that takes longer than its interval: the next run is already due and starts at once; a "skip missed runs" policy is a follow-up
- A task that hangs forever: a worker is lost; add a timeout that marks it failed (the thread can't be killed, only abandoned)
- Deep chains: skipping cascades recursively, so a chain thousands long needs an explicit stack instead of recursion
The code#
Submitting a batch: validate it, reject cycles, and release whatever is ready.
# part 2: submitting a DAG: validate, reject cycles with Kahn's algorithm, release what's ready
class CycleError(ValueError): ...
class Scheduler:
def __init__(self, workers: int = 4, clock: Callable[[], float] = time.monotonic) -> None:
self.tasks: dict[str, Task] = {}
self.log: list[str] = [] # what happened, in order
self._children: dict[str, list[str]] = {}
self._unmet: dict[str, int] = {} # task -> dependencies not yet succeeded
self._ready: list[tuple[int, int, str]] = [] # heap of (priority, seq, id)
self._delayed: list[tuple[float, int, str]] = [] # heap of (run_at, seq, id)
self._cancel: dict[str, threading.Event] = {}
self._open = 0 # tasks not yet in a final state
self._seq = itertools.count()
self._cv = threading.Condition() # one lock for all scheduler state; idle workers sleep on it
self._clock = clock
self._stopping = False
self._threads = [threading.Thread(target=self._work, daemon=True) for _ in range(workers)]
for t in self._threads:
t.start()
def submit(self, *tasks: Task) -> None:
with self._cv:
batch = {t.task_id: t for t in tasks}
if len(batch) < len(tasks) or batch.keys() & self.tasks.keys():
raise ValueError("duplicate task id")
for t in tasks:
if unknown := t.deps - batch.keys() - self.tasks.keys():
raise ValueError(f"{t.task_id} depends on unknown {sorted(unknown)}")
self._check_acyclic(batch) # old tasks can't depend on new ones, so a cycle must be in the batch
now = self._clock()
for t in tasks:
self.tasks[t.task_id], self._cancel[t.task_id] = t, threading.Event()
t.run_at = now + t.delay
self._open += 1
for d in t.deps:
self._children.setdefault(d, []).append(t.task_id)
for t in tasks:
if t.status is not Status.PENDING:
continue # already skipped by an earlier task in this loop
deps = [self.tasks[d] for d in t.deps]
if any(d.status in DONE and d.status is not Status.SUCCEEDED for d in deps):
self._settle(t, Status.SKIPPED)
continue
self._unmet[t.task_id] = sum(d.status is not Status.SUCCEEDED for d in deps)
if self._unmet[t.task_id] == 0:
self._schedule(t)
@staticmethod
def _check_acyclic(batch: dict[str, Task]) -> None:
"""Kahn's algorithm over the batch: O(V + E). Whatever it can't consume is in, or behind, a cycle."""
indeg = {tid: len(t.deps & batch.keys()) for tid, t in batch.items()}
out: dict[str, list[str]] = {tid: [] for tid in batch}
for tid, t in batch.items():
for d in t.deps & batch.keys():
out[d].append(tid)
stack = [tid for tid, n in indeg.items() if n == 0]
consumed = 0
while stack:
consumed += 1
for child in out[stack.pop()]:
indeg[child] -= 1
if indeg[child] == 0:
stack.append(child)
if consumed < len(batch):
raise CycleError(f"a cycle leaves {sorted(t for t, n in indeg.items() if n)} unable to start")
def _schedule(self, t: Task) -> None:
if t.run_at > self._clock():
heapq.heappush(self._delayed, (t.run_at, next(self._seq), t.task_id)) # stays PENDING until due
else:
t.status = Status.READY
heapq.heappush(self._ready, (t.priority, next(self._seq), t.task_id))
self._cv.notify_all()The workers, and what happens when a task ends:
# part 3, still inside Scheduler: workers, finishing, retries, recurring runs, cancellation
def _next(self) -> Task | None:
while not self._stopping:
now = self._clock()
while self._delayed and self._delayed[0][0] <= now: # move what's due to the ready heap
t = self.tasks[heapq.heappop(self._delayed)[2]]
if t.status is Status.PENDING: # a cancelled task's entry is just dropped
t.status = Status.READY
heapq.heappush(self._ready, (t.priority, next(self._seq), t.task_id))
while self._ready:
t = self.tasks[heapq.heappop(self._ready)[2]]
if t.status is Status.READY:
return t
self._cv.wait(self._delayed[0][0] - now if self._delayed else None) # until the next due time, or a notify
return None
def _work(self) -> None:
while True:
with self._cv:
t = self._next()
if t is None:
return
t.status, t.attempts = Status.RUNNING, t.attempts + 1
cancel = self._cancel[t.task_id]
try:
t.fn(cancel) # outside the lock: tasks run in parallel
error = None
except Exception as e:
error = f"{type(e).__name__}: {e}"
with self._cv:
self._after_run(t, error)
def _after_run(self, t: Task, error: str | None) -> None:
if self._cancel[t.task_id].is_set():
self._settle(t, Status.CANCELLED)
elif error is None and t.every is not None and t.runs + 1 < t.times:
t.runs, t.attempts, t.status = t.runs + 1, 0, Status.PENDING
t.run_at += t.every # fixed rate: planned from the last planned start, so runs don't drift
self.log.append(f"{t.task_id} run {t.runs}")
self._schedule(t)
elif error is None:
t.runs += 1
self._settle(t, Status.SUCCEEDED)
elif t.attempts < t.retry.max_attempts:
wait = t.retry.delay(t.attempts)
t.run_at, t.status, t.error = self._clock() + wait, Status.PENDING, error
self.log.append(f"{t.task_id} retry in {wait:.2f}s ({error})")
self._schedule(t)
else:
t.error = error
self._settle(t, Status.FAILED, f" ({error})")
def _settle(self, t: Task, status: Status, note: str = "") -> None:
t.status = status
self._open -= 1
self.log.append(f"{t.task_id} {status.value}{note}")
for c in self._children.get(t.task_id, []):
child = self.tasks[c]
if child.status is not Status.PENDING:
continue
if status is Status.SUCCEEDED:
self._unmet[c] -= 1
if self._unmet[c] == 0:
self._schedule(child)
else:
self._settle(child, Status.SKIPPED) # cascades down the graph
self._cv.notify_all()
def cancel(self, task_id: str) -> bool:
with self._cv:
t = self.tasks[task_id]
if t.status in DONE:
return False
self._cancel[task_id].set() # a running task may check it and return early
if t.status is not Status.RUNNING:
self._settle(t, Status.CANCELLED) # any heap entry it has is skipped when popped
return True
def wait(self, timeout: float = 10.0) -> bool:
with self._cv:
return self._cv.wait_for(lambda: self._open == 0, timeout)
def shutdown(self) -> None:
with self._cv:
self._stopping = True
self._cv.notify_all()
for t in self._threads:
t.join()_settle is the only way into a final state, so the open-task counter, the log and the release of children can't disagree. A success decrements each waiting child's count; any other ending skips each waiting child, which skips its own children in turn.
The demo: a pipeline on one worker, so the order is visible; a cycle; a failure and a cancel; a recurring task; then a 40-task graph on four workers.
# part 4: a CI pipeline with a flaky test, a cycle, a cascade, a recurring task, and a parallel graph
if __name__ == "__main__":
def ok(cancel: threading.Event) -> None:
pass
calls = itertools.count(1)
def flaky(cancel: threading.Event) -> None:
if next(calls) < 3:
raise ConnectionError("test db not up yet")
s = Scheduler(workers=1) # one worker, so the order below is the scheduler's choice alone
s.submit(
Task("checkout", ok),
Task("build", ok, deps={"checkout"}),
Task("lint", ok, deps={"checkout"}, priority=5),
Task("docs", ok, deps={"checkout"}, priority=9),
Task("test", flaky, deps={"build"}, retry=RetryPolicy(max_attempts=3, base_delay=0.05)),
Task("package", ok, deps={"build", "test"}),
Task("deploy", ok, deps={"package", "lint"}),
)
s.wait()
print(*s.log, sep="\n")
try:
s.submit(Task("a", ok, deps={"c"}), Task("b", ok, deps={"a"}), Task("c", ok, deps={"b"}), Task("d", ok, deps={"c"}))
except CycleError as e:
print("rejected:", e)
s.log.clear()
s.submit(Task("report", ok, delay=60.0), Task("email", ok, deps={"report"}))
print("cancel report:", s.cancel("report"), "| again:", s.cancel("report"))
def broken(cancel: threading.Event) -> None:
raise RuntimeError("disk full")
s.submit(Task("migrate", broken, retry=RetryPolicy(max_attempts=2)), Task("notify", ok, deps={"migrate"}))
s.wait()
s.submit(Task("heartbeat", ok, every=0.05, times=3))
s.wait()
print(*s.log, sep="\n")
s.shutdown()
par = Scheduler(workers=4)
stamps: dict[str, tuple[int, int]] = {}
tick, tick_lock = itertools.count(), threading.Lock()
def job(name: str) -> Callable[[threading.Event], None]:
def run(cancel: threading.Event) -> None:
with tick_lock:
start = next(tick)
time.sleep(0.005)
with tick_lock:
stamps[name] = (start, next(tick))
return run
layers = [[f"L{i}T{j}" for j in range(8)] for i in range(5)]
graph = [Task(n, job(n), deps={layers[i - 1][j], layers[i - 1][(j + 1) % 8]} if i else set()) for i, layer in enumerate(layers) for j, n in enumerate(layer)]
par.submit(*graph)
par.wait()
respected = all(stamps[t.task_id][0] > stamps[d][1] for t in graph for d in t.deps)
print(len(graph), "tasks on 4 workers: all succeeded =", all(t.status is Status.SUCCEEDED for t in graph), "| dependencies respected =", respected)
par.shutdown()checkout succeeded
build succeeded
test retry in 0.05s (ConnectionError: test db not up yet)
lint succeeded
docs succeeded
test retry in 0.10s (ConnectionError: test db not up yet)
test succeeded
package succeeded
deploy succeeded
rejected: a cycle leaves ['a', 'b', 'c', 'd'] unable to start
cancel report: True | again: False
report cancelled
email skipped
migrate retry in 0.05s (RuntimeError: disk full)
migrate failed (RuntimeError: disk full)
notify skipped
heartbeat run 1
heartbeat run 2
heartbeat succeeded
40 tasks on 4 workers: all succeeded = True | dependencies respected = TrueAfter checkout, build (priority 0) goes first. test beats lint (5) and docs (9), fails, and backs off for 0.05 s; the single worker uses the gap for lint and docs. The second wait is 0.10 s; the third attempt passes. The cycle is rejected before anything is stored, with d, which sits behind it. Cancelling report skips email; the second cancel returns False. migrate fails twice and notify never runs. The heartbeat runs three times. On four workers, every one of the 40 tasks starts after both its dependencies finished.
Extending it: the follow-ups#
- "Survive a restart." Persist tasks and states; on startup treat
RUNNINGas a failed attempt and rebuildunmetand the heaps. Tasks must then be idempotent, because one can run twice. - "Run on many machines." Workers pull ready tasks with a lease (claim until a deadline, renew while running; an expired lease lets another worker take it).
unmetmoves to the database, decremented in the transaction that marks a task succeeded. - "Real cron expressions." Replace
everywith aSchedulestrategy whosenext_after(t)returns the next fire time. The delayed heap doesn't change. - "Jitter the retries." Tasks failing on the same dependency retry in lockstep; random jitter spreads them out.
- "Some tasks need a GPU." Tasks declare resources;
_nextskips ready tasks whose resources are busy. Watch for big tasks starving behind small ones.
Common mistakes#
- Checking every task's dependencies in a loop after each finish, O(V × E) instead of O(out-degree)
- No cycle check, so a cyclic batch sits in
PENDINGforever andwait()never returns - Running
fnwhile holding the scheduler lock, so four workers run one task at a time - Sleeping in a loop (
time.sleep(0.1)and check again) instead of a condition with a timeout - Retrying immediately, in a tight loop, against a dependency that's down
- Leaving children of a failed task
PENDINGinstead of skipping them, so the run never finishes
Task scheduler
Q1/3 Tasks: a, b depends on a, c depends on b, a depends on c, and d depends on c. What does Kahn's algorithm report?
Answer & explanation
Answer: C
Kahn's algorithm starts from tasks with no unmet dependencies. None exist here, so nothing is consumed: the cycle and everything behind it are reported as unable to start.
Q2/3 test fails with base_delay 0.05 s and max_attempts 3. What are the waits before attempts 2 and 3?
Answer & explanation
Answer: B
delay(failures) = base × 2^(failures − 1): 0.05 after the first failure, 0.10 after the second, capped by max_delay.
Q3/3 A worker is about to call the task's function. Why release the scheduler lock first?
Answer & explanation
Answer: D
The lock guards bookkeeping that takes microseconds. Holding it during a task that takes seconds turns a pool of four workers into a pool of one, and blocks submit and cancel too.
New lessons by email
New lessons, side quests and case files as they ship, at most once a week. No spam, unsubscribe anytime.
Discussion · …
Public: everyone reading this lesson sees comments. Be kind and keep it on the material.