tiny.http.ThreadPool
Defined in tiny.http.
Source
Source: lib/http/src/pool.zig:24
zig
pub fn ThreadPool(comptime Task: type) type { const Owner = struct { const Self = @This(); const TaskNode = struct { link: std.DoublyLinkedList.Node = .{}, task: Task = undefined, fn fromLink(link: *std.DoublyLinkedList.Node) *TaskNode { return @fieldParentPtr("link", link); } }; const State = struct { workers: []thread.JoinHandle, tasks: []TaskNode, free: std.DoublyLinkedList, ready: std.DoublyLinkedList, queue_mutex: std.atomic.Mutex, shutdown: std.atomic.Value(bool), in_flight_tasks: std.atomic.Value(usize), wake_sem: thread.Semaphore, started_workers: usize, }; pub const claim: alloc_phase.capacity.Declaration = .{ .source = .{ .id = "http.thread_pool", .kind = .phase_static, .limit_source = .caller, .storage = .{ .covered = &.{ .{ .id = "state_and_exact_worker_handle_and_task_node_slices", .lifetime = .steady, .detail = "State and exact worker-handle and task-node slices", }, }, .excluded = &.{ "caller-owned storage reachable from task values", "allocations performed by task run and completion methods", "OS thread stacks, handles, semaphore resources, clocks, and sleeps", }, }, .capacity = .{ .inputs = &.{ alloc_phase.capacity.bindInput(Limits, "tasks", "tasks"), alloc_phase.capacity.bindInput(Limits, "workers", "workers"), }, .type_selectors = &.{}, .nodes = &.{ .{ .input = 0 }, .{ .input = 1 }, .{ .add = .{ .left = 0, .right = 1 } }, }, .assertions = &.{.{ .scope = .closure_total, .measure = .retained, .relation = .upper_bound, .expression = 2, }}, }, .overload = .{ .kind = .reject_before_mutation, .detail = "PoolFull is returned before a free slot is removed or queue state changes", }, .risks = .{ .transitive = .{ .status = .open, .detail = "the concrete task run and completion methods remain separate owner-local storage obligations", }, .foreign = .{ .status = .excluded, .detail = "sys thread, semaphore, yield, time, sleep, and worker-exit TLS cleanup cross the owner-local storage claim", }, }, .obligations = &.{ .{ .key = "http_capacity", .role = .capacity_model }, .{ .key = "http_sealed_maximum_overload", .role = .overload }, .{ .key = "http_sealed_maximum_transitive_risk", .role = .transitive_risk }, .{ .key = "http_oom_retry", .role = .custom }, .{ .key = "http_sys_tls", .role = .foreign_risk }, }, }, .bindings = .{ .owner = @This(), .seal = .{ .family = alloc_phase.capacity.selector(@This().activate), .premise = .{ .class = .checked_semantic_fact, .authority = .checker, }, }, .teardown = .{ .family = alloc_phase.capacity.selector(@This().deinit), .premise = .{ .class = .checked_semantic_fact, .authority = .checker, }, }, }, }; phase: alloc_phase.capacity.Phase, capacity: Capacity, state: *State, pub const Limits = struct { workers: usize, tasks: usize, }; pub const Capacity = struct { workers: usize, tasks: usize, state_bytes: usize, worker_bytes: usize, task_bytes: usize, working_bytes: usize, pub fn derive(limits: Limits) error{ InvalidWorkerCount, CapacityOverflow }!Capacity { if (limits.workers == 0) return error.InvalidWorkerCount; const state_bytes = @sizeOf(State); const worker_bytes = try alloc_phase.capacity.mul( usize, limits.workers, @sizeOf(thread.JoinHandle), ); const task_bytes = try alloc_phase.capacity.mul( usize, limits.tasks, @sizeOf(TaskNode), ); const partial_bytes = try alloc_phase.capacity.add( usize, state_bytes, worker_bytes, ); const working_bytes = try alloc_phase.capacity.add( usize, partial_bytes, task_bytes, ); return .{ .workers = limits.workers, .tasks = limits.tasks, .state_bytes = state_bytes, .worker_bytes = worker_bytes, .task_bytes = task_bytes, .working_bytes = working_bytes, }; } }; pub const Exhaustion = error{PoolFull}; pub const SubmitError = Exhaustion || error{ NotStarted, ShuttingDown }; pub fn init(allocator: Allocator, limits: Limits) !Self { const capacity = try Capacity.derive(limits); const state = try allocator.create(State); errdefer allocator.destroy(state); const workers = try allocator.alloc(thread.JoinHandle, capacity.workers); errdefer allocator.free(workers); const tasks = try allocator.alloc(TaskNode, capacity.tasks); errdefer allocator.free(tasks); state.* = .{ .workers = workers, .tasks = tasks, .free = .{}, .ready = .{}, .queue_mutex = .unlocked, .shutdown = std.atomic.Value(bool).init(false), .in_flight_tasks = std.atomic.Value(usize).init(0), .wake_sem = .{}, .started_workers = 0, }; for (state.tasks) |*task| { task.* = .{}; state.free.append(&task.link); } errdefer { state.shutdown.store(true, .release); for (0..state.started_workers) |_| state.wake_sem.post(); for (state.workers[0..state.started_workers]) |worker| worker.join(); } for (state.workers) |*worker| { worker.* = try thread.spawn(workerLoop, .{state}); state.started_workers += 1; } return .{ .phase = .initialization, .capacity = capacity, .state = state, }; } pub fn activate(self: *Self) error{AlreadyStarted}!void { if (self.phase != .initialization) return error.AlreadyStarted; self.phase = .steady; } pub fn deinit(self: *Self, allocator: Allocator) void { std.debug.assert(self.phase != .teardown); const state = self.state; state.shutdown.store(true, .release); for (0..state.started_workers) |_| state.wake_sem.post(); for (state.workers[0..state.started_workers]) |worker| worker.join(); self.phase = .teardown; allocator.free(state.tasks); allocator.free(state.workers); allocator.destroy(state); self.state = undefined; } pub fn submit(self: *Self, submitted: Task) SubmitError!void { if (self.phase != .steady) return error.NotStarted; const state = self.state; lock(&state.queue_mutex); if (state.shutdown.load(.acquire)) { unlock(&state.queue_mutex); return error.ShuttingDown; } const task_node = TaskNode.fromLink(state.free.popFirst() orelse { unlock(&state.queue_mutex); return error.PoolFull; }); task_node.task = submitted; state.ready.append(&task_node.link); _ = state.in_flight_tasks.fetchAdd(1, .acq_rel); unlock(&state.queue_mutex); state.wake_sem.post(); } pub fn drain( self: *Self, clock: time.AwakeClock, timeout: time.Duration, ) time.ClockError!bool { std.debug.assert(self.phase == .steady); const deadline = try DrainDeadline.init(clock, timeout); while (self.state.in_flight_tasks.load(.acquire) != 0) { if (try deadline.reached()) return false; sleepNanoseconds(1 * std.time.ns_per_ms); } return true; } fn workerLoop(state: *State) void { while (true) { state.wake_sem.wait(); if (state.shutdown.load(.acquire)) return; lock(&state.queue_mutex); const task_node = TaskNode.fromLink(state.ready.popFirst() orelse { unlock(&state.queue_mutex); continue; }); const task = task_node.task; unlock(&state.queue_mutex); task.run(); lock(&state.queue_mutex); state.free.append(&task_node.link); unlock(&state.queue_mutex); task.complete(); _ = state.in_flight_tasks.fetchSub(1, .acq_rel); } } }; comptime alloc_phase.capacity.requireAllocatorRejectingOwnerShape(Owner); return Owner;}Source: lib/http/src/root.zig:74
zig
pub const ThreadPool = pool.ThreadPool;Audit
| Definitions | 1 |
|---|---|
| Public names | 1 |
| Members | 0 |
| Version | 26.7.0 |
| Revision | daab053ee433 |