public/browser
1
0
Fork 0
mirror of https://github.com/lightpanda-io/browser.git synced 2025-10-29 15:13:28 +00:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

Removing blocking code async HTTP request

Browse Source 
The HTTP Client has a state pool. It blocks when we've exceeded max_concurrency.
This can block processing forever. A simple way to reproduce this is to go into
the demo cdp.js, and execute the XHR request 5 times (loading json/product.json)

To some degree, I think this is a result of weird / non-intuitive execution
flow. If you exec a JS with 100 XHR requests, it'll call our XHR _send function
but none of these will execute until the loop is run (after the script is done
being executed). This can result in poor utilization of our connection and
state pool.

For an async request, getting the *Request object is itself now asynchronous.
If no state is available, we use the Loop's timeout (at 20ms) to keep checking
for an available state.
This commit is contained in:
Karl Seguin
2025-06-05 20:07:51 +08:00
parent 3a5aa87853
commit 48c25c380d
4 changed files with 206 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
src/app.zig
View File

@@ -52,7 +52,8 @@ pub const App = struct {
.telemetry = undefined,
.app_dir_path = app_dir_path,
.notification = notification,
.http_client = try HttpClient.init(allocator, 5, .{
.http_client = try HttpClient.init(allocator, .{
.max_concurrent = 3,
.http_proxy = config.http_proxy,
.tls_verify_host = config.tls_verify_host,
}),

4
src/browser/page.zig
View File

@@ -113,7 +113,9 @@ pub const Page = struct {
.cookie_jar = &session.cookie_jar,
.microtask_node = .{ .func = microtaskCallback },
.window_clicked_event_node = .{ .func = windowClicked },
.request_factory = browser.http_client.requestFactory(browser.notification),
.request_factory = browser.http_client.requestFactory(.{
.notification = browser.notification,
}),
.scope = undefined,
.module_map = .empty,
};

28
src/browser/xhr/xhr.zig
View File

@@ -30,6 +30,7 @@ const Mime = @import("../mime.zig").Mime;
const parser = @import("../netsurf.zig");
const http = @import("../../http/client.zig");
const Page = @import("../page.zig").Page;
const Loop = @import("../../runtime/loop.zig").Loop;
const CookieJar = @import("../storage/storage.zig").CookieJar;
// XHR interfaces
@@ -78,6 +79,7 @@ const XMLHttpRequestBodyInit = union(enum) {
pub const XMLHttpRequest = struct {
proto: XMLHttpRequestEventTarget = XMLHttpRequestEventTarget{},
loop: *Loop,
arena: Allocator,
request: ?*http.Request = null,
@@ -91,6 +93,7 @@ pub const XMLHttpRequest = struct {
sync: bool = true,
err: ?anyerror = null,
last_dispatch: i64 = 0,
request_body: ?[]const u8 = null,
cookie_jar: *CookieJar,
// the URI of the page where this request is originating from
@@ -241,12 +244,13 @@ pub const XMLHttpRequest = struct {
pub fn constructor(page: *Page) !XMLHttpRequest {
const arena = page.arena;
return .{
.url = null,
.arena = arena,
.loop = page.loop,
.headers = Headers.init(arena),
.response_headers = Headers.init(arena),
.method = undefined,
.state = .unsent,
.url = null,
.origin_url = &page.url,
.cookie_jar = page.cookie_jar,
};
@@ -422,10 +426,16 @@ pub const XMLHttpRequest = struct {
log.debug(.http, "request", .{ .method = self.method, .url = self.url, .source = "xhr" });
self.send_flag = true;
if (body) |b| {
self.request_body = try self.arena.dupe(u8, b);
}
self.request = try page.request_factory.create(self.method, &self.url.?.uri);
var request = self.request.?;
errdefer request.deinit();
try page.request_factory.initAsync(page.arena, self.method, &self.url.?.uri, self, onHttpRequestReady, self.loop,);
}
fn onHttpRequestReady(ctx: *anyopaque, request: *http.Request) !void {
// on error, our caller will cleanup request
const self: *XMLHttpRequest = @alignCast(@ptrCast(ctx));
for (self.headers.list.items) |hdr| {
try request.addHeader(hdr.name, hdr.value, .{});
@@ -433,7 +443,7 @@ pub const XMLHttpRequest = struct {
{
var arr: std.ArrayListUnmanaged(u8) = .{};
try self.cookie_jar.forRequest(&self.url.?.uri, arr.writer(page.arena), .{
try self.cookie_jar.forRequest(&self.url.?.uri, arr.writer(self.arena), .{
.navigation = false,
.origin_uri = &self.origin_url.uri,
});
@@ -447,14 +457,15 @@ pub const XMLHttpRequest = struct {
// if the request method is GET or HEAD.
// https://xhr.spec.whatwg.org/#the-send()-method
// var used_body: ?XMLHttpRequestBodyInit = null;
if (body) |b| {
if (self.request_body) |b| {
if (self.method != .GET and self.method != .HEAD) {
request.body = try page.arena.dupe(u8, b);
request.body = b;
try request.addHeader("Content-Type", "text/plain; charset=UTF-8", .{});
}
}
try request.sendAsync(page.loop, self, .{});
try request.sendAsync(self.loop, self, .{});
self.request = request;
}
pub fn onHttpResponse(self: *XMLHttpRequest, progress_: anyerror!http.Progress) !void {
@@ -468,6 +479,7 @@ pub const XMLHttpRequest = struct {
if (progress.first) {
const header = progress.header;
log.debug(.http, "request header", .{
.source = "xhr",
.url = self.url,

219
src/http/client.zig
View File

@@ -54,16 +54,17 @@ pub const Client = struct {
request_pool: std.heap.MemoryPool(Request),
const Opts = struct {
tls_verify_host: bool = true,
max_concurrent: usize = 3,
http_proxy: ?std.Uri = null,
tls_verify_host: bool = true,
max_idle_connection: usize = 10,
};
pub fn init(allocator: Allocator, max_concurrent: usize, opts: Opts) !Client {
pub fn init(allocator: Allocator, opts: Opts) !Client {
var root_ca: tls.config.CertBundle = if (builtin.is_test) .{} else try tls.config.CertBundle.fromSystem(allocator);
errdefer root_ca.deinit(allocator);
const state_pool = try StatePool.init(allocator, max_concurrent);
const state_pool = try StatePool.init(allocator, opts.max_concurrent);
errdefer state_pool.deinit(allocator);
const connection_manager = ConnectionManager.init(allocator, opts.max_idle_connection);
@@ -92,13 +93,62 @@ pub const Client = struct {
}
pub fn request(self: *Client, method: Request.Method, uri: *const Uri) !*Request {
const state = self.state_pool.acquire();
const state = self.state_pool.acquireWait();
errdefer self.state_pool.release(state);
errdefer {
state.reset();
self.state_pool.release(state);
const req = try self.request_pool.create();
errdefer self.request_pool.destroy(req);
req.* = try Request.init(self, state, method, uri);
return req;
}
pub fn initAsync(
self: *Client,
arena: Allocator,
method: Request.Method,
uri: *const Uri,
ctx: *anyopaque,
callback: AsyncQueue.Callback,
loop: *Loop,
opts: RequestOpts,
) !void {
if (self.state_pool.acquireOrNull()) |state| {
// if we have state ready, we can skip the loop and immediately
// kick this request off.
return self.asyncRequestReady(method, uri, ctx, callback, state, opts);
}
// This cannot be a client-owned MemoryPool. The page can end before
// this is ever completed (and the check callback will never be called).
// As long as the loop doesn't guarantee that callbacks will be called,
// this _has_ to be the page arena.
const queue = try arena.create(AsyncQueue);
queue.* = .{
.ctx = ctx,
.uri = uri,
.opts = opts,
.client = self,
.method = method,
.callback = callback,
.node = .{ .func = AsyncQueue.check },
};
_ = try loop.timeout(10 * std.time.ns_per_ms, &queue.node);
}
// Either called directly from initAsync (if we have a state ready)
// Or from when the AsyncQueue(T) is ready.
fn asyncRequestReady(
self: *Client,
method: Request.Method,
uri: *const Uri,
ctx: *anyopaque,
callback: AsyncQueue.Callback,
state: *State,
opts: RequestOpts,
) !void {
errdefer self.state_pool.release(state);
// We need the request on the heap, because it can have a longer lifetime
// than the code making the request. That sounds odd, but consider the
// case of an XHR request: it can still be inflight (e.g. waiting for
@@ -110,26 +160,78 @@ pub const Client = struct {
errdefer self.request_pool.destroy(req);
req.* = try Request.init(self, state, method, uri);
return req;
if (opts.notification) |notification| {
req.notification = notification;
}
errdefer req.deinit();
try callback(ctx, req);
}
pub fn requestFactory(self: *Client, notification: ?*Notification) RequestFactory {
pub fn requestFactory(self: *Client, opts: RequestOpts) RequestFactory {
return .{
.opts = opts,
.client = self,
.notification = notification,
};
}
};
const RequestOpts = struct {
notification: ?*Notification = null,
};
// A factory for creating requests with a given set of options.
pub const RequestFactory = struct {
client: *Client,
notification: ?*Notification,
opts: RequestOpts,
pub fn create(self: RequestFactory, method: Request.Method, uri: *const Uri) !*Request {
var req = try self.client.request(method, uri);
req.notification = self.notification;
return req;
pub fn initAsync(
self: RequestFactory,
arena: Allocator,
method: Request.Method,
uri: *const Uri,
ctx: *anyopaque,
callback: AsyncQueue.Callback,
loop: *Loop,
) !void {
return self.client.initAsync(arena, method, uri, ctx, callback, loop, self.opts);
}
};
const AsyncQueue = struct {
ctx: *anyopaque,
method: Request.Method,
uri: *const Uri,
client: *Client,
opts: RequestOpts,
node: Loop.CallbackNode,
callback: Callback,
const Callback = *const fn (*anyopaque, *Request) anyerror!void;
fn check(node: *Loop.CallbackNode, repeat_delay: *?u63) void {
const self: *AsyncQueue = @fieldParentPtr("node", node);
self._check(repeat_delay) catch |err| {
log.err(.http_client, "async queue check", .{ .err = err });
};
}
fn _check(self: *AsyncQueue, repeat_delay: *?u63) !void {
const client = self.client;
const state = client.state_pool.acquireOrNull() orelse {
// re-run this function in 10 milliseconds
repeat_delay.* = 10 * std.time.ns_per_ms;
return;
};
try client.asyncRequestReady(
self.method,
self.uri,
self.ctx,
self.callback,
state,
self.opts,
);
}
};
@@ -321,7 +423,6 @@ pub const Request = struct {
pub fn deinit(self: *Request) void {
self.releaseConnection();
_ = self._state.reset();
self._client.state_pool.release(self._state);
self._client.request_pool.destroy(self);
}
@@ -1137,6 +1238,7 @@ fn AsyncHandler(comptime H: type, comptime L: type) type {
self.handleError("decompression error", err);
return .done;
};
self.handler.onHttpResponse(.{
.data = chunk,
.first = first,
@@ -2346,7 +2448,7 @@ const State = struct {
}
fn reset(self: *State) void {
_ = self.arena.reset(.{ .retain_with_limit = 1024 * 1024 });
_ = self.arena.reset(.{ .retain_with_limit = 64 * 1024 });
}
fn deinit(self: *State) void {
@@ -2399,10 +2501,11 @@ const StatePool = struct {
allocator.free(self.states);
}
pub fn acquire(self: *StatePool) *State {
pub fn acquireWait(self: *StatePool) *State {
const states = self.states;
self.mutex.lock();
while (true) {
const states = self.states;
const available = self.available;
if (available == 0) {
self.cond.wait(&self.mutex);
@@ -2416,7 +2519,25 @@ const StatePool = struct {
}
}
pub fn acquireOrNull(self: *StatePool) ?*State {
const states = self.states;
self.mutex.lock();
defer self.mutex.unlock();
const available = self.available;
if (available == 0) {
return null;
}
const index = available - 1;
const state = states[index];
self.available = index;
return state;
}
pub fn release(self: *StatePool, state: *State) void {
state.reset();
self.mutex.lock();
var states = self.states;
const available = self.available;
@@ -2823,11 +2944,19 @@ test "HttpClient: sync GET redirect" {
}
test "HttpClient: async connect error" {
defer testing.reset();
var loop = try Loop.init(testing.allocator);
defer loop.deinit();
const Handler = struct {
loop: *Loop,
reset: *Thread.ResetEvent,
fn requestReady(ctx: *anyopaque, req: *Request) !void {
const self: *@This() = @alignCast(@ptrCast(ctx));
try req.sendAsync(self.loop, self, .{});
}
fn onHttpResponse(self: *@This(), res: anyerror!Progress) !void {
_ = res catch |err| {
if (err == error.ConnectionRefused) {
@@ -2845,14 +2974,21 @@ test "HttpClient: async connect error" {
var client = try testClient();
defer client.deinit();
var handler = Handler{
.loop = &loop,
.reset = &reset,
};
const uri = try Uri.parse("HTTP://127.0.0.1:9920");
var req = try client.request(.GET, &uri);
try req.sendAsync(&loop, Handler{ .reset = &reset }, .{});
try client.initAsync(testing.arena_allocator, .GET, &uri, &handler, Handler.requestReady, &loop, .{}, );
try loop.io.run_for_ns(std.time.ns_per_ms);
try reset.timedWait(std.time.ns_per_s);
}
test "HttpClient: async no body" {
defer testing.reset();
var client = try testClient();
defer client.deinit();
@@ -2860,8 +2996,7 @@ test "HttpClient: async no body" {
defer handler.deinit();
const uri = try Uri.parse("HTTP://127.0.0.1:9582/http_client/simple");
var req = try client.request(.GET, &uri);
try req.sendAsync(&handler.loop, &handler, .{});
try client.initAsync(testing.arena_allocator, .GET, &uri, &handler, CaptureHandler.requestReady, &handler.loop, .{});
try handler.waitUntilDone();
const res = handler.response;
@@ -2871,6 +3006,8 @@ test "HttpClient: async no body" {
}
test "HttpClient: async with body" {
defer testing.reset();
var client = try testClient();
defer client.deinit();
@@ -2878,8 +3015,7 @@ test "HttpClient: async with body" {
defer handler.deinit();
const uri = try Uri.parse("HTTP://127.0.0.1:9582/http_client/echo");
var req = try client.request(.GET, &uri);
try req.sendAsync(&handler.loop, &handler, .{});
try client.initAsync(testing.arena_allocator, .GET, &uri, &handler, CaptureHandler.requestReady, &handler.loop, .{});
try handler.waitUntilDone();
const res = handler.response;
@@ -2894,6 +3030,8 @@ test "HttpClient: async with body" {
}
test "HttpClient: async with gzip body" {
defer testing.reset();
var client = try testClient();
defer client.deinit();
@@ -2901,8 +3039,7 @@ test "HttpClient: async with gzip body" {
defer handler.deinit();
const uri = try Uri.parse("HTTP://127.0.0.1:9582/http_client/gzip");
var req = try client.request(.GET, &uri);
try req.sendAsync(&handler.loop, &handler, .{});
try client.initAsync(testing.arena_allocator, .GET, &uri, &handler, CaptureHandler.requestReady, &handler.loop, .{});
try handler.waitUntilDone();
const res = handler.response;
@@ -2916,6 +3053,8 @@ test "HttpClient: async with gzip body" {
}
test "HttpClient: async redirect" {
defer testing.reset();
var client = try testClient();
defer client.deinit();
@@ -2923,8 +3062,7 @@ test "HttpClient: async redirect" {
defer handler.deinit();
const uri = try Uri.parse("HTTP://127.0.0.1:9582/http_client/redirect");
var req = try client.request(.GET, &uri);
try req.sendAsync(&handler.loop, &handler, .{});
try client.initAsync(testing.arena_allocator, .GET, &uri, &handler, CaptureHandler.requestReady, &handler.loop, .{});
// Called twice on purpose. The initial GET resutls in the # of pending
// events to reach 0. This causes our `run_for_ns` to return. But we then
@@ -2945,6 +3083,7 @@ test "HttpClient: async redirect" {
}
test "HttpClient: async tls no body" {
defer testing.reset();
var client = try testClient();
defer client.deinit();
for (0..5) |_| {
@@ -2952,8 +3091,7 @@ test "HttpClient: async tls no body" {
defer handler.deinit();
const uri = try Uri.parse("HTTPs://127.0.0.1:9581/http_client/simple");
var req = try client.request(.GET, &uri);
try req.sendAsync(&handler.loop, &handler, .{ .tls_verify_host = false });
try client.initAsync(testing.arena_allocator, .GET, &uri, &handler, CaptureHandler.requestReady, &handler.loop, .{});
try handler.waitUntilDone();
const res = handler.response;
@@ -2969,6 +3107,7 @@ test "HttpClient: async tls no body" {
}
test "HttpClient: async tls with body x" {
defer testing.reset();
for (0..5) |_| {
var client = try testClient();
defer client.deinit();
@@ -2977,8 +3116,7 @@ test "HttpClient: async tls with body x" {
defer handler.deinit();
const uri = try Uri.parse("HTTPs://127.0.0.1:9581/http_client/body");
var req = try client.request(.GET, &uri);
try req.sendAsync(&handler.loop, &handler, .{ .tls_verify_host = false });
try client.initAsync(testing.arena_allocator, .GET, &uri, &handler, CaptureHandler.requestReady, &handler.loop, .{});
try handler.waitUntilDone();
const res = handler.response;
@@ -2993,6 +3131,7 @@ test "HttpClient: async tls with body x" {
}
test "HttpClient: async redirect from TLS to Plaintext" {
defer testing.reset();
for (0..1) |_| {
var client = try testClient();
defer client.deinit();
@@ -3001,8 +3140,7 @@ test "HttpClient: async redirect from TLS to Plaintext" {
defer handler.deinit();
const uri = try Uri.parse("https://127.0.0.1:9581/http_client/redirect/insecure");
var req = try client.request(.GET, &uri);
try req.sendAsync(&handler.loop, &handler, .{ .tls_verify_host = false });
try client.initAsync(testing.arena_allocator, .GET, &uri, &handler, CaptureHandler.requestReady, &handler.loop, .{});
try handler.waitUntilDone();
const res = handler.response;
@@ -3018,6 +3156,7 @@ test "HttpClient: async redirect from TLS to Plaintext" {
}
test "HttpClient: async redirect plaintext to TLS" {
defer testing.reset();
for (0..5) |_| {
var client = try testClient();
defer client.deinit();
@@ -3026,8 +3165,7 @@ test "HttpClient: async redirect plaintext to TLS" {
defer handler.deinit();
const uri = try Uri.parse("http://127.0.0.1:9582/http_client/redirect/secure");
var req = try client.request(.GET, &uri);
try req.sendAsync(&handler.loop, &handler, .{ .tls_verify_host = false });
try client.initAsync(testing.arena_allocator, .GET, &uri, &handler, CaptureHandler.requestReady, &handler.loop, .{});
try handler.waitUntilDone();
const res = handler.response;
@@ -3149,6 +3287,11 @@ const CaptureHandler = struct {
self.loop.deinit();
}
fn requestReady(ctx: *anyopaque, req: *Request) !void {
const self: *CaptureHandler = @alignCast(@ptrCast(ctx));
try req.sendAsync(&self.loop, self, .{ .tls_verify_host = false });
}
fn onHttpResponse(self: *CaptureHandler, progress_: anyerror!Progress) !void {
self.process(progress_) catch |err| {
std.debug.print("capture handler error: {}\n", .{err});
@@ -3230,5 +3373,5 @@ fn testReader(state: *State, res: *TestResponse, data: []const u8) !void {
}
fn testClient() !Client {
return try Client.init(testing.allocator, 1, .{});
return try Client.init(testing.allocator, .{ .max_concurrent = 1 });
}