added EA's implementation of MPMC queue

lib/p2/src/structures/concurrent-queue.zig

@@ -18,131 +18,114 @@ pub fn ConcurrentQueueU(comptime T: type) type {
     return ConcurrentQueueUnmanagedAdvanced(T, .{});
 }
 
-pub fn ConcurrentQueueNoAssert(comptime T: type) type {
+pub fn ConcurrentQueueAssert(comptime T: type) type {
-    return ConcurrentQueueUnmanagedAdvanced(T, .{ .allowAsserts = false });
+    return ConcurrentQueueUnmanagedAdvanced(T, .{ .allowAsserts = true });
 }
 
 // lock-free concurrent queue, fixed capacity,
 // will never resize.
+//
+// new version based on
+// https://github.com/rigtorp/MPMCQueue
+
+pub const ConcurrentStatus = packed struct(usize) {
+    alive: bool = false,
+    generation: u63 = 0,
+};
+
 pub fn ConcurrentQueueUnmanagedAdvanced(comptime T: type, comptime opts: struct {
-    allowAsserts: bool = true,
+    allowAsserts: bool = false,
+    debug: bool = false,
 }) type {
     return struct {
-        data: []T,
+        data: []align(64) T align(64),
-        status: []Atomic(bool), // 1 = valid, 0 = invalid,
+        status: []align(64) Atomic(ConcurrentStatus) align(64), // 1 = valid, 0 = invalid,
-        head: Atomic(usize),
+        pushId: Atomic(usize) align(64),
-        tail: Atomic(usize),
+        popId: Atomic(usize) align(64),
 
         // tail points to next free slot
         // head points to the next one to pop
         pub fn initCapacity(allocator: std.mem.Allocator, cap: usize) !@This() {
             const new = @This(){
-                .data = try allocator.alloc(T, cap + 1),
+                .data = try allocator.alignedAlloc(T, .@"64", cap),
-                .status = try allocator.alloc(Atomic(bool), cap + 1),
+                .status = try allocator.alignedAlloc(Atomic(ConcurrentStatus), .@"64", cap),
-                .head = Atomic(usize).init(0),
+                .pushId = Atomic(usize).init(0),
-                .tail = Atomic(usize).init(1),
+                .popId = Atomic(usize).init(0),
             };
 
             for (new.status) |*s| {
-                s.* = Atomic(bool).init(false);
+                s.* = Atomic(ConcurrentStatus).init(std.mem.zeroes(ConcurrentStatus));
             }
 
             return new;
         }
 
-        // there is 100% an ABA problem going on here...
-
         pub fn push(self: *@This(), value: T) !void {
-            // seek the next unread bit and reserve it
+            const pushId = self.pushId.fetchAdd(1, .acq_rel);
-            const start: usize = self.tail.load(.acquire);
-            var writeIndex: usize = start;
-            while (self.status[writeIndex].cmpxchgStrong(false, true, .seq_cst, .acquire) != null) {
-                writeIndex = (writeIndex + 1) % self.data.len;
-                if (writeIndex == self.head.load(.seq_cst)) {
-                    return ConcurrentQueueError.QueueIsFull;
-                }
-            }
 
-            // writeIndex = index of newly acquired slot acquired;
+            const slot = @mod(pushId, self.data.len);
-            self.data[writeIndex] = value;
 
-            writeIndex = (writeIndex + 1) % self.data.len;
+            const newStatus = ConcurrentStatus{
-            if (writeIndex == self.head.load(.seq_cst)) {
+                .generation = @intCast(@divTrunc(pushId, self.data.len)),
-                return ConcurrentQueueError.QueueIsFull;
+                .alive = true,
-            }
+            };
-            var expected: usize = start;
 
-            // spin and resolve contention
+            if (opts.debug) std.log.warn("pushing {d}", .{pushId});
-            while (self.tail.cmpxchgStrong(expected, writeIndex, .seq_cst, .acquire)) |tail| {
+            while (true) {
-                // this is ok, update our expected value an try to CAS again
+                const status = self.status[slot].load(.acquire);
-                if ((expected > tail) or ((expected < tail) and expected < self.head.load(.acquire))) {
+                if (status.generation == newStatus.generation and status.alive == false) {
-                    expected = tail;
-                } else if ((tail > expected) or ((tail < expected) and tail < self.head.load(.acquire))) {
-                    // something else reserved a slot past ours, we can expect them to fixup the value
                     break;
                 }
+                std.atomic.spinLoopHint();
             }
+
+            if (opts.debug) std.log.warn("pushed {d}", .{pushId});
+            self.status[slot].store(newStatus, .release);
+            self.data[slot] = value;
         }
 
         // pop the value from the queue, moves the head forward
         pub fn pop(self: *@This()) ?T { // things are empty
-            var expected = self.head.load(.acquire);
+            if (self.count() == 0)
-            var popIndex = expected;
-            var newHead = (popIndex + 1) % self.data.len;
-            if (newHead == self.tail.load(.acquire)) {
                 return null;
-            }
 
-            while (self.status[popIndex].cmpxchgStrong(true, false, .seq_cst, .acquire) != null) {
+            const popId = self.popId.fetchAdd(1, .acq_rel);
-                newHead = (popIndex + 1) % self.data.len;
-                popIndex = newHead;
 
-                if (newHead == self.tail.load(.acquire)) {
+            const slot = @mod(popId, self.data.len);
-                    return null;
+            const expectedStatus = ConcurrentStatus{
+                .generation = @intCast(@divTrunc(popId, self.data.len)),
+                .alive = true,
+            };
+
+            if (opts.debug) std.log.warn("popping {d}", .{popId});
+            while (true) {
+                const status = self.status[slot].load(.acquire);
+                if (status.generation == expectedStatus.generation and status.alive) {
+                    break;
                 }
+                std.atomic.spinLoopHint();
             }
+            if (opts.debug) std.log.warn("popped {d}", .{popId});
 
-            // newHead = (popIndex + 1) % self.data.len;
+            self.status[slot].store(.{ .alive = false, .generation = expectedStatus.generation +% 1 }, .release);
-            // spin and resolve
+            return self.data[slot];
-            while (self.head.cmpxchgStrong(expected, newHead, .seq_cst, .acquire)) |head| {
-                // we failed to increment the head
-                const tail = self.tail.load(.acquire);
-
-                // something else has already incremented the head past our reservation
-                if (head > newHead or (head < newHead and head < tail)) {
-                    return self.data[popIndex];
-                }
-
-                // our new head is past what the current head is, fixup the value
-                if (head < newHead or (newHead < head and newHead < tail)) {
-                    expected = head;
-                }
-            }
-
-            return self.data[popIndex];
         }
 
         pub fn count(self: @This()) usize {
-            const head = self.head.load(.acquire);
+            const head = self.pushId.load(.acquire);
-            const tail = self.tail.load(.acquire);
+            const tail = self.popId.load(.acquire);
 
+            if (opts.debug)
+                std.log.warn("count {d}", .{head - tail});
             if (opts.allowAsserts) {
-                asserts(tail != head, "tail == head in concurrent queue, this shouldnt ever happen", .{}, "concurrent queue assert");
+                asserts(tail <= head, "tail > head in concurrent queue, this shouldnt ever happen", .{}, "concurrent queue assert");
             }
 
-            if (tail > head) {
+            return head - tail;
-                return tail - head - 1;
-            }
-
-            if (tail < head) {
-                return (self.data.len - head) + tail;
-            }
-
-            return 0;
         }
 
         pub fn capacity(self: @This()) usize {
-            return self.data.len - 1;
+            return self.data.len;
         }
 
         pub fn deinit(self: *@This(), allocator: std.mem.Allocator) void {
@@ -159,7 +142,7 @@ test "concurrent queue basic correctness test" {
 
     const allocator = std.testing.allocator;
 
-    var y = try ConcurrentQueueU(Info).initCapacity(allocator, 420);
+    var y = try ConcurrentQueueUnmanagedAdvanced(Info, .{ .allowAsserts = true, .debug = true }).initCapacity(allocator, 420);
     defer y.deinit(allocator);
 
     try y.push(.{});
@@ -192,28 +175,30 @@ test "concurrent queue basic correctness test" {
     try x.push(.{ .x = 9 });
     try x.push(.{ .x = 10 });
 
-    const maybeError = x.push(.{ .x = 11 });
+    // const maybeError = x.push(.{ .x = 11 });
 
-    try utils.assertf(maybeError == ConcurrentQueueError.QueueIsFull, "Expected queue to have an error", .{});
+    // try utils.assertf(maybeError == ConcurrentQueueError.QueueIsFull, "Expected queue to have an error", .{});
 }
 
 test "concurrent queue multiple producer single consumer" {
     // 1. create multiple threads
-    const threadCount = 12;
+    const threadCount = 24;
 
     const Payload = struct {
         x: i64 = 0,
+        arb: [4096 * 16]u8 = undefined,
     };
 
+    const QueueType = ConcurrentQueueUnmanagedAdvanced(Payload, .{ .debug = false, .allowAsserts = true });
+
     const Wrap = struct {
-        pub fn threadFunc(queueRef: *ConcurrentQueueU(Payload), id: i64, exitSignal: *Atomic(bool), pushedCountResults: *Atomic(i64)) void {
+        pub fn threadFunc(queueRef: *QueueType, id: i64, exitSignal: *Atomic(bool), pushedCountResults: *Atomic(i64)) void {
             var pushedCount: i64 = 0;
-            while (!exitSignal.load(.acquire)) {
+            while (!exitSignal.load(.monotonic)) {
                 queueRef.push(.{
                     .x = id + pushedCount,
                 }) catch unreachable;
                 pushedCount += 1;
-                std.Thread.sleep(1000 * 1000 * 100);
             }
 
             _ = pushedCountResults.fetchAdd(pushedCount, .seq_cst);
@@ -222,12 +207,11 @@ test "concurrent queue multiple producer single consumer" {
 
     var threads: [threadCount]std.Thread = undefined;
 
-    var testQueue = try ConcurrentQueueU(Payload).initCapacity(std.testing.allocator, 4096);
+    var testQueue = try QueueType.initCapacity(std.testing.allocator, 4096 * 4);
     defer testQueue.deinit(std.testing.allocator);
 
     var exitSignalAtomic = Atomic(bool).init(false);
     var pushedCountResults = Atomic(i64).init(0);
-    std.debug.print("\n\n", .{});
 
     for (0..threadCount) |i| {
         threads[i] = try std.Thread.spawn(.{}, Wrap.threadFunc, .{ &testQueue, @as(i64, @intCast(i * 10000)), &exitSignalAtomic, &pushedCountResults });
@@ -237,7 +221,8 @@ test "concurrent queue multiple producer single consumer" {
     var oldTime: f64 = test_getTime();
 
     // 10 second test, 5 seconds of input, 5 seconds of drain
-    var timeLeft: f64 = 10.0;
+    const startTime: f64 = 2.2;
+    var timeLeft: f64 = startTime;
 
     var poppedCount: i64 = 0;
 
@@ -247,7 +232,7 @@ test "concurrent queue multiple producer single consumer" {
         const newTime = test_getTime();
         const deltaTime = newTime - oldTime;
 
-        if (timeLeft - deltaTime < 5.0 and !signaled) {
+        if (timeLeft - deltaTime < 0.0 and !signaled) {
             signaled = true;
             exitSignalAtomic.store(true, .seq_cst);
         }
@@ -258,14 +243,29 @@ test "concurrent queue multiple producer single consumer" {
             _ = x;
             poppedCount += 1;
         }
-        std.Thread.sleep(1000 * 1000);
+        std.atomic.spinLoopHint();
     }
 
     for (0..threadCount) |i| {
         threads[i].join();
     }
 
-    try utils.assertf(poppedCount == pushedCountResults.load(.seq_cst), "mismatched, we popped {d} records while the workers pushed {d}", .{ poppedCount, pushedCountResults.load(.seq_cst) });
+    std.debug.print("popped {d} entries in {d} seconds payloadSize: {d} dataRate: {d:.3} MiB/s time/event {d:.3} us \n", .{
+        poppedCount,
+        startTime,
+        @sizeOf(Payload),
+        @as(f64, @floatFromInt(@as(usize, @intCast(poppedCount)) * @sizeOf(Payload))) / startTime / 1024 / 1024,
+        startTime / @as(f64, @floatFromInt(@as(usize, @intCast(poppedCount)))) * 1000 * 1000,
+    });
+
+    try utils.assertf(
+        poppedCount == pushedCountResults.load(.seq_cst),
+        "mismatched, we popped {d} records while the workers pushed {d}",
+        .{
+            poppedCount,
+            pushedCountResults.load(.seq_cst),
+        },
+    );
 }
 
 fn test_getTime() f64 {