From 6324fdb00954889754d142aa4fb617893db2b4e6 Mon Sep 17 00:00:00 2001 From: Francis Bouvier Date: Wed, 20 Nov 2024 15:43:22 +0100 Subject: [PATCH 1/6] Add a Wrapper to allow use by an external IO implementation And rename Loop -> IO for better wording Signed-off-by: Francis Bouvier --- src/Blocking.zig | 67 ++++++++++++++ src/io.zig | 192 +++++++++++++++++++++++++++++----------- src/main.zig | 7 +- src/std/http/Client.zig | 9 +- src/std/net.zig | 6 +- 5 files changed, 219 insertions(+), 62 deletions(-) create mode 100644 src/Blocking.zig diff --git a/src/Blocking.zig b/src/Blocking.zig new file mode 100644 index 0000000..b272f94 --- /dev/null +++ b/src/Blocking.zig @@ -0,0 +1,67 @@ +const std = @import("std"); + +// Blocking is an example implementation of an IO API +// following the zig-async-io model. +// As it name suggests in this implementation all operations are +// in fact blocking, the async API is just faked. +pub const Blocking = @This(); + +pub const Completion = void; + +pub const ConnectError = std.posix.ConnectError; +pub const SendError = std.posix.WriteError; +pub const RecvError = std.posix.ReadError; + +pub fn connect( + _: *Blocking, + comptime CtxT: type, + ctx: *CtxT, + _: *Completion, + comptime cbk: fn (ctx: *CtxT, _: *Completion, res: ConnectError!void) void, + socket: std.posix.socket_t, + address: std.net.Address, +) void { + std.posix.connect(socket, &address.any, address.getOsSockLen()) catch |err| { + cbk(ctx, @constCast(&{}), err); + return; + }; + cbk(ctx, @constCast(&{}), {}); +} + +pub fn onConnect(_: *Blocking, _: ConnectError!void) void {} + +pub fn send( + _: *Blocking, + comptime CtxT: type, + ctx: *CtxT, + _: *Completion, + comptime cbk: fn (ctx: *CtxT, _: *Completion, res: SendError!usize) void, + socket: std.posix.socket_t, + buf: []const u8, +) void { + const len = std.posix.write(socket, buf) catch |err| { + cbk(ctx, @constCast(&{}), err); + return; + }; + cbk(ctx, @constCast(&{}), len); +} + +pub fn onSend(_: *Blocking, _: SendError!usize) void {} + +pub fn recv( + _: *Blocking, + comptime CtxT: type, + ctx: *CtxT, + _: *Completion, + comptime cbk: fn (ctx: *CtxT, _: *Completion, res: RecvError!usize) void, + socket: std.posix.socket_t, + buf: []u8, +) void { + const len = std.posix.read(socket, buf) catch |err| { + cbk(ctx, @constCast(&{}), err); + return; + }; + cbk(ctx, @constCast(&{}), len); +} + +pub fn onRecv(_: *Blocking, _: RecvError!usize) void {} diff --git a/src/io.zig b/src/io.zig index b4d1841..c273e6d 100644 --- a/src/io.zig +++ b/src/io.zig @@ -1,59 +1,147 @@ const std = @import("std"); -pub const Ctx = @import("std/http/Client.zig").Ctx; -pub const Cbk = @import("std/http/Client.zig").Cbk; - -pub const Blocking = struct { - pub fn connect( - _: *Blocking, - comptime ctxT: type, - ctx: *ctxT, - comptime cbk: Cbk, - socket: std.posix.socket_t, - address: std.net.Address, - ) void { - std.posix.connect(socket, &address.any, address.getOsSockLen()) catch |err| { - std.posix.close(socket); - cbk(ctx, err) catch |e| { - ctx.setErr(e); - }; - }; - cbk(ctx, {}) catch |e| ctx.setErr(e); +// IO is a type defined via a root declaration. +// It must implements the following methods: +// - connect, onConnect +// - send, onSend +// - recv, onRecv +// It must also define the following types: +// - Completion +// - ConnectError +// - SendError +// - RecvError +// see Blocking.io for an implementation example. +pub const IO = blk: { + const root = @import("root"); + if (@hasDecl(root, "IO")) { + break :blk root.IO; } + @compileError("no IO API defined at root"); +}; + +// Wrapper for a base IO API. +pub fn Wrapper(IO_T: type) type { + return struct { + io: *IO_T, + completion: IO_T.Completion, + + const Self = @This(); + + pub fn init(io: *IO_T) Self { + return .{ .io = io, .completion = undefined }; + } + + // NOTE: Business methods connect, send, recv expect a Ctx + // who should reference the base IO API in Ctx.io field + + // NOTE: Ctx is already none (ie. @import("std/http/Client.zig").Ctx) + // but we require to provide it's type (comptime) as argument + // to avoid dependancy loop + // ie. Wrapper requiring Ctx and Ctx requiring Wrapper + + fn Cbk(comptime Ctx: type) type { + return *const fn (ctx: *Ctx, res: anyerror!void) anyerror!void; + } - pub fn send( - _: *Blocking, - comptime ctxT: type, - ctx: *ctxT, - comptime cbk: Cbk, - socket: std.posix.socket_t, - buf: []const u8, - ) void { - const len = std.posix.write(socket, buf) catch |err| { - cbk(ctx, err) catch |e| { - return ctx.setErr(e); + pub fn connect( + self: *Self, + comptime Ctx: type, + ctx: *Ctx, + comptime cbk: Cbk(Ctx), + socket: std.posix.socket_t, + address: std.net.Address, + ) void { + self.io.connect(Ctx, ctx, &self.completion, onConnect(Ctx, cbk), socket, address); + } + + fn onConnectFn(comptime Ctx: type) type { + return fn ( + ctx: *Ctx, + _: *IO_T.Completion, + result: IO_T.ConnectError!void, + ) void; + } + fn onConnect(comptime Ctx: type, comptime cbk: Cbk(Ctx)) onConnectFn(Ctx) { + const s = struct { + fn on( + ctx: *Ctx, + _: *IO_T.Completion, + result: IO_T.ConnectError!void, + ) void { + ctx.io.io.onConnect(result); // base IO callback + _ = result catch |err| return ctx.setErr(err); + cbk(ctx, {}) catch |err| return ctx.setErr(err); + } }; - return ctx.setErr(err); - }; - ctx.setLen(len); - cbk(ctx, {}) catch |e| ctx.setErr(e); - } + return s.on; + } - pub fn recv( - _: *Blocking, - comptime ctxT: type, - ctx: *ctxT, - comptime cbk: Cbk, - socket: std.posix.socket_t, - buf: []u8, - ) void { - const len = std.posix.read(socket, buf) catch |err| { - cbk(ctx, err) catch |e| { - return ctx.setErr(e); + pub fn send( + self: *Self, + comptime Ctx: type, + ctx: *Ctx, + comptime cbk: Cbk(Ctx), + socket: std.posix.socket_t, + buf: []const u8, + ) void { + self.io.send(Ctx, ctx, &self.completion, onSend(Ctx, cbk), socket, buf); + } + + fn onSendFn(comptime Ctx: type) type { + return fn ( + ctx: *Ctx, + _: *IO_T.Completion, + result: IO_T.SendError!usize, + ) void; + } + fn onSend(comptime Ctx: type, comptime cbk: Cbk(Ctx)) onSendFn(Ctx) { + const s = struct { + fn on( + ctx: *Ctx, + _: *IO_T.Completion, + result: IO_T.SendError!usize, + ) void { + ctx.io.io.onSend(result); // base IO callback + const len = result catch |err| return ctx.setErr(err); + ctx.setLen(len); + cbk(ctx, {}) catch |e| ctx.setErr(e); + } }; - return ctx.setErr(err); - }; - ctx.setLen(len); - cbk(ctx, {}) catch |e| ctx.setErr(e); - } -}; + return s.on; + } + + pub fn recv( + self: *Self, + comptime Ctx: type, + ctx: *Ctx, + comptime cbk: Cbk(Ctx), + socket: std.posix.socket_t, + buf: []u8, + ) void { + self.io.recv(Ctx, ctx, &self.completion, onRecv(Ctx, cbk), socket, buf); + } + + fn onRecvFn(comptime Ctx: type) type { + return fn ( + ctx: *Ctx, + _: *IO_T.Completion, + result: IO_T.RecvError!usize, + ) void; + } + fn onRecv(comptime Ctx: type, comptime cbk: Cbk(Ctx)) onRecvFn(Ctx) { + const s = struct { + fn do( + ctx: *Ctx, + _: *IO_T.Completion, + result: IO_T.RecvError!usize, + ) void { + ctx.io.io.onRecv(result); // base IO callback + const len = result catch |err| return ctx.setErr(err); + ctx.setLen(len); + cbk(ctx, {}) catch |err| return ctx.setErr(err); + } + }; + return s.do; + } + }; +} diff --git a/src/main.zig b/src/main.zig index 0a4a7ae..adddbe8 100644 --- a/src/main.zig +++ b/src/main.zig @@ -3,7 +3,9 @@ const std = @import("std"); const stack = @import("stack.zig"); pub const Client = @import("std/http/Client.zig"); const Ctx = Client.Ctx; -const Loop = @import("io.zig").Blocking; +pub const Wrapper = @import("io.zig").Wrapper; +const Blocking = @import("blocking.zig").Blocking; +pub const IO = Wrapper(Blocking); const root = @import("root"); @@ -53,7 +55,8 @@ pub fn run() !void { }; const alloc = gpa.allocator(); - var loop = Loop{}; + var blocking = Blocking{}; + var loop = IO.init(&blocking); var client = Client{ .allocator = alloc }; defer client.deinit(); diff --git a/src/std/http/Client.zig b/src/std/http/Client.zig index 9ff02c7..0506dc1 100644 --- a/src/std/http/Client.zig +++ b/src/std/http/Client.zig @@ -21,8 +21,7 @@ const proto = @import("protocol.zig"); const tls23 = @import("../../tls.zig/main.zig"); const VecPut = @import("../../tls.zig/connection.zig").VecPut; const GenericStack = @import("../../stack.zig").Stack; -const async_io = @import("../../io.zig"); -const Loop = async_io.Blocking; +pub const IO = @import("../../io.zig").IO; const cipher = @import("../../tls.zig/cipher.zig"); @@ -2390,7 +2389,7 @@ pub const Ctx = struct { userData: *anyopaque = undefined, - loop: *Loop, + io: *IO, data: Data, stack: ?*Stack = null, err: ?anyerror = null, @@ -2419,7 +2418,7 @@ pub const Ctx = struct { _tls_write_index: usize = 0, _tls_write_buf: [cipher.max_ciphertext_record_len]u8 = undefined, - pub fn init(loop: *Loop, req: *Request) !Ctx { + pub fn init(io: *IO, req: *Request) !Ctx { const connection = try req.client.allocator.create(Connection); connection.* = .{ .stream = undefined, @@ -2430,7 +2429,7 @@ pub const Ctx = struct { }; return .{ .req = req, - .loop = loop, + .io = io, .data = .{ .conn = connection }, }; } diff --git a/src/std/net.zig b/src/std/net.zig index 86863de..d6a1509 100644 --- a/src/std/net.zig +++ b/src/std/net.zig @@ -1908,7 +1908,7 @@ pub const Stream = struct { ctx: *Ctx, comptime cbk: Cbk, ) !void { - return ctx.loop.recv(Ctx, ctx, cbk, self.handle, buffer); + return ctx.io.recv(Ctx, ctx, cbk, self.handle, buffer); } pub fn async_readv( @@ -1924,7 +1924,7 @@ pub const Stream = struct { // TODO: why not take a buffer here? pub fn async_write(self: Stream, buffer: []const u8, ctx: *Ctx, comptime cbk: Cbk) void { - return ctx.loop.send(Ctx, ctx, cbk, self.handle, buffer); + return ctx.io.send(Ctx, ctx, cbk, self.handle, buffer); } fn onWriteAll(ctx: *Ctx, res: anyerror!void) anyerror!void { @@ -2033,7 +2033,7 @@ pub fn async_tcpConnectToAddress(address: std.net.Address, ctx: *Ctx, comptime c ctx.data.socket = sockfd; ctx.push(cbk) catch |e| return ctx.pop(e); - ctx.loop.connect( + ctx.io.connect( Ctx, ctx, setStream, From beec41ae2a9294d6f9647dfdc5b317a0b27b77c2 Mon Sep 17 00:00:00 2001 From: Francis Bouvier Date: Wed, 20 Nov 2024 18:40:27 +0100 Subject: [PATCH 2/6] Fix tests Signed-off-by: Francis Bouvier --- .github/workflows/zig-test.yml | 1 - build.zig | 6 +++++- src/lib.zig | 2 ++ src/test_runner.zig | 18 ++++++++++++++++++ src/{main.zig => tests.zig} | 23 +++++------------------ 5 files changed, 30 insertions(+), 20 deletions(-) create mode 100644 src/lib.zig create mode 100644 src/test_runner.zig rename src/{main.zig => tests.zig} (83%) diff --git a/.github/workflows/zig-test.yml b/.github/workflows/zig-test.yml index e557252..c8cefb4 100644 --- a/.github/workflows/zig-test.yml +++ b/.github/workflows/zig-test.yml @@ -31,4 +31,3 @@ jobs: fetch-depth: 0 - run: zig build test - - run: zig build run diff --git a/build.zig b/build.zig index a67d058..f62cb4b 100644 --- a/build.zig +++ b/build.zig @@ -46,12 +46,16 @@ pub fn build(b: *std.Build) void { run_step.dependOn(&run_cmd.step); const exe_unit_tests = b.addTest(.{ - .root_source_file = b.path("src/main.zig"), + .root_source_file = b.path("src/tests.zig"), + .test_runner = b.path("src/test_runner.zig"), .target = target, .optimize = optimize, }); const run_exe_unit_tests = b.addRunArtifact(exe_unit_tests); + if (b.args) |args| { + run_exe_unit_tests.addArgs(args); + } // Similar to creating the run step earlier, this exposes a `test` step to // the `zig build --help` menu, providing a way for the user to request diff --git a/src/lib.zig b/src/lib.zig new file mode 100644 index 0000000..701a393 --- /dev/null +++ b/src/lib.zig @@ -0,0 +1,2 @@ +pub const Wrapper = @import("io.zig").Wrapper; +pub const Client = @import("std/http/Client.zig"); diff --git a/src/test_runner.zig b/src/test_runner.zig new file mode 100644 index 0000000..5208d6e --- /dev/null +++ b/src/test_runner.zig @@ -0,0 +1,18 @@ +const std = @import("std"); +const builtin = @import("builtin"); + +pub const Blocking = @import("Blocking.zig").Blocking; +pub const IO = @import("io.zig").Wrapper(Blocking); + +pub const tests = @import("tests.zig"); + +pub fn main() !void { + std.testing.refAllDecls(tests); + for (builtin.test_functions) |test_fn| { + test_fn.func() catch |err| { + if (err == error.SkipZigTest) continue; + return err; + }; + std.debug.print("{s}\tOK\n", .{test_fn.name}); + } +} diff --git a/src/main.zig b/src/tests.zig similarity index 83% rename from src/main.zig rename to src/tests.zig index adddbe8..7f9755b 100644 --- a/src/main.zig +++ b/src/tests.zig @@ -1,13 +1,11 @@ const std = @import("std"); +const Ctx = Client.Ctx; const stack = @import("stack.zig"); pub const Client = @import("std/http/Client.zig"); -const Ctx = Client.Ctx; -pub const Wrapper = @import("io.zig").Wrapper; -const Blocking = @import("blocking.zig").Blocking; -pub const IO = Wrapper(Blocking); -const root = @import("root"); +const IO = @import("root").IO; +const Blocking = @import("root").Blocking; fn onRequestWait(ctx: *Ctx, res: anyerror!void) !void { res catch |e| { @@ -39,12 +37,7 @@ pub fn onRequestConnect(ctx: *Ctx, res: anyerror!void) anyerror!void { return ctx.req.async_send(ctx, onRequestSend); } -pub fn main() !void { - return run(); -} - -pub fn run() !void { - +test "example.com" { // const url = "http://127.0.0.1:8080"; const url = "https://www.example.com"; @@ -79,11 +72,5 @@ pub fn run() !void { onRequestConnect, ); - std.log.debug("Final error: {any}", .{ctx.err}); -} - -test { - _ = stack.Stack(fn () void); - std.testing.refAllDecls(@This()); - try run(); + try std.testing.expect(ctx.err == null); } From d996742c00f518be4f088af69d81912b8df94d58 Mon Sep 17 00:00:00 2001 From: Francis Bouvier Date: Wed, 20 Nov 2024 19:38:33 +0100 Subject: [PATCH 3/6] Remove run cmd Signed-off-by: Francis Bouvier --- build.zig | 38 ++++---------------------------------- 1 file changed, 4 insertions(+), 34 deletions(-) diff --git a/build.zig b/build.zig index f62cb4b..743d99a 100644 --- a/build.zig +++ b/build.zig @@ -15,51 +15,21 @@ pub fn build(b: *std.Build) void { // set a preferred release mode, allowing the user to decide how to optimize. const optimize = b.standardOptimizeOption(.{}); - const exe = b.addExecutable(.{ - .name = "zigio", - .root_source_file = b.path("src/main.zig"), - .target = target, - .optimize = optimize, - }); - - // This *creates* a Run step in the build graph, to be executed when another - // step is evaluated that depends on it. The next line below will establish - // such a dependency. - const run_cmd = b.addRunArtifact(exe); - - // By making the run step depend on the install step, it will be run from the - // installation directory rather than directly from within the cache directory. - // This is not necessary, however, if the application depends on other installed - // files, this ensures they will be present and in the expected location. - run_cmd.step.dependOn(b.getInstallStep()); - - // This allows the user to pass arguments to the application in the build - // command itself, like this: `zig build run -- arg1 arg2 etc` - if (b.args) |args| { - run_cmd.addArgs(args); - } - - // This creates a build step. It will be visible in the `zig build --help` menu, - // and can be selected like this: `zig build run` - // This will evaluate the `run` step rather than the default, which is "install". - const run_step = b.step("run", "Run the app"); - run_step.dependOn(&run_cmd.step); - - const exe_unit_tests = b.addTest(.{ + const tests = b.addTest(.{ .root_source_file = b.path("src/tests.zig"), .test_runner = b.path("src/test_runner.zig"), .target = target, .optimize = optimize, }); - const run_exe_unit_tests = b.addRunArtifact(exe_unit_tests); + const run_tests = b.addRunArtifact(tests); if (b.args) |args| { - run_exe_unit_tests.addArgs(args); + run_tests.addArgs(args); } // Similar to creating the run step earlier, this exposes a `test` step to // the `zig build --help` menu, providing a way for the user to request // running the unit tests. const test_step = b.step("test", "Run unit tests"); - test_step.dependOn(&run_exe_unit_tests.step); + test_step.dependOn(&run_tests.step); } From b111de532692eae0bd79b9b4790ff620380cbb0c Mon Sep 17 00:00:00 2001 From: Francis Bouvier Date: Thu, 21 Nov 2024 16:43:42 +0100 Subject: [PATCH 4/6] Update src/io.zig Co-authored-by: Pierre Tachoire --- src/io.zig | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/io.zig b/src/io.zig index c273e6d..63a9226 100644 --- a/src/io.zig +++ b/src/io.zig @@ -34,7 +34,7 @@ pub fn Wrapper(IO_T: type) type { // NOTE: Business methods connect, send, recv expect a Ctx // who should reference the base IO API in Ctx.io field - // NOTE: Ctx is already none (ie. @import("std/http/Client.zig").Ctx) + // NOTE: Ctx is already known (ie. @import("std/http/Client.zig").Ctx) // but we require to provide it's type (comptime) as argument // to avoid dependancy loop // ie. Wrapper requiring Ctx and Ctx requiring Wrapper From 2df5438f8bce070925442be6a0c2e93ef320dc06 Mon Sep 17 00:00:00 2001 From: Francis Bouvier Date: Thu, 21 Nov 2024 16:43:49 +0100 Subject: [PATCH 5/6] Update src/io.zig Co-authored-by: Pierre Tachoire --- src/io.zig | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/io.zig b/src/io.zig index 63a9226..b7172d3 100644 --- a/src/io.zig +++ b/src/io.zig @@ -36,7 +36,7 @@ pub fn Wrapper(IO_T: type) type { // NOTE: Ctx is already known (ie. @import("std/http/Client.zig").Ctx) // but we require to provide it's type (comptime) as argument - // to avoid dependancy loop + // to avoid dependency loop // ie. Wrapper requiring Ctx and Ctx requiring Wrapper fn Cbk(comptime Ctx: type) type { From 2831fbe58b716d1430b8aa008618e9c285a7eb22 Mon Sep 17 00:00:00 2001 From: Francis Bouvier Date: Thu, 21 Nov 2024 16:43:56 +0100 Subject: [PATCH 6/6] Update src/io.zig Co-authored-by: Pierre Tachoire --- src/io.zig | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/io.zig b/src/io.zig index b7172d3..4bbb7be 100644 --- a/src/io.zig +++ b/src/io.zig @@ -35,7 +35,7 @@ pub fn Wrapper(IO_T: type) type { // who should reference the base IO API in Ctx.io field // NOTE: Ctx is already known (ie. @import("std/http/Client.zig").Ctx) - // but we require to provide it's type (comptime) as argument + // but we require to provide its type (comptime) as argument // to avoid dependency loop // ie. Wrapper requiring Ctx and Ctx requiring Wrapper