Thread-safe task queue on a linked list

Question: Thread-safe task queue on a linked list

Companies: Anthropic

Difficulty: Medium

Problem statement

A task queue was written for one thread: a singly linked list with head and tail, where put appends at the tail and get removes from the head. Explain what goes wrong when many producer and consumer threads share it, then make it safe:

• put(value, timeout=None) blocks while the queue is at capacity (if one is set) and raises Full on timeout.
• get(timeout=None) blocks while the queue is empty and raises Empty on timeout.
• close() makes later put calls raise Closed; consumers drain what is left, then get raises Closed. Waiting threads must wake up.
• Every task is delivered exactly once; one producer's tasks come out in the order it put them.

class Empty(Exception): ...
class Full(Exception): ...
class Closed(Exception): ...


class TaskQueue:
    def __init__(self, capacity=None): ...
    def put(self, value, timeout=None): ...
    def get(self, timeout=None): ...
    def close(self): ...
    def __len__(self): ...

Approach

The starting point. A task queue written for one thread: a singly linked list with head and tail, put appends at the tail, get removes from the head. Under several producer and consumer threads it breaks in specific ways; naming them is the first part of the answer.

What goes wrong without synchronization.

• Lost tasks. Two producers both read the same tail, both set tail.next; one node is overwritten and its task vanishes.
• Double delivery. Two consumers both read the same head and both return its task.
• Broken empty/one-element transitions. get on the last node must clear tail too; a producer appending at that moment can attach its node to a node that is no longer in the list.
• Busy waiting. A consumer on an empty queue spins on while head is None, burning a core.

The GIL does not save you: it makes single bytecodes atomic, not a read-modify-write of tail spread over several bytecodes.

Design.

• One lock, two conditions (not_empty, not_full) on that lock. put waits while full, links the node, notifies not_empty. get waits while empty, unlinks, notifies not_full. Waits are in while loops because a thread can wake with nothing to do (another consumer got there first, or a spurious wakeup).
• Timeouts: get(timeout) raises Empty if nothing arrives in time; compute a deadline once and wait for the remaining time, so repeated wakeups do not extend the total wait.
• Shutdown: close() makes put fail and lets consumers drain what is left, then get raises Closed. notify_all on close, or waiting consumers sleep forever.
• A dummy head node removes the empty-list special case: head always points at a sentinel; the first real task is head.next. Then put touches only tail and get touches only head, which is what makes the two-lock version possible (follow-up 1).

Invariants (checked in the tests): every task put is returned by exactly one get; a single producer's tasks come out in order; size equals the number of nodes after the sentinel.

Code

import threading
import time


class Empty(Exception):
    pass


class Full(Exception):
    pass


class Closed(Exception):
    pass


class _Node:
    __slots__ = ("value", "next")

    def __init__(self, value=None):
        self.value, self.next = value, None


class TaskQueue:
    """Blocking FIFO queue on a singly linked list with a sentinel head."""

    def __init__(self, capacity=None):
        self.head = self.tail = _Node()  # sentinel
        self.size, self.capacity, self.closed = 0, capacity, False
        self.lock = threading.Lock()
        self.not_empty = threading.Condition(self.lock)
        self.not_full = threading.Condition(self.lock)

    @staticmethod
    def _remaining(deadline):
        return None if deadline is None else deadline - time.monotonic()

    def put(self, value, timeout=None):
        deadline = None if timeout is None else time.monotonic() + timeout
        with self.not_full:
            while not self.closed and self.capacity is not None and self.size >= self.capacity:
                left = self._remaining(deadline)
                if left is not None and left <= 0:
                    raise Full
                self.not_full.wait(left)
            if self.closed:
                raise Closed
            node = _Node(value)
            self.tail.next = node
            self.tail = node
            self.size += 1
            self.not_empty.notify()

    def get(self, timeout=None):
        deadline = None if timeout is None else time.monotonic() + timeout
        with self.not_empty:
            while self.size == 0:
                if self.closed:
                    raise Closed
                left = self._remaining(deadline)
                if left is not None and left <= 0:
                    raise Empty
                self.not_empty.wait(left)
            first = self.head.next
            self.head = first          # first becomes the new sentinel
            value, first.value = first.value, None
            self.size -= 1
            self.not_full.notify()
            return value

    def close(self):
        with self.lock:
            self.closed = True
            self.not_empty.notify_all()
            self.not_full.notify_all()

    def __len__(self):
        with self.lock:
            return self.size

get makes the removed node the new sentinel instead of unlinking it: one pointer write, and the tail never needs fixing when the queue empties.

Complexity

• put, get: O(1) plus waiting. notify (not notify_all) on each operation wakes one waiter, avoiding a thundering herd.
• Throughput: one lock serializes all operations; with very short critical sections it handles millions of operations per second in C, far fewer in Python (a few hundred thousand per second). Contention, not the list, is the limit.
• Memory: one node per queued task.

Test cases

import threading
import time

from solution import Closed, Empty, Full, TaskQueue


def test_fifo_single_thread():
    q = TaskQueue()
    for i in range(5):
        q.put(i)
    assert [q.get() for _ in range(5)] == [0, 1, 2, 3, 4]
    assert len(q) == 0


def test_get_times_out_on_empty():
    q = TaskQueue()
    start = time.monotonic()
    try:
        q.get(timeout=0.05)
    except Empty:
        pass
    else:
        assert False
    assert time.monotonic() - start >= 0.04


def test_many_producers_and_consumers_deliver_each_task_once():
    q = TaskQueue(capacity=50)
    producers, consumers, per = 4, 4, 2000
    got, got_lock = [], threading.Lock()

    def produce(p):
        for i in range(per):
            q.put((p, i))

    def consume():
        for _ in range(producers * per + 1):  # bounded, so a broken queue cannot hang the test
            try:
                item = q.get(timeout=5)
            except (Closed, Empty):
                return
            with got_lock:
                got.append(item)

    cs = [threading.Thread(target=consume, daemon=True) for _ in range(consumers)]
    ps = [threading.Thread(target=produce, args=(p,), daemon=True) for p in range(producers)]
    for t in cs + ps:
        t.start()
    for t in ps:
        t.join(10)
    q.close()
    for t in cs:
        t.join(5)
    assert sorted(got) == sorted((p, i) for p in range(producers) for i in range(per))
    for p in range(producers):  # every task of every producer arrived
        mine = [i for (pp, i) in got if pp == p]
        assert len(mine) == per


def test_bounded_put_blocks_until_space_or_times_out():
    q = TaskQueue(capacity=1)
    q.put("a")
    try:
        q.put("b", timeout=0.02)
    except Full:
        pass
    else:
        assert False
    threading.Timer(0.05, q.get).start()
    q.put("b", timeout=1)  # succeeds once the timer frees a slot
    assert q.get() == "b"


def test_close_wakes_waiters_and_drains():
    q = TaskQueue()
    q.put("last")
    q.close()
    assert q.get() == "last"
    try:
        q.get()
    except Closed:
        pass
    else:
        assert False
    try:
        q.put("x")
    except Closed:
        pass
    else:
        assert False

    q2 = TaskQueue()
    outcome = []

    def wait():
        try:
            q2.get()
        except Closed:
            outcome.append("closed")

    t = threading.Thread(target=wait)
    t.start()
    time.sleep(0.02)
    q2.close()
    t.join(1)
    assert outcome == ["closed"]

Follow-ups

1. Two locks. Because the sentinel separates the ends, producers only touch tail and consumers only touch head. A head lock and a tail lock let one producer and one consumer run at the same time. The shared size (for capacity) then needs an atomic counter or its own small lock, and the not-empty signal must be sent under the head lock. This roughly doubles throughput under mixed load in languages without a global lock.
2. Lock-free. The Michael–Scott queue uses compare-and-swap on tail.next and tail, with helping (a thread that sees a lagging tail advances it). It needs safe memory reclamation (hazard pointers or epochs) in languages without garbage collection. Not available in pure Python; mention it, do not write it.
3. Batching. get_many(max_items) takes up to n tasks under one lock acquisition, cutting lock traffic for small tasks.
4. Fairness and priority. Python's Condition does not promise FIFO wakeup order among waiters. For strict fairness keep a queue of per-waiter events; for priorities, replace the list with a heap and keep the same locking.
5. Processes, not threads. Across processes, use a pipe- or socket-backed queue, or a broker; the semantics to preserve are the same: at-most-once removal, blocking with timeouts, and clean shutdown.

Interviewer expectations

Mid-level (L4)

• Explains the races (lost append, double removal) and fixes them with a lock around both operations.
• Replaces busy waiting with a condition variable, waiting in a loop.
• Tests with several producer and consumer threads.

Senior (L5)

• Adds capacity with two conditions, timeouts with a deadline, and a clean close that wakes waiters.
• Uses a sentinel node and explains how it removes the empty-list special case.
• States the invariants and checks them in tests (each task once, per-producer order).

Staff (L6+)

• Describes the two-lock queue and why the sentinel enables it, and the lock-free variant with its memory-reclamation problem.
• Discusses fairness, batching and contention as the real throughput limits.
• Relates the design to process-level and distributed queues.