serveWebSocket

fn serveWebSocket(ws: *WebServer, sock: *http.Server.WebSocket) !noreturn

File

Code

fn serveWebSocket(ws: *WebServer, sock: *http.Server.WebSocket) !noreturn {
    const graph = ws.graph;
    const gpa = graph.cache.gpa;
    const io = graph.io;
    log.err("TODO serve a different message when the configuration changes", .{});
    const configured = &ws.configured.?;
    const maker = configured.maker;
    const all_steps = maker.step_stack.keys();

    var prev_build_status = ws.build_status.load(.monotonic);

    const prev_step_status_bits = try gpa.alloc(u8, configured.step_status_bits.len);
    defer gpa.free(prev_step_status_bits);
    for (prev_step_status_bits, configured.step_status_bits) |*copy, *shared| {
        copy.* = @atomicLoad(u8, shared, .monotonic);
    }

    var recv_thread = try io.concurrent(recvWebSocketMessages, .{ ws, sock });
    defer recv_thread.cancel(io);

    {
        const hello_header: abi.Hello = .{
            .status = prev_build_status,
            .flags = .{
                .time_report = graph.time_report,
            },
            .timestamp = ws.now(),
            .steps_len = @intCast(all_steps.len),
        };
        var bufs: [3][]const u8 = .{ @ptrCast(&hello_header), configured.step_names_trailing, prev_step_status_bits };
        try sock.writeMessageVec(&bufs, .binary);
    }

    var prev_fuzz: Fuzz.Previous = .init;
    var prev_time: i64 = std.math.minInt(i64);
    while (true) {
        const start_time = ws.now();
        const start_update_id = ws.update_id.load(.acquire);

        if (ws.fuzz) |*fuzz| {
            try fuzz.sendUpdate(sock, &prev_fuzz);
        }

        {
            try configured.time_report_mutex.lock(io);
            defer configured.time_report_mutex.unlock(io);
            for (configured.time_report_msgs, configured.time_report_update_times) |msg, update_time| {
                if (update_time <= prev_time) continue;
                // We want to send `msg`, but shouldn't block `configured.time_report_mutex` while we do, so
                // that we don't hold up the build system on the client accepting this packet.
                const owned_msg = try gpa.dupe(u8, msg);
                defer gpa.free(owned_msg);
                // Temporarily unlock, then re-lock after the message is sent.
                configured.time_report_mutex.unlock(io);
                defer configured.time_report_mutex.lockUncancelable(io);
                try sock.writeMessage(owned_msg, .binary);
            }
        }

        {
            const build_status = ws.build_status.load(.monotonic);
            if (build_status != prev_build_status) {
                prev_build_status = build_status;
                const msg: abi.StatusUpdate = .{ .new = build_status };
                try sock.writeMessage(@ptrCast(&msg), .binary);
            }
        }

        for (prev_step_status_bits, configured.step_status_bits, 0..) |*prev_byte, *shared, byte_idx| {
            const cur_byte = @atomicLoad(u8, shared, .monotonic);
            if (prev_byte.* == cur_byte) continue;
            const cur: [4]abi.StepUpdate.Status = .{
                @fromBackingInt(@intCast(@as(u2, @truncate(cur_byte >> 0)))),
                @fromBackingInt(@intCast(@as(u2, @truncate(cur_byte >> 2)))),
                @fromBackingInt(@intCast(@as(u2, @truncate(cur_byte >> 4)))),
                @fromBackingInt(@intCast(@as(u2, @truncate(cur_byte >> 6)))),
            };
            const prev: [4]abi.StepUpdate.Status = .{
                @fromBackingInt(@intCast(@as(u2, @truncate(prev_byte.* >> 0)))),
                @fromBackingInt(@intCast(@as(u2, @truncate(prev_byte.* >> 2)))),
                @fromBackingInt(@intCast(@as(u2, @truncate(prev_byte.* >> 4)))),
                @fromBackingInt(@intCast(@as(u2, @truncate(prev_byte.* >> 6)))),
            };
            for (cur, prev, byte_idx * 4..) |cur_status, prev_status, step_idx| {
                const msg: abi.StepUpdate = .{ .step_idx = @intCast(step_idx), .bits = .{ .status = cur_status } };
                if (cur_status != prev_status) try sock.writeMessage(@ptrCast(&msg), .binary);
            }
            prev_byte.* = cur_byte;
        }

        prev_time = start_time;

        const old_cp = io.swapCancelProtection(.blocked);
        defer _ = io.swapCancelProtection(old_cp);
        io.futexWaitTimeout(
            u32,
            &ws.update_id.raw,
            start_update_id,
            .{ .duration = .{
                .clock = .awake,
                .raw = .fromMilliseconds(default_update_interval_ms),
            } },
        ) catch |err| switch (err) {
            error.Canceled => unreachable,
        };
    }
}