async: refacto with comptime generation

src/async/http.zig

@@ -3,27 +3,34 @@ const http = std.http;
 const stdcli = @import("Client.zig");
 
 pub const Loop = @import("jsruntime").Loop;
+const YieldImpl = Loop.Yield(Request);
 
 pub const Client = struct {
     cli: stdcli,
 
     pub fn init(alloc: std.mem.Allocator, loop: *Loop) Client {
-        return .{ .cli = .{
+        return .{
+            .cli = .{
                 .allocator = alloc,
                 .loop = loop,
-        } };
+            },
+        };
     }
 
     pub fn deinit(self: *Client) void {
         self.cli.deinit();
     }
 
-    pub fn create(self: *Client, uri: std.Uri) Request {
-        return .{
+    pub fn create(self: *Client, uri: std.Uri) !*Request {
+        var req = try self.cli.allocator.create(Request);
+        req.* = Request{
+            .impl = undefined,
             .cli = &self.cli,
             .uri = uri,
             .headers = .{ .allocator = self.cli.allocator, .owned = false },
         };
+        req.impl = YieldImpl.init(self.cli.loop, req);
+        return req;
     }
 };
 
@@ -32,24 +39,26 @@ pub const Request = struct {
     uri: std.Uri,
     headers: std.http.Headers,
 
+    impl: YieldImpl,
     done: bool = false,
     err: ?anyerror = null,
 
     pub fn deinit(self: *Request) void {
         self.headers.deinit();
+        self.cli.allocator.destroy(self);
     }
 
-    pub fn fetch(self: *Request) !void {
-        self.cli.loop.yield(*Request, self, callback);
+    pub fn fetch(self: *Request) void {
+        return self.impl.yield();
     }
 
     fn onerr(self: *Request, err: anyerror) void {
         self.err = err;
     }
 
-    fn callback(self: *Request, err: ?anyerror) void {
-        if (err) |e| return self.onerr(e);
+    pub fn onYield(self: *Request, err: ?anyerror) void {
         defer self.done = true;
+        if (err) |e| return self.onerr(e);
         var req = self.cli.open(.GET, self.uri, self.headers, .{}) catch |e| return self.onerr(e);
         defer req.deinit();
 
@@ -59,7 +68,7 @@ pub const Request = struct {
     }
 
     pub fn wait(self: *Request) !void {
-        while (!self.done) try self.cli.loop.tick();
+        while (!self.done) try self.impl.tick();
         if (self.err) |err| return err;
     }
 };

src/async/stream.zig

@@ -4,65 +4,17 @@ const os = std.os;
 const io = std.io;
 const assert = std.debug.assert;
 
-const Loop = @import("jsruntime").Loop;
-
-const WriteCmd = struct {
-    const Self = @This();
-
-    stream: Stream,
-    done: bool = false,
-    res: usize = undefined,
-    err: ?anyerror = null,
-
-    fn run(self: *Self, buffer: []const u8) void {
-        self.stream.loop.send(*Self, self, callback, self.stream.handle, buffer);
-    }
-
-    fn callback(self: *Self, err: ?anyerror, res: usize) void {
-        self.res = res;
-        self.err = err;
-        self.done = true;
-    }
-
-    fn wait(self: *Self) !usize {
-        while (!self.done) try self.stream.loop.tick();
-        if (self.err) |err| return err;
-        return self.res;
-    }
-};
-
-const ReadCmd = struct {
-    const Self = @This();
-
-    stream: Stream,
-    done: bool = false,
-    res: usize = undefined,
-    err: ?anyerror = null,
-
-    fn run(self: *Self, buffer: []u8) void {
-        self.stream.loop.receive(*Self, self, callback, self.stream.handle, buffer);
-    }
-
-    fn callback(self: *Self, _: []const u8, err: ?anyerror, res: usize) void {
-        self.res = res;
-        self.err = err;
-        self.done = true;
-    }
-
-    fn wait(self: *Self) !usize {
-        while (!self.done) try self.stream.loop.tick();
-        if (self.err) |err| return err;
-        return self.res;
-    }
-};
+const tcp = @import("tcp.zig");
 
 pub const Stream = struct {
-    loop: *Loop,
+    alloc: std.mem.Allocator,
+    conn: *tcp.Conn,
 
     handle: std.os.socket_t,
 
     pub fn close(self: Stream) void {
         os.closeSocket(self.handle);
+        self.alloc.destroy(self.conn);
     }
 
     pub const ReadError = os.ReadError;
@@ -80,9 +32,7 @@ pub const Stream = struct {
     }
 
     pub fn read(self: Stream, buffer: []u8) ReadError!usize {
-        var cmd = ReadCmd{ .stream = self };
-        cmd.run(buffer);
-        return cmd.wait() catch |err| switch (err) {
+        return self.conn.receive(self.handle, buffer) catch |err| switch (err) {
             else => return error.Unexpected,
         };
     }
@@ -118,10 +68,7 @@ pub const Stream = struct {
     /// file system thread instead of non-blocking. It needs to be reworked to properly
     /// use non-blocking I/O.
     pub fn write(self: Stream, buffer: []const u8) WriteError!usize {
-        var cmd = WriteCmd{ .stream = self };
-        cmd.run(buffer);
-
-        return cmd.wait() catch |err| switch (err) {
+        return self.conn.send(self.handle, buffer) catch |err| switch (err) {
             error.AccessDenied => error.AccessDenied,
             error.WouldBlock => error.WouldBlock,
             error.ConnectionResetByPeer => error.ConnectionResetByPeer,

src/async/tcp.zig

@@ -2,27 +2,64 @@ const std = @import("std");
 const net = std.net;
 const Stream = @import("stream.zig").Stream;
 const Loop = @import("jsruntime").Loop;
+const NetworkImpl = Loop.Network(Conn.Command);
 
-const ConnectCmd = struct {
-    const Self = @This();
+// Conn is a TCP connection using jsruntime Loop async I/O.
+// connect, send and receive are blocking, but use async I/O in the background.
+// Client doesn't own the socket used for the connection, the caller is
+// responsible for closing it.
+pub const Conn = struct {
+    const Command = struct {
+        impl: NetworkImpl,
 
-    loop: *Loop,
-    socket: std.os.socket_t,
-    err: ?anyerror = null,
         done: bool = false,
+        err: ?anyerror = null,
+        ln: usize = 0,
 
-    fn run(self: *Self, addr: std.net.Address) !void {
-        self.loop.connect(*Self, self, callback, self.socket, addr);
-    }
-
-    fn callback(self: *Self, _: std.os.socket_t, err: ?anyerror) void {
+        fn ok(self: *Command, err: ?anyerror, ln: usize) void {
             self.err = err;
+            self.ln = ln;
             self.done = true;
         }
 
-    fn wait(self: *Self) !void {
-        while (!self.done) try self.loop.tick();
+        fn wait(self: *Command) !usize {
+            while (!self.done) try self.impl.tick();
+
             if (self.err) |err| return err;
+            return self.ln;
+        }
+        pub fn onConnect(self: *Command, err: ?anyerror) void {
+            self.ok(err, 0);
+        }
+        pub fn onSend(self: *Command, ln: usize, err: ?anyerror) void {
+            self.ok(err, ln);
+        }
+        pub fn onReceive(self: *Command, ln: usize, err: ?anyerror) void {
+            self.ok(err, ln);
+        }
+    };
+
+    loop: *Loop,
+
+    pub fn connect(self: *Conn, socket: std.os.socket_t, address: std.net.Address) !void {
+        var cmd = Command{ .impl = undefined };
+        cmd.impl = NetworkImpl.init(self.loop, &cmd);
+        cmd.impl.connect(socket, address);
+        _ = try cmd.wait();
+    }
+
+    pub fn send(self: *Conn, socket: std.os.socket_t, buffer: []const u8) !usize {
+        var cmd = Command{ .impl = undefined };
+        cmd.impl = NetworkImpl.init(self.loop, &cmd);
+        cmd.impl.send(socket, buffer);
+        return try cmd.wait();
+    }
+
+    pub fn receive(self: *Conn, socket: std.os.socket_t, buffer: []u8) !usize {
+        var cmd = Command{ .impl = undefined };
+        cmd.impl = NetworkImpl.init(self.loop, &cmd);
+        cmd.impl.receive(socket, buffer);
+        return try cmd.wait();
     }
 };
 
@@ -34,7 +71,7 @@ pub fn tcpConnectToHost(alloc: std.mem.Allocator, loop: *Loop, name: []const u8,
     if (list.addrs.len == 0) return error.UnknownHostName;
 
     for (list.addrs) |addr| {
-        return tcpConnectToAddress(loop, addr) catch |err| switch (err) {
+        return tcpConnectToAddress(alloc, loop, addr) catch |err| switch (err) {
             error.ConnectionRefused => {
                 continue;
             },
@@ -44,19 +81,17 @@ pub fn tcpConnectToHost(alloc: std.mem.Allocator, loop: *Loop, name: []const u8,
     return std.os.ConnectError.ConnectionRefused;
 }
 
-pub fn tcpConnectToAddress(loop: *Loop, addr: net.Address) !Stream {
-    const sockfd = try loop.open(addr.any.family, std.os.SOCK.STREAM, std.os.IPPROTO.TCP);
+pub fn tcpConnectToAddress(alloc: std.mem.Allocator, loop: *Loop, addr: net.Address) !Stream {
+    const sockfd = try std.os.socket(addr.any.family, std.os.SOCK.STREAM, std.os.IPPROTO.TCP);
     errdefer std.os.closeSocket(sockfd);
 
-    var cmd = ConnectCmd{
-        .loop = loop,
-        .socket = sockfd,
-    };
-    try cmd.run(addr);
-    try cmd.wait();
+    var conn = try alloc.create(Conn);
+    conn.* = Conn{ .loop = loop };
+    try conn.connect(sockfd, addr);
 
     return Stream{
-        .loop = loop,
+        .alloc = alloc,
+        .conn = conn,
         .handle = sockfd,
     };
 }

src/async/test.zig

@@ -6,7 +6,9 @@ const AsyncRequest = @import("http.zig").Request;
 
 pub const Loop = @import("jsruntime").Loop;
 
-const url = "https://www.w3.org/";
+const TCPClient = @import("tcp.zig").Client;
+
+const url = "https://w3.org";
 
 test "blocking mode fetch API" {
     const alloc = std.testing.allocator;
@@ -69,10 +71,10 @@ test "non blocking mode API" {
     var client = AsyncClient.init(alloc, &loop);
     defer client.deinit();
 
-    var reqs: [10]AsyncRequest = undefined;
+    var reqs: [10]*AsyncRequest = undefined;
     for (0..reqs.len) |i| {
-        reqs[i] = client.create(try std.Uri.parse(url));
-        try reqs[i].fetch();
+        reqs[i] = try client.create(try std.Uri.parse(url));
+        reqs[i].fetch();
     }
 
     for (0..reqs.len) |i| {