tiny.sys.event

1pub const Async = struct {2    read_fd: fd.Descriptor,3    write_fd: fd.Descriptor,45    pub const WaitError: type = ReadError;67    pub fn init() !Async {8        const pipe_fds = try fd.pipeWithOptions(.{ .close_on_exec = true, .nonblocking = true });9        return .{ .read_fd = pipe_fds[0], .write_fd = pipe_fds[1] };10    }1112    pub fn deinit(self: *Async) void {13        fd.close(self.read_fd);14        fd.close(self.write_fd);15        self.* = undefined;16    }1718    pub fn wait(19        self: Async,20        loop: *Loop,21        completion: *Completion,22        comptime Userdata: type,23        userdata: ?*Userdata,24        comptime cb: *const fn (25            ud: ?*Userdata,26            loop: *Loop,27            completion: *Completion,28            result: WaitError!void,29        ) CallbackAction,30    ) void {31        loop.arm(32            completion,33            .{ .async_wait = self },34            userdata,35            AsyncWaitCallback(Userdata, cb).callback,36        );37    }3839    pub fn notify(self: Async) !void {40        _ = fd.write(self.write_fd, &.{1}) catch |err| switch (err) {41            error.WouldBlock => return,42            else => return err,43        };44    }4546    fn drain(self: Async) void {47        var buffer: [64]u8 = undefined;48        while (true) {49            _ = fd.read(self.read_fd, &buffer) catch |err| switch (err) {50                error.WouldBlock => return,51                else => return,52            };53        }54    }55};

1pub const Completion = struct {2    active: bool = false,3    op: Operation = .none,4    userdata: ?*anyopaque = null,5    callback: ?Callback = null,6};

1pub const File = struct {2    fd: fd.Descriptor,34    pub fn initFd(file_fd: std.Io.File.Handle) File {5        return .{ .fd = file_fd };6    }78    pub fn poll(9        self: File,10        loop: *Loop,11        completion: *Completion,12        event: PollEvent,13        comptime Userdata: type,14        userdata: ?*Userdata,15        comptime cb: *const fn (16            ud: ?*Userdata,17            loop: *Loop,18            completion: *Completion,19            file: File,20            result: PollError!PollEvent,21        ) CallbackAction,22    ) void {23        loop.arm(24            completion,25            .{ .file_poll = .{ .file = self, .event = event } },26            userdata,27            FilePollCallback(Userdata, cb).callback,28        );29    }30};

1pub const Loop = struct {2    allocator: std.mem.Allocator,3    awake_clock: time.AwakeClock,4    last_instant: ?time.AwakeInstant,5    entries: []?*Completion,6    pollfds: []descriptor_poll.Descriptor,7    poll_completions: []*Completion,89    pub fn init(options: Options) !Loop {10        const allocator = options.allocator;11        const count: usize = @intCast(options.entries);12        const entries = try allocator.alloc(?*Completion, count);13        errdefer allocator.free(entries);14        @memset(entries, null);1516        const pollfds = try allocator.alloc(descriptor_poll.Descriptor, count);17        errdefer allocator.free(pollfds);1819        const poll_completions = try allocator.alloc(*Completion, count);20        errdefer allocator.free(poll_completions);2122        return .{23            .allocator = allocator,24            .awake_clock = options.awake_clock,25            .last_instant = null,26            .entries = entries,27            .pollfds = pollfds,28            .poll_completions = poll_completions,29        };30    }3132    pub fn deinit(self: *Loop) void {33        self.allocator.free(self.poll_completions);34        self.allocator.free(self.pollfds);35        self.allocator.free(self.entries);36        self.* = undefined;37    }3839    pub fn run(self: *Loop, mode: RunMode) !void {40        while (self.hasActive()) {41            const now = try self.readInstant();42            if (try self.fireDueTimer(now)) {43                if (mode != .until_done) return;44                continue;45            }4647            const poll_count = self.preparePoll();48            const timeout_ms = self.timeoutMillis(mode, now);49            if (poll_count == 0) {50                if (timeout_ms < 0) return;51                if (timeout_ms > 0) {52                    const timeout_ns = @as(u64, @intCast(timeout_ms)) *53                        std.time.ns_per_ms;54                    std.Io.sleep(55                        std.Options.debug_io,56                        .fromNanoseconds(timeout_ns),57                        .awake,58                    ) catch {};59                }60                if (mode != .until_done) return;61                continue;62            }6364            const ready = descriptor_poll.wait(self.pollfds[0..poll_count], timeout_ms) catch65                return error.Unexpected;66            if (ready == 0) {67                if (mode != .until_done) return;68                continue;69            }7071            _ = try self.readInstant();72            if (try self.fireFirstReady(poll_count) and mode != .until_done) return;73            if (mode == .no_wait) return;74        }75    }7677    pub fn currentInstant(self: *const Loop) time.AwakeInstant {78        return self.last_instant orelse @panic("sys.event loop has not sampled its awake clock");79    }8081    pub fn awakeClock(self: *const Loop) time.AwakeClock {82        return self.awake_clock;83    }8485    pub fn disarm(self: *Loop, completion: *Completion) void {86        self.deactivate(completion);87        completion.* = .{};88    }8990    fn activate(self: *Loop, completion: *Completion) void {91        if (completion.active) {92            for (self.entries) |slot| {93                if (slot == completion) return;94            }95            @panic("sys.event active completion belongs to another loop");96        }97        for (self.entries) |*slot| {98            if (slot.* == null) {99                slot.* = completion;100                completion.active = true;101                return;102            }103        }104        @panic("sys.event loop completion capacity exhausted");105    }106107    fn arm(108        self: *Loop,109        completion: *Completion,110        op: Operation,111        userdata: ?*anyopaque,112        callback: Callback,113    ) void {114        var replacement: Completion = .{115            .op = op,116            .userdata = userdata,117            .callback = callback,118        };119        if (completion.active) {120            for (self.entries) |slot| {121                if (slot == completion) {122                    replacement.active = true;123                    completion.* = replacement;124                    return;125                }126            }127            @panic("sys.event active completion belongs to another loop");128        }129        completion.* = replacement;130        self.activate(completion);131    }132133    fn deactivate(self: *Loop, completion: *Completion) void {134        if (!completion.active) return;135        for (self.entries) |*slot| {136            if (slot.* == completion) {137                slot.* = null;138                completion.active = false;139                return;140            }141        }142        @panic("sys.event active completion belongs to another loop");143    }144145    fn deactivateDescriptor(self: *Loop, descriptor: fd.Descriptor) void {146        for (self.entries) |*slot| {147            const completion = slot.* orelse continue;148            if (completion.op.descriptor()) |active_descriptor| {149                if (active_descriptor == descriptor) {150                    completion.active = false;151                    slot.* = null;152                }153            }154        }155    }156157    fn hasActive(self: *Loop) bool {158        for (self.entries) |slot| {159            if (slot != null) return true;160        }161        return false;162    }163164    fn preparePoll(self: *Loop) usize {165        var count: usize = 0;166        for (self.entries) |slot| {167            const completion = slot orelse continue;168            const descriptor = completion.op.descriptor() orelse continue;169            self.pollfds[count] = .{170                .fd = descriptor,171                .events = completion.op.events(),172                .revents = 0,173            };174            self.poll_completions[count] = completion;175            count += 1;176        }177        return count;178    }179180    fn timeoutMillis(self: *Loop, mode: RunMode, now: time.AwakeInstant) i32 {181        if (mode == .no_wait) return 0;182        var best: ?u64 = null;183        for (self.entries) |slot| {184            const completion = slot orelse continue;185            switch (completion.op) {186                .timer => |timer| {187                    const remaining = now.remainingUntil(timer.deadline).asMillisecondsCeil();188                    best = if (best) |current| @min(current, remaining) else remaining;189                },190                else => {},191            }192        }193        const value = best orelse return -1;194        return @intCast(@min(value, std.math.maxInt(i32)));195    }196197    fn fireDueTimer(self: *Loop, now: time.AwakeInstant) time.ClockError!bool {198        var due: ?*Completion = null;199        for (self.entries) |slot| {200            const completion = slot orelse continue;201            switch (completion.op) {202                .timer => |timer| {203                    if (!now.reached(timer.deadline)) continue;204                    const current = due orelse {205                        due = completion;206                        continue;207                    };208                    const current_timer = switch (current.op) {209                        .timer => |value| value,210                        else => unreachable,211                    };212                    if (timer.deadline.isBefore(current_timer.deadline)) due = completion;213                },214                else => {},215            }216        }217        const completion = due orelse return false;218        try self.fire(completion, .{ .timer = {} });219        return true;220    }221222    fn fireFirstReady(self: *Loop, count: usize) time.ClockError!bool {223        var index: usize = 0;224        while (index < count) : (index += 1) {225            if (self.pollfds[index].revents == 0) continue;226            const completion = self.poll_completions[index];227            if (!completion.active) continue;228            try self.fireReady(completion);229            return true;230        }231        return false;232    }233234    fn fireReady(self: *Loop, completion: *Completion) time.ClockError!void {235        switch (completion.op) {236            .async_wait => |async_watcher| {237                async_watcher.drain();238                try self.fire(completion, .{ .async_wait = {} });239            },240            .tcp_accept => |socket| {241                const result = acceptSocket(socket);242                try self.fire(completion, .{ .tcp_accept = result });243            },244            .tcp_read => |*state| {245                const result = readSocket(state);246                try self.fire(completion, .{ .tcp_read = .{247                    .socket = state.socket,248                    .buffer = state.buffer,249                    .result = result,250                } });251            },252            .tcp_write => |*state| {253                const result = writeSocket(state);254                try self.fire(completion, .{ .tcp_write = .{255                    .socket = state.socket,256                    .buffer = state.buffer,257                    .result = result,258                } });259            },260            .udp_recv_from => |*state| try self.fireUdpRecvFrom(completion, state),261            .udp_send_to => |*state| try self.fireUdpSendTo(completion, state),262            .file_poll => |state| {263                try self.fire(completion, .{ .file_poll = .{264                    .file = state.file,265                    .result = state.event,266                } });267            },268            .none, .timer => {},269        }270    }271272    fn fireUdpRecvFrom(273        self: *Loop,274        completion: *Completion,275        state: *UdpRecvFromState,276    ) time.ClockError!void {277        const received = net.recvFromIpAddress(278            state.socket.fd,279            readSlice(&state.buffer),280            0,281        ) catch |err| {282            if (err == error.WouldBlock) return;283            const result: RecvError = switch (err) {284                error.ConnectionResetByPeer => error.ConnectionResetByPeer,285                else => error.Unexpected,286            };287            try self.fire(completion, .{ .udp_recv_from = .{288                .socket = state.socket,289                .buffer = state.buffer,290                .result = result,291            } });292            return;293        };294        try self.fire(completion, .{ .udp_recv_from = .{295            .socket = state.socket,296            .buffer = state.buffer,297            .result = Datagram{298                .address = received.address,299                .bytes = received.bytes,300                .truncated = received.truncated,301            },302        } });303    }304305    fn fireUdpSendTo(306        self: *Loop,307        completion: *Completion,308        state: *const UdpSendToState,309    ) time.ClockError!void {310        const bytes = writeSlice(&state.buffer);311        const sent = net.sendToIpAddress(state.socket.fd, bytes, 0, state.address) catch |err| {312            if (err == error.WouldBlock) return;313            const result: SendError = switch (err) {314                error.BrokenPipe => error.BrokenPipe,315                error.ConnectionResetByPeer => error.ConnectionResetByPeer,316                error.MessageTooLarge => error.MessageTooLarge,317                else => error.Unexpected,318            };319            try self.fire(completion, .{ .udp_send_to = .{320                .socket = state.socket,321                .buffer = state.buffer,322                .result = result,323            } });324            return;325        };326        std.debug.assert(sent == bytes.len);327        try self.fire(completion, .{ .udp_send_to = .{328            .socket = state.socket,329            .buffer = state.buffer,330            .result = sent,331        } });332    }333334    fn fire(self: *Loop, completion: *Completion, result: Result) time.ClockError!void {335        const callback = completion.callback orelse return;336        self.deactivate(completion);337        const action = callback(completion.userdata, self, completion, result);338        if (action == .rearm) {339            switch (completion.op) {340                .timer => |*timer| timer.deadline = try self.rearmDeadline(timer.*),341                else => {},342            }343            self.activate(completion);344        }345    }346347    fn readInstant(self: *Loop) time.ClockError!time.AwakeInstant {348        const current = try self.awake_clock.now();349        if (self.last_instant) |previous| _ = try current.elapsedSince(previous);350        self.last_instant = current;351        return current;352    }353354    fn rearmDeadline(self: *Loop, timer: TimerState) time.ClockError!time.AwakeInstant {355        const repeat = timer.repeat orelse @panic("sys.event timer rearm needs a repeat policy");356        const now = try self.readInstant();357        const interval = switch (repeat) {358            .fixed_delay, .fixed_rate => |value| value,359        };360        const deadline = switch (repeat) {361            .fixed_delay => now.deadlineAfter(interval),362            .fixed_rate => fixedRateDeadline(timer.deadline, now, interval),363        };364        if (!interval.isZero() and now.reached(deadline)) return error.ClockOverflow;365        return deadline;366    }367};

1pub const TCP = struct {2    fd: net.Socket,34    pub fn initFd(socket_fd: anytype) TCP {5        return .{ .fd = @intCast(socket_fd) };6    }78    pub fn accept(9        self: TCP,10        loop: *Loop,11        completion: *Completion,12        comptime Userdata: type,13        userdata: ?*Userdata,14        comptime cb: *const fn (15            ud: ?*Userdata,16            loop: *Loop,17            completion: *Completion,18            result: AcceptError!TCP,19        ) CallbackAction,20    ) void {21        loop.arm(22            completion,23            .{ .tcp_accept = self },24            userdata,25            TcpAcceptCallback(Userdata, cb).callback,26        );27    }2829    pub fn read(30        self: TCP,31        loop: *Loop,32        completion: *Completion,33        buffer: ReadBuffer,34        comptime Userdata: type,35        userdata: ?*Userdata,36        comptime cb: *const fn (37            ud: ?*Userdata,38            loop: *Loop,39            completion: *Completion,40            socket: TCP,41            buffer: ReadBuffer,42            result: ReadError!usize,43        ) CallbackAction,44    ) void {45        loop.arm(46            completion,47            .{ .tcp_read = .{ .socket = self, .buffer = buffer } },48            userdata,49            TcpReadCallback(Userdata, cb).callback,50        );51    }5253    pub fn write(54        self: TCP,55        loop: *Loop,56        completion: *Completion,57        buffer: WriteBuffer,58        comptime Userdata: type,59        userdata: ?*Userdata,60        comptime cb: *const fn (61            ud: ?*Userdata,62            loop: *Loop,63            completion: *Completion,64            socket: TCP,65            buffer: WriteBuffer,66            result: WriteError!usize,67        ) CallbackAction,68    ) void {69        loop.arm(70            completion,71            .{ .tcp_write = .{ .socket = self, .buffer = buffer } },72            userdata,73            TcpWriteCallback(Userdata, cb).callback,74        );75    }7677    pub fn close(78        self: TCP,79        loop: *Loop,80        completion: *Completion,81        comptime Userdata: type,82        userdata: ?*Userdata,83        comptime cb: *const fn (84            ud: ?*Userdata,85            loop: *Loop,86            completion: *Completion,87            socket: TCP,88            result: CloseError!void,89        ) CallbackAction,90    ) void {91        loop.deactivateDescriptor(self.fd);92        loop.deactivate(completion);93        net.close(self.fd);94        completion.* = .{95            .userdata = userdata,96            .callback = TcpCloseCallback(Userdata, cb).callback,97        };98        loop.fire(99            completion,100            .{ .tcp_close = .{ .socket = self, .result = {} } },101        ) catch unreachable;102    }103};

1pub const UDP = struct {2    fd: net.Socket,34    pub const BindError = net.SocketError || net.BindError;56    pub fn initFd(socket_fd: anytype) UDP {7        return .{ .fd = @intCast(socket_fd) };8    }910    pub fn bind(address: net.IpAddress) BindError!UDP {11        const socket_fd = try net.udpDatagramSocketForAddress(address, .{12            .close_on_exec = true,13            .nonblocking = true,14        });15        errdefer net.close(socket_fd);16        try net.bindIpAddress(socket_fd, address);17        return initFd(socket_fd);18    }1920    pub fn recvFrom(21        self: UDP,22        loop: *Loop,23        completion: *Completion,24        buffer: ReadBuffer,25        comptime Userdata: type,26        userdata: ?*Userdata,27        comptime cb: *const fn (28            ud: ?*Userdata,29            loop: *Loop,30            completion: *Completion,31            socket: UDP,32            buffer: ReadBuffer,33            result: RecvError!Datagram,34        ) CallbackAction,35    ) void {36        std.debug.assert(readBufferLength(buffer) > 0);37        loop.arm(38            completion,39            .{ .udp_recv_from = .{ .socket = self, .buffer = buffer } },40            userdata,41            UdpRecvFromCallback(Userdata, cb).callback,42        );43    }4445    pub fn sendTo(46        self: UDP,47        loop: *Loop,48        completion: *Completion,49        buffer: WriteBuffer,50        address: net.IpAddress,51        comptime Userdata: type,52        userdata: ?*Userdata,53        comptime cb: *const fn (54            ud: ?*Userdata,55            loop: *Loop,56            completion: *Completion,57            socket: UDP,58            buffer: WriteBuffer,59            result: SendError!usize,60        ) CallbackAction,61    ) void {62        loop.arm(63            completion,64            .{ .udp_send_to = .{65                .socket = self,66                .buffer = buffer,67                .address = address,68            } },69            userdata,70            UdpSendToCallback(Userdata, cb).callback,71        );72    }7374    pub fn close(self: UDP) void {75        net.close(self.fd);76    }77};

1pub const AcceptError = error{2    Canceled,3    Again,4    Unexpected,5};

1pub const CallbackAction = enum {2    disarm,3    rearm,4};

1pub const CloseError = error{2    Canceled,3    Unexpected,4};

1/// Record of one UDP message that arrived, read once a UDP receive completes to2/// learn who sent the message and whether all of it arrived. A caller hands a3/// buffer to the event loop when it arms a receive. The payload byte count4/// reports what landed in that armed buffer and stops at its length, and a5/// separate flag reports whether the message ran past the end of it. On Linux6/// the flag is exact, because the kernel reports the message's full length and7/// the receive compares it against the buffer length, while on every other host8/// the flag is set whenever the byte count equals the armed buffer's length, so9/// a message that exactly fills the buffer reports as truncated.10pub const Datagram = struct {11    address: sys.net.IpAddress,12    bytes: usize,13    truncated: bool,14};

1pub const Options = struct {2    entries: u32 = 256,3    allocator: std.mem.Allocator = sys.allocator.processAllocator(),4    awake_clock: sys.time.AwakeClock = .system(),5};

1pub const PollError = error{2    Canceled,3    Unexpected,4};

1pub const PollEvent = enum {2    read,3};

1pub const ReadBuffer = union(enum) {2    slice: []u8,3    array: [32]u8,4};

1pub const ReadError = error{2    EOF,3    Canceled,4    ConnectionResetByPeer,5    Unexpected,6};

1pub const RecvError = error{2    Canceled,3    ConnectionResetByPeer,4    Unexpected,5};

1pub const RunMode = enum {2    no_wait,3    once,4    until_done,5};

1pub const SendError = error{2    Canceled,3    BrokenPipe,4    ConnectionResetByPeer,5    MessageTooLarge,6    Unexpected,7};

1pub const WriteBuffer = union(enum) {2    slice: []const u8,3    array: struct {4        array: [32]u8,5        len: usize,6    },7};

1pub const WriteError = error{2    Canceled,3    BrokenPipe,4    ConnectionResetByPeer,5    Unexpected,6};

1const sys = @import("../root.zig");2const types = @import("types.zig");3const backend = @import("poll.zig");45pub const required_capabilities = sys.capabilities.host(&.{ .descriptors, .event_loop, .networking, .time });67pub const AcceptError = types.AcceptError;8pub const CallbackAction = types.CallbackAction;9pub const CloseError = types.CloseError;10pub const Datagram = types.Datagram;11pub const Options = types.Options;12pub const PollError = types.PollError;13pub const PollEvent = types.PollEvent;14pub const ReadBuffer = types.ReadBuffer;15pub const ReadError = types.ReadError;16pub const RecvError = types.RecvError;17pub const RunMode = types.RunMode;18pub const SendError = types.SendError;19pub const WriteBuffer = types.WriteBuffer;20pub const WriteError = types.WriteError;2122pub const Async = backend.Async;23pub const Completion = backend.Completion;24pub const File = backend.File;25pub const Loop = backend.Loop;26pub const TCP = backend.TCP;27pub const Timer = backend.Timer;28pub const UDP = backend.UDP;

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