tiny.wayland.transport

1pub const transport = @import("transport.zig");

1const std = @import("std");2const sys = @import("sys");3const wayland = @import("root.zig");45pub const ReceiveStatus = enum {6    data,7    pending,8    closed,9};1011pub const Limits = wayland.stream.Limits;12pub const Capacity = wayland.stream.Capacity;13pub const CapacityError = wayland.stream.CapacityError;14pub const StorageError = wayland.stream.StorageError;15pub const Status = wayland.stream.Status;16pub const default_byte_count = wayland.stream.default_byte_count;17pub const default_descriptor_count = wayland.stream.default_descriptor_count;18const test_capacity = Capacity.derive(.{}) catch unreachable;1920pub const Storage = struct {21    inbox: wayland.stream.Inbox,22    outbox: wayland.stream.Outbox,2324    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    }3536    pub fn deinit(self: *Storage) void {37        self.inbox.deinit();38        self.outbox.deinit();39        self.* = undefined;40    }4142    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};5859pub const Transport = struct {60    descriptor: sys.fd.Descriptor,61    inbox: wayland.stream.Inbox,62    outbox: wayland.stream.Outbox,6364    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    }7677    pub fn deinit(self: *Transport) void {78        self.inbox.deinit();79        self.outbox.deinit();80        sys.fd.close(self.descriptor);81        self.* = undefined;82    }8384    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    }9394    pub fn flush(self: *Transport) !wayland.stream.FlushStatus {95        return self.outbox.flush(self.descriptor);96    }9798    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    }118119    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};130131fn closeAll(descriptors: []const sys.fd.Descriptor) void {132    for (descriptors) |descriptor| sys.fd.close(descriptor);133}134135test "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}162163test "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}179180test "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();192193    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    };202203    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());207208    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);212213    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"));220221    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}228229test "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();239240    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 });247248    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());252253    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}259260test "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();279280    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}300301test "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]);323324    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}