authorgravatar for mlugg@mlugg.co.ukMatthew Lugg <mlugg@mlugg.co.uk> 2025-12-05 11:46:13+00:00
committergravatar for jacobly@ziglang.orgJacob Young <jacobly@ziglang.org> 2025-12-14 18:42:51-05:00
logbf53a2de51858e034435b83c47ce0174b14b1e1f
tree83cf994d5d252a96a3ab714961cf9ea820bece64
parentb599ac5e82930c83b0d7232258c0c208a189c73e

DO NOT MERGE: jobserver usage example with `std.Thread.Pool`

Co-authored-by: Jacob Young <jacobly0@users.noreply.github.com>

3 files changed, 84 insertions(+), 0 deletions(-)

lib/std/Thread/Pool.zig+11
...@@ -15,6 +15,7 @@ ids: if (builtin.single_threaded) struct {...@@ -15,6 +15,7 @@ ids: if (builtin.single_threaded) struct {
15 return 0;15 return 0;
16 }16 }
17} else std.AutoArrayHashMapUnmanaged(std.Thread.Id, void),17} else std.AutoArrayHashMapUnmanaged(std.Thread.Id, void),
18job_client: ?*std.job.Client,
1819
19const Runnable = struct {20const Runnable = struct {
20 runFn: RunProto,21 runFn: RunProto,
...@@ -28,6 +29,7 @@ pub const Options = struct {...@@ -28,6 +29,7 @@ pub const Options = struct {
28 n_jobs: ?usize = null,29 n_jobs: ?usize = null,
29 track_ids: bool = false,30 track_ids: bool = false,
30 stack_size: usize = std.Thread.SpawnConfig.default_stack_size,31 stack_size: usize = std.Thread.SpawnConfig.default_stack_size,
32 job_client: ?*std.job.Client = null,
31};33};
3234
33pub fn init(pool: *Pool, options: Options) !void {35pub fn init(pool: *Pool, options: Options) !void {
...@@ -37,6 +39,7 @@ pub fn init(pool: *Pool, options: Options) !void {...@@ -37,6 +39,7 @@ pub fn init(pool: *Pool, options: Options) !void {
37 .allocator = allocator,39 .allocator = allocator,
38 .threads = if (builtin.single_threaded) .{} else &.{},40 .threads = if (builtin.single_threaded) .{} else &.{},
39 .ids = .{},41 .ids = .{},
42 .job_client = options.job_client,
40 };43 };
4144
42 if (builtin.single_threaded) {45 if (builtin.single_threaded) {
...@@ -284,6 +287,13 @@ fn worker(pool: *Pool) void {...@@ -284,6 +287,13 @@ fn worker(pool: *Pool) void {
284 if (id) |_| pool.ids.putAssumeCapacityNoClobber(std.Thread.getCurrentId(), {});287 if (id) |_| pool.ids.putAssumeCapacityNoClobber(std.Thread.getCurrentId(), {});
285288
286 while (true) {289 while (true) {
290 const job_token: ?std.job.Client.Token = if (pool.job_client) |job_client| token: {
291 pool.mutex.unlock();
292 defer pool.mutex.lock();
293 break :token job_client.acquire() catch |err| {
294 std.debug.panic("failed to acquire job token: {t}", .{err});
295 };
296 } else null;
287 while (pool.run_queue.popFirst()) |run_node| {297 while (pool.run_queue.popFirst()) |run_node| {
288 // Temporarily unlock the mutex in order to execute the run_node298 // Temporarily unlock the mutex in order to execute the run_node
289 pool.mutex.unlock();299 pool.mutex.unlock();
...@@ -292,6 +302,7 @@ fn worker(pool: *Pool) void {...@@ -292,6 +302,7 @@ fn worker(pool: *Pool) void {
292 const runnable: *Runnable = @fieldParentPtr("node", run_node);302 const runnable: *Runnable = @fieldParentPtr("node", run_node);
293 runnable.runFn(runnable, id);303 runnable.runFn(runnable, id);
294 }304 }
305 if (job_token) |t| t.release();
295306
296 // Stop executing instead of waiting if the thread pool is no longer running.307 // Stop executing instead of waiting if the thread pool is no longer running.
297 if (pool.is_running) {308 if (pool.is_running) {
server.zig created+32
...@@ -0,0 +1,32 @@
1const num_children = 25;
2
3pub fn main() !void {
4 var arena_instance: std.heap.ArenaAllocator = .init(std.heap.page_allocator);
5 defer arena_instance.deinit();
6 const arena = arena_instance.allocator();
7
8 const cpu_count = try std.Thread.getCpuCount();
9 var env = try std.process.getEnvMap(arena);
10
11 var job_server: std.job.Server = try .init(arena, @intCast(cpu_count), &env);
12 defer job_server.deinit();
13
14 var children: [num_children]std.process.Child = undefined;
15 for (&children, 0..) |*c, child_num| {
16 const child_num_str = try std.fmt.allocPrint(arena, "{d}", .{child_num});
17 const argv = try arena.dupe([]const u8, &.{ switch (builtin.os.tag) {
18 else => "./worker",
19 .windows => ".\\worker.exe",
20 }, child_num_str });
21 c.* = .init(argv, arena);
22 c.env_map = &env;
23 }
24
25 std.log.info("Spawning {d} workers on {d} CPUs", .{ children.len, cpu_count });
26 for (&children) |*c| try c.spawn();
27 for (&children) |*c| _ = try c.wait();
28 std.log.info("All {d} workers exited", .{children.len});
29}
30
31const builtin = @import("builtin");
32const std = @import("std");
worker.zig created+41
...@@ -0,0 +1,41 @@
1const num_tasks = 25;
2
3pub fn main() !void {
4 var arena_instance: std.heap.ArenaAllocator = .init(std.heap.page_allocator);
5 defer arena_instance.deinit();
6 const arena = arena_instance.allocator();
7
8 const args = try std.process.argsAlloc(arena);
9 const env = try std.process.getEnvMap(arena);
10
11 const process_index = try std.fmt.parseInt(usize, args[1], 10);
12
13 var job_client: std.job.Client = try .init(arena, &env);
14 defer job_client.deinit();
15
16 var thread_pool: std.Thread.Pool = undefined;
17 try thread_pool.init(.{ .allocator = arena, .job_client = &job_client });
18 defer thread_pool.deinit();
19
20 // Spawn a bunch of tasks with a variable amount of CPU-intensive work
21 var rng: std.Random.DefaultPrng = .init(process_index);
22 const r = rng.random();
23 for (0..num_tasks) |_| {
24 try thread_pool.spawn(doWork, .{ process_index, r.intRangeAtMost(u32, 300_000, 1_000_000) });
25 }
26}
27
28fn doWork(
29 process_index: usize,
30 work_amount: u32,
31) void {
32 std.log.info("[process {d}] start", .{process_index});
33 defer std.log.info("[process {d}] stop", .{process_index});
34
35 // Badly simulate single-threaded CPU-intensive work
36 for (0..work_amount) |_| {
37 for (0..1000) |_| asm volatile ("nop");
38 }
39}
40
41const std = @import("std");