Multi-producer multi-consumer queues

With one producer and one consumer a position has one writer. With many, the positions must be handed out atomically, and a ticket is not enough: the producer that holds position 5 has not written its item yet. A sequence number on every cell says whose turn the cell is. Over 2,756 schedules of two producers and 1,600 of two consumers, this queue never breaks.

A bounded queue of capacity cells uses two counters that only grow. A producer takes the next position with a compare-and-swap on enq; a consumer takes the next one with a compare-and-swap on deq. Position p lives in cell p % capacity. The compare-and-swap makes the claims safe: no two producers get the same position. But claiming a position is not writing the item. Between the swap and the store the cell is claimed and empty, and a consumer that reads it now would read garbage. Each cell therefore carries a sequence number that is moved on only at the right moment.

def init_queue(capacity):    shared = {"enq": 0, "deq": 0}    for cell in range(capacity):        shared["seq%d" % cell] = cell    return shareddef enqueue(shared, item, capacity):    while True:        yield        pos = shared["enq"]        cell = pos % capacity        diff = shared["seq%d" % cell] - pos        if diff < 0:            return False        if diff > 0:            continue        yield        if compare_and_swap(shared, "enq", pos, pos + 1):            break    yield    shared["data%d" % cell] = item    shared["seq%d" % cell] = pos + 1    shared["put%d" % pos] = item    return Truedef dequeue(shared, capacity):    while True:        yield        pos = shared["deq"]        cell = pos % capacity        diff = shared["seq%d" % cell] - (pos + 1)        if diff < 0:            return None        if diff > 0:            continue        yield        if compare_and_swap(shared, "deq", pos, pos + 1):            break    yield    item = shared.get("data%d" % cell)    shared["seq%d" % cell] = pos + capacity    shared["got%d" % pos] = item    return itemsolo = lambda shared: [producer_thread((7,), 2)(shared), consumer_thread(1, 2)(shared)]for prefix in ([0, 0, 0], [0, 0, 0, 1, 1, 1], [0, 0, 1, 1, 0, 1, 1]):    print(prefix, run_prefix(solo, prefix, init_queue(2))[0])

[0, 0, 0] {'enq': 1, 'deq': 0, 'seq0': 1, 'seq1': 1, 'data0': 7, 'put0': 7}
[0, 0, 0, 1, 1, 1] {'enq': 1, 'deq': 1, 'seq0': 2, 'seq1': 1, 'data0': 7, 'put0': 7, 'got0': 7}
[0, 0, 1, 1, 0, 1, 1] {'enq': 1, 'deq': 0, 'seq0': 1, 'seq1': 1, 'data0': 7, 'put0': 7}

Read the three runs. After [0, 0, 0] the producer has claimed position 0 (enq: 1), stored data0, and raised seq0 to 1, so the item is ready. After the consumer's three turns, got0: 7 and seq0: 2: the cell is free for the next lap. In the last run the consumer looks between the producer's claim and its publish, sees seq0 = 0 instead of 1, and goes away empty-handed. It does not read garbage. It just says "nothing ready yet", and the item will be found by a later attempt.

What must the queue promise now? The producers record each item under its position (put3), the consumers record what they took under the position they claimed (got3). If the queue works, every position that was got holds the item that was put there (no mismatch), and the positions got are 0, 1, 2, ... without holes or repeats (no gap). That gives a queue its order: the order of positions. violations reads these two facts out of a final state and verify runs the explorer for any number of producers and consumers. Here a single producer with two items against two consumers, and a single item against one consumer.

def violations(final):    puts = {int(key[3:]): value for key, value in final.items() if key.startswith("put")}    gots = {int(key[3:]): value for key, value in final.items() if key.startswith("got")}    broken = []    if any(gots[p] != puts.get(p) for p in gots):        broken.append("mismatch")    if sorted(gots) != list(range(len(gots))):        broken.append("gap")    return sorted(broken)def verify(producers, consumers, capacity):    def make_threads(shared):        return [producer_thread(items, capacity)(shared) for items in producers] + [consumer_thread(n, capacity)(shared) for n in consumers]    outcomes = explore_all(make_threads, init_queue(capacity))    broken = {}    pushed = popped = 0    for key, count in outcomes.items():        final = dict(key)        pushed = max(pushed, sum(1 for k in final if k.startswith("put")))        popped = max(popped, sum(1 for k in final if k.startswith("got")))        for name in violations(final):            broken[name] = broken.get(name, 0) + count    return {"schedules": sum(outcomes.values()), "broken": dict(sorted(broken.items())), "pushed": pushed, "popped": popped}print(verify([(7,)], [1], 2))print(verify([(7, 8)], [1, 1], 2))

{'schedules': 4, 'broken': {}, 'pushed': 1, 'popped': 1}
{'schedules': 1600, 'broken': {}, 'pushed': 2, 'popped': 2}

To see its value, take it away. The careless consumer below claims a position with the compare-and-swap and then reads the cell at once, without looking at seq. Against one producer with one item it has 20 schedules, and in 10 of them it claims a position before the producer has published anything and returns None as an item, or the wrong item. The real queue has 4 schedules for the same program and none breaks. The sequence number is exactly the check that turns a claimed ticket into a delivered item.

def careless_dequeue(shared, capacity):    yield    pos = shared["deq"]    cell = pos % capacity    yield    if not compare_and_swap(shared, "deq", pos, pos + 1):        return None    yield    item = shared.get("data%d" % cell)    shared["got%d" % pos] = item    return itemdef careless_consumer(attempts, capacity):    def thread(shared):        for _ in range(attempts):            yield from careless_dequeue(shared, capacity)    return threadmake = lambda shared: [producer_thread((7,), 2)(shared), careless_consumer(1, 2)(shared)]outcomes = explore_all(make, init_queue(2))broken = sum(n for key, n in outcomes.items() if violations(dict(key)))print(sum(outcomes.values()), "schedules,", broken, "break the invariants")

20 schedules, 10 break the invariants

ticks = [int(round(float(c) * 100)) for c in load_bars()["close"].iloc[:2]]print(ticks)print("two producers:", verify([(ticks[0],), (ticks[1],)], [2], 2))print("two consumers:", verify([tuple(ticks)], [1, 1], 2))

[9369, 9305]
two producers: {'schedules': 2756, 'broken': {}, 'pushed': 2, 'popped': 2}
two consumers: {'schedules': 1600, 'broken': {}, 'pushed': 2, 'popped': 2}

Both shapes of the queue are correct over every schedule, with the explorer as the witness, while a version without the sequence check fails in half of its (much smaller) space. Producers and consumers never wait for each other: a consumer that finds nothing ready simply returns, and a producer that finds the queue full returns False. That is the price and the gift of a lock-free queue: operations may fail, but they never block.

The lab follows the lecture: the empty queue with its sequence numbers, the enqueue and the dequeue with a compare-and-swap claim and a publish, the check of the two invariants, the verifier that runs the explorer for any number of producers and consumers, and the study on two SPY closes.

• Function · questionBuilds on
• init_queue(capacity)What does an empty queue look like?Plain arithmetic
• enqueue(shared, item, capacity)How does a producer claim and publish?Plain arithmetic
• dequeue(shared, capacity)How does a consumer claim and release?Plain arithmetic
• violations(final)Which invariants does a final state break?Plain arithmetic
• verify(producers, consumers, capacity)Is the queue right under every schedule?explore_allviolations
• study(bars)Do both shapes hold on SPY closes?verify

Next topic: you have now proved small programs correct by running every schedule. Testing concurrent code compares that with the tools real teams use: a race detector that watches one run and flags accesses that could race, and a model checker that does what your explorer does, but with the tricks that let it scale.