DSA Task Scheduler

The problem

Task Scheduler arranges a multiset of tasks into time slots so that two runs of the same task sit at least n slots apart. The gap n is the cooldown. A slot with no task allowed to run stays idle.

The answer is the total slot count, idle slots included. Tasks are given as letters, and only the counts per letter matter, never the input order.

cases = [
    ("AAAABBBCC", 2, 10, 1),
    ("AAAAAABBBBBB", 2, 17, 5),
    ("ABCDEF", 2, 6, 0),
    ("AAAA", 0, 4, 0),
]
for tasks, n, slots, idles in cases:
    print("tasks", tasks, "cooldown", n, "-> slots:", slots, "idles:", idles)

Read one row at a time. The first row runs nine tasks in ten slots, so exactly one slot idles. The last row has a zero cooldown, so four runs of A pack into four slots.

Intuition: running tasks in input order wastes the gaps

The first idea runs the tasks in the order given, waiting whenever the next task is still cooling down. Waiting feels safe. The flaw is that other tasks could have filled those waiting slots.

The trap shows on six tasks. Input order runs A three times first, and B waits behind them even while A is cooling down. Taking a ready task in each slot instead cuts the schedule nearly in half.

tasks = "AAABBB"
n = 2
time = 0
last_run = {}
plan = []
for t in tasks:
    while t in last_run and time - last_run[t] <= n:
        plan.append("idle")
        time += 1
    plan.append(t)
    last_run[t] = time
    time += 1
print("input order plan:", " ".join(plan), "- slots:", len(plan))
print("interleaved plan: A B idle A B idle A B - slots: 8")

The input-order plan takes 14 slots for 6 tasks. B sits ready through the early idle slots and runs in none of them. A scheduler must pick the next task, not inherit one.

Brute force: simulate every slot, scanning every task kind

The brute force simulates the schedule one slot at a time. Each slot, it scans all task kinds and picks the ready kind with the largest remaining count. Ties break alphabetically, so the result is fixed.

A kind is ready when its cooldown has fully passed. A slot with no ready kind is idle. The counter records every kind examined, and the scan runs again on every slot.

def schedule_brute(tasks, n):
    counts = {}
    for t in tasks:
        counts[t] = counts.get(t, 0) + 1
    ready_at = {t: 1 for t in counts}
    time = 0
    idles = 0
    scanned = 0
    plan = []
    remaining = len(tasks)
    while remaining:
        time += 1
        best = None
        for t in sorted(counts):
            scanned += 1
            if counts[t] and ready_at[t] <= time:
                if best is None or counts[t] > counts[best]:
                    best = t
        if best is None:
            plan.append("idle")
            idles += 1
        else:
            plan.append(best)
            counts[best] -= 1
            ready_at[best] = time + n + 1
            remaining -= 1
    return time, idles, "".join(plan), scanned

total, idles, plan, scanned = schedule_brute("AAAABBBCC", 2)
print("plan:", plan)
print("slots:", total, "idles:", idles, "kinds scanned:", scanned)
total, idles, plan, scanned = schedule_brute("AAAAAABBBBBB", 2)
print("slots:", total, "idles:", idles, "kinds scanned:", scanned)

The plans are correct. The cost is the rescan: ten slots over three kinds scan 30 entries. A long schedule with many kinds pays that scan thousands of times.

The optimal solution: a max-heap of counts and a cooldown queue

The method in one sentence: always run a ready task with the largest remaining count, keep ready counts in a max-heap, and park each cooling task in a queue with the time it may run again. A max-heap is a heap whose top value is the largest one inside, and heapq is a min-heap, so counts are stored negated.

A cooldown queue is a first-in first-out line of resting tasks. Each entry carries a release time and the negated count left. When the clock reaches a release time, the entry moves back into the heap.

import heapq
from collections import deque

def schedule(tasks, n):
    counts = {}
    for t in tasks:
        counts[t] = counts.get(t, 0) + 1
    heap = [(-c, t) for t, c in counts.items()]
    heapq.heapify(heap)
    waiting = deque()
    time = 0
    idles = 0
    plan = []
    while heap or waiting:
        time += 1
        if waiting and waiting[0][0] == time:
            release, neg_left, t = waiting.popleft()
            heapq.heappush(heap, (neg_left, t))
        if heap:
            neg, t = heapq.heappop(heap)
            left = -neg - 1
            plan.append(t)
            if left > 0:
                waiting.append((time + n + 1, -left, t))
        else:
            plan.append(".")
            idles += 1
    return time, idles, "".join(plan)

cases = [
    ("AAAABBBCC", 2, 10, 1),
    ("AAAAAABBBBBB", 2, 17, 5),
    ("ABCDEF", 2, 6, 0),
    ("AAAA", 0, 4, 0),
]
for tasks, n, slots, idles in cases:
    total, idle_count, plan = schedule(tasks, n)
    ok = (total, idle_count) == (slots, idles)
    print("tasks", tasks, "cooldown", n, "-> slots:", total, "idles:", idle_count, "match:", ok, "plan:", plan)

All four cases pass, and the plans print with a dot for each idle slot. The first plan reads ABCABCAB, one idle dot, then A. The most frequent task claims the first slot of every cooldown round.

Walkthrough: AAAB with a cooldown of 2

The walk schedules three runs of A and one run of B. The state at each slot is the heap of ready counts and the queue of resting tasks. A release time of 4 means the task may run again at time 4.

import heapq
from collections import deque

counts = {"A": 3, "B": 1}
n = 2
heap = [(-c, t) for t, c in counts.items()]
heapq.heapify(heap)
waiting = deque()
time = 0
plan = []
while heap or waiting:
    time += 1
    if waiting and waiting[0][0] == time:
        release, neg_left, t = waiting.popleft()
        heapq.heappush(heap, (neg_left, t))
    if heap:
        neg, t = heapq.heappop(heap)
        left = -neg - 1
        plan.append(t)
        if left > 0:
            waiting.append((time + n + 1, -left, t))
        ran = t
    else:
        plan.append(".")
        ran = "idle"
    ready = sorted((-neg, t) for neg, t in heap)
    print("time", time, "- ran:", ran, "- ready:", ready, "- waiting:", list(waiting))
print("plan:", "".join(plan), "- slots:", time)

B runs once and leaves the system. After that, A cools down alone, and two slots idle. The total is read off the clock when the last A leaves the queue and runs.

Complexity

Let S be the total slot count and m the number of distinct task kinds. The brute force scans all m kinds on every slot, so its time is O(S m). The heap version pays one pop and at most one push per slot, each costing about log2 m, so its time is O(S log m).

Both versions store one count per kind plus the resting queue, so both use O(m) extra space. The plan itself is the output and is not counted as extra space.

import heapq
from collections import deque

def brute_scans(tasks, n):
    counts = {}
    for t in tasks:
        counts[t] = counts.get(t, 0) + 1
    ready_at = {t: 1 for t in counts}
    time = 0
    scanned = 0
    remaining = len(tasks)
    while remaining:
        time += 1
        best = None
        for t in sorted(counts):
            scanned += 1
            if counts[t] and ready_at[t] <= time:
                if best is None or counts[t] > counts[best]:
                    best = t
        if best is not None:
            counts[best] -= 1
            ready_at[best] = time + n + 1
            remaining -= 1
    return time, scanned

def heap_ops(tasks, n):
    counts = {}
    for t in tasks:
        counts[t] = counts.get(t, 0) + 1
    heap = [(-c, t) for t, c in counts.items()]
    heapq.heapify(heap)
    waiting = deque()
    time = 0
    ops = 0
    while heap or waiting:
        time += 1
        if waiting and waiting[0][0] == time:
            release, neg_left, t = waiting.popleft()
            heapq.heappush(heap, (neg_left, t))
            ops += 1
        if heap:
            neg, t = heapq.heappop(heap)
            ops += 1
            left = -neg - 1
            if left > 0:
                waiting.append((time + n + 1, -left, t))
    return time, ops

for m in (3, 6, 12):
    tasks = "".join(chr(ord("A") + i) * 3 for i in range(m))
    slots_b, scanned = brute_scans(tasks, 2)
    slots_h, ops = heap_ops(tasks, 2)
    print("kinds", m, "- slots:", slots_b, " brute scans:", scanned, " heap pushes+pops:", ops, " same slots:", slots_b == slots_h)

Counts beat clocks here. Doubling the kinds roughly quadruples the brute scans. The heap operations grow with the slot count alone, a few operations per slot.

Edge cases and pitfalls

The release time is the classic off-by-one trap. A task that runs at time t may run again at time t + n + 1, not t + n. The second run needs n full slots between the two runs.

A zero cooldown removes every gap. An empty task list needs zero slots. One kind alone can never fill its own gaps, so its schedule is mostly idle.

import heapq
from collections import deque

def schedule(tasks, n):
    counts = {}
    for t in tasks:
        counts[t] = counts.get(t, 0) + 1
    heap = [(-c, t) for t, c in counts.items()]
    heapq.heapify(heap)
    waiting = deque()
    time = 0
    idles = 0
    plan = []
    while heap or waiting:
        time += 1
        if waiting and waiting[0][0] == time:
            release, neg_left, t = waiting.popleft()
            heapq.heappush(heap, (neg_left, t))
        if heap:
            neg, t = heapq.heappop(heap)
            left = -neg - 1
            plan.append(t)
            if left > 0:
                waiting.append((time + n + 1, -left, t))
        else:
            plan.append(".")
            idles += 1
    return time, idles, "".join(plan)

for label, tasks, n in (("no tasks", "", 2), ("zero cooldown", "AAAA", 0), ("one kind alone", "AA", 2), ("all distinct", "ABCDEF", 2)):
    total, idles, plan = schedule(tasks, n)
    print(label, "- slots:", total, "idles:", idles, "plan:", plan or "(empty)")
total, idles, plan = schedule("AAA", 2)
print("A three times, cooldown 2:", plan, "- A runs at slots:", [i + 1 for i, t in enumerate(plan) if t == "A"])

The last row places A at slots 1, 4 and 7. Exactly two slots sit between each pair of runs. That spacing is the cooldown rule, visible directly in the plan.