lib/wayland/src/stream/inbox.zig
daab053ee43316e1809a84551d573ddd1e5bf3d2
1 const std = @import("std");
2 const sys = @import("sys");
3 const wayland = @import("../root.zig");
4 const storage = @import("storage.zig");
5
6 const wire = wayland.wire;
7
8 pub const Frame = struct {
9 header: wire.Header,
10 bytes: []const u8,
11 payload: []const u8,
12 };
13
14 pub const Inbox = struct {
15 session_allocator: std.mem.Allocator,
16 bytes: []u8,
17 descriptors: []sys.fd.Descriptor,
18 byte_count: usize = 0,
19 descriptor_count: usize = 0,
20 byte_offset: usize = 0,
21 descriptor_offset: usize = 0,
22 byte_capacity_rejection_count: u64 = 0,
23 descriptor_capacity_rejection_count: u64 = 0,
24
25 pub fn init(
26 session_allocator: std.mem.Allocator,
27 capacity: storage.Capacity,
28 ) std.mem.Allocator.Error!Inbox {
29 const bytes = try session_allocator.alloc(u8, capacity.inbound_byte_count);
30 errdefer session_allocator.free(bytes);
31 return .{
32 .session_allocator = session_allocator,
33 .bytes = bytes,
34 .descriptors = try session_allocator.alloc(
35 sys.fd.Descriptor,
36 capacity.inbound_descriptor_count,
37 ),
38 };
39 }
40
41 pub fn deinit(self: *Inbox) void {
42 self.assertValid();
43 closeAll(self.descriptors[self.descriptor_offset..self.descriptor_count]);
44 self.session_allocator.free(self.bytes);
45 if (self.descriptors.len != 0) self.session_allocator.free(self.descriptors);
46 self.* = undefined;
47 }
48
49 pub fn appendOwned(
50 self: *Inbox,
51 bytes: []const u8,
52 descriptors: []const sys.fd.Descriptor,
53 ) storage.StorageError!void {
54 self.compact();
55 var failure: ?storage.StorageError = null;
56 if (bytes.len > self.bytes.len - self.byte_count) {
57 self.byte_capacity_rejection_count +|= 1;
58 failure = error.InboundByteCapacityExceeded;
59 }
60 if (descriptors.len > self.descriptors.len - self.descriptor_count) {
61 self.descriptor_capacity_rejection_count +|= 1;
62 if (failure == null) failure = error.InboundDescriptorCapacityExceeded;
63 }
64 if (failure) |err| return err;
65 @memcpy(self.bytes[self.byte_count..][0..bytes.len], bytes);
66 @memcpy(
67 self.descriptors[self.descriptor_count..][0..descriptors.len],
68 descriptors,
69 );
70 self.byte_count += bytes.len;
71 self.descriptor_count += descriptors.len;
72 self.assertValid();
73 }
74
75 pub fn prepareReceive(
76 self: *Inbox,
77 maximum_byte_count: usize,
78 ) storage.StorageError![]u8 {
79 std.debug.assert(maximum_byte_count > 0);
80 self.compact();
81 const available = self.bytes.len - self.byte_count;
82 if (available == 0) {
83 self.byte_capacity_rejection_count +|= 1;
84 return error.InboundByteCapacityExceeded;
85 }
86 return self.bytes[self.byte_count..][0..@min(available, maximum_byte_count)];
87 }
88
89 pub fn commitReceived(
90 self: *Inbox,
91 byte_count: usize,
92 descriptors: []const sys.fd.Descriptor,
93 ) storage.StorageError!void {
94 std.debug.assert(byte_count > 0);
95 std.debug.assert(byte_count <= self.bytes.len - self.byte_count);
96 if (descriptors.len > self.descriptors.len - self.descriptor_count) {
97 self.descriptor_capacity_rejection_count +|= 1;
98 return error.InboundDescriptorCapacityExceeded;
99 }
100 @memcpy(
101 self.descriptors[self.descriptor_count..][0..descriptors.len],
102 descriptors,
103 );
104 self.byte_count += byte_count;
105 self.descriptor_count += descriptors.len;
106 self.assertValid();
107 }
108
109 pub fn peek(self: *const Inbox) wire.Error!?Frame {
110 self.assertValid();
111 const available = self.bytes[self.byte_offset..self.byte_count];
112 if (available.len < wire.header_size) return null;
113 const header = try wire.decode(available[0..wire.header_size]);
114 if (available.len < header.size) return null;
115 return .{
116 .header = header,
117 .bytes = available[0..header.size],
118 .payload = available[wire.header_size..header.size],
119 };
120 }
121
122 pub fn consume(self: *Inbox) (wire.Error || error{IncompleteMessage})!void {
123 self.assertValid();
124 const frame = (try self.peek()) orelse return error.IncompleteMessage;
125 self.byte_offset += frame.header.size;
126 if (self.byte_offset == self.byte_count) {
127 self.byte_count = 0;
128 self.byte_offset = 0;
129 }
130 self.assertValid();
131 }
132
133 pub fn takeDescriptor(self: *Inbox) ?sys.fd.Descriptor {
134 self.assertValid();
135 if (self.descriptor_offset == self.descriptor_count) return null;
136 const descriptor = self.descriptors[self.descriptor_offset];
137 self.descriptor_offset += 1;
138 if (self.descriptor_offset == self.descriptor_count) {
139 self.descriptor_count = 0;
140 self.descriptor_offset = 0;
141 }
142 self.assertValid();
143 return descriptor;
144 }
145
146 pub fn peekDescriptors(self: *const Inbox, count: usize) ?[]const sys.fd.Descriptor {
147 if (count > self.queuedDescriptorCount()) return null;
148 return self.descriptors[self.descriptor_offset..self.descriptor_count][0..count];
149 }
150
151 pub fn queuedByteCount(self: *const Inbox) usize {
152 return self.byte_count - self.byte_offset;
153 }
154
155 pub fn queuedDescriptorCount(self: *const Inbox) usize {
156 return self.descriptor_count - self.descriptor_offset;
157 }
158
159 pub fn status(self: *const Inbox) storage.Status {
160 return .{
161 .inbound_byte_capacity_rejection_count = self.byte_capacity_rejection_count,
162 .inbound_descriptor_capacity_rejection_count = self.descriptor_capacity_rejection_count,
163 };
164 }
165
166 fn compact(self: *Inbox) void {
167 storage.compactSlice(u8, self.bytes, &self.byte_count, &self.byte_offset);
168 storage.compactSlice(
169 sys.fd.Descriptor,
170 self.descriptors,
171 &self.descriptor_count,
172 &self.descriptor_offset,
173 );
174 }
175
176 fn assertValid(self: *const Inbox) void {
177 std.debug.assert(self.byte_offset <= self.byte_count);
178 std.debug.assert(self.byte_count <= self.bytes.len);
179 std.debug.assert(self.descriptor_offset <= self.descriptor_count);
180 std.debug.assert(self.descriptor_count <= self.descriptors.len);
181 }
182 };
183
184 fn closeAll(descriptors: []const sys.fd.Descriptor) void {
185 for (descriptors) |descriptor| sys.fd.close(descriptor);
186 }
187
188 test "inbox preserves fragmented and coalesced message boundaries" {
189 const first_header = try wire.Header.init(2, 7, 4);
190 const second_header = try wire.Header.init(3, 9, 0);
191 var first: [12]u8 = undefined;
192 var second: [8]u8 = undefined;
193 try first_header.encode(first[0..8]);
194 @memcpy(first[8..], &[_]u8{ 1, 2, 3, 4 });
195 try second_header.encode(&second);
196
197 const capacity = try storage.Capacity.derive(.{ .inbound_byte_count = 20 });
198 var inbox = try Inbox.init(std.testing.allocator, capacity);
199 defer inbox.deinit();
200 try inbox.appendOwned(first[0..3], &.{});
201 try std.testing.expect((try inbox.peek()) == null);
202 try inbox.appendOwned(first[3..], &.{});
203 try inbox.appendOwned(second[0..2], &.{});
204 const first_frame = (try inbox.peek()).?;
205 try std.testing.expectEqual(first_header, first_frame.header);
206 try std.testing.expectEqualSlices(u8, &.{ 1, 2, 3, 4 }, first_frame.payload);
207 try inbox.consume();
208 try std.testing.expect((try inbox.peek()) == null);
209 try inbox.appendOwned(second[2..], &.{});
210 try std.testing.expectEqual(second_header, (try inbox.peek()).?.header);
211 try inbox.consume();
212 try std.testing.expectEqual(@as(usize, 0), inbox.queuedByteCount());
213 }
214
215 test "descriptor lookahead does not transfer ownership" {
216 const pipe = try sys.fd.pipeWithOptions(.{ .close_on_exec = true });
217 defer sys.fd.close(pipe[0]);
218
219 const capacity = try storage.Capacity.derive(.{ .inbound_descriptor_count = 1 });
220 var inbox = try Inbox.init(std.testing.allocator, capacity);
221 defer inbox.deinit();
222 try inbox.appendOwned(&.{}, &.{pipe[1]});
223 try std.testing.expect(inbox.peekDescriptors(2) == null);
224 try std.testing.expectEqualSlices(sys.fd.Descriptor, &.{pipe[1]}, inbox.peekDescriptors(1).?);
225 try std.testing.expectEqual(pipe[1], inbox.takeDescriptor().?);
226 sys.fd.close(pipe[1]);
227 }
228
229 test "inbox max plus one preserves retained bytes and saturates status" {
230 const capacity = try storage.Capacity.derive(.{ .inbound_byte_count = 8 });
231 var inbox = try Inbox.init(std.testing.allocator, capacity);
232 defer inbox.deinit();
233
234 try inbox.appendOwned(&@as([8]u8, @splat(1)), &.{});
235 try std.testing.expectError(
236 error.InboundByteCapacityExceeded,
237 inbox.appendOwned(&.{2}, &.{}),
238 );
239 try std.testing.expectEqual(@as(usize, 8), inbox.queuedByteCount());
240 try std.testing.expectEqualSlices(u8, &@as([8]u8, @splat(1)), inbox.bytes[0..inbox.byte_count]);
241 try std.testing.expectEqual(@as(u64, 1), inbox.status().inbound_byte_capacity_rejection_count);
242
243 inbox.byte_capacity_rejection_count = std.math.maxInt(u64);
244 try std.testing.expectError(
245 error.InboundByteCapacityExceeded,
246 inbox.prepareReceive(1),
247 );
248 try std.testing.expectEqual(
249 std.math.maxInt(u64),
250 inbox.status().inbound_byte_capacity_rejection_count,
251 );
252 }
253
254 test "inbox max plus one preserves retained descriptor ownership" {
255 const first_pipe = try sys.fd.pipeWithOptions(.{ .close_on_exec = true });
256 defer sys.fd.close(first_pipe[0]);
257 const second_pipe = try sys.fd.pipeWithOptions(.{ .close_on_exec = true });
258 defer sys.fd.close(second_pipe[0]);
259 defer sys.fd.close(second_pipe[1]);
260
261 const capacity = try storage.Capacity.derive(.{ .inbound_descriptor_count = 1 });
262 var inbox = try Inbox.init(std.testing.allocator, capacity);
263 defer inbox.deinit();
264 try inbox.appendOwned(&.{}, &.{first_pipe[1]});
265 try std.testing.expectError(
266 error.InboundDescriptorCapacityExceeded,
267 inbox.appendOwned(&.{}, &.{second_pipe[1]}),
268 );
269 try std.testing.expectEqual(@as(usize, 1), inbox.queuedDescriptorCount());
270 try std.testing.expectEqual(first_pipe[1], inbox.takeDescriptor().?);
271 sys.fd.close(first_pipe[1]);
272 try std.testing.expectEqual(
273 @as(u64, 1),
274 inbox.status().inbound_descriptor_capacity_rejection_count,
275 );
276 }