tiny.wayland.stream

1pub const Frame = struct {2    header: wire.Header,3    bytes: []const u8,4    payload: []const u8,5};

1pub const Inbox = struct {2    session_allocator: std.mem.Allocator,3    bytes: []u8,4    descriptors: []sys.fd.Descriptor,5    byte_count: usize = 0,6    descriptor_count: usize = 0,7    byte_offset: usize = 0,8    descriptor_offset: usize = 0,9    byte_capacity_rejection_count: u64 = 0,10    descriptor_capacity_rejection_count: u64 = 0,1112    pub fn init(13        session_allocator: std.mem.Allocator,14        capacity: storage.Capacity,15    ) std.mem.Allocator.Error!Inbox {16        const bytes = try session_allocator.alloc(u8, capacity.inbound_byte_count);17        errdefer session_allocator.free(bytes);18        return .{19            .session_allocator = session_allocator,20            .bytes = bytes,21            .descriptors = try session_allocator.alloc(22                sys.fd.Descriptor,23                capacity.inbound_descriptor_count,24            ),25        };26    }2728    pub fn deinit(self: *Inbox) void {29        self.assertValid();30        closeAll(self.descriptors[self.descriptor_offset..self.descriptor_count]);31        self.session_allocator.free(self.bytes);32        if (self.descriptors.len != 0) self.session_allocator.free(self.descriptors);33        self.* = undefined;34    }3536    pub fn appendOwned(37        self: *Inbox,38        bytes: []const u8,39        descriptors: []const sys.fd.Descriptor,40    ) storage.StorageError!void {41        self.compact();42        var failure: ?storage.StorageError = null;43        if (bytes.len > self.bytes.len - self.byte_count) {44            self.byte_capacity_rejection_count +|= 1;45            failure = error.InboundByteCapacityExceeded;46        }47        if (descriptors.len > self.descriptors.len - self.descriptor_count) {48            self.descriptor_capacity_rejection_count +|= 1;49            if (failure == null) failure = error.InboundDescriptorCapacityExceeded;50        }51        if (failure) |err| return err;52        @memcpy(self.bytes[self.byte_count..][0..bytes.len], bytes);53        @memcpy(54            self.descriptors[self.descriptor_count..][0..descriptors.len],55            descriptors,56        );57        self.byte_count += bytes.len;58        self.descriptor_count += descriptors.len;59        self.assertValid();60    }6162    pub fn prepareReceive(63        self: *Inbox,64        maximum_byte_count: usize,65    ) storage.StorageError![]u8 {66        std.debug.assert(maximum_byte_count > 0);67        self.compact();68        const available = self.bytes.len - self.byte_count;69        if (available == 0) {70            self.byte_capacity_rejection_count +|= 1;71            return error.InboundByteCapacityExceeded;72        }73        return self.bytes[self.byte_count..][0..@min(available, maximum_byte_count)];74    }7576    pub fn commitReceived(77        self: *Inbox,78        byte_count: usize,79        descriptors: []const sys.fd.Descriptor,80    ) storage.StorageError!void {81        std.debug.assert(byte_count > 0);82        std.debug.assert(byte_count <= self.bytes.len - self.byte_count);83        if (descriptors.len > self.descriptors.len - self.descriptor_count) {84            self.descriptor_capacity_rejection_count +|= 1;85            return error.InboundDescriptorCapacityExceeded;86        }87        @memcpy(88            self.descriptors[self.descriptor_count..][0..descriptors.len],89            descriptors,90        );91        self.byte_count += byte_count;92        self.descriptor_count += descriptors.len;93        self.assertValid();94    }9596    pub fn peek(self: *const Inbox) wire.Error!?Frame {97        self.assertValid();98        const available = self.bytes[self.byte_offset..self.byte_count];99        if (available.len < wire.header_size) return null;100        const header = try wire.decode(available[0..wire.header_size]);101        if (available.len < header.size) return null;102        return .{103            .header = header,104            .bytes = available[0..header.size],105            .payload = available[wire.header_size..header.size],106        };107    }108109    pub fn consume(self: *Inbox) (wire.Error || error{IncompleteMessage})!void {110        self.assertValid();111        const frame = (try self.peek()) orelse return error.IncompleteMessage;112        self.byte_offset += frame.header.size;113        if (self.byte_offset == self.byte_count) {114            self.byte_count = 0;115            self.byte_offset = 0;116        }117        self.assertValid();118    }119120    pub fn takeDescriptor(self: *Inbox) ?sys.fd.Descriptor {121        self.assertValid();122        if (self.descriptor_offset == self.descriptor_count) return null;123        const descriptor = self.descriptors[self.descriptor_offset];124        self.descriptor_offset += 1;125        if (self.descriptor_offset == self.descriptor_count) {126            self.descriptor_count = 0;127            self.descriptor_offset = 0;128        }129        self.assertValid();130        return descriptor;131    }132133    pub fn peekDescriptors(self: *const Inbox, count: usize) ?[]const sys.fd.Descriptor {134        if (count > self.queuedDescriptorCount()) return null;135        return self.descriptors[self.descriptor_offset..self.descriptor_count][0..count];136    }137138    pub fn queuedByteCount(self: *const Inbox) usize {139        return self.byte_count - self.byte_offset;140    }141142    pub fn queuedDescriptorCount(self: *const Inbox) usize {143        return self.descriptor_count - self.descriptor_offset;144    }145146    pub fn status(self: *const Inbox) storage.Status {147        return .{148            .inbound_byte_capacity_rejection_count = self.byte_capacity_rejection_count,149            .inbound_descriptor_capacity_rejection_count = self.descriptor_capacity_rejection_count,150        };151    }152153    fn compact(self: *Inbox) void {154        storage.compactSlice(u8, self.bytes, &self.byte_count, &self.byte_offset);155        storage.compactSlice(156            sys.fd.Descriptor,157            self.descriptors,158            &self.descriptor_count,159            &self.descriptor_offset,160        );161    }162163    fn assertValid(self: *const Inbox) void {164        std.debug.assert(self.byte_offset <= self.byte_count);165        std.debug.assert(self.byte_count <= self.bytes.len);166        std.debug.assert(self.descriptor_offset <= self.descriptor_count);167        std.debug.assert(self.descriptor_count <= self.descriptors.len);168    }169};

1pub const FlushStatus = enum {2    drained,3    pending,4};

1pub const Outbox = struct {2    session_allocator: std.mem.Allocator,3    bytes: []u8,4    descriptors: []sys.fd.Descriptor,5    byte_count: usize = 0,6    descriptor_count: usize = 0,7    byte_offset: usize = 0,8    descriptor_offset: usize = 0,9    byte_capacity_rejection_count: u64 = 0,10    descriptor_capacity_rejection_count: u64 = 0,1112    pub fn init(13        session_allocator: std.mem.Allocator,14        capacity: storage.Capacity,15    ) std.mem.Allocator.Error!Outbox {16        const bytes = try session_allocator.alloc(u8, capacity.outbound_byte_count);17        errdefer session_allocator.free(bytes);18        return .{19            .session_allocator = session_allocator,20            .bytes = bytes,21            .descriptors = try session_allocator.alloc(22                sys.fd.Descriptor,23                capacity.outbound_descriptor_count,24            ),25        };26    }2728    pub fn deinit(self: *Outbox) void {29        self.assertValid();30        closeAll(self.descriptors[self.descriptor_offset..self.descriptor_count]);31        self.session_allocator.free(self.bytes);32        if (self.descriptors.len != 0) self.session_allocator.free(self.descriptors);33        self.* = undefined;34    }3536    pub fn queueOwned(37        self: *Outbox,38        object_id: u32,39        opcode: u16,40        payload: []const u8,41        descriptors: []const sys.fd.Descriptor,42    ) (wire.Error || storage.StorageError || error{TooManyDescriptors})!void {43        if (descriptors.len > sys.ancillary.maximum_descriptors) return error.TooManyDescriptors;44        const header = try wire.Header.init(object_id, opcode, payload.len);45        var encoded: [wire.header_size]u8 = undefined;46        try header.encode(&encoded);4748        self.compact();49        const message_byte_count = encoded.len + payload.len;50        var failure: ?storage.StorageError = null;51        if (message_byte_count > self.bytes.len - self.byte_count) {52            self.byte_capacity_rejection_count +|= 1;53            failure = error.OutboundByteCapacityExceeded;54        }55        if (descriptors.len > self.descriptors.len - self.descriptor_count) {56            self.descriptor_capacity_rejection_count +|= 1;57            if (failure == null) failure = error.OutboundDescriptorCapacityExceeded;58        }59        if (failure) |err| return err;60        @memcpy(self.bytes[self.byte_count..][0..encoded.len], &encoded);61        self.byte_count += encoded.len;62        @memcpy(self.bytes[self.byte_count..][0..payload.len], payload);63        self.byte_count += payload.len;64        @memcpy(65            self.descriptors[self.descriptor_count..][0..descriptors.len],66            descriptors,67        );68        self.descriptor_count += descriptors.len;69        self.assertValid();70    }7172    pub fn flush(self: *Outbox, socket: sys.fd.Descriptor) !FlushStatus {73        while (self.byte_offset < self.byte_count) {74            const pending_bytes = self.bytes[self.byte_offset..self.byte_count];75            const pending_descriptors = self.descriptors[self.descriptor_offset..self.descriptor_count];76            const descriptor_count = @min(pending_descriptors.len, sys.ancillary.maximum_descriptors);77            const bytes = if (pending_descriptors.len > descriptor_count)78                pending_bytes[0..1]79            else80                pending_bytes;81            const descriptors = pending_descriptors[0..descriptor_count];82            const sent = sys.ancillary.send(socket, bytes, descriptors) catch |err| switch (err) {83                error.WouldBlock => return .pending,84                else => return err,85            };86            if (sent == 0) return error.SendFailed;87            closeAll(descriptors);88            self.descriptor_offset += descriptors.len;89            self.byte_offset += sent;90            self.assertValid();91        }9293        std.debug.assert(self.descriptor_offset == self.descriptor_count);94        self.byte_count = 0;95        self.descriptor_count = 0;96        self.byte_offset = 0;97        self.descriptor_offset = 0;98        self.assertValid();99        return .drained;100    }101102    pub fn pendingByteCount(self: *const Outbox) usize {103        return self.byte_count - self.byte_offset;104    }105106    pub fn pendingDescriptorCount(self: *const Outbox) usize {107        return self.descriptor_count - self.descriptor_offset;108    }109110    pub fn status(self: *const Outbox) storage.Status {111        return .{112            .outbound_byte_capacity_rejection_count = self.byte_capacity_rejection_count,113            .outbound_descriptor_capacity_rejection_count = self.descriptor_capacity_rejection_count,114        };115    }116117    fn compact(self: *Outbox) void {118        storage.compactSlice(u8, self.bytes, &self.byte_count, &self.byte_offset);119        storage.compactSlice(120            sys.fd.Descriptor,121            self.descriptors,122            &self.descriptor_count,123            &self.descriptor_offset,124        );125    }126127    fn assertValid(self: *const Outbox) void {128        std.debug.assert(self.byte_offset <= self.byte_count);129        std.debug.assert(self.byte_count <= self.bytes.len);130        std.debug.assert(self.descriptor_offset <= self.descriptor_count);131        std.debug.assert(self.descriptor_count <= self.descriptors.len);132    }133};

1pub const Capacity = struct {2    inbound_byte_count: usize,3    inbound_descriptor_count: usize,4    inbound_descriptor_bytes: usize,5    inbound_storage_bytes: usize,6    outbound_byte_count: usize,7    outbound_descriptor_count: usize,8    outbound_descriptor_bytes: usize,9    outbound_storage_bytes: usize,10    total_requested_bytes: usize,1112    pub fn derive(limits: Limits) CapacityError!Capacity {13        if (limits.inbound_byte_count < wire.header_size) {14            return error.InboundByteStorageTooSmall;15        }16        if (limits.outbound_byte_count < wire.header_size) {17            return error.OutboundByteStorageTooSmall;18        }19        const inbound_descriptor_bytes = std.math.mul(20            usize,21            limits.inbound_descriptor_count,22            @sizeOf(sys.fd.Descriptor),23        ) catch return error.CapacityOverflow;24        const inbound_storage_bytes = std.math.add(25            usize,26            limits.inbound_byte_count,27            inbound_descriptor_bytes,28        ) catch return error.CapacityOverflow;29        const outbound_descriptor_bytes = std.math.mul(30            usize,31            limits.outbound_descriptor_count,32            @sizeOf(sys.fd.Descriptor),33        ) catch return error.CapacityOverflow;34        const outbound_storage_bytes = std.math.add(35            usize,36            limits.outbound_byte_count,37            outbound_descriptor_bytes,38        ) catch return error.CapacityOverflow;39        const total_requested_bytes = std.math.add(40            usize,41            inbound_storage_bytes,42            outbound_storage_bytes,43        ) catch return error.CapacityOverflow;44        return .{45            .inbound_byte_count = limits.inbound_byte_count,46            .inbound_descriptor_count = limits.inbound_descriptor_count,47            .inbound_descriptor_bytes = inbound_descriptor_bytes,48            .inbound_storage_bytes = inbound_storage_bytes,49            .outbound_byte_count = limits.outbound_byte_count,50            .outbound_descriptor_count = limits.outbound_descriptor_count,51            .outbound_descriptor_bytes = outbound_descriptor_bytes,52            .outbound_storage_bytes = outbound_storage_bytes,53            .total_requested_bytes = total_requested_bytes,54        };55    }56};

1pub const CapacityError = error{2    InboundByteStorageTooSmall,3    OutboundByteStorageTooSmall,4    CapacityOverflow,5};

1pub const Limits = struct {2    inbound_byte_count: usize = default_byte_count,3    inbound_descriptor_count: usize = default_descriptor_count,4    outbound_byte_count: usize = default_byte_count,5    outbound_descriptor_count: usize = default_descriptor_count,6};

1pub const Status = struct {2    inbound_byte_capacity_rejection_count: u64 = 0,3    inbound_descriptor_capacity_rejection_count: u64 = 0,4    outbound_byte_capacity_rejection_count: u64 = 0,5    outbound_descriptor_capacity_rejection_count: u64 = 0,6};

1pub const StorageError = error{2    InboundByteCapacityExceeded,3    InboundDescriptorCapacityExceeded,4    OutboundByteCapacityExceeded,5    OutboundDescriptorCapacityExceeded,6};

1pub const stream = @import("stream/root.zig");

1const inbox = @import("inbox.zig");2const outbox = @import("outbox.zig");3const storage = @import("storage.zig");45pub const FlushStatus = outbox.FlushStatus;6pub const Frame = inbox.Frame;7pub const Inbox = inbox.Inbox;8pub const Outbox = outbox.Outbox;9pub const Limits = storage.Limits;10pub const Capacity = storage.Capacity;11pub const CapacityError = storage.CapacityError;12pub const StorageError = storage.StorageError;13pub const Status = storage.Status;14pub const default_byte_count = storage.default_byte_count;15pub const default_descriptor_count = storage.default_descriptor_count;

1pub const default_descriptor_count: usize = sys.ancillary.maximum_descriptors;