Merge pull request #302 from lightpanda-io/zig-async-io

Use zig-async-io for xhr requests

.gitmodules

@@ -25,3 +25,6 @@
 [submodule "vendor/tls.zig"]
 	path = vendor/tls.zig
 	url = git@github.com:ianic/tls.zig.git
+[submodule "vendor/zig-async-io"]
+	path = vendor/zig-async-io
+	url = git@github.com:lightpanda-io/zig-async-io.git

build.zig

@@ -159,6 +159,11 @@ fn common(
     netsurf.addImport("jsruntime", jsruntimemod);
     step.root_module.addImport("netsurf", netsurf);
 
+    const asyncio = b.addModule("asyncio", .{
+        .root_source_file = b.path("vendor/zig-async-io/src/lib.zig"),
+    });
+    step.root_module.addImport("asyncio", asyncio);
+
     const tlsmod = b.addModule("tls", .{
         .root_source_file = b.path("vendor/tls.zig/src/main.zig"),
     });

src/async/stream.zig

@@ -1,133 +0,0 @@
-// Copyright (C) 2023-2024  Lightpanda (Selecy SAS)
-//
-// Francis Bouvier <francis@lightpanda.io>
-// Pierre Tachoire <pierre@lightpanda.io>
-//
-// This program is free software: you can redistribute it and/or modify
-// it under the terms of the GNU Affero General Public License as
-// published by the Free Software Foundation, either version 3 of the
-// License, or (at your option) any later version.
-//
-// This program is distributed in the hope that it will be useful,
-// but WITHOUT ANY WARRANTY; without even the implied warranty of
-// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
-// GNU Affero General Public License for more details.
-//
-// You should have received a copy of the GNU Affero General Public License
-// along with this program.  If not, see <https://www.gnu.org/licenses/>.
-
-const std = @import("std");
-const builtin = @import("builtin");
-const posix = std.posix;
-const io = std.io;
-const assert = std.debug.assert;
-
-const tcp = @import("tcp.zig");
-
-pub const Stream = struct {
-    alloc: std.mem.Allocator,
-    conn: *tcp.Conn,
-
-    handle: posix.socket_t,
-
-    pub fn close(self: Stream) void {
-        posix.close(self.handle);
-        self.alloc.destroy(self.conn);
-    }
-
-    pub const ReadError = posix.ReadError;
-    pub const WriteError = posix.WriteError;
-
-    pub const Reader = io.Reader(Stream, ReadError, read);
-    pub const Writer = io.Writer(Stream, WriteError, write);
-
-    pub fn reader(self: Stream) Reader {
-        return .{ .context = self };
-    }
-
-    pub fn writer(self: Stream) Writer {
-        return .{ .context = self };
-    }
-
-    pub fn read(self: Stream, buffer: []u8) ReadError!usize {
-        return self.conn.receive(self.handle, buffer) catch |err| switch (err) {
-            else => return error.Unexpected,
-        };
-    }
-
-    pub fn readv(s: Stream, iovecs: []const posix.iovec) ReadError!usize {
-        return posix.readv(s.handle, iovecs);
-    }
-
-    /// Returns the number of bytes read. If the number read is smaller than
-    /// `buffer.len`, it means the stream reached the end. Reaching the end of
-    /// a stream is not an error condition.
-    pub fn readAll(s: Stream, buffer: []u8) ReadError!usize {
-        return readAtLeast(s, buffer, buffer.len);
-    }
-
-    /// Returns the number of bytes read, calling the underlying read function
-    /// the minimal number of times until the buffer has at least `len` bytes
-    /// filled. If the number read is less than `len` it means the stream
-    /// reached the end. Reaching the end of the stream is not an error
-    /// condition.
-    pub fn readAtLeast(s: Stream, buffer: []u8, len: usize) ReadError!usize {
-        assert(len <= buffer.len);
-        var index: usize = 0;
-        while (index < len) {
-            const amt = try s.read(buffer[index..]);
-            if (amt == 0) break;
-            index += amt;
-        }
-        return index;
-    }
-
-    /// TODO in evented I/O mode, this implementation incorrectly uses the event loop's
-    /// 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 {
-        return self.conn.send(self.handle, buffer) catch |err| switch (err) {
-            error.AccessDenied => error.AccessDenied,
-            error.WouldBlock => error.WouldBlock,
-            error.ConnectionResetByPeer => error.ConnectionResetByPeer,
-            error.MessageTooBig => error.FileTooBig,
-            error.BrokenPipe => error.BrokenPipe,
-            else => return error.Unexpected,
-        };
-    }
-
-    pub fn writeAll(self: Stream, bytes: []const u8) WriteError!void {
-        var index: usize = 0;
-        while (index < bytes.len) {
-            index += try self.write(bytes[index..]);
-        }
-    }
-
-    /// See https://github.com/ziglang/zig/issues/7699
-    /// See equivalent function: `std.fs.File.writev`.
-    pub fn writev(self: Stream, iovecs: []const posix.iovec_const) WriteError!usize {
-        if (iovecs.len == 0) return 0;
-        const first_buffer = iovecs[0].base[0..iovecs[0].len];
-        return try self.write(first_buffer);
-    }
-
-    /// The `iovecs` parameter is mutable because this function needs to mutate the fields in
-    /// order to handle partial writes from the underlying OS layer.
-    /// See https://github.com/ziglang/zig/issues/7699
-    /// See equivalent function: `std.fs.File.writevAll`.
-    pub fn writevAll(self: Stream, iovecs: []posix.iovec_const) WriteError!void {
-        if (iovecs.len == 0) return;
-
-        var i: usize = 0;
-        while (true) {
-            var amt = try self.writev(iovecs[i..]);
-            while (amt >= iovecs[i].len) {
-                amt -= iovecs[i].len;
-                i += 1;
-                if (i >= iovecs.len) return;
-            }
-            iovecs[i].base += amt;
-            iovecs[i].len -= amt;
-        }
-    }
-};

src/async/tcp.zig

@@ -1,112 +0,0 @@
-// Copyright (C) 2023-2024  Lightpanda (Selecy SAS)
-//
-// Francis Bouvier <francis@lightpanda.io>
-// Pierre Tachoire <pierre@lightpanda.io>
-//
-// This program is free software: you can redistribute it and/or modify
-// it under the terms of the GNU Affero General Public License as
-// published by the Free Software Foundation, either version 3 of the
-// License, or (at your option) any later version.
-//
-// This program is distributed in the hope that it will be useful,
-// but WITHOUT ANY WARRANTY; without even the implied warranty of
-// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
-// GNU Affero General Public License for more details.
-//
-// You should have received a copy of the GNU Affero General Public License
-// along with this program.  If not, see <https://www.gnu.org/licenses/>.
-
-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);
-
-// 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,
-
-        done: bool = false,
-        err: ?anyerror = null,
-        ln: usize = 0,
-
-        fn ok(self: *Command, err: ?anyerror, ln: usize) void {
-            self.err = err;
-            self.ln = ln;
-            self.done = true;
-        }
-
-        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.posix.socket_t, address: std.net.Address) !void {
-        var cmd = Command{ .impl = NetworkImpl.init(self.loop) };
-        cmd.impl.connect(&cmd, socket, address);
-        _ = try cmd.wait();
-    }
-
-    pub fn send(self: *Conn, socket: std.posix.socket_t, buffer: []const u8) !usize {
-        var cmd = Command{ .impl = NetworkImpl.init(self.loop) };
-        cmd.impl.send(&cmd, socket, buffer);
-        return try cmd.wait();
-    }
-
-    pub fn receive(self: *Conn, socket: std.posix.socket_t, buffer: []u8) !usize {
-        var cmd = Command{ .impl = NetworkImpl.init(self.loop) };
-        cmd.impl.receive(&cmd, socket, buffer);
-        return try cmd.wait();
-    }
-};
-
-pub fn tcpConnectToHost(alloc: std.mem.Allocator, loop: *Loop, name: []const u8, port: u16) !Stream {
-    // TODO async resolve
-    const list = try net.getAddressList(alloc, name, port);
-    defer list.deinit();
-
-    if (list.addrs.len == 0) return error.UnknownHostName;
-
-    for (list.addrs) |addr| {
-        return tcpConnectToAddress(alloc, loop, addr) catch |err| switch (err) {
-            error.ConnectionRefused => {
-                continue;
-            },
-            else => return err,
-        };
-    }
-    return std.posix.ConnectError.ConnectionRefused;
-}
-
-pub fn tcpConnectToAddress(alloc: std.mem.Allocator, loop: *Loop, addr: net.Address) !Stream {
-    const sockfd = try std.posix.socket(addr.any.family, std.posix.SOCK.STREAM, std.posix.IPPROTO.TCP);
-    errdefer std.posix.close(sockfd);
-
-    var conn = try alloc.create(Conn);
-    conn.* = Conn{ .loop = loop };
-    try conn.connect(sockfd, addr);
-
-    return Stream{
-        .alloc = alloc,
-        .conn = conn,
-        .handle = sockfd,
-    };
-}

src/async/test.zig

@@ -1,189 +0,0 @@
-// Copyright (C) 2023-2024  Lightpanda (Selecy SAS)
-//
-// Francis Bouvier <francis@lightpanda.io>
-// Pierre Tachoire <pierre@lightpanda.io>
-//
-// This program is free software: you can redistribute it and/or modify
-// it under the terms of the GNU Affero General Public License as
-// published by the Free Software Foundation, either version 3 of the
-// License, or (at your option) any later version.
-//
-// This program is distributed in the hope that it will be useful,
-// but WITHOUT ANY WARRANTY; without even the implied warranty of
-// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
-// GNU Affero General Public License for more details.
-//
-// You should have received a copy of the GNU Affero General Public License
-// along with this program.  If not, see <https://www.gnu.org/licenses/>.
-
-const std = @import("std");
-const http = std.http;
-const Client = @import("Client.zig");
-const Request = @import("Client.zig").Request;
-
-pub const Loop = @import("jsruntime").Loop;
-
-const url = "https://w3.org";
-
-test "blocking mode fetch API" {
-    const alloc = std.testing.allocator;
-
-    var loop = try Loop.init(alloc);
-    defer loop.deinit();
-
-    var client: Client = .{
-        .allocator = alloc,
-        .loop = &loop,
-    };
-    defer client.deinit();
-
-    // force client's CA cert scan from system.
-    try client.ca_bundle.rescan(client.allocator);
-
-    const res = try client.fetch(.{
-        .location = .{ .uri = try std.Uri.parse(url) },
-    });
-
-    try std.testing.expect(res.status == .ok);
-}
-
-test "blocking mode open/send/wait API" {
-    const alloc = std.testing.allocator;
-
-    var loop = try Loop.init(alloc);
-    defer loop.deinit();
-
-    var client: Client = .{
-        .allocator = alloc,
-        .loop = &loop,
-    };
-    defer client.deinit();
-
-    // force client's CA cert scan from system.
-    try client.ca_bundle.rescan(client.allocator);
-
-    var buf: [2014]u8 = undefined;
-    var req = try client.open(.GET, try std.Uri.parse(url), .{
-        .server_header_buffer = &buf,
-    });
-    defer req.deinit();
-
-    try req.send();
-    try req.finish();
-    try req.wait();
-
-    try std.testing.expect(req.response.status == .ok);
-}
-
-// Example how to write an async http client using the modified standard client.
-const AsyncClient = struct {
-    cli: Client,
-
-    const YieldImpl = Loop.Yield(AsyncRequest);
-    const AsyncRequest = struct {
-        const State = enum { new, open, send, finish, wait, done };
-
-        cli: *Client,
-        uri: std.Uri,
-
-        req: ?Request = undefined,
-        state: State = .new,
-
-        impl: YieldImpl,
-        err: ?anyerror = null,
-
-        buf: [2014]u8 = undefined,
-
-        pub fn deinit(self: *AsyncRequest) void {
-            if (self.req) |*r| r.deinit();
-        }
-
-        pub fn fetch(self: *AsyncRequest) void {
-            self.state = .new;
-            return self.impl.yield(self);
-        }
-
-        fn onerr(self: *AsyncRequest, err: anyerror) void {
-            self.state = .done;
-            self.err = err;
-        }
-
-        pub fn onYield(self: *AsyncRequest, err: ?anyerror) void {
-            if (err) |e| return self.onerr(e);
-
-            switch (self.state) {
-                .new => {
-                    self.state = .open;
-                    self.req = self.cli.open(.GET, self.uri, .{
-                        .server_header_buffer = &self.buf,
-                    }) catch |e| return self.onerr(e);
-                },
-                .open => {
-                    self.state = .send;
-                    self.req.?.send() catch |e| return self.onerr(e);
-                },
-                .send => {
-                    self.state = .finish;
-                    self.req.?.finish() catch |e| return self.onerr(e);
-                },
-                .finish => {
-                    self.state = .wait;
-                    self.req.?.wait() catch |e| return self.onerr(e);
-                },
-                .wait => {
-                    self.state = .done;
-                    return;
-                },
-                .done => return,
-            }
-
-            return self.impl.yield(self);
-        }
-
-        pub fn wait(self: *AsyncRequest) !void {
-            while (self.state != .done) try self.impl.tick();
-            if (self.err) |err| return err;
-        }
-    };
-
-    pub fn init(alloc: std.mem.Allocator, loop: *Loop) AsyncClient {
-        return .{
-            .cli = .{
-                .allocator = alloc,
-                .loop = loop,
-            },
-        };
-    }
-
-    pub fn deinit(self: *AsyncClient) void {
-        self.cli.deinit();
-    }
-
-    pub fn createRequest(self: *AsyncClient, uri: std.Uri) !AsyncRequest {
-        return .{
-            .impl = YieldImpl.init(self.cli.loop),
-            .cli = &self.cli,
-            .uri = uri,
-        };
-    }
-};
-
-test "non blocking client" {
-    const alloc = std.testing.allocator;
-
-    var loop = try Loop.init(alloc);
-    defer loop.deinit();
-
-    var client = AsyncClient.init(alloc, &loop);
-    defer client.deinit();
-
-    var reqs: [3]AsyncClient.AsyncRequest = undefined;
-    for (0..reqs.len) |i| {
-        reqs[i] = try client.createRequest(try std.Uri.parse(url));
-        reqs[i].fetch();
-    }
-    for (0..reqs.len) |i| {
-        try reqs[i].wait();
-        reqs[i].deinit();
-    }
-}

src/browser/browser.zig

@@ -40,7 +40,7 @@ const storage = @import("../storage/storage.zig");
 const FetchResult = @import("../http/Client.zig").Client.FetchResult;
 
 const UserContext = @import("../user_context.zig").UserContext;
-const HttpClient = @import("../async/Client.zig");
+const HttpClient = @import("asyncio").Client;
 
 const log = std.log.scoped(.browser);
 
@@ -116,7 +116,7 @@ pub const Session = struct {
         };
 
         Env.init(&self.env, self.arena.allocator(), loop, null);
-        self.httpClient = .{ .allocator = alloc, .loop = loop };
+        self.httpClient = .{ .allocator = alloc };
         try self.env.load(&self.jstypes);
     }
 

src/main.zig

@@ -30,6 +30,7 @@ const apiweb = @import("apiweb.zig");
 
 pub const Types = jsruntime.reflect(apiweb.Interfaces);
 pub const UserContext = apiweb.UserContext;
+pub const IO = @import("asyncio").Wrapper(jsruntime.Loop);
 
 const log = std.log.scoped(.cli);
 

src/main_shell.zig

@@ -24,12 +24,13 @@ const parser = @import("netsurf");
 const apiweb = @import("apiweb.zig");
 const Window = @import("html/window.zig").Window;
 const storage = @import("storage/storage.zig");
+const Client = @import("asyncio").Client;
 
 const html_test = @import("html_test.zig").html;
 
 pub const Types = jsruntime.reflect(apiweb.Interfaces);
 pub const UserContext = apiweb.UserContext;
-const Client = @import("async/Client.zig");
+pub const IO = @import("asyncio").Wrapper(jsruntime.Loop);
 
 var doc: *parser.DocumentHTML = undefined;
 
@@ -41,7 +42,7 @@ fn execJS(
     try js_env.start();
     defer js_env.stop();
 
-    var cli = Client{ .allocator = alloc, .loop = js_env.nat_ctx.loop };
+    var cli = Client{ .allocator = alloc };
     defer cli.deinit();
 
     try js_env.setUserContext(UserContext{

src/main_wpt.zig

@@ -50,6 +50,7 @@ const Out = enum {
 pub const Types = jsruntime.reflect(apiweb.Interfaces);
 pub const GlobalType = apiweb.GlobalType;
 pub const UserContext = apiweb.UserContext;
+pub const IO = @import("asyncio").Wrapper(jsruntime.Loop);
 
 // TODO For now the WPT tests run is specific to WPT.
 // It manually load js framwork libs, and run the first script w/ js content in

src/run_tests.zig

@@ -30,7 +30,7 @@ const xhr = @import("xhr/xhr.zig");
 const storage = @import("storage/storage.zig");
 const url = @import("url/url.zig");
 const urlquery = @import("url/query.zig");
-const Client = @import("async/Client.zig");
+const Client = @import("asyncio").Client;
 
 const documentTestExecFn = @import("dom/document.zig").testExecFn;
 const HTMLDocumentTestExecFn = @import("html/document.zig").testExecFn;
@@ -59,6 +59,7 @@ const MutationObserverTestExecFn = @import("dom/mutation_observer.zig").testExec
 
 pub const Types = jsruntime.reflect(apiweb.Interfaces);
 pub const UserContext = @import("user_context.zig").UserContext;
+pub const IO = @import("asyncio").Wrapper(jsruntime.Loop);
 
 var doc: *parser.DocumentHTML = undefined;
 
@@ -86,7 +87,7 @@ fn testExecFn(
         std.debug.print("documentHTMLClose error: {s}\n", .{@errorName(err)});
     };
 
-    var cli = Client{ .allocator = alloc, .loop = js_env.nat_ctx.loop };
+    var cli = Client{ .allocator = alloc };
     defer cli.deinit();
 
     try js_env.setUserContext(.{
@@ -298,9 +299,6 @@ test {
     const msgTest = @import("msg.zig");
     std.testing.refAllDecls(msgTest);
 
-    const asyncTest = @import("async/test.zig");
-    std.testing.refAllDecls(asyncTest);
-
     const dumpTest = @import("browser/dump.zig");
     std.testing.refAllDecls(dumpTest);
 

src/server.zig

@@ -482,7 +482,7 @@ pub fn listen(
     // - cmd from incoming connection on server socket
     // - JS callbacks events from scripts
     while (true) {
-        try loop.io.tick();
+        try loop.io.run_for_ns(10 * std.time.ns_per_ms);
         if (loop.cbk_error) {
             log.err("JS error", .{});
             // if (try try_catch.exception(alloc, js_env.*)) |msg| {

src/test_runner.zig

@@ -22,6 +22,7 @@ const tests = @import("run_tests.zig");
 
 pub const Types = tests.Types;
 pub const UserContext = tests.UserContext;
+pub const IO = tests.IO;
 
 pub fn main() !void {
     try tests.main();

src/user_context.zig

@@ -1,6 +1,6 @@
 const std = @import("std");
 const parser = @import("netsurf");
-const Client = @import("async/Client.zig");
+const Client = @import("asyncio").Client;
 
 pub const UserContext = struct {
     document: *parser.DocumentHTML,

src/wpt/run.zig

@@ -28,10 +28,10 @@ const Loop = jsruntime.Loop;
 const Env = jsruntime.Env;
 const Window = @import("../html/window.zig").Window;
 const storage = @import("../storage/storage.zig");
+const Client = @import("asyncio").Client;
 
 const Types = @import("../main_wpt.zig").Types;
 const UserContext = @import("../main_wpt.zig").UserContext;
-const Client = @import("../async/Client.zig");
 
 // runWPT parses the given HTML file, starts a js env and run the first script
 // tags containing javascript sources.
@@ -53,7 +53,7 @@ pub fn run(arena: *std.heap.ArenaAllocator, comptime dir: []const u8, f: []const
     var loop = try Loop.init(alloc);
     defer loop.deinit();
 
-    var cli = Client{ .allocator = alloc, .loop = &loop };
+    var cli = Client{ .allocator = alloc };
     defer cli.deinit();
 
     var js_env: Env = undefined;

src/xhr/xhr.zig

@@ -32,8 +32,7 @@ const XMLHttpRequestEventTarget = @import("event_target.zig").XMLHttpRequestEven
 const Mime = @import("../browser/mime.zig");
 
 const Loop = jsruntime.Loop;
-const YieldImpl = Loop.Yield(XMLHttpRequest);
+const Client = @import("asyncio").Client;
-const Client = @import("../async/Client.zig");
 
 const parser = @import("netsurf");
 
@@ -98,10 +97,11 @@ pub const XMLHttpRequest = struct {
     proto: XMLHttpRequestEventTarget = XMLHttpRequestEventTarget{},
     alloc: std.mem.Allocator,
     cli: *Client,
-    impl: YieldImpl,
+    io: Client.IO,
 
     priv_state: PrivState = .new,
     req: ?Client.Request = null,
+    ctx: ?Client.Ctx = null,
 
     method: std.http.Method,
     state: u16,
@@ -135,7 +135,13 @@ pub const XMLHttpRequest = struct {
     response_header_buffer: [1024 * 16]u8 = undefined,
 
     response_status: u10 = 0,
-    response_override_mime_type: ?[]const u8 = null,
+
+    // TODO uncomment this field causes casting issue with
+    // XMLHttpRequestEventTarget. I think it's dueto an alignement issue, but
+    // not sure. see
+    // https://lightpanda.slack.com/archives/C05TRU6RBM1/p1707819010681019
+    // response_override_mime_type: ?[]const u8 = null,
+
     response_mime: Mime = undefined,
     response_obj: ?ResponseObj = null,
     send_flag: bool = false,
@@ -288,7 +294,7 @@ pub const XMLHttpRequest = struct {
             .alloc = alloc,
             .headers = Headers.init(alloc),
             .response_headers = Headers.init(alloc),
-            .impl = YieldImpl.init(loop),
+            .io = Client.IO.init(loop),
             .method = undefined,
             .url = null,
             .uri = undefined,
@@ -320,10 +326,11 @@ pub const XMLHttpRequest = struct {
 
         self.priv_state = .new;
 
-        if (self.req) |*r| {
+        if (self.ctx) |*c| c.deinit();
-            r.deinit();
+        self.ctx = null;
-            self.req = null;
+
-        }
+        if (self.req) |*r| r.deinit();
+        self.req = null;
     }
 
     pub fn deinit(self: *XMLHttpRequest, alloc: std.mem.Allocator) void {
@@ -498,138 +505,160 @@ pub const XMLHttpRequest = struct {
         log.debug("{any} {any}", .{ self.method, self.uri });
 
         self.send_flag = true;
-        self.impl.yield(self);
-    }
 
-    // onYield is a callback called between each request's steps.
+        self.priv_state = .open;
-    // Between each step, the code is blocking.
-    // Yielding allows pseudo-async and gives a chance to other async process
-    // to be called.
-    pub fn onYield(self: *XMLHttpRequest, err: ?anyerror) void {
-        if (err) |e| return self.onErr(e);
 
-        switch (self.priv_state) {
+        self.req = try self.cli.create(self.method, self.uri, .{
-            .new => {
+            .server_header_buffer = &self.response_header_buffer,
-                self.priv_state = .open;
+            .extra_headers = self.headers.all(),
-                self.req = self.cli.open(self.method, self.uri, .{
+        });
-                    .server_header_buffer = &self.response_header_buffer,
+        errdefer {
-                    .extra_headers = self.headers.all(),
+            self.req.?.deinit();
-                }) catch |e| return self.onErr(e);
+            self.req = null;
-            },
-            .open => {
-                // prepare payload transfert.
-                if (self.payload) |v| self.req.?.transfer_encoding = .{ .content_length = v.len };
-
-                self.priv_state = .send;
-                self.req.?.send() catch |e| return self.onErr(e);
-            },
-            .send => {
-                if (self.payload) |payload| {
-                    self.priv_state = .write;
-                    self.req.?.writeAll(payload) catch |e| return self.onErr(e);
-                } else {
-                    self.priv_state = .finish;
-                    self.req.?.finish() catch |e| return self.onErr(e);
-                }
-            },
-            .write => {
-                self.priv_state = .finish;
-                self.req.?.finish() catch |e| return self.onErr(e);
-            },
-            .finish => {
-                self.priv_state = .wait;
-                self.req.?.wait() catch |e| return self.onErr(e);
-            },
-            .wait => {
-                log.info("{any} {any} {d}", .{ self.method, self.uri, self.req.?.response.status });
-
-                self.priv_state = .done;
-                var it = self.req.?.response.iterateHeaders();
-                self.response_headers.load(&it) catch |e| return self.onErr(e);
-
-                // extract a mime type from headers.
-                const ct = self.response_headers.getFirstValue("Content-Type") orelse "text/xml";
-                self.response_mime = Mime.parse(ct) catch |e| return self.onErr(e);
-
-                // TODO handle override mime type
-
-                self.state = HEADERS_RECEIVED;
-                self.dispatchEvt("readystatechange");
-
-                self.response_status = @intFromEnum(self.req.?.response.status);
-
-                var buf: std.ArrayListUnmanaged(u8) = .{};
-
-                // TODO set correct length
-                const total = 0;
-                var loaded: u64 = 0;
-
-                // dispatch a progress event loadstart.
-                self.dispatchProgressEvent("loadstart", .{ .loaded = loaded, .total = total });
-
-                const reader = self.req.?.reader();
-                var buffer: [1024]u8 = undefined;
-                var ln = buffer.len;
-                var prev_dispatch: ?std.time.Instant = null;
-                while (ln > 0) {
-                    ln = reader.read(&buffer) catch |e| {
-                        buf.deinit(self.alloc);
-                        return self.onErr(e);
-                    };
-                    buf.appendSlice(self.alloc, buffer[0..ln]) catch |e| {
-                        buf.deinit(self.alloc);
-                        return self.onErr(e);
-                    };
-                    loaded = loaded + ln;
-
-                    // Dispatch only if 50ms have passed.
-                    const now = std.time.Instant.now() catch |e| {
-                        buf.deinit(self.alloc);
-                        return self.onErr(e);
-                    };
-                    if (prev_dispatch != null and now.since(prev_dispatch.?) < min_delay) continue;
-                    defer prev_dispatch = now;
-
-                    self.state = LOADING;
-                    self.dispatchEvt("readystatechange");
-
-                    // dispatch a progress event progress.
-                    self.dispatchProgressEvent("progress", .{
-                        .loaded = loaded,
-                        .total = total,
-                    });
-                }
-                self.response_bytes = buf.items;
-                self.send_flag = false;
-
-                self.state = DONE;
-                self.dispatchEvt("readystatechange");
-
-                // dispatch a progress event load.
-                self.dispatchProgressEvent("load", .{ .loaded = loaded, .total = total });
-                // dispatch a progress event loadend.
-                self.dispatchProgressEvent("loadend", .{ .loaded = loaded, .total = total });
-            },
-            .done => {
-                if (self.req) |*r| {
-                    r.deinit();
-                    self.req = null;
-                }
-
-                // finalize fetch process.
-                return;
-            },
         }
 
-        self.impl.yield(self);
+        self.ctx = try Client.Ctx.init(&self.io, &self.req.?);
+        errdefer {
+            self.ctx.?.deinit();
+            self.ctx = null;
+        }
+        self.ctx.?.userData = self;
+
+        try self.cli.async_open(
+            self.method,
+            self.uri,
+            .{ .server_header_buffer = &self.response_header_buffer },
+            &self.ctx.?,
+            onRequestConnect,
+        );
+    }
+
+    fn onRequestWait(ctx: *Client.Ctx, res: anyerror!void) !void {
+        var self = selfCtx(ctx);
+        res catch |err| return self.onErr(err);
+
+        log.info("{any} {any} {d}", .{ self.method, self.uri, self.req.?.response.status });
+
+        self.priv_state = .done;
+        var it = self.req.?.response.iterateHeaders();
+        self.response_headers.load(&it) catch |e| return self.onErr(e);
+
+        // extract a mime type from headers.
+        const ct = self.response_headers.getFirstValue("Content-Type") orelse "text/xml";
+        self.response_mime = Mime.parse(ct) catch |e| return self.onErr(e);
+
+        // TODO handle override mime type
+
+        self.state = HEADERS_RECEIVED;
+        self.dispatchEvt("readystatechange");
+
+        self.response_status = @intFromEnum(self.req.?.response.status);
+
+        var buf: std.ArrayListUnmanaged(u8) = .{};
+
+        // TODO set correct length
+        const total = 0;
+        var loaded: u64 = 0;
+
+        // dispatch a progress event loadstart.
+        self.dispatchProgressEvent("loadstart", .{ .loaded = loaded, .total = total });
+
+        // TODO read async
+        const reader = self.req.?.reader();
+        var buffer: [1024]u8 = undefined;
+        var ln = buffer.len;
+        var prev_dispatch: ?std.time.Instant = null;
+        while (ln > 0) {
+            ln = reader.read(&buffer) catch |e| {
+                buf.deinit(self.alloc);
+                return self.onErr(e);
+            };
+            buf.appendSlice(self.alloc, buffer[0..ln]) catch |e| {
+                buf.deinit(self.alloc);
+                return self.onErr(e);
+            };
+            loaded = loaded + ln;
+
+            // Dispatch only if 50ms have passed.
+            const now = std.time.Instant.now() catch |e| {
+                buf.deinit(self.alloc);
+                return self.onErr(e);
+            };
+            if (prev_dispatch != null and now.since(prev_dispatch.?) < min_delay) continue;
+            defer prev_dispatch = now;
+
+            self.state = LOADING;
+            self.dispatchEvt("readystatechange");
+
+            // dispatch a progress event progress.
+            self.dispatchProgressEvent("progress", .{
+                .loaded = loaded,
+                .total = total,
+            });
+        }
+        self.response_bytes = buf.items;
+        self.send_flag = false;
+
+        self.state = DONE;
+        self.dispatchEvt("readystatechange");
+
+        // dispatch a progress event load.
+        self.dispatchProgressEvent("load", .{ .loaded = loaded, .total = total });
+        // dispatch a progress event loadend.
+        self.dispatchProgressEvent("loadend", .{ .loaded = loaded, .total = total });
+
+        if (self.ctx) |*c| c.deinit();
+        self.ctx = null;
+
+        if (self.req) |*r| r.deinit();
+        self.req = null;
+    }
+
+    fn onRequestFinish(ctx: *Client.Ctx, res: anyerror!void) !void {
+        var self = selfCtx(ctx);
+        res catch |err| return self.onErr(err);
+
+        self.priv_state = .wait;
+        return ctx.req.async_wait(ctx, onRequestWait) catch |e| return self.onErr(e);
+    }
+
+    fn onRequestSend(ctx: *Client.Ctx, res: anyerror!void) !void {
+        var self = selfCtx(ctx);
+        res catch |err| return self.onErr(err);
+
+        if (self.payload) |payload| {
+            self.priv_state = .write;
+            return ctx.req.async_writeAll(payload, ctx, onRequestWrite) catch |e| return self.onErr(e);
+        }
+
+        self.priv_state = .finish;
+        return ctx.req.async_finish(ctx, onRequestFinish) catch |e| return self.onErr(e);
+    }
+
+    fn onRequestWrite(ctx: *Client.Ctx, res: anyerror!void) !void {
+        var self = selfCtx(ctx);
+        res catch |err| return self.onErr(err);
+        self.priv_state = .finish;
+        return ctx.req.async_finish(ctx, onRequestFinish) catch |e| return self.onErr(e);
+    }
+
+    fn onRequestConnect(ctx: *Client.Ctx, res: anyerror!void) anyerror!void {
+        var self = selfCtx(ctx);
+        res catch |err| return self.onErr(err);
+
+        // prepare payload transfert.
+        if (self.payload) |v| self.req.?.transfer_encoding = .{ .content_length = v.len };
+
+        self.priv_state = .send;
+        return ctx.req.async_send(ctx, onRequestSend) catch |err| return self.onErr(err);
+    }
+
+    fn selfCtx(ctx: *Client.Ctx) *XMLHttpRequest {
+        return @ptrCast(@alignCast(ctx.userData));
     }
 
     fn onErr(self: *XMLHttpRequest, err: anyerror) void {
         self.priv_state = .done;
-        if (self.req) |*r| {
-            r.deinit();
-            self.req = null;
-        }
 
         self.err = err;
         self.state = DONE;
@@ -639,6 +668,12 @@ pub const XMLHttpRequest = struct {
         self.dispatchProgressEvent("loadend", .{});
 
         log.debug("{any} {any} {any}", .{ self.method, self.uri, self.err });
+
+        if (self.ctx) |*c| c.deinit();
+        self.ctx = null;
+
+        if (self.req) |*r| r.deinit();
+        self.req = null;
     }
 
     pub fn _abort(self: *XMLHttpRequest) void {
@@ -886,7 +921,7 @@ pub fn testExecFn(
         // .{ .src = "req.onload", .ex = "function cbk(event) { nb ++; evt = event; }" },
         //.{ .src = "req.onload = cbk", .ex = "function cbk(event) { nb ++; evt = event; }" },
 
-        .{ .src = "req.open('GET', 'http://httpbin.io/html')", .ex = "undefined" },
+        .{ .src = "req.open('GET', 'https://httpbin.io/html')", .ex = "undefined" },
         .{ .src = "req.setRequestHeader('User-Agent', 'lightpanda/1.0')", .ex = "undefined" },
 
         // ensure open resets values
@@ -916,7 +951,7 @@ pub fn testExecFn(
 
     var document = [_]Case{
         .{ .src = "const req2 = new XMLHttpRequest()", .ex = "undefined" },
-        .{ .src = "req2.open('GET', 'http://httpbin.io/html')", .ex = "undefined" },
+        .{ .src = "req2.open('GET', 'https://httpbin.io/html')", .ex = "undefined" },
         .{ .src = "req2.responseType = 'document'", .ex = "document" },
 
         .{ .src = "req2.send()", .ex = "undefined" },
@@ -932,7 +967,7 @@ pub fn testExecFn(
 
     var json = [_]Case{
         .{ .src = "const req3 = new XMLHttpRequest()", .ex = "undefined" },
-        .{ .src = "req3.open('GET', 'http://httpbin.io/json')", .ex = "undefined" },
+        .{ .src = "req3.open('GET', 'https://httpbin.io/json')", .ex = "undefined" },
         .{ .src = "req3.responseType = 'json'", .ex = "json" },
 
         .{ .src = "req3.send()", .ex = "undefined" },
@@ -947,7 +982,7 @@ pub fn testExecFn(
 
     var post = [_]Case{
         .{ .src = "const req4 = new XMLHttpRequest()", .ex = "undefined" },
-        .{ .src = "req4.open('POST', 'http://httpbin.io/post')", .ex = "undefined" },
+        .{ .src = "req4.open('POST', 'https://httpbin.io/post')", .ex = "undefined" },
         .{ .src = "req4.send('foo')", .ex = "undefined" },
 
         // Each case executed waits for all loop callaback calls.
@@ -960,7 +995,7 @@ pub fn testExecFn(
 
     var cbk = [_]Case{
         .{ .src = "const req5 = new XMLHttpRequest()", .ex = "undefined" },
-        .{ .src = "req5.open('GET', 'http://httpbin.io/json')", .ex = "undefined" },
+        .{ .src = "req5.open('GET', 'https://httpbin.io/json')", .ex = "undefined" },
         .{ .src = "var status = 0; req5.onload = function () { status = this.status };", .ex = "function () { status = this.status }" },
         .{ .src = "req5.send()", .ex = "undefined" },