MPMC Ring Buffers

In the previous lesson , we built a fast data pipeline spanning two CPU cores. By padding our Head and Tail atomics to avoid false sharing, and by using acquire and release fences, we created lock-free synchronization.

But our implementation relied on a strict rule: Single-Producer, Single-Consumer (SPSC). The design assumes only one thread would ever update the Tail variable, and only one thread would ever update the Head.

In many highly parallel systems, this rule is impossible to enforce. If we are building a global job scheduler for a thread pool, we might have 16 threads finishing tasks and attempting to Push() new jobs onto the queue at the same time.

In this lesson, we will upgrade our ring buffer to support Multiple-Producer, Multiple-Consumer (MPMC) environments. We will use Compare-And-Swap (CAS) loops to claim indices, and we will introduce Sequence Tickets to synchronize the chaotic transfer of physical payloads.

The Cost of Unrestricted Concurrency

Over the next three lessons in this chapter, we will build data structures that safely allow any number of threads to fight over them simultaneously. We are moving into the realm of unrestricted, chaotic concurrency.

But just because something is possible doesn't mean it is a good design. As we covered in the previous concurrency chapters, one of the primary goals of high-performance multithreaded architecture is to reduce synchronization as much as possible. Using thread-local storage and isolating data to individual threads remains the most effective way to do that.

Lock-free data structures do not magically solve the physical realities of hardware contention. Additionally, the defensive logic required to support unrestricted concurrency makes simple algorithms like Push() and Pop() significantly more complex and computationally expensive than their single-threaded or SPSC counterparts.

However, some systems, like global job schedulers, thread pools, or work-stealing algorithms, genuinely require a container that supports full, unbridled multithreading. For those unavoidable bottlenecks, using a full Multi-Producer Multi-Consumer (MPMC) design is often unavoidable.

Recommended Reading

This is a relatively advanced lesson, pulling together many techniques we covered in the previous chapters. To follow along, being particularly comfortable with the following topics is recommended:

• Implementing a basic ring buffer and an SPSC ring buffer
• The Read-Modify-Write trap and how to prevent it using std::atomic.
• Implementing custom atomic logic using compare-and-swap (CAS)
• Optimizing the memory ordering around atomic instructions.

Let's look at why our previous SPSC code to understand why it breaks if we allow multiple producers to use it. In the SPSC queue, our push logic looked like this:

1bool Push(const T& Value) {
2  // Relaxed read: safe ONLY if we are the only producer
3  size_t CurrentTail = Tail.load(); 
4
5  // check if full ...
6
7  Buffer[CurrentTail] = Value; 
8
9  // update tail ...
10}

If multiple threads run this function concurrently, they might all execute the load() instruction before any of them writes to the Buffer.

If producers running on Thread 1 and Thread 2 both read Tail at the same time, they will both get the same index (e.g., 4). Thread 1 writes its data to Buffer[4]. One microsecond later, Thread 2 blindly writes its data to Buffer[4]. Thread 1's data is permanently erased from memory.

Our Pop() function would have similar problems if multiple consumers were trying to use it concurrently.

To prevent these collisions, reading a Tail or Head index is no longer enough. A thread must claim the index.

The Atomic Index Claim

To claim an index, we can rely on the silicon's native capability to read and modify a memory address in a single, indivisible step. We do this using the optimstic Compare-And-Swap (CAS) loop we covered in earlier lessons.

Instead of just loading the Tail, we optimistically attempt to push it forward. If another thread snuck in and claimed our expected index, the CPU rejects our update, hands us the fresh Tail index, and we loop around to try again.

The final logic will be more complicated, but a basic CAS loop for claiming a Tail index would look like this:

1size_t ExpectedTail;
2
3do {
4  ExpectedTail = Tail.load();
5
6  // Attempt to claim ExpectedTail by swinging the
7  // Tail forward.  If another thread has updated,
8  // ExpectedTail, our CAS fails, ExpectedTail is
9  // automatically updated to the latest value,
10  // and the loop spins around and tries again
11} while (!Tail.compare_exchange_weak( 
12  ExpectedTail,                       
13  ExpectedTail + 1                    
14));                                   
15
16// If the CAS succeeds, our loop ends and we 
17// exclusively own ExpectedTail. We can now 
18// safely write to Buffer[ExpectedTail]

The Race to the Payload

By using CAS, we can safely claim a slot, but that's not the biggest challenge. The more complex problem is the Race to the Payload. Just because Thread 1 successfully claimed index 4 and advanced the Tail to 5, does not mean Thread 1 has finished writing its data into physical RAM.

Imagine Thread 1 executes its CAS instruction and claims index 4. Immediately, the OS suspends Thread 1. The Tail is now officially 5. Thread 2 (a Consumer) wakes up, checks the Tail, sees that it is 5, and assumes indices 0 through 4 are ready to read. It pops index 4 and reads the payload.

But Thread 1 is asleep. It hasn't written the payload yet. Thread 2 just consumed uninitialized garbage data. Claiming an index is no longer enough. We need a way to synchronize the exact moment the payload data is fully written to memory and safe to read.

To solve the payload race, we need a way to lock every single slot in our buffer independently. We can achieve this by giving every slot its own dedicated Sequence Ticket.

Instead of just trusting the global Head and Tail counters, threads must physically inspect the sequence ticket of the specific slot they want to use. This ticket tells us exactly which "lap" the slot is currently on, and whether it is ready for a producer or a consumer.

One key difference between an MPMC circular queue and our previous versions is that our Head and Tail counters will no longer "wrap around" to 0. Instead, we just let them keep counting up towards infinity. This creates a pattern where we have generations (or laps) around our circle.

For example, imagine we use a ring buffer with a Capacity of 4 slots. Then, our Head and Tail would increment with the following pattern:

• Lap 1: Tickets 0, 1, 2, 3
• Lap 2: Tickets 4, 5, 6, 7
• Lap 3: Tickets 8, 9, 10, 11

When a producer wants to push data, it first reads the global Tail counter. Let's say it reads 4. This means the producer wants to claim ticket 4. After using modulo math or the bitwise masking techniques, it will know that ticket 4 corresponds to slot 0 of the second lap around our buffer.

But in a concurrent system, multiple producers might read a Tail of 4 at around the same time, and all of them will rush to slot 0. How do they negotiate who gets to write? And how do we prevent them from overwriting data that no consumer has read yet?

The Producer (+1) and Consumer (+Capacity) Dance

The access in each slot alternates between a producer and a consumer, so we can update the ticket in those slots accordingly to keep track of who has visited it. When a producer finishes writing to a slot, we increment the ticket by 1. When a consumer looks at that slot, if the ticket is incremented, it knows a producer has populated it, and the data is ready to be consumed.

When the consumer finishes processing the data, it resets the slot for the next producer, which will be on the next lap around the circle. So, for example, if we look at slot 0 of our 4-slot buffer, then the ticket starts at 0, is incremented to 1 when a producer has finished populating it, and then set to 4 when a consumer has finished processing it.

The next producer increments it to 5, the next consumer resets it to 8, and so on.

Below, we illustrate this pattern with a single producer and single consumer (SPSPC), and we'll explain how this helps us in MPMC scenarios in the next section.

So, once we have our producers and consumers dutifully updating our sequence tickets using this agreed pattern, how does that help us?

Because we are in a heavily parallel environment, threads will frequently look at tickets at the wrong time. A producer might read Head as 4, calculate that it corresponds to slot 0 in the buffer, but by the time they read what's in that slot, something could have changed.

To figure out what to do, threads calculate the difference between the number on the ticket and the number they expected to see.

For Producers (Pushing Data)

A producer expects the ticket to exactly match the global Tail it just read. It calculates the difference using the formula Diff = Ticket - Tail

• If Diff == 0 (Perfect Match): The ticket is waiting for us. We fire our CAS instruction to try to claim the slot by incrementing the global Tail. If the CAS succeeds, the slot is ours, and we can write our data. If it failed, another producer beat us to updating the tail. We need to start again - figure out what slot the new Tail corresponds to, read its ticket, and recalculate the Diff.
• If Diff < 0 (Queue is Full): The ticket is behind us. If we want ticket 4, but the ticket says 1, it means the consumers haven't finished reading Lap 1 yet. The buffer is effectively full, so we abort the push.
• If Diff > 0 (We are Stale): The ticket is ahead of us. If we want ticket 4, but the ticket already says 5, it means another producer is way ahead of us - they've already claimed the slot, written their data and clicked the ticket forward. Our expected Tail is stale, so we must reload the global Tail and try again.

For Consumers (Popping Data)

A consumer expects the ticket to be exactly one step ahead of the global Head it just read, because the producer adds 1 when it finishes writing. Therefore, it uses the formula Diff = Ticket - (Head + 1)

• If Diff == 0 (Perfect Match): The producer has finished writing. We fire our CAS instruction to officially claim the slot by incrementing the Head index. If it succeeds, the slot is ours, and we can process the data. If it failed, another consumer beat us to the slot, and we need to retry using the latest Head.
• If Diff < 0 (Queue is Empty): The ticket hasn't been clicked forward yet. If we want to read ticket 0, but the ticket still says 0, a producer hasn't finished writing yet. The queue is effectively empty, so we abort the pop.
• If Diff > 0 (We are Stale): The ticket is ahead of us. Another consumer beat us - they've already read this data and pushed the ticket forward for the next lap. Our expected Head is stale, so we must reload the global Head and loop around to try again.

The Unsigned Trap

Most of the time, we'll want to store our Head and Tail indicies as unsigned integers, such as size_t. This creates a problem: unsigned integers cannot be negative, so our Diff < 0 check will always fail. If our calculation results in a negative Diff, it underflows and wraps around to a massive positive number.

To solve this, we can cast the variables to intptr_t before performing the subtraction. The intptr_t type is a signed integer guaranteed to be the same bit-width as a memory pointer on the target system, which makes it the same size as size_t.

Now that we understand the sequence ticket mechanics conceptually, let's establish our class structure.

We will stop treating our buffer as just an array of T payloads. Instead, we bundle the payload alongside a std::atomic<size_t> sequence integer to act as our sequence ticket. We will call this bundle a Cell:

1#pragma once
2#include <vector>
3#include <cstdint>
4#include <atomic>
5#include <new>
6#include <bit>
7
8template <typename T>
9class MpmcQueue {
10private:
11  static constexpr size_t CacheLineSize =
12    std::hardware_destructive_interference_size;
13
14  // Bundle the data with an atomic sequence ticket
15  struct alignas(CacheLineSize) Cell {
16    std::atomic<size_t> Sequence;
17    T Payload;
18  };
19
20  std::vector<Cell> Buffer;
21  size_t Capacity;
22  size_t Mask;
23
24  alignas(CacheLineSize) std::atomic<size_t> Head{0};
25  alignas(CacheLineSize) std::atomic<size_t> Tail{0};
26  // ...
27};

Notice the alignas(CacheLineSize) on the Cell struct. If we didn't pad the cell, Buffer[0] and Buffer[1] would likely share the same physical 64-byte cache line in RAM. If Thread 1 is writing to slot 0 and Thread 2 is writing to slot 1, the CPU cores would thrash the cache line back and forth, crippling performance via false sharing.

By padding the cell to the cache line size (usually 64 bytes), every slot gets its own dedicated cache line, even when the payload type T is small.

The Constructor

When we initialize our queue, we also need to initialize our sequence tickets for the first lap by pre-filling every cell's ticket with its starting physical index:

1// ...
2
3template <typename T>
4class MpmcQueue {
5  // ...
6public:
7  MpmcQueue(size_t RequestedCapacity) {
8    if (!std::has_single_bit(RequestedCapacity)) {
9      throw std::invalid_argument(
10        "Capacity must be power of 2"
11      );
12    }
13    Capacity = RequestedCapacity;
14    Mask = Capacity - 1;
15    Buffer.resize(Capacity);
16
17    // Initialize the tickets for Lap 1
18    for (size_t i = 0; i < Capacity; ++i) {
19      Buffer[i].Sequence.store(i);
20    }
21  }
22};

Preventing Exceptions

Lock-free structures that need to copy or move data typically restrict support to only types that cannot throw during these actions, as it makes the implementation much easier and safer.

We can assert our template type T meets this requirement with the help of a type trait:

1// ...
2#include <type_traits> 
3
4template <typename T>
5class MpmcQueue {
6private:
7  // Guarantee exception safety to prevent memory leaks
8  static_assert(
9    std::is_nothrow_copy_assignable_v<T>,
10    "Type T must not throw to guarantee exception safety"
11  );
12
13  // ...
14};

If we used move semantics in our container, we'd also want to add std::is_nothrow_move_assignable_v<T> to this safety checks.

Let's write the Push() function step-by-step. This is where we combine our CAS index claim with the sequence ticket logic.

Step 1: Inspecting the Ticket

We need three things to implement our Push() algorithm, which we'll store in the following variables:

• ExpectedTail - the Tail value that we're trying to claim
• TargetCell - the slot within our Buffer that this tail corresponds to, based on our bitwise masking logic
• Seq - the sequence ticket within that cell

Let's start by retrieving this data:

1// ...
2
3template <typename T>
4class MpmcQueue {
5  // ...
6public:
7  // ...
8  bool Push(const T& Value) {
9    Cell* TargetCell;
10    size_t ExpectedTail;
11
12    do {
13      ExpectedTail = Tail.load();
14      TargetCell = &Buffer[ExpectedTail & Mask];
15
16      // Read the ticket to determine the cell's state
17      size_t Seq = TargetCell->Sequence.load();
18
19      // ...
20    } while (true);
21
22    // ...
23  }
24};

Step 2: Claiming the Index

Now that we have everything we need, we calculate the difference between our sequence ticket and our expected tail and react to it using the logic we outlined earlier in the lesson.

A producer expects the ticket to exactly match the global Tail it just read, so it calculates the difference using the formula Diff = Ticket - ExpectedTail

• If Diff == 0 (Perfect Match): The ticket is waiting for us. We fire our CAS instruction to try to claim the slot by incrementing the global Tail. If the CAS succeeds, the slot is ours, and we can write our data. If it failed, another producer beat us to updating the tail. We need to start again - figure out what slot the new Tail corresponds to, read its ticket, and recalculate the Diff.
• If Diff < 0 (Queue is Full): The ticket is behind us. If we want ticket 4, but the ticket says 1, it means the consumers haven't finished reading Lap 1 yet. The buffer is effectively full, so we abort the push.
• If Diff > 0 (We are Stale): The ticket is ahead of us. If we want ticket 4, but the ticket already says 5, it means another producer is way ahead of us - they've already claimed the slot, written their data and clicked the ticket forward. Our expected Tail is stale, so we must reload the global Tail and try again.

Remember, Seq and ExpectedTail are unsigned integers, so we need to cast them to a signed type like intptr_t to make our Diff < 0 check work. Combining all of these steps in code looks like this:

1// ...
2
3template <typename T>
4class MpmcQueue {
5  // ...
6public:
7  // ...
8  bool Push(const T& Value) {
9    // ...
10
11    do {
12      // ...
13
14      // Does the ticket match our expected tail?
15      intptr_t Diff = (intptr_t)Seq - (intptr_t)ExpectedTail;
16
17      if (Diff == 0) {
18        // Yes! It's our turn. Attempt to claim the index.
19        if (Tail.compare_exchange_weak(
20          ExpectedTail,
21          ExpectedTail + 1
22        )) {
23          break; // We successfully claimed the slot!
24        }
25      } else if (Diff < 0) {
26        // The ticket is behind us. The queue is full.
27        return false;
28      }
29
30      // If Diff > 0, another thread won. Loop restarts and
31      // ExpectedTail reloads.
32    } while (true);
33
34    // We exclusively own TargetCell if we get here
35    // ...
36  }
37};

Step 3: Publishing the Payload

Once we break out of the loop, we exclusively own TargetCell. We can safely write our data to it.

After writing the data, we must advance the ticket so the consumer knows the data is ready. We advance the ticket to ExpectedTail + 1.

1// ...
2
3template <typename T>
4class MpmcQueue {
5  // ...
6public:
7  // ...
8  bool Push(const T& Value) {
9    // ...
10    do {
11      // ...
12    } while (true);
13
14    // We exclusively own TargetCell at this point
15
16    // Write the payload to memory
17    TargetCell->Payload = Value;
18
19    // Advance the ticket for the Consumer
20    TargetCell->Sequence.store(ExpectedTail + 1);
21
22    return true;
23  }
24};

Let's implement the Pop() function step-by-step. It follows a fairly similar pattern to the Push() implementation, with the key differences being that we're working with the Head instead of the Tail, and we treat the sequence ticket Seq slightly differently.

Step 1: Waiting for the Producer

Again, we'll start by gathering the key values we need. This is the same as our Push() logic, except they're based on the Head rather than the Tail. We'll store them in three variables:

• ExpectedHead - the Head value that we're trying to claim
• TargetCell - the slot within our Buffer that this head corresponds to, based on our bitwise masking logic
• Seq - the sequence ticket within that cell

Collecting these variables would look something like this:

1// ...
2
3template <typename T>
4class MpmcQueue {
5  // ...
6public:
7  // ...
8  bool Pop(T& OutValue) {
9    Cell* TargetCell;
10
11    // Pop cares about the Head rather than the tail
12    size_t ExpectedHead;
13
14    do {
15      ExpectedHead = Head.load();
16      TargetCell = &Buffer[ExpectedHead & Mask];
17
18      // Read the ticket
19      size_t Seq = TargetCell->Sequence.load();
20
21      // ...
22    } while (true);
23
24    // ...
25  }
26};

Step 2: Claiming the Read Index

Then, we implement the Diff calculation and logic we described previously.

A consumer expects the ticket to be exactly one step ahead of the global Head it just read (because the producer adds 1 when it finishes writing). It uses the formula Diff = Ticket - (ExpectedHead + 1)

• If Diff == 0 (Perfect Match): The producer has finished writing. We fire our CAS instruction to officially claim the slot by incrementing the Head index. If it succeeds, the slot is ours, and we can process the data. If it failed, another consumer beat us to the slot, and we need to retry using the latest Head.
• If Diff < 0 (Queue is Empty): The ticket hasn't been incremented yet. If the Head is slot 5, but the ticket still says 5 in that slot, a producer hasn't finished writing to it yet. The queue is effectively empty, so we abort the pop.
• If Diff > 0 (We are Stale): The ticket is ahead of us. Another consumer beat us - they've already read this data and pushed the ticket forward for the next lap. Our expected Head is stale, so we must reload the global Head and loop around to try again.

Implementing it in code would look something like below. The key difference compared to the Pop() logic is that we expect the sequence ticket to have been incremented relative to the head - that is, we want it to equal ExpectedHead + 1:

1// ...
2
3template <typename T>
4class MpmcQueue {
5  // ...
6public:
7  // ...
8  bool Pop(T& OutValue) {
9    // ...
10    do {
11      // ...
12
13      // The consumer expects the ticket to be exactly Head + 1
14      // (which is what the producer sets it to after writing)
15      intptr_t Diff = (intptr_t)Seq - (intptr_t)(ExpectedHead + 1);
16
17      if (Diff == 0) {
18        // The data is ready! Attempt to claim the read index.
19        if (Head.compare_exchange_weak(
20          ExpectedHead,
21          ExpectedHead + 1
22        )) {
23          break; // We successfully claimed the slot!
24        }
25      } else if (Diff < 0) {
26        // The ticket hasn't advanced yet. The queue is empty.
27        return false;
28      }
29
30    } while (true);
31
32    // We exclusively own TargetCell at this point
33    // ...
34  }
35};

Step 3: Extracting and Resetting

Once we claim the cell, we safely copy the payload into our OutValue.

Finally, we advance the ticket for the next generation of producers. We do this in the same way we did with Push(), except we reset the ticket based on the Capacity of the buffer rather than simply incrementing it by 1. This prepares the slot for a new producer to use it on the next lap around our circle:

1// ...
2
3template <typename T>
4class MpmcQueue {
5  // ...
6public:
7  // ...
8  bool Pop(T& OutValue) {
9    // ...
10
11    do {
12      // ...
13    } while (true);
14
15    // We exclusively own TargetCell at this point
16
17    // Safely read the payload from memory
18    OutValue = TargetCell->Payload;
19
20    // Reset the ticket for the next generation
21    TargetCell->Sequence.store(ExpectedHead + Capacity);
22
23    return true;
24  }
25};

Up to this point, we have used the default memory ordering (std::memory_order_seq_cst) for all our atomic operations. While this is safe, it forces the CPU to issue heavy memory barriers for every single read and write, reducing the performance of our lock-free queue.

To address this, we can manually tune the memory fences using acquire, release, and relaxed ordering. Let's revisit each of our functions and optimize them.

Optimizing the Constructor

In the constructor, we initialize the sequence tickets for Lap 1. Because this happens before any threads are spawned or have access to the queue, no other threads can possibly race with these writes. Therefore, we can safely use std::memory_order_relaxed:

1// ...
2
3template <typename T>
4class MpmcQueue {
5  // ...
6public:
7  MpmcQueue(size_t RequestedCapacity) {
8    // ...
9    // Initialize the tickets for Lap 1
10    for (size_t i = 0; i < Capacity; ++i) {
11      Buffer[i].Sequence.store(i, std::memory_order_relaxed); 
12    }
13  }
14  // ...
15};

Optimizing Push()

In Push(), we perform several atomic operations. We need to carefully establish a memory contract with the consumer to prevent data races on the payload.

• Loading the Tail: We load the global Tail index and attempt to increment it via CAS. Since the sequence ticket acts as our true synchronization guard for the payload, the Tail itself doesn't need to synchronize memory around it. We can use std::memory_order_relaxed for both the load and the compare_exchange_weak().
• Reading the Ticket: We should use std::memory_order_acquire when loading the sequence ticket. This half-fence tells the CPU: "Absolutely no memory reads that appear below this line are allowed to float above it." This prevents the CPU from speculatively pre-fetching the cell's payload before we have officially verified the ticket.
• Writing the Ticket: After writing the payload, we advance the ticket to signal the consumer. We should use std::memory_order_release. This tells the CPU: "Absolutely no memory writes that appear above this line are allowed to sink below it." If we relaxed this, the CPU might update the ticket before it flushes the payload to RAM, causing a consumer to read uninitialized garbage.

Implementing it would look like this:

1// ...
2
3template <typename T>
4class MpmcQueue {
5  // ...
6public:
7  // ...
8  bool Push(const T& Value) {
9    Cell* TargetCell;
10    size_t ExpectedTail;
11
12    do {
13      ExpectedTail = Tail.load(std::memory_order_relaxed); 
14      TargetCell = &Buffer[ExpectedTail & Mask];
15
16      // Read the ticket with an acquire fence
17      size_t Seq = TargetCell->Sequence.load(
18        std::memory_order_acquire 
19      );
20
21      intptr_t Diff = (intptr_t)Seq - (intptr_t)ExpectedTail;
22
23      if (Diff == 0) {
24        if (Tail.compare_exchange_weak(
25          ExpectedTail,
26          ExpectedTail + 1,
27          std::memory_order_relaxed 
28        )) {
29          break;
30        }
31      } else if (Diff < 0) {
32        return false;
33      }
34    } while (true);
35
36    TargetCell->Payload = Value;
37
38    // Advance the ticket with a release fence
39    TargetCell->Sequence.store(
40      ExpectedTail + 1,
41      std::memory_order_release 
42    );
43
44    return true;
45  }
46  // ...
47};

Optimizing Pop()

Pop() follows the exact same memory contract as Push(), just mirrored for reading.

• Loading the Head: Just like the Tail, the Head index purely governs contention between consumers, not the payload synchronization. We use std::memory_order_relaxed for loading and CAS.
• Reading the Ticket: We use std::memory_order_acquire to read the ticket. This ensures that the payload is fully synchronized from RAM after the ticket confirms the data is ready.
• Writing the Ticket: After copying the payload out, we reset the ticket for the next generation of producers. We use std::memory_order_release to guarantee our payload read is completely finished before the green light is shown to future producers.

We can implement it like this:

1// ...
2
3template <typename T>
4class MpmcQueue {
5  // ...
6public:
7  // ...
8  bool Pop(T& OutValue) {
9    Cell* TargetCell;
10    size_t ExpectedHead;
11
12    do {
13      ExpectedHead = Head.load(std::memory_order_relaxed); 
14      TargetCell = &Buffer[ExpectedHead & Mask];
15
16      // Read the ticket with an acquire fence
17      size_t Seq = TargetCell->Sequence.load(
18        std::memory_order_acquire 
19      );
20
21      intptr_t Diff = (intptr_t)Seq - (intptr_t)(ExpectedHead + 1);
22
23      if (Diff == 0) {
24        if (Head.compare_exchange_weak(
25          ExpectedHead,
26          ExpectedHead + 1,
27          std::memory_order_relaxed 
28        )) {
29          break;
30        }
31      } else if (Diff < 0) {
32        return false;
33      }
34    } while (true);
35
36    OutValue = TargetCell->Payload;
37
38    // Reset the ticket with a release fence
39    TargetCell->Sequence.store(
40      ExpectedHead + Capacity,
41      std::memory_order_release 
42    );
43
44    return true;
45  }
46  // ...
47};

We can now safely expand our architecture from two cores to an entire thread pool. Below, we spawn 2 producers and 2 consumers to interact with our MpmcQueue simultaneously:

1#include <print>
2#include <thread>
3#include <dsa/MpmcQueue.h>
4
5int main() {
6  MpmcQueue<int> JobQueue(1024);
7
8  auto producer = & {
9    for (int i = 1; i <= 3; ++i) {
10      int job_id = (producer_id * 100) + i;
11      std::println(
12        "Producer {} queued job {}",
13        producer_id, job_id
14      );
15      while (!JobQueue.Push(job_id)) {}
16    }
17  };
18
19  auto consumer = & {
20    int job_id;
21    int count = 0;
22
23    while (count < 3) {
24      if (JobQueue.Pop(job_id)) {
25        std::println(
26          "Consumer {} finished job {}",
27          consumer_id, job_id
28        );
29        ++count;
30      }
31    }
32  };
33
34  std::jthread p1(producer, 1);
35  std::jthread p2(producer, 2);
36  std::jthread c1(consumer, 1);
37  std::jthread c2(consumer, 2);
38}

The exact order of the output will vary on every run, but the hardware will flawlessly route every job from the producers to the consumers without any dropped jobs or data races:

1Producer 1 queued job 101
2Producer 2 queued job 201
3Consumer 1 finished job 101
4Producer 1 queued job 102
5Consumer 2 finished job 201
6Producer 2 queued job 202
7Consumer 1 finished job 102
8Producer 1 queued job 103
9Consumer 2 finished job 202
10Consumer 1 finished job 103
11Producer 2 queued job 203
12Consumer 2 finished job 203

Let's test the performance of our queue. We will create 8 producer threads and 8 consumer threads to pit our lock-free MpmcQueue against a std::queue protected by a std::mutex:

1#include <benchmark/benchmark.h>
2#include <queue>
3#include <mutex>
4#include <thread>
5#include <vector>
6#include <dsa/MpmcQueue.h>
7
8// A thread-safe wrapper around std::queue
9template <typename T>
10class MutexQueue {
11  std::queue<T> Q;
12  std::mutex Mtx;
13public:
14  void Push(const T& val) {
15    std::lock_guard<std::mutex> lock(Mtx);
16    Q.push(val);
17  }
18  bool Pop(T& out) {
19    std::lock_guard<std::mutex> lock(Mtx);
20    if (Q.empty()) return false;
21    out = Q.front();
22    Q.pop();
23    return true;
24  }
25};
26
27static void BM_MutexMPMC(benchmark::State& state) {
28  for (auto _ : state) {
29    MutexQueue<int> MQ;
30    std::vector<std::jthread> threads;
31
32    // 8 Producers
33    for (int t = 0; t < 8; ++t) {
34      threads.emplace_back([&] {
35        for (int i = 0; i < 1'000'000; ++i) {
36          MQ.Push(i);
37        }
38      });
39    }
40
41    // 8 Consumers
42    for (int t = 0; t < 8; ++t) {
43      threads.emplace_back([&] {
44        int count = 0;
45        int val;
46        while (count < 1'000'000) {
47          if (MQ.Pop(val)) count++;
48        }
49      });
50    }
51  }
52}
53BENCHMARK(BM_MutexMPMC)->Unit(benchmark::kMillisecond);
54
55static void BM_LockFreeMPMC(benchmark::State& state) {
56  for (auto _ : state) {
57    MpmcQueue<int> LFQ(1024 * 1024);
58    std::vector<std::jthread> threads;
59
60    // 8 Producers
61    for (int t = 0; t < 8; ++t) {
62      threads.emplace_back([&] {
63        for (int i = 0; i < 1'000'000; ++i) {
64          while (!LFQ.Push(i)) {} // Spin if full
65        }
66      });
67    }
68
69    // 8 Consumers
70    for (int t = 0; t < 8; ++t) {
71      threads.emplace_back([&] {
72        int count = 0;
73        int val;
74        while (count < 1'000'000) {
75          if (LFQ.Pop(val)) count++;
76        }
77      });
78    }
79  }
80}
81BENCHMARK(BM_LockFreeMPMC)->Unit(benchmark::kMillisecond);

1------------------------------------
2Benchmark             Time       CPU
3------------------------------------
4BM_MutexMPMC       1450 ms   8200 ms
5BM_LockFreeMPMC     310 ms   2410 ms

In the std::mutex benchmark, 16 threads are endlessly colliding with the operating system's locks. The OS frantically context-switches, flushing registers and putting threads to sleep.

In the MpmcQueue, the OS is completely unaware of the contention. The threads remain awake, spinning locally in their CAS loops and efficiently negotiating cache-line ownership via the hardware bus. It processes the same workload 4 times faster in these tests.

The Scaling Caveats

While our MPMC queue is better than a mutex, it is important to understand its limitations. If we compare the throughput of this MPMC queue to the throughput of our SPSC queue from the previous lesson, the SPSC queue will win every time.

In our SPSC queue, the producer owned the Tail cache line exclusively. In this MPMC queue, 8 producers are fighting for ownership of the Tail cache line to execute their CAS loops. This hardware contention causes cache ping-ponging on the silicon bus, wasting CPU cycles.

Furthermore, "optimistic spinning" (like our while (!LFQ.Push(i)) loop) burns 100% of the CPU core's power while it waits.

Because of this, architectural pragmatism is required. If you need to route data from 4 threads to 1 thread, it is often faster to create 4 independent SPSC queues than to force them all into 1 MPMC queue. Reserve the heavy MPMC architecture for truly many-to-many communication networks that must share the same queue, like global job schedulers.

Here is the complete implementation of our bounded lock-free MpmcQueue:

In this lesson, we tackled the immense complexity of synchronizing many-to-many communication channels without OS locks.

• The Index Claim: We used atomic compare_exchange_weak() (CAS) loops to safely allow multiple threads to contend for and claim indices from the Head and Tail counters.
• The Payload Race: We discovered that simply claiming an index is insufficient, as the thread hasn't actually written its data yet. We solved this using Sequence Tickets to synchronize the physical payload transfer per slot.
• Generational Wrap-Around: We allowed our Head and Tail integers to grow toward infinity, creating a natural generation tag that seamlessly manages the ring buffer's wrap-around logic.
• Padded Cells: We bundled the payload and the atomic ticket into a Cell struct, and aligned it to the CPU's cache line size to prevent false sharing and bus thrashing between adjacent array slots.
• Memory Fences: We optimized the memory order. The producer relies on an acquire fence to check the ticket, and a release fence to publish the payload. The consumer relies on the exact opposite to safely consume the data.