Design a lazy tee: one iterator into n independent ones

01The design, in the six-step framework

1. Requirements and use cases

Reported in 2026 phone screens and as a less common L5 coding problem: split one iterator into n independent ones, reading the source only when somebody needs a new element and forgetting elements once everybody has seen them.

• A caller splits one source iterator into n consumer iterators.
• Each consumer sees every element once, in source order, at its own pace.
• The source is read lazily, and each source element is read exactly once however many consumers want it.
• Elements already read by all consumers are released from memory.
• Follow-up: consumers run on different threads.
• Follow-up: a slow consumer may be detached so that it stops pinning the buffer.

Out of scope: rewinding, adding consumers after the split, and a bounded buffer that blocks (discussed below). Non-functional constraints: amortised O(1) per element per consumer; memory proportional to the gap between the fastest and slowest consumer, not to the stream length; and no loss or duplication under concurrent reads.

2. Core entities and relationships

• TeeBuffer: the shared core; owns the source, the buffered elements, the position of each consumer and the lock.
• TeeConsumer: a thin iterator with an id; every call goes to the buffer.
• tee(): a small factory function returning the n consumers.

The buffer composes the deque and the positions. Consumers aggregate the shared buffer, and they are deliberately stateless apart from their id, so the rule "who has read what" lives in one place. An element's absolute index is base plus its offset in the deque, where base counts everything already released. Absolute positions never change, which makes each consumer's position a single integer that needs no adjusting when the deque shrinks.

3. Interfaces and key methods

from typing import Generic, Iterator, List, TypeVar

T = TypeVar("T")


class TeeBuffer(Generic[T]):
    def read(self, cid: int) -> T:
        """Next element for this consumer; pulls the source only at the frontier."""

    def detach(self, cid: int) -> None:
        """Forget a consumer so it no longer holds the buffer back."""


class TeeConsumer(Generic[T]):
    def __next__(self) -> T: ...
    def close(self) -> None:
        """Detach; the consumer raises StopIteration from then on."""


def tee(source: Iterator[T], n: int) -> List[TeeConsumer[T]]:
    """Exactly n independent consumers over one source; n >= 1."""

Invariants: base is at most the smallest live position; every index from base up to base + len(items) is in the deque; a consumer's position never decreases; and the source is only advanced while holding the lock, by the consumer whose position equals the frontier.

4. Design patterns and the choices behind them

One shared deque rather than a queue per consumer

The force is memory with many consumers. itertools.tee and many answers give each consumer its own queue, which is simple: the leader pushes each element into every other queue. That stores n references per element and costs O(n) per read at the frontier. One shared deque with integer positions stores each element once. Its cost is working out how far to release, which is a minimum over positions.

Absolute positions and the release rule

The force is knowing when an element can go. After a read, the buffer checks whether the reader was the only one sitting at base. Releasing while base is below the smallest position keeps the invariant. The naive minimum is O(n) per read; a count of consumers per position makes it O(1), which the code does with a dictionary so a read touches two counters. Mention this as the point where a design for 3 consumers and a design for 3,000 differ, a judgement covered under trade-offs and over-engineering.

A lock around the frontier

The force is two threads reaching the frontier together and both pulling the source, which loses an element for one of them. A thread-safe collection does not help, because read-then-pull-then-append is a sequence. One lock around read makes it atomic. The source may be slow, and holding the lock during the pull blocks other consumers who only need buffered data; the rejected refinement releases the lock during the pull, and it needs a "fetch in progress" state and a condition variable, which is more machinery than an hour allows. Name it; do not build it. This is an iterator with a producer-consumer problem inside it.

5. Core flow and code

Flow for two consumers, A and B, over [10, 20]:

1. A reads at position 0, the frontier: the source yields 10, it joins the deque, A moves to 1.
2. A reads at position 1, also the frontier: 20 joins, A moves to 2. The deque holds both because B is at 0.
3. B reads at 0 and gets 10. No consumer remains at 0, so base moves to 1 and 10 is released.
4. B reads 20 and it is released in the same way. The deque is empty, and the next read pulls the source and raises StopIteration.

import threading
from collections import Counter, deque
from typing import Deque, Dict, Generic, Iterator, List, TypeVar

T = TypeVar("T")


class TeeBuffer(Generic[T]):
    def __init__(self, source: Iterator[T], n: int) -> None:
        self.source = source
        self.items: Deque[T] = deque()
        self.base = 0                                  # absolute index of items[0]
        self.pos: Dict[int, int] = {cid: 0 for cid in range(n)}
        self.at: Counter = Counter({0: n})             # how many consumers sit at each position
        self.lock = threading.Lock()
        self.done = False

    def read(self, cid: int) -> T:
        with self.lock:
            if cid not in self.pos:
                raise StopIteration
            p = self.pos[cid]
            offset = p - self.base
            if offset == len(self.items):              # at the frontier
                if self.done:
                    raise StopIteration
                try:
                    self.items.append(next(self.source))
                except StopIteration:
                    self.done = True
                    raise
            value = self.items[offset]
            self._move(p, p + 1)
            self.pos[cid] = p + 1
            self._release()
            return value

    def detach(self, cid: int) -> None:
        with self.lock:
            p = self.pos.pop(cid, None)
            if p is not None:
                self.at[p] -= 1
                if self.at[p] == 0:
                    del self.at[p]
                self._release()

    def _move(self, old: int, new: int) -> None:
        self.at[old] -= 1
        if self.at[old] == 0:
            del self.at[old]
        self.at[new] += 1

    def _release(self) -> None:
        """Drop elements below the slowest live consumer; amortised O(1) per element."""
        while self.items and self.base not in self.at:
            self.items.popleft()
            self.base += 1


class TeeConsumer(Generic[T]):
    def __init__(self, buffer: TeeBuffer[T], cid: int) -> None:
        self.buffer, self.cid = buffer, cid

    def __next__(self) -> T:
        return self.buffer.read(self.cid)

    def __iter__(self) -> "TeeConsumer[T]":
        return self

    def close(self) -> None:
        self.buffer.detach(self.cid)


def tee(source: Iterator[T], n: int) -> List[TeeConsumer[T]]:
    if n < 1:
        raise ValueError("n must be at least 1")
    buffer: TeeBuffer[T] = TeeBuffer(source, n)
    return [TeeConsumer(buffer, i) for i in range(n)]


if __name__ == "__main__":
    pulled: List[int] = []

    def counted() -> Iterator[int]:
        for i in range(5):
            pulled.append(i)
            yield i

    a, b = tee(counted(), 2)
    assert next(a) == 0 and next(a) == 1 and pulled == [0, 1]   # lazy
    assert len(a.buffer.items) == 2                             # b still needs both
    assert next(b) == 0 and len(a.buffer.items) == 1            # 0 released
    b.close()
    assert len(a.buffer.items) == 0                             # detached consumer stops pinning
    assert list(a) == [2, 3, 4] and pulled == [0, 1, 2, 3, 4]

    results: Dict[int, List[int]] = {0: [], 1: [], 2: []}
    consumers = tee(iter(range(2000)), 3)

    def drain(i: int) -> None:
        results[i].extend(consumers[i])

    threads = [threading.Thread(target=drain, args=(i,)) for i in range(3)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    assert all(r == list(range(2000)) for r in results.values())

6. Extensions, concurrency and what interviewers probe

• The memory answer. Unbounded in the worst case: if one consumer never reads, the buffer holds the whole stream. Say that plainly, then offer close and a bounded version.
• Bounded buffer. A maximum gap means the fastest consumer blocks on a condition until the slowest moves, which is backpressure; it deadlocks if one thread drains consumers in sequence, so make it opt-in. See bounded queues.
• Exceptions from the source. A source that raises should raise for every consumer that reaches that point; store the exception as a final element rather than letting one consumer swallow it. The code handles normal end only; say so.
• Is a lock needed at all? In a single-threaded program, no. Offer to remove it, and note that itertools.tee is documented as not thread-safe.
• What to test first. Laziness (the pulled list), then release, then the three-thread drain.
• Common mistake. Storing relative offsets per consumer, which must all be decremented on every release and so make it O(n).

02Where this question was reported

• Land the Lab · 'Lazy tee: split one iterator into n independent iterators'
• aceoffer.app · 'Implementing a tee iterator' named as a less-common L5 coding problem
• PracHub · Google Senior+, Hard; iterator class design for independent consumers