lib/wayland/src/transport.zig
daab053ee43316e1809a84551d573ddd1e5bf3d2
1 const std = @import("std");
2 const sys = @import("sys");
3 const wayland = @import("root.zig");
4
5 pub const ReceiveStatus = enum {
6 data,
7 pending,
8 closed,
9 };
10
11 pub const Limits = wayland.stream.Limits;
12 pub const Capacity = wayland.stream.Capacity;
13 pub const CapacityError = wayland.stream.CapacityError;
14 pub const StorageError = wayland.stream.StorageError;
15 pub const Status = wayland.stream.Status;
16 pub const default_byte_count = wayland.stream.default_byte_count;
17 pub const default_descriptor_count = wayland.stream.default_descriptor_count;
18 const test_capacity = Capacity.derive(.{}) catch unreachable;
19
20 pub const Storage = struct {
21 inbox: wayland.stream.Inbox,
22 outbox: wayland.stream.Outbox,
23
24 pub fn init(
25 session_allocator: std.mem.Allocator,
26 capacity: Capacity,
27 ) std.mem.Allocator.Error!Storage {
28 var inbox = try wayland.stream.Inbox.init(session_allocator, capacity);
29 errdefer inbox.deinit();
30 return .{
31 .inbox = inbox,
32 .outbox = try wayland.stream.Outbox.init(session_allocator, capacity),
33 };
34 }
35
36 pub fn deinit(self: *Storage) void {
37 self.inbox.deinit();
38 self.outbox.deinit();
39 self.* = undefined;
40 }
41
42 pub fn attach(
43 self: *Storage,
44 owned_descriptor: sys.fd.Descriptor,
45 ) sys.fd.FlagError!Transport {
46 errdefer sys.fd.close(owned_descriptor);
47 try sys.fd.setCloseOnExec(owned_descriptor);
48 try sys.fd.setNonBlocking(owned_descriptor);
49 const transport: Transport = .{
50 .descriptor = owned_descriptor,
51 .inbox = self.inbox,
52 .outbox = self.outbox,
53 };
54 self.* = undefined;
55 return transport;
56 }
57 };
58
59 pub const Transport = struct {
60 descriptor: sys.fd.Descriptor,
61 inbox: wayland.stream.Inbox,
62 outbox: wayland.stream.Outbox,
63
64 pub fn initOwned(
65 session_allocator: std.mem.Allocator,
66 descriptor: sys.fd.Descriptor,
67 capacity: Capacity,
68 ) (sys.fd.FlagError || std.mem.Allocator.Error)!Transport {
69 var storage = Storage.init(session_allocator, capacity) catch |err| {
70 sys.fd.close(descriptor);
71 return err;
72 };
73 errdefer storage.deinit();
74 return storage.attach(descriptor);
75 }
76
77 pub fn deinit(self: *Transport) void {
78 self.inbox.deinit();
79 self.outbox.deinit();
80 sys.fd.close(self.descriptor);
81 self.* = undefined;
82 }
83
84 pub fn queueOwned(
85 self: *Transport,
86 object_id: u32,
87 opcode: u16,
88 payload: []const u8,
89 descriptors: []const sys.fd.Descriptor,
90 ) !void {
91 try self.outbox.queueOwned(object_id, opcode, payload, descriptors);
92 }
93
94 pub fn flush(self: *Transport) !wayland.stream.FlushStatus {
95 return self.outbox.flush(self.descriptor);
96 }
97
98 pub fn receive(self: *Transport) !ReceiveStatus {
99 const bytes = try self.inbox.prepareReceive(16 * 1024);
100 var descriptors: [sys.ancillary.maximum_descriptors]sys.fd.Descriptor = undefined;
101 const received = sys.ancillary.receive(self.descriptor, bytes, &descriptors) catch |err| switch (err) {
102 error.WouldBlock => return .pending,
103 else => return err,
104 };
105 if (received.byte_count == 0) {
106 std.debug.assert(received.descriptor_count == 0);
107 return .closed;
108 }
109 self.inbox.commitReceived(
110 received.byte_count,
111 descriptors[0..received.descriptor_count],
112 ) catch |err| {
113 closeAll(descriptors[0..received.descriptor_count]);
114 return err;
115 };
116 return .data;
117 }
118
119 pub fn status(self: *const Transport) Status {
120 const inbound = self.inbox.status();
121 const outbound = self.outbox.status();
122 return .{
123 .inbound_byte_capacity_rejection_count = inbound.inbound_byte_capacity_rejection_count,
124 .inbound_descriptor_capacity_rejection_count = inbound.inbound_descriptor_capacity_rejection_count,
125 .outbound_byte_capacity_rejection_count = outbound.outbound_byte_capacity_rejection_count,
126 .outbound_descriptor_capacity_rejection_count = outbound.outbound_descriptor_capacity_rejection_count,
127 };
128 }
129 };
130
131 fn closeAll(descriptors: []const sys.fd.Descriptor) void {
132 for (descriptors) |descriptor| sys.fd.close(descriptor);
133 }
134
135 test "transport storage acquires all four regions before attachment" {
136 const capacity = try Capacity.derive(.{
137 .inbound_byte_count = 8,
138 .inbound_descriptor_count = 1,
139 .outbound_byte_count = 8,
140 .outbound_descriptor_count = 1,
141 });
142 for (0..4) |fail_index| {
143 var failing = std.testing.FailingAllocator.init(std.testing.allocator, .{
144 .fail_index = fail_index,
145 });
146 try std.testing.expectError(
147 error.OutOfMemory,
148 Storage.init(failing.allocator(), capacity),
149 );
150 }
151 var failing = std.testing.FailingAllocator.init(std.testing.allocator, .{
152 .fail_index = 4,
153 });
154 var storage = try Storage.init(failing.allocator(), capacity);
155 defer storage.deinit();
156 try std.testing.expectEqual(@as(usize, 4), failing.allocations);
157 try std.testing.expectEqual(@as(usize, 8), storage.inbox.bytes.len);
158 try std.testing.expectEqual(@as(usize, 1), storage.inbox.descriptors.len);
159 try std.testing.expectEqual(@as(usize, 8), storage.outbox.bytes.len);
160 try std.testing.expectEqual(@as(usize, 1), storage.outbox.descriptors.len);
161 }
162
163 test "owned descriptor closes when initial storage acquisition fails" {
164 if (comptime @import("builtin").os.tag != .linux) return error.SkipZigTest;
165 const sockets = sys.fd.socketPairUnixStream(.{ .close_on_exec = true }) catch |err| switch (err) {
166 error.UnsupportedPlatform => return error.SkipZigTest,
167 else => return err,
168 };
169 defer sys.fd.close(sockets[1]);
170 var failing = std.testing.FailingAllocator.init(std.testing.allocator, .{
171 .fail_index = 0,
172 });
173 try std.testing.expectError(
174 error.OutOfMemory,
175 Transport.initOwned(failing.allocator(), sockets[0], test_capacity),
176 );
177 try std.testing.expect(!sys.fd.isOpen(sockets[0]));
178 }
179
180 test "transport preserves descriptor order over a Unix stream" {
181 const sockets = sys.fd.socketPairUnixStream(.{ .close_on_exec = true }) catch |err| switch (err) {
182 error.UnsupportedPlatform => return error.SkipZigTest,
183 else => return err,
184 };
185 var sender = Transport.initOwned(std.testing.allocator, sockets[0], test_capacity) catch |err| {
186 sys.fd.close(sockets[1]);
187 return err;
188 };
189 defer sender.deinit();
190 var receiver = try Transport.initOwned(std.testing.allocator, sockets[1], test_capacity);
191 defer receiver.deinit();
192
193 const first_pipe = try sys.fd.pipeWithOptions(.{ .close_on_exec = true });
194 defer sys.fd.close(first_pipe[0]);
195 const second_pipe = try sys.fd.pipeWithOptions(.{ .close_on_exec = true });
196 defer sys.fd.close(second_pipe[0]);
197 var queued = false;
198 defer if (!queued) {
199 sys.fd.close(first_pipe[1]);
200 sys.fd.close(second_pipe[1]);
201 };
202
203 try sender.queueOwned(2, 5, &.{ 9, 8, 7, 6 }, &.{ first_pipe[1], second_pipe[1] });
204 queued = true;
205 try std.testing.expectEqual(wayland.stream.FlushStatus.drained, try sender.flush());
206 try std.testing.expectEqual(ReceiveStatus.data, try receiver.receive());
207
208 const frame = (try receiver.inbox.peek()).?;
209 try std.testing.expectEqual(@as(u32, 2), frame.header.object_id);
210 try std.testing.expectEqual(@as(u16, 5), frame.header.opcode);
211 try std.testing.expectEqualSlices(u8, &.{ 9, 8, 7, 6 }, frame.payload);
212
213 const received_first = receiver.inbox.takeDescriptor().?;
214 defer sys.fd.close(received_first);
215 const received_second = receiver.inbox.takeDescriptor().?;
216 defer sys.fd.close(received_second);
217 try std.testing.expect(receiver.inbox.takeDescriptor() == null);
218 try std.testing.expectEqual(@as(usize, 1), try sys.fd.write(received_first, "a"));
219 try std.testing.expectEqual(@as(usize, 1), try sys.fd.write(received_second, "b"));
220
221 var byte: [1]u8 = undefined;
222 try std.testing.expectEqual(@as(usize, 1), try sys.fd.read(first_pipe[0], &byte));
223 try std.testing.expectEqual(@as(u8, 'a'), byte[0]);
224 try std.testing.expectEqual(@as(usize, 1), try sys.fd.read(second_pipe[0], &byte));
225 try std.testing.expectEqual(@as(u8, 'b'), byte[0]);
226 try receiver.inbox.consume();
227 }
228
229 test "received descriptors queue independently of fragmented bytes" {
230 const sockets = sys.fd.socketPairUnixStream(.{ .close_on_exec = true }) catch |err| switch (err) {
231 error.UnsupportedPlatform => return error.SkipZigTest,
232 else => return err,
233 };
234 defer sys.fd.close(sockets[0]);
235 var receiver = Transport.initOwned(std.testing.allocator, sockets[1], test_capacity) catch |err| {
236 return err;
237 };
238 defer receiver.deinit();
239
240 const pipe = try sys.fd.pipeWithOptions(.{ .close_on_exec = true });
241 defer sys.fd.close(pipe[0]);
242 defer sys.fd.close(pipe[1]);
243 const header = try wayland.wire.Header.init(2, 1, 4);
244 var message: [12]u8 = undefined;
245 try header.encode(message[0..8]);
246 @memcpy(message[8..], &[_]u8{ 1, 2, 3, 4 });
247
248 try std.testing.expectEqual(@as(usize, 1), try sys.ancillary.send(sockets[0], message[0..1], &.{pipe[1]}));
249 try std.testing.expectEqual(ReceiveStatus.data, try receiver.receive());
250 try std.testing.expect((try receiver.inbox.peek()) == null);
251 try std.testing.expectEqual(@as(usize, 1), receiver.inbox.queuedDescriptorCount());
252
253 try std.testing.expectEqual(message.len - 1, try sys.fd.write(sockets[0], message[1..]));
254 try std.testing.expectEqual(ReceiveStatus.data, try receiver.receive());
255 try std.testing.expectEqual(header, (try receiver.inbox.peek()).?.header);
256 const received = receiver.inbox.takeDescriptor().?;
257 sys.fd.close(received);
258 }
259
260 test "inbound byte max plus one preserves the full retained prefix" {
261 const sockets = sys.fd.socketPairUnixStream(.{ .close_on_exec = true }) catch |err| switch (err) {
262 error.UnsupportedPlatform => return error.SkipZigTest,
263 else => return err,
264 };
265 var sender = Transport.initOwned(std.testing.allocator, sockets[0], test_capacity) catch |err| {
266 sys.fd.close(sockets[1]);
267 return err;
268 };
269 defer sender.deinit();
270 const receiver_capacity = try Capacity.derive(.{
271 .inbound_byte_count = wayland.wire.header_size,
272 });
273 var receiver = try Transport.initOwned(
274 std.testing.allocator,
275 sockets[1],
276 receiver_capacity,
277 );
278 defer receiver.deinit();
279
280 try sender.queueOwned(2, 1, &.{ 1, 2, 3, 4 }, &.{});
281 try std.testing.expectEqual(wayland.stream.FlushStatus.drained, try sender.flush());
282 try std.testing.expectEqual(ReceiveStatus.data, try receiver.receive());
283 try std.testing.expectEqual(
284 @as(usize, wayland.wire.header_size),
285 receiver.inbox.queuedByteCount(),
286 );
287 var retained: [wayland.wire.header_size]u8 = undefined;
288 @memcpy(&retained, receiver.inbox.bytes[0..receiver.inbox.byte_count]);
289 try std.testing.expectError(error.InboundByteCapacityExceeded, receiver.receive());
290 try std.testing.expectEqualSlices(
291 u8,
292 &retained,
293 receiver.inbox.bytes[0..receiver.inbox.byte_count],
294 );
295 try std.testing.expectEqual(
296 @as(u64, 1),
297 receiver.status().inbound_byte_capacity_rejection_count,
298 );
299 }
300
301 test "inbound descriptor max plus one closes the rejected foreign batch" {
302 const sockets = sys.fd.socketPairUnixStream(.{ .close_on_exec = true }) catch |err| switch (err) {
303 error.UnsupportedPlatform => return error.SkipZigTest,
304 else => return err,
305 };
306 const sender_capacity = try Capacity.derive(.{ .outbound_descriptor_count = 2 });
307 var sender = Transport.initOwned(std.testing.allocator, sockets[0], sender_capacity) catch |err| {
308 sys.fd.close(sockets[1]);
309 return err;
310 };
311 defer sender.deinit();
312 const receiver_capacity = try Capacity.derive(.{ .inbound_descriptor_count = 1 });
313 var receiver = try Transport.initOwned(
314 std.testing.allocator,
315 sockets[1],
316 receiver_capacity,
317 );
318 defer receiver.deinit();
319 const first_pipe = try sys.fd.pipeWithOptions(.{ .close_on_exec = true });
320 defer sys.fd.close(first_pipe[0]);
321 const second_pipe = try sys.fd.pipeWithOptions(.{ .close_on_exec = true });
322 defer sys.fd.close(second_pipe[0]);
323
324 try sender.queueOwned(2, 1, &.{}, &.{ first_pipe[1], second_pipe[1] });
325 try std.testing.expectEqual(wayland.stream.FlushStatus.drained, try sender.flush());
326 try std.testing.expectError(
327 error.InboundDescriptorCapacityExceeded,
328 receiver.receive(),
329 );
330 try std.testing.expectEqual(@as(usize, 0), receiver.inbox.queuedByteCount());
331 try std.testing.expectEqual(@as(usize, 0), receiver.inbox.queuedDescriptorCount());
332 try std.testing.expectEqual(
333 @as(u64, 1),
334 receiver.status().inbound_descriptor_capacity_rejection_count,
335 );
336 var byte: [1]u8 = undefined;
337 try std.testing.expectEqual(@as(usize, 0), try sys.fd.read(first_pipe[0], &byte));
338 try std.testing.expectEqual(@as(usize, 0), try sys.fd.read(second_pipe[0], &byte));
339 }