mirror/browser
1
0
Fork 0
mirror of https://github.com/lightpanda-io/browser.git synced 2026-07-30 17:25:58 -04:00
Code Issues Packages Projects Releases Wiki Activity

telemetry: Move telemetry worker to its own thread

Browse Source 
This reverts recent(ish) changes to telemetry which moved it from its own thread
onto the main thread.

The downside is: we have an extra thread.

The upside is largely that Network.zig becomes drastically simpler and more
efficient. There's a bunch of machinery in Network.zig to support arbitrary
workers, of which Telemetry is the only one. There's also a lot of code to
support an optional multi and requests made to is. This is all removed.

Also, fetch, agent and mcp without a cdp server no longer even need to start
the network loop. And, it IS started (e.g. serve/cdp), there's no longer an
arbitrary 250ms wakeup on poll to progress workers. Nor can telemtry block CDP.

Telemetry's implementation itself was changed. The ring buffer was removed in
favor of a double-buffer arraylist. When telemetry is disabled, this saves
64Kb of memory. When it's enabled, it creates more allocator churn, but should
still use less memory in most cases (and never more). Finally, Telemetry is
given its own easy connection rather than using one out of the pool (which
workers would maybe like to use).
This commit is contained in:
Karl Seguin
2026-06-25 17:06:39 +08:00
parent 5fcdc1c5ad
commit 13eb35720a
4 changed files with 195 additions and 316 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
src/main.zig
View File

@@ -184,9 +184,7 @@ fn run(allocator: Allocator, main_arena: Allocator) !void {
}
var worker_thread = try std.Thread.spawn(.{}, fetchThread, .{ app, &ft, urls, fetch_opts });
defer worker_thread.join();
app.network.run();
worker_thread.join();
},
.mcp => |opts| {
log.info(.mcp, "starting server", .{});
@@ -207,7 +205,12 @@ fn run(allocator: Allocator, main_arena: Allocator) !void {
var worker_thread = try std.Thread.spawn(.{}, mcpThread, .{ allocator, app });
defer worker_thread.join();
app.network.run();
// mcp talks over stdio on mcpThread. Only run the CDP accept/read
// loop when an optional CDP server was started; otherwise the main
// thread just waits for the worker.
if (cdp_server != null) {
app.network.run();
}
},
.agent => |opts| {
log.info(.app, "starting agent", .{});
@@ -230,9 +233,7 @@ fn run(allocator: Allocator, main_arena: Allocator) !void {
&cancelled,
&sig_bridge,
});
defer worker_thread.join();
app.network.run();
worker_thread.join();
}
if (cancelled) return error.UserCancelled;

276
src/network/Network.zig
View File

@@ -81,8 +81,6 @@ pub const CdpLink = struct {
// Number of fixed pollfds entries (wakeup pipe + listener).
const PSEUDO_POLLFDS = 2;
const MAX_TICK_CALLBACKS = 16;
allocator: Allocator,
app: *App,
@@ -110,17 +108,6 @@ wakeup_pipe: [2]posix.fd_t = .{ -1, -1 },
shutdown: std.atomic.Value(bool) = .init(false),
// Multi is a heavy structure that can consume up to 2MB of RAM.
// Currently, Network is used sparingly, and we only create it on demand.
// When Network becomes truly shared, it should become a regular field.
multi: ?*libcurl.CurlM = null,
submission_mutex: std.Thread.Mutex = .{},
submission_queue: DoublyLinkedList = .{},
callbacks: [MAX_TICK_CALLBACKS]TickCallback = undefined,
callbacks_len: usize = 0,
callbacks_mutex: std.Thread.Mutex = .{},
// Registered CDP read endpoints. Producer-side (the worker doing
// register/unregister) and consumer-side (this thread's run loop) are
// serialized by cdp_mutex. cdp_unregister signals when a link
@@ -149,11 +136,6 @@ cdp_start: usize,
/// Optional IP filter for blocking requests to private/internal networks (--block-private-networks).
ip_filter: ?*IpFilter = null,
const TickCallback = struct {
ctx: *anyopaque,
fun: *const fn (*anyopaque) void,
};
fn globalInit(allocator: Allocator) void {
// Only route curl's own allocations through our allocator in Debug, so the
// leak detector sees them. In Release it'd just wrap c_allocator (curl's
@@ -178,19 +160,12 @@ pub fn init(allocator: Allocator, app: *App, config: *const Config) !Network {
const pipe = try posix.pipe2(.{ .NONBLOCK = true, .CLOEXEC = true });
// IMPORTANT: This is a bit messy, and it exists specifically because
// self.multi is optional. self.multi is optional so that, when telemetry is
// disabled, we don't need the overhead of a multi. If self.multi wasn't
// optional, then we wouldn't need to use posix.poll, we could use
// curl_multi_poll. This is to do in a follow up.
// The structure is: 0 is wakeup, 1 is listener, rest for curl fds:
// [0] wakeup pipe
// [1] listener
// [PSEUDO_POLLFDS .. + httpMaxConcurrent] curl multi fds
// [.. + maxConnections] CDP socket fds
// pollfds layout:
// [0] wakeup pipe
// [1] listener
// [PSEUDO_POLLFDS .. + max_cdp] CDP socket fds
const max_cdp = config.maxConnections();
const pollfds = try allocator.alloc(posix.pollfd, PSEUDO_POLLFDS + config.httpMaxConcurrent() + max_cdp);
const pollfds = try allocator.alloc(posix.pollfd, PSEUDO_POLLFDS + max_cdp);
errdefer allocator.free(pollfds);
const cdp_poll_snapshot = try allocator.alloc(?*CdpLink, max_cdp);
@@ -262,7 +237,7 @@ pub fn init(allocator: Allocator, app: *App, config: *const Config) !Network {
.pollfds = pollfds,
.wakeup_pipe = pipe,
.cdp_poll_snapshot = cdp_poll_snapshot,
.cdp_start = PSEUDO_POLLFDS + config.httpMaxConcurrent(),
.cdp_start = PSEUDO_POLLFDS,
.available = available,
.connections = connections,
@@ -281,10 +256,6 @@ pub fn init(allocator: Allocator, app: *App, config: *const Config) !Network {
}
pub fn deinit(self: *Network) void {
if (self.multi) |multi| {
libcurl.curl_multi_cleanup(multi) catch {};
}
for (&self.wakeup_pipe) |*fd| {
if (fd.* >= 0) {
posix.close(fd.*);
@@ -368,30 +339,6 @@ pub fn unbind(self: *Network) void {
self.wakeupPoll();
}
pub fn onTick(self: *Network, ctx: *anyopaque, callback: *const fn (*anyopaque) void) void {
self.callbacks_mutex.lock();
defer self.callbacks_mutex.unlock();
lp.assert(self.callbacks_len < MAX_TICK_CALLBACKS, "too many ticks", .{});
self.callbacks[self.callbacks_len] = .{
.ctx = ctx,
.fun = callback,
};
self.callbacks_len += 1;
self.wakeupPoll();
}
pub fn fireTicks(self: *Network) void {
self.callbacks_mutex.lock();
defer self.callbacks_mutex.unlock();
for (self.callbacks[0..self.callbacks_len]) |*callback| {
callback.fun(callback.ctx);
}
}
// Hand a CDP WebSocket's read side over to the main network thread. The caller
// owns the link and must keep it alive until unregisterCdp is called.
// The caller must not read from the socket.
@@ -609,15 +556,15 @@ fn shutdownCdpLinks(self: *Network) void {
pub fn run(self: *Network) void {
var drain_buf: [64]u8 = undefined;
var running_handles: c_int = 0;
const poll_fd = &self.pollfds[0];
const listen_fd = &self.pollfds[1];
// Please note that receiving a shutdown command does not terminate all connections.
// When gracefully shutting down a server, we at least want to send the remaining
// telemetry, but we stop accepting new connections. It is the responsibility
// of external code to terminate its requests upon shutdown.
// Receiving a shutdown command does not terminate existing connections: we
// stop accepting new ones but leave in-flight requests to external code to
// terminate. This loop only services the listener and the CDP read sockets;
// page fetches run on per-worker HttpClient multis and telemetry on its own
// thread, so nothing here drives libcurl.
while (true) {
if (self.listener != null and !self.accept.load(.acquire)) {
posix.close(self.listener.?.socket);
@@ -625,43 +572,10 @@ pub fn run(self: *Network) void {
self.pollfds[1] = .{ .fd = -1, .events = 0, .revents = 0 };
}
self.drainQueue();
if (self.multi) |multi| {
// Kickstart newly added handles (DNS/connect) so that
// curl registers its sockets before we poll.
libcurl.curl_multi_perform(multi, &running_handles) catch |err| {
lp.log.err(.app, "curl perform", .{ .err = err });
};
self.preparePollFds(multi);
}
self.prepareCdpPollFds();
// for ontick to work, you need to wake up periodically
const timeout = blk: {
const min_timeout = 250; // 250ms
if (self.multi == null) {
break :blk min_timeout;
}
// curl_multi_timeout reports -1 when curl has no timeout
// preference (idle) and 0 when it wants to be serviced
// immediately. Treat both as "no curl-imposed deadline" and
// fall back to min_timeout — otherwise @min(min_timeout, -1)
// would be -1, i.e. poll() blocks forever, starving onTick
// (telemetry's periodic flush) and removing the safety net
// that bounds any missed wakeup to min_timeout.
const curl_timeout = self.getCurlTimeout();
if (curl_timeout <= 0) {
break :blk min_timeout;
}
break :blk @min(min_timeout, curl_timeout);
};
_ = posix.poll(self.pollfds, timeout) catch |err| {
// wait until we get a CDP message or a signal on the wakeup pipe
_ = posix.poll(self.pollfds, -1) catch |err| {
lp.log.err(.app, "poll", .{ .err = err });
continue;
};
@@ -679,35 +593,14 @@ pub fn run(self: *Network) void {
self.acceptConnections();
}
if (self.multi) |multi| {
// Drive transfers and process completions.
libcurl.curl_multi_perform(multi, &running_handles) catch |err| {
lp.log.err(.app, "curl perform", .{ .err = err });
};
self.processCompletions(multi);
}
self.processCdpEvents();
self.fireTicks();
if (self.shutdown.load(.acquire)) {
// Drain any live CDP links so their workers can exit (issue #2510).
// Idempotent — no-op once drained, safe to call every iteration
// Drain any live CDP links so their workers can exit (issue #2510),
// then stop. Page fetches and telemetry don't run on this loop, so
// there is nothing else to flush here.
self.shutdownCdpLinks();
if (running_handles == 0) {
// Check if fireTicks submitted new requests (e.g. telemetry
// flush). If so, continue the loop to drain and send them
// before exiting.
self.submission_mutex.lock();
const has_pending = self.submission_queue.first != null;
self.submission_mutex.unlock();
if (!has_pending) {
break;
}
}
break;
}
}
@@ -725,46 +618,10 @@ pub fn run(self: *Network) void {
}
}
pub fn submitRequest(self: *Network, conn: *http.Connection) void {
self.submission_mutex.lock();
self.submission_queue.append(&conn.node);
self.submission_mutex.unlock();
self.wakeupPoll();
}
fn wakeupPoll(self: *Network) void {
_ = posix.write(self.wakeup_pipe[1], &.{1}) catch {};
}
fn drainQueue(self: *Network) void {
self.submission_mutex.lock();
defer self.submission_mutex.unlock();
if (self.submission_queue.first == null) return;
const multi = self.multi orelse blk: {
const m = libcurl.curl_multi_init() orelse {
lp.assert(false, "curl multi init failed", .{});
unreachable;
};
self.multi = m;
break :blk m;
};
while (self.submission_queue.popFirst()) |node| {
const conn: *http.Connection = @fieldParentPtr("node", node);
conn.setPrivate(conn) catch |err| {
lp.log.err(.app, "curl set private", .{ .err = err });
self.releaseConnection(conn);
continue;
};
libcurl.curl_multi_add_handle(multi, conn._easy) catch |err| {
lp.log.err(.app, "curl multi add", .{ .err = err });
self.releaseConnection(conn);
};
}
}
pub fn stop(self: *Network) void {
self.shutdown.store(true, .release);
self.wakeupPoll();
@@ -800,68 +657,6 @@ fn acceptConnections(self: *Network) void {
}
}
fn preparePollFds(self: *Network, multi: *libcurl.CurlM) void {
// Only the curl slice — NOT through to the end of pollfds. The CDP
// socket fds live in [cdp_start..] and are owned by
// prepareCdpPollFds, which only rebuilds them when cdp_dirty is set
// (a steady-state optimization). Slicing to the end here would
// @memset those fds to -1 every iteration once a multi exists (which
// happens as soon as telemetry sends its first request), silently
// dropping every live CDP socket from the poll set — Network then
// never reads another CDP message (#2508) nor observes peer
// EOF/shutdown (#2507).
const curl_fds = self.pollfds[PSEUDO_POLLFDS..self.cdp_start];
@memset(curl_fds, .{ .fd = -1, .events = 0, .revents = 0 });
var fd_count: c_uint = 0;
const wait_fds: []libcurl.CurlWaitFd = @ptrCast(curl_fds);
libcurl.curl_multi_waitfds(multi, wait_fds, &fd_count) catch |err| {
lp.log.err(.app, "curl waitfds", .{ .err = err });
};
}
fn getCurlTimeout(self: *Network) i32 {
const multi = self.multi orelse return -1;
var timeout_ms: c_long = -1;
libcurl.curl_multi_timeout(multi, &timeout_ms) catch return -1;
return @intCast(@min(timeout_ms, std.math.maxInt(i32)));
}
fn processCompletions(self: *Network, multi: *libcurl.CurlM) void {
var msgs_in_queue: c_int = 0;
while (libcurl.curl_multi_info_read(multi, &msgs_in_queue)) |msg| {
switch (msg.data) {
.done => |maybe_err| {
if (maybe_err) |err| {
lp.log.warn(.app, "curl transfer error", .{ .err = err });
}
},
else => continue,
}
const easy: *libcurl.Curl = msg.easy_handle;
var ptr: *anyopaque = undefined;
libcurl.curl_easy_getinfo(easy, .private, &ptr) catch
lp.assert(false, "curl getinfo private", .{});
const conn: *http.Connection = @ptrCast(@alignCast(ptr));
libcurl.curl_multi_remove_handle(multi, easy) catch {};
self.releaseConnection(conn);
}
}
comptime {
if (@sizeOf(posix.pollfd) != @sizeOf(libcurl.CurlWaitFd)) {
@compileError("pollfd and CurlWaitFd size mismatch");
}
if (@offsetOf(posix.pollfd, "fd") != @offsetOf(libcurl.CurlWaitFd, "fd") or
@offsetOf(posix.pollfd, "events") != @offsetOf(libcurl.CurlWaitFd, "events") or
@offsetOf(posix.pollfd, "revents") != @offsetOf(libcurl.CurlWaitFd, "revents"))
{
@compileError("pollfd and CurlWaitFd layout mismatch");
}
}
pub fn getConnection(self: *Network) ?*http.Connection {
self.conn_mutex.lock();
defer self.conn_mutex.unlock();
@@ -1003,40 +798,3 @@ fn loadCerts(allocator: Allocator) !libcurl.CurlBlob {
.flags = 0,
};
}
const testing = @import("../testing.zig");
test "Network: preparePollFds leaves the CDP fd region untouched" {
// Regression for #2507 / #2508. Once a multi exists (telemetry creates
// one in optimized builds), preparePollFds runs every loop iteration.
// It rebuilds only the curl slice [PSEUDO_POLLFDS..cdp_start]; the CDP
// region [cdp_start..] is owned by prepareCdpPollFds, which keeps its
// entries across iterations and only rebuilds when cdp_dirty is set.
// A slice that ran to the end of pollfds @memset those CDP sockets to
// -1, silently dropping every live CDP connection from the poll set —
// so Network stopped reading CDP messages (#2508) and never observed
// peer EOF/shutdown (#2507). curl global is initialized by the test
// harness (App.init -> Network.init).
const multi = libcurl.curl_multi_init() orelse return error.FailedToInitMulti;
defer libcurl.curl_multi_cleanup(multi) catch {};
const curl_slots = 4;
const cdp_slots = 3;
var pollfds: [PSEUDO_POLLFDS + curl_slots + cdp_slots]posix.pollfd = undefined;
@memset(&pollfds, .{ .fd = -1, .events = 0, .revents = 0 });
// preparePollFds only reads self.pollfds and self.cdp_start.
var nw: Network = undefined;
nw.pollfds = &pollfds;
nw.cdp_start = PSEUDO_POLLFDS + curl_slots;
// Two live CDP sockets parked in the CDP region, mimicking the steady
// state between cdp_dirty rebuilds.
pollfds[nw.cdp_start] = .{ .fd = 4242, .events = posix.POLL.IN, .revents = 0 };
pollfds[nw.cdp_start + 1] = .{ .fd = 4243, .events = posix.POLL.IN, .revents = 0 };
nw.preparePollFds(multi);
try testing.expectEqual(@as(posix.fd_t, 4242), pollfds[nw.cdp_start].fd);
try testing.expectEqual(@as(posix.fd_t, 4243), pollfds[nw.cdp_start + 1].fd);
}

8
src/network/http.zig
View File

@@ -668,6 +668,14 @@ pub const Connection = struct {
return self.getResponseCode();
}
// Synchronous transfer that adds no request headers. request() injects the
// browser User-Agent / sec-ch-ua machinery meant for page fetches; callers
// that manage their own connection (telemetry) use this leaner path.
pub fn perform(self: *const Connection) !u16 {
try libcurl.curl_easy_perform(self._easy);
return self.getResponseCode();
}
pub fn wsStartFrame(self: *const Connection, frame_type: libcurl.WsFrameType, size: usize) !void {
try libcurl.curl_ws_start_frame(self._easy, frame_type, @intCast(size));
}

212
src/telemetry/lightpanda.zig
View File

@@ -5,33 +5,55 @@ const build_config = @import("build_config");
const App = @import("../App.zig");
const Config = @import("../Config.zig");
const telemetry = @import("telemetry.zig");
const http = @import("../network/http.zig");
const Network = @import("../network/Network.zig");
const telemetry = @import("telemetry.zig");
const log = lp.log;
const Allocator = std.mem.Allocator;
const IS_DEBUG = builtin.mode == .Debug;
const BUFFER_SIZE = 1024;
const MAX_BODY_SIZE = 500 * 1024; // 500KB server limit
const MAX_PENDING = 4096; // hard cap: drop + count beyond this (reported via buffer_overflow)
const RECLAIM_CAPACITY = 64; // reclaim the drain buffer once a burst grows it past this
const MAX_BODY_SIZE = 500 * 1024; // 500KB
const REQUEST_TIMEOUT_MS = 5000;
const URL = "https://telemetry.lightpanda.io";
const LightPanda = @This();
allocator: Allocator,
network: *Network,
writer: std.Io.Writer.Allocating,
/// Protects concurrent producers in send().
mutex: std.Thread.Mutex = .{},
// Owned and used only by the sender thread; never touched by producers.
writer: std.Io.Writer.Allocating,
iid: ?[36]u8 = null,
run_mode: Config.RunMode = .serve,
interactive: bool = false,
head: std.atomic.Value(usize) = .init(0),
tail: std.atomic.Value(usize) = .init(0),
dropped: std.atomic.Value(u32) = .init(0),
buffer: [BUFFER_SIZE]telemetry.Event = undefined,
// `mutex` guards the ring buffer (head/tail/dropped), `running`, and the lazy
// `thread` creation. The sender thread blocks on `cond` while idle.
mutex: std.Thread.Mutex = .{},
cond: std.Thread.Condition = .{},
// The sender thread is spawned on the first event, so a process that emits no
// telemetry pays for none of this
thread: ?std.Thread = null,
running: bool = true,
// Pending events. Producers append under `mutex`; the sender thread swaps the
// whole list out in O(1) and drains it. The list grows from empty under load
// and is reclaimed afterward, so memory tracks actual queue depth and never
// pre-commits a fixed count × sizeOf(Event). Events must be self-contained
// (inline storage, no borrowed slices) since they outlive the send() call on
// the sender thread. Past MAX_PENDING (or on alloc failure) we drop and bump
// `dropped`, which is reported as a buffer_overflow event on the next
// successful POST and only cleared then — our signal that the cap is too low or
// the endpoint is failing.
pending: std.ArrayList(telemetry.Event) = .empty,
dropped: u32 = 0,
pub fn init(self: *LightPanda, app: *App, iid: ?[36]u8, run_mode: Config.RunMode, interactive: bool) !void {
self.* = .{
@@ -42,11 +64,19 @@ pub fn init(self: *LightPanda, app: *App, iid: ?[36]u8, run_mode: Config.RunMode
.network = &app.network,
.writer = std.Io.Writer.Allocating.init(app.allocator),
};
self.network.onTick(@ptrCast(self), flushCallback);
}
pub fn deinit(self: *LightPanda) void {
if (self.thread) |thread| {
self.mutex.lock();
self.running = false;
self.mutex.unlock();
self.cond.signal();
// The thread drains anything still queued before returning, so this is
// also our graceful flush-on-shutdown. Bounded by REQUEST_TIMEOUT_MS.
thread.join();
}
self.pending.deinit(self.allocator);
self.writer.deinit();
}
@@ -54,60 +84,142 @@ pub fn send(self: *LightPanda, raw_event: telemetry.Event) !void {
self.mutex.lock();
defer self.mutex.unlock();
const t = self.tail.load(.monotonic);
const h = self.head.load(.acquire);
if (t - h >= BUFFER_SIZE) {
_ = self.dropped.fetchAdd(1, .monotonic);
if (self.thread == null) {
// First event: bring the sender thread online. If it can't spawn, drop
// the event rather than blocking the producer.
self.thread = std.Thread.spawn(.{}, run, .{self}) catch |err| {
log.warn(.telemetry, "thread spawn", .{ .err = err });
return;
};
}
if (self.pending.items.len >= MAX_PENDING) {
self.dropped +|= 1;
return;
}
self.buffer[t % BUFFER_SIZE] = raw_event;
self.tail.store(t + 1, .release);
}
fn flushCallback(ctx: *anyopaque) void {
const self: *LightPanda = @ptrCast(@alignCast(ctx));
self.postEvent() catch |err| {
log.warn(.telemetry, "flush error", .{ .err = err });
};
}
fn postEvent(self: *LightPanda) !void {
const conn = self.network.getConnection() orelse {
self.pending.append(self.allocator, raw_event) catch {
// Allocation failure growing the queue: drop and count, same as a cap hit.
self.dropped +|= 1;
return;
};
errdefer self.network.releaseConnection(conn);
self.cond.signal();
}
const h = self.head.load(.monotonic);
const t = self.tail.load(.acquire);
const dropped = self.dropped.swap(0, .monotonic);
if (h == t and dropped == 0) {
self.network.releaseConnection(conn);
fn run(self: *LightPanda) void {
// The connection is created, owned, and torn down entirely on this thread;
// the network thread never sees it (Transport == .none).
var conn = http.Connection.init(self.network.ca_blob, self.network.config, self.network.ip_filter) catch |err| {
// Essentially OOM — the process is already in trouble. The thread
// handle stays set so send() won't respawn; events drop at the cap.
log.warn(.telemetry, "connection init", .{ .err = err });
return;
};
defer conn.deinit();
// Static for the lifetime of the connection.
conn.setURL(URL) catch |err| return log.warn(.telemetry, "set url", .{ .err = err });
conn.setMethod(.POST) catch |err| return log.warn(.telemetry, "set method", .{ .err = err });
conn.setTimeout(REQUEST_TIMEOUT_MS) catch |err| return log.warn(.telemetry, "set timeout", .{ .err = err });
// Consumer-owned buffer, swapped with `pending` each cycle: producers always
// get an empty list back and we own the batch lock-free across the POST.
var batch: std.ArrayList(telemetry.Event) = .empty;
defer batch.deinit(self.allocator);
self.mutex.lock();
defer self.mutex.unlock();
while (true) {
while (self.pending.items.len == 0) {
if (self.running == false) {
return;
}
self.cond.wait(&self.mutex);
}
// Swap the batch out and release the lock for the (blocking) serialize +
// POST so producers never wait on the network.
std.mem.swap(std.ArrayList(telemetry.Event), &self.pending, &batch);
const dropped = self.dropped;
self.dropped = 0;
self.mutex.unlock();
var sent: usize = 0;
self.postEvents(&conn, batch.items, dropped, &sent) catch |err| {
log.warn(.telemetry, "postEvents", .{ .err = err, .events = batch.items.len, .dropped = dropped });
};
const lost: u32 = @intCast(batch.items.len - sent);
// batch is owned by this thread and can be mutated without the lock
if (batch.capacity > RECLAIM_CAPACITY) {
batch.clearAndFree(self.allocator);
} else {
batch.clearRetainingCapacity();
}
self.mutex.lock();
// Re-fold whatever didn't reach the wire back into `dropped`, so the next
// successful POST still reports it — our only failure signal.
self.dropped +|= lost;
if (sent == 0) {
// if nothing was sent, than we didn't get to send the buffer_overflow
// message, so whatever was dropped is still dropped.
self.dropped +|= dropped;
}
}
errdefer _ = self.dropped.fetchAdd(dropped, .monotonic);
}
self.writer.clearRetainingCapacity();
if (dropped > 0) {
fn postEvents(self: *LightPanda, conn: *http.Connection, events: []const telemetry.Event, dropped: u32, sent: *usize) !void {
// The overflow report rides ahead of the first body; the rest of that body
// and any subsequent ones are filled from `events` below.
const has_overflow = dropped > 0;
if (has_overflow) {
_ = try self.writeEvent(.{ .buffer_overflow = .{ .dropped = dropped } });
}
var sent: usize = 0;
for (h..t) |i| {
const fit = try self.writeEvent(self.buffer[i % BUFFER_SIZE]);
if (!fit) break;
var queued: usize = 0;
for (events) |event| {
if (try self.writeEvent(event) == false) {
try self.flush(conn);
sent.* += queued;
queued = 0;
sent += 1;
// now re-write the message that didn't fit
const fit = try self.writeEvent(event);
if (comptime IS_DEBUG) {
std.debug.assert(fit);
}
if (fit == false) {
// we have a single event that can never be serialized. This
// should never happen, but I'm not willing to crash on that
// belief.
continue;
}
}
queued += 1;
}
try conn.setURL(URL);
try conn.setMethod(.POST);
try conn.setBody(self.writer.written());
if (self.writer.written().len != 0) {
try self.flush(conn);
sent.* += queued;
}
}
self.head.store(h + sent, .release);
self.network.submitRequest(conn);
fn flush(self: *LightPanda, conn: *http.Connection) !void {
self._flush(conn) catch |err| {
log.warn(.telemetry, "flush", .{ .err = err, .size = self.writer.written().len });
return err;
};
}
fn _flush(self: *LightPanda, conn: *http.Connection) !void {
defer self.writer.clearRetainingCapacity();
try conn.setBody(self.writer.written());
const status = try conn.perform();
if (status != 200) {
return error.ServerError;
}
}
fn writeEvent(self: *LightPanda, event: telemetry.Event) !bool {