Concurrent queue in Python
A FIFO queue supporting concurrent enqueue and dequeue from many threads.
- File
concurrent_queue.py, 105 lines- Level
- I0 not checked yet
- Toolchain
- Python 3.14, pytest 9
- Test framework
- pytest
- Run the tests
python3 -m pytest
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/concurrent-queue/. The authors are listed in prov/EBTAW1.jsonl.
import threading
import time
from collections.abc import Generator
from typing import Generic, TypeVar
T = TypeVar("T")
R = TypeVar("R")
EMPTY = object() # the result of a dequeue on an empty queue
SPIN = "spin" # what a step generator yields while it waits for a lock
Steps = Generator[str | None, None, R]
def acquire(lock: threading.Lock) -> Generator[str | None, None, None]:
"""Take the lock, one try per step; yield SPIN while another holder has it."""
while not lock.acquire(blocking=False):
yield SPIN
def run(steps: "Steps[R]") -> R:
"""Run a step generator to its end in the calling thread and return its result."""
while True:
try:
if next(steps) == SPIN:
time.sleep(0)
except StopIteration as stop:
return stop.value
class _Node(Generic[T]):
__slots__ = ("value", "next")
def __init__(self, value: "T | None") -> None:
self.value = value
self.next: "_Node[T] | None" = None
class ConcurrentQueue(Generic[T]):
"""An unbounded FIFO queue that many threads can use, with one lock.
The list starts with a sentinel node; head is the sentinel and tail is
the last node. Each operation runs its sequential code while it holds
the lock, so the queue is linearizable and deadlock-free. Each operation
takes O(1) steps plus the time it waits for the lock.
The *_steps methods return generators that yield after each step on
shared memory, so a test can choose the interleaving. enqueue and
dequeue run the same steps directly and are safe to call from threads.
"""
def __init__(self) -> None:
self.head: _Node[T] = _Node(None)
self.tail: _Node[T] = self.head
self.lock = threading.Lock()
def enqueue_steps(self, x: T) -> "Steps[None]":
n: _Node[T] = _Node(x) # @line L1
yield from acquire(self.lock) # @line L2
self.tail.next = n # @line L3
yield None
self.tail = n # @line L4
yield None
self.lock.release() # @line L5
def dequeue_steps(self) -> "Steps[object]":
yield from acquire(self.lock) # @line L6
first = self.head.next # @line L7
yield None
if first is None: # @line L8
self.lock.release()
return EMPTY
self.head = first # @line L9
yield None
x = first.value # @line L10
first.value = None # the node is the new sentinel; drop the reference
self.lock.release() # @line L11
return x # @line L12
def enqueue(self, x: T) -> None:
run(self.enqueue_steps(x))
def dequeue(self) -> object:
"""Return the oldest item, or EMPTY when the queue has no item."""
return run(self.dequeue_steps())
def items(self) -> list[T]:
"""The items from the oldest to the newest. Call it only when no operation runs."""
out: list[T] = []
n = self.head.next
while n is not None:
out.append(n.value) # type: ignore[arg-type]
n = n.next
return out
def check_invariant(self) -> None:
"""Check, at a quiet moment, that tail is the last node and the lock is free."""
n = self.head
while n.next is not None:
n = n.next
assert n is self.tail
assert not self.lock.locked()
test_concurrent_queue.py 201 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/concurrent-queue/. The authors are listed in prov/EBTAW1.jsonl.
import random
import threading
from concurrent_queue import EMPTY, ConcurrentQueue
# ---- deterministic interleaving driver and linearizability check ----
Op = tuple # (thread, name, arg, result, invoke time, response time)
def interleave(threads, schedule, limit=100000):
"""Run the operations of each thread, one atomic step at a time.
threads[t] is a list of (name, arg, make) where make() returns the step
generator of the operation. schedule is a sequence of thread numbers; a
pick of a thread with no work left is skipped. After the schedule, the
remaining steps run round robin. Returns the history as a list of Op.
"""
todo = [list(reversed(ops)) for ops in threads]
cur: list = [None] * len(threads)
history: list = []
clock = 0
def step(t):
nonlocal clock
if cur[t] is None:
if not todo[t]:
return False
name, arg, make = todo[t].pop()
cur[t] = (name, arg, make(), clock)
name, arg, gen, inv = cur[t]
clock += 1
try:
next(gen)
except StopIteration as stop:
history.append((t, name, arg, stop.value, inv, clock))
cur[t] = None
return True
for t in schedule:
step(t)
steps = 0
while any(cur) or any(todo):
for t in range(len(threads)):
step(t)
steps += 1
assert steps < limit, "no progress"
return history
def linearizable(history, init, apply):
"""Wing and Gong search: is there a legal order that respects real time?
apply(state, name, arg) returns (new state, result); states are hashable.
"""
n = len(history)
full = (1 << n) - 1
seen = set()
def dfs(done, state):
if done == full:
return True
if (done, state) in seen:
return False
seen.add((done, state))
for i in range(n):
if done >> i & 1:
continue
inv_i = history[i][4]
if any(not done >> j & 1 and history[j][5] < inv_i for j in range(n)):
continue
s2, r = apply(state, history[i][1], history[i][2])
if r == history[i][3] and dfs(done | 1 << i, s2):
return True
return False
return dfs(0, init)
def queue_spec(state, name, arg):
if name == "enq":
return state + (arg,), None
if not state:
return state, EMPTY
return state[1:], state[0]
def make_ops(q, plan):
ops = []
for name, arg in plan:
if name == "enq":
ops.append((name, arg, lambda a=arg: q.enqueue_steps(a)))
else:
ops.append((name, None, q.dequeue_steps))
return ops
def test_example() -> None:
q: ConcurrentQueue[int] = ConcurrentQueue()
a = make_ops(q, [("enq", 1), ("enq", 2)])
b = make_ops(q, [("deq", None), ("deq", None)])
h = interleave([a, b], [0, 1, 1, 0, 0, 0, 1, 1, 1, 1, 0, 0, 0, 0])
results = {(t, name, arg): r for t, name, arg, r, _, _ in h}
# A takes the lock first; B spins until A releases it, then dequeues 1.
assert [r for t, name, _, r, _, _ in h if t == 1] == [1, 2]
assert results[(0, "enq", 1)] is None
assert linearizable(h, (), queue_spec)
q.check_invariant()
assert q.items() == []
def test_edges() -> None:
q: ConcurrentQueue[str] = ConcurrentQueue()
assert q.dequeue() is EMPTY
q.enqueue("a")
q.check_invariant()
assert q.items() == ["a"]
assert q.dequeue() == "a"
assert q.dequeue() is EMPTY
q.check_invariant()
q.enqueue(None) # type: ignore[arg-type]
assert q.dequeue() is None
def test_lock_excludes() -> None:
q: ConcurrentQueue[int] = ConcurrentQueue()
g = q.enqueue_steps(5)
next(g) # g now holds the lock
d = q.dequeue_steps()
assert all(next(d) == "spin" for _ in range(10))
for _ in g:
pass
assert next(d) is None # the dequeue got the lock and read head.next
assert list(d) == [None]
def test_random_against_model() -> None:
rng = random.Random(4711)
for _ in range(500):
q: ConcurrentQueue[int] = ConcurrentQueue()
model: list[int] = []
for _ in range(rng.randint(0, 40)):
if rng.random() < 0.55:
x = rng.randint(-5, 5)
q.enqueue(x)
model.append(x)
else:
want = model.pop(0) if model else EMPTY
assert q.dequeue() == want
q.check_invariant()
assert q.items() == model
def test_random_interleavings_are_linearizable() -> None:
rng = random.Random(90210)
for _ in range(500):
q: ConcurrentQueue[int] = ConcurrentQueue()
start = [rng.randint(0, 9) for _ in range(rng.randint(0, 2))]
for x in start:
q.enqueue(x)
threads = []
for t in range(3):
plan = [("enq", 10 * t + i) if rng.random() < 0.5 else ("deq", None) for i in range(rng.randint(1, 3))]
threads.append(make_ops(q, plan))
schedule = [rng.randrange(3) for _ in range(rng.randint(0, 60))]
h = interleave(threads, schedule)
assert linearizable(h, tuple(start), queue_spec)
q.check_invariant()
def test_threads() -> None:
q: ConcurrentQueue[tuple[int, int]] = ConcurrentQueue()
per, producers = 300, 3
got: list[list[tuple[int, int]]] = [[] for _ in range(3)]
def produce(p: int) -> None:
for i in range(per):
q.enqueue((p, i))
def consume(c: int) -> None:
while len(got[c]) < per:
x = q.dequeue()
if x is not EMPTY:
got[c].append(x) # type: ignore[arg-type]
ts = [threading.Thread(target=produce, args=(p,)) for p in range(producers)]
ts += [threading.Thread(target=consume, args=(c,)) for c in range(3)]
for t in ts:
t.start()
for t in ts:
t.join()
allgot = [x for g in got for x in g]
assert sorted(allgot) == sorted((p, i) for p in range(producers) for i in range(per))
for g in got: # one consumer sees the items of one producer in order
for p in range(producers):
seq = [i for (pp, i) in g if pp == p]
assert seq == sorted(seq)
assert q.dequeue() is EMPTY
What the next level needs
- the tests did not run in this build