29. Peak Concurrent Voice Sessions

Problem

A capacity-planning job looks at a batch of streaming voice sessions, each with a start and end time, and needs the single number that matters most for provisioning: the maximum number of sessions that were ever open at the same instant. Given the sessions as (start, end) intervals, compute that peak concurrency.

Sessions are half-open intervals: a session ending at time t does not count as overlapping with one starting at exactly t — the outgoing session has already freed its slot.

Source: src/29_peak_concurrent_sessions.py

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

>>> peak_concurrent_sessions([(0, 5), (1, 3), (4, 7)])
2
>>> peak_concurrent_sessions([(0, 10), (0, 10), (0, 10)])
3

Step-by-Step Approach

1. Reframe each session interval as two independent events on a timeline: a "+1" event at its start time, and a "-1" event at its end time.
2. Collect all 2n events into one list and sort them by time.
3. Break ties carefully: when a session's end and another session's start land on the exact same timestamp, the end event must be processed first, since the interval is half-open — sorting by (time, delta) achieves this automatically because -1 sorts before +1.
4. Sweep through the sorted events left to right, maintaining a running concurrent counter: add the event's delta to it at each step.
5. Track the maximum value concurrent ever reaches during the sweep — that maximum is the answer.
6. An empty session list produces no events, so the running max stays at its initial value of 0.

The key insight is that peak concurrency only changes at a session's start or end — it's constant everywhere in between — so it's enough to evaluate the count at those O(n) event points rather than at every real-valued instant, turning a continuous problem into a discrete O(n log n) sort-and-sweep.

Reference solution

def peak_concurrent_sessions(sessions: list[tuple[float, float]]) -> int:
    # Turn each interval into a +1 (open) event at start and a -1 (close)
    # event at end, then sweep time left to right. Sorting by (time, delta)
    # puts -1 before +1 at an identical timestamp, which encodes the
    # half-open [start, end) semantics: a session closing at t frees its
    # slot before a session starting at t claims one.
    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

Key Functions & Tricks

• Event-pair encoding — every interval becomes two timeline events, turning interval overlap into a simple running-sum sweep.
• events.sort(key=lambda e: (e[0], e[1])) — sorting by (time, delta) gets the half-open tie-break for free since -1 < +1.
• Running concurrent counter — O(1) update per event instead of re-counting overlaps from scratch.
• Tracking a running peak alongside the sweep — avoids a second pass over the events.
• O(n log n) total (dominated by the sort) versus a naive O(n²) pairwise-overlap check.

How to Recognize This Pattern