Always monitor the CDP client socket, even on page.wait

This commit is contained in:
Karl Seguin
2025-09-02 19:45:49 +08:00
parent a791212d89
commit 2ac9b2088a
6 changed files with 54 additions and 24 deletions

View File

@@ -285,7 +285,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(.{ .timeout_ms = 200 });
_ = try client.tick(200);
switch (blocking.state) {
.running => {},
.done => |result| return result,

View File

@@ -313,7 +313,10 @@ pub const Page = struct {
}
// There should only be 1 active http transfer, the main page
_ = try http_client.tick(.{ .timeout_ms = ms_remaining });
if (try http_client.tick(ms_remaining) == .extra_socket) {
// data on a socket we aren't handling, return to caller
return;
}
},
.html, .parsed => {
// The HTML page was parsed. We now either have JS scripts to
@@ -374,7 +377,10 @@ pub const Page = struct {
// inflight requests
else @min(ms_remaining, ms_to_next_task orelse 1000);
_ = try http_client.tick(.{ .timeout_ms = ms_to_wait });
if (try http_client.tick(ms_to_wait) == .extra_socket) {
// data on a socket we aren't handling, return to caller
return;
}
if (request_intercepted) {
// Again, proritizing intercepted requests. Exit this

View File

@@ -103,7 +103,7 @@ pub fn CDPT(comptime TypeProvider: type) type {
pub fn handleMessage(self: *Self, msg: []const u8) bool {
// if there's an error, it's already been logged
self.processMessage(msg) catch return false;
self.pageWait();
// self.pageWait();
return true;
}

View File

@@ -87,6 +87,11 @@ notification: ?*Notification = null,
// restoring, this originally-configured value is what it goes to.
http_proxy: ?[:0]const u8 = null,
// libcurl can monitor arbitrary sockets. Currently, we ever [maybe] want to
// monitor the CDP client socket, so we've done the simplest thing possible
// by having this single optional field
extra_socket: ?posix.socket_t = null,
const TransferQueue = std.DoublyLinkedList;
pub fn init(allocator: Allocator, ca_blob: ?c.curl_blob, opts: Http.Opts) !*Client {
@@ -162,11 +167,7 @@ pub fn abort(self: *Client) void {
}
}
const TickOpts = struct {
timeout_ms: i32 = 0,
poll_socket: ?posix.socket_t = null,
};
pub fn tick(self: *Client, opts: TickOpts) !bool {
pub fn tick(self: *Client, timeout_ms: i32) !PerformStatus {
while (true) {
if (self.handles.hasAvailable() == false) {
break;
@@ -178,7 +179,7 @@ pub fn tick(self: *Client, opts: TickOpts) !bool {
const handle = self.handles.getFreeHandle().?;
try self.makeRequest(handle, transfer);
}
return self.perform(opts.timeout_ms, opts.poll_socket);
return self.perform(timeout_ms);
}
pub fn request(self: *Client, req: Request) !void {
@@ -342,15 +343,25 @@ fn makeRequest(self: *Client, handle: *Handle, transfer: *Transfer) !void {
}
self.active += 1;
_ = try self.perform(0, null);
_ = try self.perform(0);
}
fn perform(self: *Client, timeout_ms: c_int, socket: ?posix.socket_t) !bool {
pub const PerformStatus = enum{
extra_socket,
normal,
};
fn perform(self: *Client, timeout_ms: c_int) !PerformStatus {
const multi = self.multi;
var running: c_int = undefined;
try errorMCheck(c.curl_multi_perform(multi, &running));
if (socket) |s| {
// We're potentially going to block for a while until we get data. Process
// whatever messages we have waiting ahead of time.
try self.processMessages();
var status = PerformStatus.normal;
if (self.extra_socket) |s| {
var wait_fd = c.curl_waitfd{
.fd = s,
.events = c.CURL_WAIT_POLLIN,
@@ -359,12 +370,18 @@ fn perform(self: *Client, timeout_ms: c_int, socket: ?posix.socket_t) !bool {
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;
status = .extra_socket;
}
} else if (running > 0 and timeout_ms > 0) {
} else if (running > 0) {
try errorMCheck(c.curl_multi_poll(multi, null, 0, timeout_ms, null));
}
try self.processMessages();
return status;
}
fn processMessages(self: *Client) !void {
const multi = self.multi;
var messages_count: c_int = 0;
while (c.curl_multi_info_read(multi, &messages_count)) |msg_| {
const msg: *c.CURLMsg = @ptrCast(msg_);
@@ -422,8 +439,6 @@ fn perform(self: *Client, timeout_ms: c_int, socket: ?posix.socket_t) !bool {
self.requestFailed(transfer, err);
}
}
return false;
}
fn endTransfer(self: *Client, transfer: *Transfer) void {

View File

@@ -83,16 +83,21 @@ pub fn deinit(self: *Http) void {
self.arena.deinit();
}
pub fn poll(self: *Http, timeout_ms: i32, socket: posix.socket_t) bool {
return self.client.tick(.{
.timeout_ms = timeout_ms,
.poll_socket = socket,
}) catch |err| {
pub fn poll(self: *Http, timeout_ms: i32) Client.PerformStatus {
return self.client.tick(timeout_ms) catch |err| {
log.err(.app, "http poll", .{ .err = err });
return false;
return .normal;
};
}
pub fn monitorSocket(self: *Http, socket: posix.socket_t) void {
self.client.extra_socket = socket;
}
pub fn unmonitorSocket(self: *Http) void {
self.client.extra_socket = null;
}
pub fn newConnection(self: *Http) !Connection {
return Connection.init(self.ca_blob, &self.opts);
}

View File

@@ -126,8 +126,12 @@ pub const Server = struct {
var last_message = timestamp();
var http = &self.app.http;
http.monitorSocket(socket);
defer http.unmonitorSocket();
while (true) {
if (http.poll(20, socket)) {
if (http.poll(10) == .extra_socket) {
const n = posix.read(socket, client.readBuf()) catch |err| {
log.warn(.app, "CDP read", .{ .err = err });
return;