lib/machine/src/explore/distributed/queue.zig

daab053ee43316e1809a84551d573ddd1e5bf3d2

  1 const std = @import("std");
  2 const types = @import("types.zig");
  3 
  4 /// Stores sent messages awaiting delivery for one position of the three-node
  5 /// protocol: at most 12, twice the depth of the choice tree. The three-node
  6 /// commit protocol under test (*workload*) pushes each message it sends here
  7 /// and takes it out at delivery. `push` appends one message and returns
  8 /// `MessageCapacityExceeded` when the queue is full. Messages stay in the order
  9 /// sent: `find` returns the index of the earliest message for one destination,
 10 /// and `take` removes it and keeps the others in order. `cancel` drops every
 11 /// message of one kind from one sender. The workload calls `cancel` to withdraw
 12 /// the coordinator's old proposals when the coordinator, node 0, opens a new
 13 /// round, a new attempt to get a value accepted.
 14 pub const Queue = struct {
 15     storage: [types.message_capacity]types.Message = undefined,
 16     count: u8 = 0,
 17 
 18     pub const capacity: u8 = types.message_capacity;
 19 
 20     pub fn push(self: *Queue, message: types.Message) types.Error!void {
 21         self.assertValid();
 22         std.debug.assert(message.sender < types.node_count);
 23         std.debug.assert(message.destination < types.node_count);
 24         if (self.count == capacity) return error.MessageCapacityExceeded;
 25         self.storage[self.count] = message;
 26         self.count += 1;
 27         self.assertValid();
 28     }
 29 
 30     pub fn find(self: *const Queue, destination: u8) ?u8 {
 31         self.assertValid();
 32         std.debug.assert(destination < types.node_count);
 33         for (self.messages(), 0..) |message, index| {
 34             if (message.destination == destination) return @intCast(index);
 35         }
 36         return null;
 37     }
 38 
 39     pub fn take(self: *Queue, index: u8) types.Message {
 40         self.assertValid();
 41         std.debug.assert(index < self.count);
 42         const message = self.storage[index];
 43         const remaining = self.count - index - 1;
 44         std.mem.copyForwards(
 45             types.Message,
 46             self.storage[index .. index + remaining],
 47             self.storage[index + 1 .. self.count],
 48         );
 49         self.count -= 1;
 50         self.assertValid();
 51         return message;
 52     }
 53 
 54     pub fn cancel(self: *Queue, sender: u8, kind: types.MessageKind) void {
 55         self.assertValid();
 56         std.debug.assert(sender < types.node_count);
 57         var kept: u8 = 0;
 58         var index: u8 = 0;
 59         while (index < self.count) : (index += 1) {
 60             const message = self.storage[index];
 61             if (message.sender == sender and message.kind == kind) continue;
 62             std.debug.assert(kept <= index);
 63             self.storage[kept] = message;
 64             kept += 1;
 65         }
 66         self.count = kept;
 67         self.assertValid();
 68     }
 69 
 70     pub fn messages(self: *const Queue) []const types.Message {
 71         std.debug.assert(self.count <= capacity);
 72         return self.storage[0..self.count];
 73     }
 74 
 75     fn assertValid(self: *const Queue) void {
 76         std.debug.assert(self.count <= capacity);
 77         std.debug.assert(self.count <= self.storage.len);
 78     }
 79 };
 80 
 81 test "distributed queue keeps first-in order and refuses past capacity" {
 82     var queue: Queue = .{};
 83     for (0..Queue.capacity) |index| {
 84         try queue.push(.{
 85             .kind = if (index % 2 == 0) .propose else .ack,
 86             .round = @intCast(index + 1),
 87             .value = index,
 88             .sender = 0,
 89             .destination = @intCast(index % types.node_count),
 90         });
 91     }
 92     try std.testing.expectError(error.MessageCapacityExceeded, queue.push(.{
 93         .kind = .ack,
 94         .round = 1,
 95         .value = 0,
 96         .sender = 0,
 97         .destination = 0,
 98     }));
 99     const first = queue.find(1).?;
100     try std.testing.expectEqual(@as(u64, 1), queue.storage[first].value);
101     const taken = queue.take(first);
102     try std.testing.expectEqual(@as(u64, 1), taken.value);
103     try std.testing.expectEqual(Queue.capacity - 1, queue.count);
104     queue.cancel(0, .propose);
105     for (queue.messages()) |message| {
106         try std.testing.expectEqual(types.MessageKind.ack, message.kind);
107     }
108 }