Skip to content

Task Scheduler

Problem

Given a list of tasks (characters) and a cooldown period n, find the minimum number of intervals needed to execute all tasks. The same task must wait at least n intervals before being executed again.

Approach

Use a max-heap (negated counts) to always schedule the most frequent task first. After executing a task, place it in a cooldown queue with the time it becomes available. When that time arrives, push it back onto the heap.

When to Use

Scheduling with cooldown constraints — "minimum time to complete all tasks with cooldown", "CPU scheduling", "rate-limited job execution". Max-heap ensures the most frequent task is scheduled first. Aviation: runway scheduling with minimum separation times.

Complexity

Time O(t) where t = total intervals (bounded by len(tasks) * (n+1))
Space O(1) — at most 26 distinct task types

Implementation

task_scheduler

Task scheduler with cooldown.

Problem

Given a list of tasks (characters) and a cooldown period n, find the minimum number of intervals needed to execute all tasks. The same task must wait at least n intervals before being executed again.

Approach

Use a max-heap (negated counts) to always schedule the most frequent task first. After executing a task, place it in a cooldown queue with the time it becomes available. When that time arrives, push it back onto the heap.

When to use

Scheduling with cooldown constraints — "minimum time to complete all tasks with cooldown", "CPU scheduling", "rate-limited job execution". Max-heap ensures the most frequent task is scheduled first. Aviation: runway scheduling with minimum separation times.

Complexity

Time: O(t) where t = total intervals (bounded by len(tasks) * (n+1)) Space: O(1) — at most 26 distinct task types

least_interval

least_interval(tasks: Sequence[str], n: int) -> int

Return the minimum number of intervals to complete all tasks.

least_interval(["A", "A", "A", "B", "B", "B"], 2) 8

Source code in src/algo/heaps/task_scheduler.py
def least_interval(tasks: Sequence[str], n: int) -> int:
    """Return the minimum number of intervals to complete all tasks.

    >>> least_interval(["A", "A", "A", "B", "B", "B"], 2)
    8
    """
    if not tasks:
        return 0

    counts = list(Counter(tasks).values())
    max_heap = [-c for c in counts]
    heapq.heapify(max_heap)

    time = 0
    cooldown: deque[tuple[int, int]] = deque()  # (available_time, neg_count)

    while max_heap or cooldown:
        time += 1

        if max_heap:
            cnt = heapq.heappop(max_heap) + 1  # execute one (cnt is negative)
            if cnt:
                cooldown.append((time + n, cnt))

        if cooldown and cooldown[0][0] == time:
            heapq.heappush(max_heap, cooldown.popleft()[1])

    return time
tests/heaps/test_task_scheduler.py
"""Tests for task scheduler."""

from algo.heaps.task_scheduler import least_interval


class TestLeastInterval:
    def test_basic(self) -> None:
        assert least_interval(["A", "A", "A", "B", "B", "B"], 2) == 8

    def test_no_cooldown(self) -> None:
        assert least_interval(["A", "A", "A", "B", "B", "B"], 0) == 6

    def test_single_task_type(self) -> None:
        assert least_interval(["A", "A", "A"], 2) == 7

    def test_all_unique(self) -> None:
        assert least_interval(["A", "B", "C", "D"], 2) == 4

    def test_empty_tasks(self) -> None:
        assert least_interval([], 5) == 0

    def test_one_task(self) -> None:
        assert least_interval(["A"], 100) == 1

    def test_cooldown_one(self) -> None:
        # A B A B A => 5 intervals (cooldown of 1 means 1 gap between same tasks)
        assert least_interval(["A", "A", "A", "B", "B"], 1) == 5

Implement it yourself

Run: just challenge heaps task_scheduler

Then implement the functions to make all tests pass. Use just study heaps for watch mode.

Reveal Solution

task_scheduler

Task scheduler with cooldown.

Problem

Given a list of tasks (characters) and a cooldown period n, find the minimum number of intervals needed to execute all tasks. The same task must wait at least n intervals before being executed again.

Approach

Use a max-heap (negated counts) to always schedule the most frequent task first. After executing a task, place it in a cooldown queue with the time it becomes available. When that time arrives, push it back onto the heap.

When to use

Scheduling with cooldown constraints — "minimum time to complete all tasks with cooldown", "CPU scheduling", "rate-limited job execution". Max-heap ensures the most frequent task is scheduled first. Aviation: runway scheduling with minimum separation times.

Complexity

Time: O(t) where t = total intervals (bounded by len(tasks) * (n+1)) Space: O(1) — at most 26 distinct task types

least_interval

least_interval(tasks: Sequence[str], n: int) -> int

Return the minimum number of intervals to complete all tasks.

least_interval(["A", "A", "A", "B", "B", "B"], 2) 8

Source code in src/algo/heaps/task_scheduler.py
def least_interval(tasks: Sequence[str], n: int) -> int:
    """Return the minimum number of intervals to complete all tasks.

    >>> least_interval(["A", "A", "A", "B", "B", "B"], 2)
    8
    """
    if not tasks:
        return 0

    counts = list(Counter(tasks).values())
    max_heap = [-c for c in counts]
    heapq.heapify(max_heap)

    time = 0
    cooldown: deque[tuple[int, int]] = deque()  # (available_time, neg_count)

    while max_heap or cooldown:
        time += 1

        if max_heap:
            cnt = heapq.heappop(max_heap) + 1  # execute one (cnt is negative)
            if cnt:
                cooldown.append((time + n, cnt))

        if cooldown and cooldown[0][0] == time:
            heapq.heappush(max_heap, cooldown.popleft()[1])

    return time