From 9574a744fefaeb43c5d9cc193e2ff30d3d388d27 Mon Sep 17 00:00:00 2001 From: Tim Culverhouse Date: Tue, 22 Apr 2025 15:33:11 -0500 Subject: [PATCH] s/Runtime/Ring --- README.md | 18 ++++++------- src/Kqueue.zig | 10 +++---- src/Mock.zig | 2 +- src/Task.zig | 6 ++--- src/Uring.zig | 2 +- src/main.zig | 72 +++++++++++++++++++++++++------------------------- src/net.zig | 10 +++---- src/tls.zig | 20 +++++++------- 8 files changed, 70 insertions(+), 70 deletions(-) diff --git a/README.md b/README.md index ab8c8de..4607bd5 100644 --- a/README.md +++ b/README.md @@ -16,7 +16,7 @@ Ourio has io_uring and kqueue backends. Ourio supports the `msg_ring` capability of io_uring to pass a completion from one ring to another. This allows a multithreaded application to implement message passing using io_uring (or kqueue, if that's your flavor). Multithreaded applications should plan to -use one `Runtime` per thread. Submission onto the runtime is not thread safe, +use one `Ring` per thread. Submission onto the runtime is not thread safe, any message passing must occur using `msg_ring` rather than directly submitting a task to another @@ -61,17 +61,17 @@ target_task.* { .req = .{ .userfd = fd }, }; -// Send target_task from the main_rt thread to the thread_rt Runtime. The -// thread_rt Runtime will then // process the task as a completion, ie +// Send target_task from the main_rt thread to the thread_rt Ring. The +// thread_rt Ring will then // process the task as a completion, ie // Worker.onCompletion will be called with // this task. That thread can then // schedule a recv, a write, etc on the file // descriptor it just received. _ = try main_rt.msgRing(thread_rt, target_task, .{}); ``` -### Multiple Runtimes on the same thread +### Multiple Rings on the same thread -You can have multiple Runtimes in a single thread. One could be a priority -Runtime, or handle specific types of tasks, etc. Poll any runtime from any other +You can have multiple Rings in a single thread. One could be a priority +Ring, or handle specific types of tasks, etc. Poll any runtime from any other runtime. ```zig @@ -110,7 +110,7 @@ pub const MultiWriter = struct { try self.buf.appendSlice(gpa, bytes); } - pub fn flush(self: *MultiWriter, rt: *io.Runtime) !void { + pub fn flush(self: *MultiWriter, rt: *io.Ring) !void { if (self.fd1_written < self.buf.items.len) { _ = try rt.write(self.fd1, self.buf.items[self.fd1_written..], .{ .ptr = self, @@ -128,7 +128,7 @@ pub const MultiWriter = struct { } } - pub fn onCompletion(rt: *io.Runtime, task: io.Task) anyerror!void { + pub fn onCompletion(rt: *io.Ring, task: io.Task) anyerror!void { const self = task.userdataCast(MultiWriter); const result = task.result.?; @@ -151,7 +151,7 @@ pub const MultiWriter = struct { pub fn main() !void { var gpa: std.heap.DebugAllocator(.{}) = .init; - var rt: io.Runtime = try .init(gpa.allocator(), 16); + var rt: io.Ring = try .init(gpa.allocator(), 16); defer rt.deinit(); // Pretend I created some files diff --git a/src/Kqueue.zig b/src/Kqueue.zig index 1a1dfa9..17b20f8 100644 --- a/src/Kqueue.zig +++ b/src/Kqueue.zig @@ -548,7 +548,7 @@ pub fn pollableFd(self: Kqueue) !posix.fd_t { return self.kq; } -pub fn reapCompletions(self: *Kqueue, rt: *io.Runtime) anyerror!void { +pub fn reapCompletions(self: *Kqueue, rt: *io.Ring) anyerror!void { defer self.event_idx = 0; if (self.event_idx == 0) { @@ -604,7 +604,7 @@ pub fn reapCompletions(self: *Kqueue, rt: *io.Runtime) anyerror!void { /// to call the callback and return the task(s) to the free list fn handleSynchronousCompletion( self: *Kqueue, - rt: *io.Runtime, + rt: *io.Ring, task: *io.Task, ) !void { switch (task.req) { @@ -685,7 +685,7 @@ fn unexpectedError(err: posix.E) posix.UnexpectedError { fn handleCompletion( self: *Kqueue, - rt: *io.Runtime, + rt: *io.Ring, task: *io.Task, event: posix.Kevent, ) !void { @@ -790,7 +790,7 @@ fn handleCompletion( } } -fn releaseTask(self: *Kqueue, rt: *io.Runtime, task: *io.Task) void { +fn releaseTask(self: *Kqueue, rt: *io.Ring, task: *io.Task) void { rt.free_q.push(task); if (task.deadline) |d| { // remove the deadline @@ -805,7 +805,7 @@ fn releaseTask(self: *Kqueue, rt: *io.Runtime, task: *io.Task) void { } } -fn handleExpiredTimer(self: *Kqueue, rt: *io.Runtime, t: Timer) !void { +fn handleExpiredTimer(self: *Kqueue, rt: *io.Ring, t: Timer) !void { switch (t) { .deadline => |deadline| { defer self.releaseTask(rt, deadline.task); diff --git a/src/Mock.zig b/src/Mock.zig index 1d84e2c..b7e3219 100644 --- a/src/Mock.zig +++ b/src/Mock.zig @@ -81,7 +81,7 @@ pub fn submit(self: *Mock, queue: *Queue(io.Task, .in_flight)) !void { } } -pub fn reapCompletions(self: *Mock, rt: *io.Runtime) anyerror!void { +pub fn reapCompletions(self: *Mock, rt: *io.Ring) anyerror!void { while (self.completions.pop()) |task| { try task.callback(rt, task.*); rt.free_q.push(task); diff --git a/src/Task.zig b/src/Task.zig index 9a7075d..7dc2d3c 100644 --- a/src/Task.zig +++ b/src/Task.zig @@ -4,7 +4,7 @@ const std = @import("std"); const io = @import("main.zig"); const Allocator = std.mem.Allocator; -const Runtime = io.Runtime; +const Ring = io.Ring; userdata: ?*anyopaque = null, msg: u16 = 0, @@ -37,7 +37,7 @@ prev: ?*Task = null, pub fn setDeadline( self: *Task, - rt: *Runtime, + rt: *Ring, deadline: io.Timespec, ) Allocator.Error!void { std.debug.assert(!deadline.isZero()); @@ -55,7 +55,7 @@ pub fn setDeadline( pub fn cancel( self: *Task, - rt: *Runtime, + rt: *Ring, ctx: io.Context, ) Allocator.Error!void { const task = try rt.getTask(); diff --git a/src/Uring.zig b/src/Uring.zig index d52dac1..62d7354 100644 --- a/src/Uring.zig +++ b/src/Uring.zig @@ -243,7 +243,7 @@ fn getSqe(self: *Uring) *linux.io_uring_sqe { return self.ring.get_sqe() catch unreachable; } -pub fn reapCompletions(self: *Uring, rt: *io.Runtime) anyerror!void { +pub fn reapCompletions(self: *Uring, rt: *io.Ring) anyerror!void { var cqes: [64]linux.io_uring_cqe = undefined; const n = self.ring.copy_cqes(&cqes, 0) catch |err| { switch (err) { diff --git a/src/main.zig b/src/main.zig index 06bda3a..276131e 100644 --- a/src/main.zig +++ b/src/main.zig @@ -24,8 +24,8 @@ pub const has_kqueue = switch (builtin.os.tag) { pub const has_io_uring = builtin.os.tag == .linux; pub const Task = @import("Task.zig"); -pub const Callback = *const fn (*Runtime, Task) anyerror!void; -pub fn noopCallback(_: *Runtime, _: Task) anyerror!void {} +pub const Callback = *const fn (*Ring, Task) anyerror!void; +pub fn noopCallback(_: *Ring, _: Task) anyerror!void {} pub const RunCondition = enum { once, @@ -103,7 +103,7 @@ pub const Backend = union(enum) { pub fn reapCompletions( self: *Backend, - rt: *Runtime, + rt: *Ring, ) !void { return switch (self.*) { inline else => |*backend| backend.reapCompletions(rt), @@ -121,7 +121,7 @@ pub const CompletionQueue = Queue(Task, .complete); pub const FreeQueue = Queue(Task, .free); pub const SubmissionQueue = Queue(Task, .in_flight); -pub const Runtime = struct { +pub const Ring = struct { backend: Backend, gpa: Allocator, @@ -129,21 +129,21 @@ pub const Runtime = struct { submission_q: SubmissionQueue = .{}, free_q: FreeQueue = .{}, - pub fn init(gpa: Allocator, entries: u16) !Runtime { + pub fn init(gpa: Allocator, entries: u16) !Ring { return .{ .backend = .{ .platform = try .init(gpa, entries) }, .gpa = gpa, }; } - pub fn initChild(self: *Runtime, entries: u16) !Runtime { + pub fn initChild(self: *Ring, entries: u16) !Ring { return .{ .backend = try self.backend.initChild(entries), .gpa = self.gpa, }; } - pub fn initMock(gpa: Allocator, entries: u16) !Runtime { + pub fn initMock(gpa: Allocator, entries: u16) !Ring { return .{ .backend = .{ .mock = try .init(entries) }, .gpa = gpa, @@ -153,14 +153,14 @@ pub const Runtime = struct { }; } - pub fn deinit(self: *Runtime) void { + pub fn deinit(self: *Ring) void { self.backend.deinit(self.gpa); while (self.free_q.pop()) |task| self.gpa.destroy(task); while (self.submission_q.pop()) |task| self.gpa.destroy(task); while (self.completion_q.pop()) |task| self.gpa.destroy(task); } - pub fn run(self: *Runtime, condition: RunCondition) !void { + pub fn run(self: *Ring, condition: RunCondition) !void { while (true) { try self.backend.submitAndWait(&self.submission_q); try self.backend.reapCompletions(self); @@ -172,12 +172,12 @@ pub const Runtime = struct { } } - pub fn getTask(self: *Runtime) Allocator.Error!*Task { + pub fn getTask(self: *Ring) Allocator.Error!*Task { return self.free_q.pop() orelse try self.gpa.create(Task); } pub fn noop( - self: *Runtime, + self: *Ring, ctx: Context, ) Allocator.Error!*Task { const task = try self.getTask(); @@ -193,7 +193,7 @@ pub const Runtime = struct { } pub fn timer( - self: *Runtime, + self: *Ring, duration: Timespec, ctx: Context, ) Allocator.Error!*Task { @@ -209,7 +209,7 @@ pub const Runtime = struct { return task; } - pub fn cancelAll(self: *Runtime) Allocator.Error!void { + pub fn cancelAll(self: *Ring) Allocator.Error!void { const task = try self.getTask(); task.* = .{ .req = .{ .cancel = .all }, @@ -219,7 +219,7 @@ pub const Runtime = struct { } pub fn accept( - self: *Runtime, + self: *Ring, fd: posix.fd_t, ctx: Context, ) Allocator.Error!*Task { @@ -236,8 +236,8 @@ pub const Runtime = struct { } pub fn msgRing( - self: *Runtime, - target: *Runtime, + self: *Ring, + target: *Ring, target_task: *Task, // The task that the target ring will receive. The callbacks of // this task are what will be called when the target receives the message @@ -260,7 +260,7 @@ pub const Runtime = struct { } pub fn recv( - self: *Runtime, + self: *Ring, fd: posix.fd_t, buffer: []u8, ctx: Context, @@ -281,7 +281,7 @@ pub const Runtime = struct { } pub fn write( - self: *Runtime, + self: *Ring, fd: posix.fd_t, buffer: []const u8, ctx: Context, @@ -302,7 +302,7 @@ pub const Runtime = struct { } pub fn writev( - self: *Runtime, + self: *Ring, fd: posix.fd_t, vecs: []const posix.iovec_const, ctx: Context, @@ -323,7 +323,7 @@ pub const Runtime = struct { } pub fn close( - self: *Runtime, + self: *Ring, fd: posix.fd_t, ctx: Context, ) Allocator.Error!*Task { @@ -340,7 +340,7 @@ pub const Runtime = struct { } pub fn poll( - self: *Runtime, + self: *Ring, fd: posix.fd_t, mask: u32, ctx: Context, @@ -358,7 +358,7 @@ pub const Runtime = struct { } pub fn socket( - self: *Runtime, + self: *Ring, domain: u32, socket_type: u32, protocol: u32, @@ -377,7 +377,7 @@ pub const Runtime = struct { } pub fn connect( - self: *Runtime, + self: *Ring, fd: posix.socket_t, addr: *posix.sockaddr, addr_len: posix.socklen_t, @@ -411,7 +411,7 @@ pub const Op = enum { socket, connect, - /// userfd is meant to send file descriptors between Runtime instances (using msgRing) + /// userfd is meant to send file descriptors between Ring instances (using msgRing) userfd, /// usermsg is meant to send a u16 between runtime instances (using msgRing) usermsg, @@ -429,7 +429,7 @@ pub const Request = union(Op) { }, accept: posix.fd_t, msg_ring: struct { - target: *Runtime, + target: *Ring, task: *Task, }, recv: struct { @@ -521,14 +521,14 @@ test { const Foo = struct { bar: usize = 0, - fn callback(_: *io.Runtime, task: io.Task) anyerror!void { + fn callback(_: *io.Ring, task: io.Task) anyerror!void { const self = task.userdataCast(Foo); self.bar += 1; } }; test "runtime: noop" { - var rt: io.Runtime = try .init(std.testing.allocator, 16); + var rt: io.Ring = try .init(std.testing.allocator, 16); defer rt.deinit(); var foo: Foo = .{}; @@ -546,7 +546,7 @@ test "runtime: noop" { } test "runtime: timer" { - var rt: io.Runtime = try .init(std.testing.allocator, 16); + var rt: io.Ring = try .init(std.testing.allocator, 16); defer rt.deinit(); var foo: Foo = .{}; @@ -562,7 +562,7 @@ test "runtime: timer" { } test "runtime: poll" { - var rt: io.Runtime = try .init(std.testing.allocator, 16); + var rt: io.Ring = try .init(std.testing.allocator, 16); defer rt.deinit(); var foo: Foo = .{}; @@ -580,7 +580,7 @@ test "runtime: poll" { test "runtime: deadline doesn't call user callback" { const gpa = std.testing.allocator; - var rt = try io.Runtime.init(gpa, 16); + var rt = try io.Ring.init(gpa, 16); defer rt.deinit(); var foo: Foo = .{}; @@ -596,7 +596,7 @@ test "runtime: deadline doesn't call user callback" { test "runtime: timeout" { const gpa = std.testing.allocator; - var rt = try io.Runtime.init(gpa, 16); + var rt = try io.Ring.init(gpa, 16); defer rt.deinit(); var foo: Foo = .{}; @@ -613,7 +613,7 @@ test "runtime: timeout" { test "runtime: cancel" { const gpa = std.testing.allocator; - var rt = try io.Runtime.init(gpa, 16); + var rt = try io.Ring.init(gpa, 16); defer rt.deinit(); var foo: Foo = .{}; @@ -633,13 +633,13 @@ test "runtime: cancel" { test "runtime: cancel all" { const gpa = std.testing.allocator; - var rt = try io.Runtime.init(gpa, 16); + var rt = try io.Ring.init(gpa, 16); defer rt.deinit(); const Foo2 = struct { bar: usize = 0, - fn callback(_: *io.Runtime, task: io.Task) anyerror!void { + fn callback(_: *io.Ring, task: io.Task) anyerror!void { const self = task.userdataCast(@This()); const result = task.result.?; _ = result.timer catch |err| { @@ -669,7 +669,7 @@ test "runtime: cancel all" { test "runtime: msgRing" { const gpa = std.testing.allocator; - var rt1 = try io.Runtime.init(gpa, 16); + var rt1 = try io.Ring.init(gpa, 16); defer rt1.deinit(); var rt2 = try rt1.initChild(16); @@ -681,7 +681,7 @@ test "runtime: msgRing" { const Msg = enum { rt1, rt2 }; - fn callback(_: *io.Runtime, task: io.Task) anyerror!void { + fn callback(_: *io.Ring, task: io.Task) anyerror!void { const self = task.userdataCast(@This()); const msg = task.msgToEnum(Msg); switch (msg) { diff --git a/src/net.zig b/src/net.zig index 13aee9a..ae635eb 100644 --- a/src/net.zig +++ b/src/net.zig @@ -8,7 +8,7 @@ const Uri = std.Uri; const assert = std.debug.assert; pub fn tcpConnectToHost( - rt: *io.Runtime, + rt: *io.Ring, host: []const u8, port: u16, ctx: io.Context, @@ -26,7 +26,7 @@ pub fn tcpConnectToHost( } pub fn tcpConnectToAddr( - rt: *io.Runtime, + rt: *io.Ring, addr: std.net.Address, ctx: io.Context, ) Allocator.Error!*ConnectTask { @@ -68,11 +68,11 @@ pub const ConnectTask = struct { /// Cancels the current task. Not guaranteed to actually cancel. User's callback will get an /// error.Canceled if cancelation was successful, otherwise the operation will complete as /// normal and this is essentially a no-op - pub fn cancel(self: *ConnectTask, rt: *io.Runtime) void { + pub fn cancel(self: *ConnectTask, rt: *io.Ring) void { _ = self.task.cancel(rt, null, 0, io.noopCallback) catch {}; } - pub fn handleMsg(rt: *io.Runtime, task: io.Task) anyerror!void { + pub fn handleMsg(rt: *io.Ring, task: io.Task) anyerror!void { const self = task.userdataCast(ConnectTask); const result = task.result.?; switch (task.msgToEnum(Msg)) { @@ -127,7 +127,7 @@ pub const ConnectTask = struct { }; test "tcp connect" { - var rt: io.Runtime = try .init(std.testing.allocator, 16); + var rt: io.Ring = try .init(std.testing.allocator, 16); defer rt.deinit(); const addr: std.net.Address = try .parseIp4("127.0.0.1", 80); diff --git a/src/tls.zig b/src/tls.zig index d22c937..700ad19 100644 --- a/src/tls.zig +++ b/src/tls.zig @@ -22,7 +22,7 @@ pub const Client = struct { written: usize = 0, userdata: ?*anyopaque = null, - callback: *const fn (*io.Runtime, io.Task) anyerror!void = io.noopCallback, + callback: *const fn (*io.Ring, io.Task) anyerror!void = io.noopCallback, close_msg: u16 = 0, write_msg: u16 = 0, recv_msg: u16 = 0, @@ -38,7 +38,7 @@ pub const Client = struct { handshake: tls.nonblock.Client, task: *io.Task, - pub fn handleMsg(rt: *io.Runtime, task: io.Task) anyerror!void { + pub fn handleMsg(rt: *io.Ring, task: io.Task) anyerror!void { const self = task.userdataCast(HandshakeTask); const result = task.result.?; @@ -148,7 +148,7 @@ pub const Client = struct { /// Tries to cancel the handshake. Callback will receive an error.Canceled if cancelation /// was successful, otherwise handhsake will proceed - pub fn cancel(self: *HandshakeTask, rt: *io.Runtime) void { + pub fn cancel(self: *HandshakeTask, rt: *io.Ring) void { self.task.cancel(rt, null, 0, io.noopCallback) catch {}; } }; @@ -162,7 +162,7 @@ pub const Client = struct { /// Initializes a handshake, which will ultimately deliver a Client to the callback via a /// userptr result pub fn init( - rt: *io.Runtime, + rt: *io.Ring, fd: posix.fd_t, opts: tls.config.Client, ctx: io.Context, @@ -189,7 +189,7 @@ pub const Client = struct { self.cleartext_buf.deinit(gpa); } - pub fn close(self: *Client, gpa: Allocator, rt: *io.Runtime) !void { + pub fn close(self: *Client, gpa: Allocator, rt: *io.Ring) !void { // close notify is 2 bytes long const len = self.tls.encryptedLength(2); try self.ciphertext_buf.ensureUnusedCapacity(gpa, len); @@ -209,7 +209,7 @@ pub const Client = struct { } } - fn onCompletion(rt: *io.Runtime, task: io.Task) anyerror!void { + fn onCompletion(rt: *io.Ring, task: io.Task) anyerror!void { const self = task.userdataCast(Client); const result = task.result.?; @@ -317,7 +317,7 @@ pub const Client = struct { } } - pub fn recv(self: *Client, rt: *io.Runtime) !void { + pub fn recv(self: *Client, rt: *io.Ring) !void { if (self.recv_task != null) return; self.recv_task = try rt.recv( self.fd, @@ -330,7 +330,7 @@ pub const Client = struct { try self.cleartext_buf.appendSlice(gpa, bytes); } - pub fn flush(self: *Client, gpa: Allocator, rt: *io.Runtime) !void { + pub fn flush(self: *Client, gpa: Allocator, rt: *io.Ring) !void { const len = self.tls.encryptedLength(self.cleartext_buf.items.len); try self.ciphertext_buf.ensureUnusedCapacity(gpa, len); const slice = self.ciphertext_buf.unusedCapacitySlice(); @@ -372,7 +372,7 @@ test "tls: Client" { const net = @import("net.zig"); const gpa = std.testing.allocator; - var rt = try io.Runtime.init(gpa, 16); + var rt = try io.Ring.init(gpa, 16); defer rt.deinit(); const Foo = struct { @@ -389,7 +389,7 @@ test "tls: Client" { recv, }; - fn callback(_: *io.Runtime, task: io.Task) anyerror!void { + fn callback(_: *io.Ring, task: io.Task) anyerror!void { const self = task.userdataCast(Self); const result = task.result.?; errdefer { -- 2.51.2