lib/http/src/profiling/pool.zig

daab053ee43316e1809a84551d573ddd1e5bf3d2

 1 const std = @import("std");
 2 const bench = @import("bench");
 3 const http = @import("http");
 4 
 5 const task_capacity = 256;
 6 const wave_count = 16;
 7 
 8 const CountTask = struct {
 9     counter: *std.atomic.Value(usize),
10 
11     pub fn run(task: CountTask) void {
12         _ = task.counter.fetchAdd(1, .acq_rel);
13     }
14 
15     pub fn complete(_: CountTask) void {}
16 };
17 
18 const ThreadPool = http.ThreadPool(CountTask);
19 
20 fn dispatchBounded(allocator: std.mem.Allocator) void {
21     var pool = ThreadPool.init(allocator, .{
22         .workers = 4,
23         .tasks = task_capacity,
24     }) catch unreachable;
25     defer pool.deinit(allocator);
26     pool.activate() catch unreachable;
27 
28     const phase = bench.phaseAt("http.pool.dispatch.bounded", @src());
29     defer phase.end();
30 
31     var counter = std.atomic.Value(usize).init(0);
32     for (0..wave_count) |_| {
33         for (0..task_capacity) |_| {
34             pool.submit(.{ .counter = &counter }) catch unreachable;
35         }
36         const drained = pool.drain(
37             .system(),
38             .fromNanoseconds(5 * std.time.ns_per_s),
39         ) catch @panic("http bounded dispatch benchmark clock failed");
40         if (!drained) @panic("http bounded dispatch benchmark timed out");
41         bench.coz.progressNamed("http.pool.dispatch.bounded.wave");
42     }
43     if (counter.load(.acquire) != task_capacity * wave_count) {
44         @panic("http bounded dispatch benchmark lost work");
45     }
46 }
47 
48 pub fn addTo(suite: *bench.Suite) !void {
49     try suite.add("http.pool.dispatch.bounded.4096", dispatchBounded, .{});
50 }