Remove the loop

Previously, the IO loop was doing three things:
1 - Managing timeouts (either from scripts or for our own needs)
2 - Handling browser IO events (page/script/xhr)
3 - Handling CDP events (accept, read, write, timeout)

With the libcurl merge, 1 was moved to an in-process scheduler and 2 was moved
to libcurl's own event loop. That means the entire loop code, including
the dependency on tigerbeetle-io existed for handling a single TCP client.
Not only is that a lot of code, there was also friction between the two loops
(the libcurl one and our IO loop), which would result in latency - while one
loop is waiting for the events, any events on the other loop go un-processed.

This PR removes our IO loop. To accomplish this:

1 - The main accept loop is blocking. This is simpler and works perfectly well,
given we only allow 1 active connection.
2 - The client socket is passed to libcurl - yes, libcurl's loop can take
arbitrary FDs and poll them along with its own.

In addition to having one less dependency, the CDP code is quite a bit simpler,
especially around shutdowns and writes. This also removes _some_ of the latency
caused by the friction between page process and CDP processing. Specifically,
when CDP now blocks for input, http page events (script loading, xhr, ...) will
still be processed.

There's still friction. For one, the reverse isn't true: when the page is
waiting for events, CDP events aren't going to be processed. But the page.wait
already have some sensitivity to this (e.g. the page.request_intercepted flag).
Also, when CDP waits, while we will process network events, page timeouts are
still not processed. Because of both these remaining issues, we still need to
jump between the two loops - but being able to block on CDP (even for a short
time) WITHOUT stopping the page's network I/O, should reduce some latency.

Author: karlseguin

build.zig

@@ -153,8 +153,6 @@ fn addDependencies(b: *Build, mod: *Build.Module, opts: *Build.Step.Options) !vo
         .optimize = mod.optimize.?,
     };
 
-    mod.addImport("tigerbeetle-io", b.dependency("tigerbeetle_io", .{}).module("tigerbeetle_io"));
-
     mod.addIncludePath(b.path("vendor/lightpanda"));
 
     {

build.zig.zon

@@ -4,15 +4,10 @@
     .version = "0.0.0",
     .fingerprint = 0xda130f3af836cea0,
     .dependencies = .{
-        .tigerbeetle_io = .{
-            .url = "https://github.com/lightpanda-io/tigerbeetle-io/archive/19ae89eb3814d48c202ac9e0495fc5cadb29dfe7.tar.gz",
-            .hash = "tigerbeetle_io-0.0.0-ViLgxjqSBADhuHO_RZm4yNzuoKDXWP39hDn60Kht40OC",
-        },
         .v8 = .{
             .url = "https://github.com/lightpanda-io/zig-v8-fork/archive/cf412d5b3d9d608582571d821e0d552337ef690d.tar.gz",
             .hash = "v8-0.0.0-xddH69zDAwA4fp1dBo_jEDjS5bhXycPwRlZHp6_X890t",
         },
         //.v8 = .{ .path = "../zig-v8-fork" },
-        //.tigerbeetle_io = .{ .path = "../tigerbeetle-io" },
     },
 }

src/app.zig

@@ -4,7 +4,6 @@ const Allocator = std.mem.Allocator;
 
 const log = @import("log.zig");
 const Http = @import("http/Http.zig");
-const Loop = @import("runtime/loop.zig").Loop;
 const Platform = @import("runtime/js.zig").Platform;
 
 const Telemetry = @import("telemetry/telemetry.zig").Telemetry;
@@ -14,7 +13,6 @@ const Notification = @import("notification.zig").Notification;
 // might need.
 pub const App = struct {
     http: Http,
-    loop: *Loop,
     config: Config,
     platform: ?*const Platform,
     allocator: Allocator,
@@ -45,12 +43,6 @@ pub const App = struct {
         const app = try allocator.create(App);
         errdefer allocator.destroy(app);
 
-        const loop = try allocator.create(Loop);
-        errdefer allocator.destroy(loop);
-
-        loop.* = try Loop.init(allocator);
-        errdefer loop.deinit();
-
         const notification = try Notification.init(allocator, null);
         errdefer notification.deinit();
 
@@ -68,7 +60,6 @@ pub const App = struct {
         const app_dir_path = getAndMakeAppDir(allocator);
 
         app.* = .{
-            .loop = loop,
             .http = http,
             .allocator = allocator,
             .telemetry = undefined,
@@ -92,8 +83,6 @@ pub const App = struct {
             allocator.free(app_dir_path);
         }
         self.telemetry.deinit();
-        self.loop.deinit();
-        allocator.destroy(self.loop);
         self.notification.deinit();
         self.http.deinit();
         allocator.destroy(self);

src/browser/Scheduler.zig

@@ -59,15 +59,15 @@ pub fn add(self: *Scheduler, ctx: *anyopaque, func: Task.Func, ms: u32, opts: Ad
     });
 }
 
-pub fn runHighPriority(self: *Scheduler) !?u32 {
+pub fn runHighPriority(self: *Scheduler) !?i32 {
     return self.runQueue(&self.primary);
 }
 
-pub fn runLowPriority(self: *Scheduler) !?u32 {
+pub fn runLowPriority(self: *Scheduler) !?i32 {
     return self.runQueue(&self.secondary);
 }
 
-fn runQueue(self: *Scheduler, queue: *Queue) !?u32 {
+fn runQueue(self: *Scheduler, queue: *Queue) !?i32 {
     // this is O(1)
     if (queue.count() == 0) {
         return null;

src/browser/ScriptManager.zig

@@ -317,7 +317,7 @@ pub fn blockingGet(self: *ScriptManager, url: [:0]const u8) !BlockingResult {
 
     // rely on http's timeout settings to avoid an endless/long loop.
     while (true) {
-        try client.tick(200);
+        _ = try client.tick(.{ .timeout_ms = 200 });
         switch (blocking.state) {
             .running => {},
             .done => |result| return result,

src/browser/page.zig

@@ -51,10 +51,6 @@ const polyfill = @import("polyfill/polyfill.zig");
 pub const Page = struct {
     cookie_jar: *storage.CookieJar,
 
-    // Pre-configured http/cilent.zig used to make HTTP requests.
-    // @newhttp
-    // request_factory: RequestFactory,
-
     session: *Session,
 
     // An arena with a lifetime for the entire duration of the page
@@ -146,12 +142,9 @@ pub const Page = struct {
             .scheduler = Scheduler.init(arena),
             .keydown_event_node = .{ .func = keydownCallback },
             .window_clicked_event_node = .{ .func = windowClicked },
-            // @newhttp
-            // .request_factory = browser.http_client.requestFactory(.{
-            //     .notification = browser.notification,
-            // }),
             .main_context = undefined,
         };
+
         self.main_context = try session.executor.createJsContext(&self.window, self, self, true, Env.GlobalMissingCallback.init(&self.polyfill_loader));
         try polyfill.preload(self.arena, self.main_context);
 
@@ -269,7 +262,7 @@ pub const Page = struct {
         return self.script_manager.blockingGet(src);
     }
 
-    pub fn wait(self: *Page, wait_sec: usize) void {
+    pub fn wait(self: *Page, wait_sec: u16) void {
         self._wait(wait_sec) catch |err| switch (err) {
             error.JsError => {}, // already logged (with hopefully more context)
             else => {
@@ -283,9 +276,9 @@ pub const Page = struct {
         };
     }
 
-    fn _wait(self: *Page, wait_sec: usize) !void {
-        var ms_remaining = wait_sec * 1000;
+    fn _wait(self: *Page, wait_sec: u16) !void {
         var timer = try std.time.Timer.start();
+        var ms_remaining: i32 = @intCast(wait_sec * 1000);
 
         var try_catch: Env.TryCatch = undefined;
         try_catch.init(self.main_context);
@@ -320,7 +313,7 @@ pub const Page = struct {
                     }
 
                     // There should only be 1 active http transfer, the main page
-                    try http_client.tick(ms_remaining);
+                    _ = try http_client.tick(.{ .timeout_ms = ms_remaining });
                 },
                 .html, .parsed => {
                     // The HTML page was parsed. We now either have JS scripts to
@@ -381,7 +374,7 @@ pub const Page = struct {
                         // inflight requests
                         else @min(ms_remaining, ms_to_next_task orelse 1000);
 
-                    try http_client.tick(ms_to_wait);
+                    _ = try http_client.tick(.{ .timeout_ms = ms_to_wait });
 
                     if (request_intercepted) {
                         // Again, proritizing intercepted requests. Exit this
@@ -401,7 +394,7 @@ pub const Page = struct {
             if (ms_elapsed >= ms_remaining) {
                 return;
             }
-            ms_remaining -= ms_elapsed;
+            ms_remaining -= @intCast(ms_elapsed);
         }
     }
 

src/browser/session.zig

@@ -137,7 +137,7 @@ pub const Session = struct {
         return &(self.page orelse return null);
     }
 
-    pub fn wait(self: *Session, wait_sec: usize) void {
+    pub fn wait(self: *Session, wait_sec: u16) void {
         if (self.queued_navigation) |qn| {
             // This was already aborted on the page, but it would be pretty
             // bad if old requests went to the new page, so let's make double sure

src/cdp/cdp.zig

@@ -114,12 +114,9 @@ pub fn CDPT(comptime TypeProvider: type) type {
         }
 
         // @newhttp
-        // A bit hacky right now. The main server loop blocks only for CDP
-        // messages. It no longer blocks for page timeouts of page HTTP
-        // transfers. So we need to call this more ourselves.
-        // This is called after every message and [very hackily] from the server
-        // loop.
-        // This is hopefully temporary.
+        // A bit hacky right now. The main server loop doesn't unblock for
+        // scheduled task. So we run this directly in order to process any
+        // timeouts (or http events) which are ready to be processed.
         pub fn pageWait(self: *Self) void {
             const session = &(self.browser.session orelse return);
             // exits early if there's nothing to do, so a large value like
@@ -592,8 +589,7 @@ pub fn BrowserContext(comptime CDP_T: type) type {
             };
 
             const cdp = self.cdp;
-            var arena = std.heap.ArenaAllocator.init(cdp.allocator);
-            errdefer arena.deinit();
+            const allocator = cdp.client.send_arena.allocator();
 
             const field = ",\"sessionId\":\"";
 
@@ -602,7 +598,7 @@ pub fn BrowserContext(comptime CDP_T: type) type {
             const message_len = msg.len + session_id.len + 1 + field.len + 10;
 
             var buf: std.ArrayListUnmanaged(u8) = .{};
-            buf.ensureTotalCapacity(arena.allocator(), message_len) catch |err| {
+            buf.ensureTotalCapacity(allocator, message_len) catch |err| {
                 log.err(.cdp, "inspector buffer", .{ .err = err });
                 return;
             };
@@ -617,7 +613,7 @@ pub fn BrowserContext(comptime CDP_T: type) type {
             buf.appendSliceAssumeCapacity("\"}");
             std.debug.assert(buf.items.len == message_len);
 
-            try cdp.client.sendJSONRaw(arena, buf);
+            try cdp.client.sendJSONRaw(buf);
         }
     };
 }

src/cdp/testing.zig

@@ -39,12 +39,14 @@ pub const Document = @import("../testing.zig").Document;
 
 const Client = struct {
     allocator: Allocator,
+    send_arena: ArenaAllocator,
     sent: std.ArrayListUnmanaged(json.Value) = .{},
     serialized: std.ArrayListUnmanaged([]const u8) = .{},
 
     fn init(alloc: Allocator) Client {
         return .{
             .allocator = alloc,
+            .send_arena = ArenaAllocator.init(alloc),
         };
     }
 
@@ -58,7 +60,7 @@ const Client = struct {
         try self.sent.append(self.allocator, value);
     }
 
-    pub fn sendJSONRaw(self: *Client, _: ArenaAllocator, buf: std.ArrayListUnmanaged(u8)) !void {
+    pub fn sendJSONRaw(self: *Client, buf: std.ArrayListUnmanaged(u8)) !void {
         const value = try json.parseFromSliceLeaky(json.Value, self.allocator, buf.items, .{});
         try self.sent.append(self.allocator, value);
     }

src/http/Client.zig

@@ -27,6 +27,7 @@ const CookieJar = @import("../browser/storage/storage.zig").CookieJar;
 const urlStitch = @import("../url.zig").stitch;
 
 const c = Http.c;
+const posix = std.posix;
 
 const Allocator = std.mem.Allocator;
 const ArenaAllocator = std.heap.ArenaAllocator;
@@ -167,22 +168,24 @@ pub fn abort(self: *Client) void {
     }
 }
 
-pub fn tick(self: *Client, timeout_ms: usize) !void {
-    var handles = &self.handles;
+const TickOpts = struct {
+    timeout_ms: i32 = 0,
+    poll_socket: ?posix.socket_t = null,
+};
+pub fn tick(self: *Client, opts: TickOpts) !bool {
     while (true) {
-        if (handles.hasAvailable() == false) {
+        if (self.handles.hasAvailable() == false) {
             break;
         }
         const queue_node = self.queue.popFirst() orelse break;
         const req = queue_node.data;
         self.queue_node_pool.destroy(queue_node);
 
         // we know this exists, because we checked isEmpty() above
-        const handle = handles.getFreeHandle().?;
+        const handle = self.handles.getFreeHandle().?;
         try self.makeRequest(handle, req);
     }
-
-    try self.perform(@intCast(timeout_ms));
+    return self.perform(opts.timeout_ms, opts.poll_socket);
 }
 
 pub fn request(self: *Client, req: Request) !void {
@@ -343,16 +346,26 @@ fn makeRequest(self: *Client, handle: *Handle, transfer: *Transfer) !void {
     }
 
     self.active += 1;
-    return self.perform(0);
+    _ = try self.perform(0, null);
 }
 
-fn perform(self: *Client, timeout_ms: c_int) !void {
+fn perform(self: *Client, timeout_ms: c_int, socket: ?posix.socket_t) !bool {
     const multi = self.multi;
-
     var running: c_int = undefined;
     try errorMCheck(c.curl_multi_perform(multi, &running));
 
-    if (running > 0 and timeout_ms > 0) {
+    if (socket) |s| {
+        var wait_fd = c.curl_waitfd{
+            .fd = s,
+            .events = c.CURL_WAIT_POLLIN,
+            .revents = 0,
+        };
+        try errorMCheck(c.curl_multi_poll(multi, &wait_fd, 1, timeout_ms, null));
+        if (wait_fd.revents != 0) {
+            // the extra socket we passed in is ready, let's signal our caller
+            return true;
+        }
+    } else if (running > 0 and timeout_ms > 0) {
         try errorMCheck(c.curl_multi_poll(multi, null, 0, timeout_ms, null));
     }
 
@@ -388,6 +401,8 @@ fn perform(self: *Client, timeout_ms: c_int) !void {
             self.requestFailed(transfer, err);
         }
     }
+
+    return false;
 }
 
 fn endTransfer(self: *Client, transfer: *Transfer) void {