34. Minimum GPU Workers to Cover Peak Load

Problem

Capacity planning for a real-time voice pipeline needs to answer a concrete provisioning question: given a batch of session intervals and the number of concurrent sessions a single GPU worker can serve, how many workers must be running to guarantee every session gets a worker at every point in time?

This reduces to finding the peak number of concurrent sessions, then dividing by each worker's capacity, rounding up — a partially-full worker still needs to exist.

Source: src/34_min_gpu_workers_for_peak.py

def min_gpu_workers(sessions: list[tuple[float, float]], capacity: int) -> int: ...

>>> min_gpu_workers([(0, 5), (1, 3), (4, 7)], capacity=1)
2
>>> min_gpu_workers([(0, 10), (0, 10), (0, 10), (0, 10), (0, 10)], capacity=2)
3

Step-by-Step Approach

1. Recognize that the number of workers is entirely determined by the single worst moment in time — the peak concurrent session count — not by the total number of sessions or their total duration.
2. Find that peak using the same sweep-line technique as counting peak concurrency directly: turn each session into a +1 event at its start and a -1 event at its end, sort all events by (time, delta) so closes are processed before opens at an identical timestamp, then sweep and track the running maximum.
3. Once the peak concurrent count is known, each GPU worker can absorb up to capacity of those simultaneous sessions.
4. The minimum number of workers needed is therefore ceil(peak / capacity) — rounding up matters because a worker serving even one session still needs to fully exist.
5. Handle the empty-input edge case explicitly: zero sessions means zero peak concurrency, which needs zero workers (not ceil(0/capacity) = 0 by coincidence of integer math, but worth confirming rather than assuming).
6. Compute the ceiling division without importing math.ceil: negate, floor-divide, negate back (-(-peak // capacity)) — a common integer-ceiling-division idiom that works because Python's // always floors toward negative infinity.

The key insight is that this is really two smaller, separately-recognizable problems chained together: a sweep-line peak-concurrency computation (see the peak-concurrent-sessions problem in this set) feeding into a simple capacity-division formula — you don't need to simulate assigning individual sessions to individual workers at all.

Reference solution

def _peak_concurrency(sessions: list[tuple[float, float]]) -> int:
    # same sweep-line technique as "peak concurrent sessions": +1 at each
    # start, -1 at each end, sorted so a close at time t is processed
    # before an open at the same t (half-open interval semantics)
    events: list[tuple[float, int]] = []
    for start, end in sessions:
        events.append((start, 1))
        events.append((end, -1))
    events.sort(key=lambda e: (e[0], e[1]))

    concurrent = 0
    peak = 0
    for _, delta in events:
        concurrent += delta
        peak = max(peak, concurrent)
    return peak


def min_gpu_workers(sessions: list[tuple[float, float]], capacity: int) -> int:
    peak = _peak_concurrency(sessions)
    if peak == 0:
        return 0
    # ceil(peak / capacity) without importing math.ceil: negate, floor-divide, negate back
    return -(-peak // capacity)

Key Functions & Tricks

• Sweep-line event encoding — identical to the peak-concurrency sweep, reused here as a building block rather than re-derived.
• -(-peak // capacity) — the negate/floor-divide/negate idiom for integer ceiling division, avoiding a math import and floating-point rounding.
• Explicit if peak == 0: return 0 guard — makes the empty-input behavior deliberate rather than an accident of how the ceiling-division idiom happens to behave at zero.
• Splitting the solution into a private helper (_peak_concurrency) plus a thin public function — keeps the capacity-planning formula separate from and easy to unit-test against the concurrency computation itself.
• O(n log n), dominated by the event sort; the capacity division afterward is O(1).

How to Recognize This Pattern