authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2023-02-13 13:39:06-07:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2023-03-15 10:48:12-07:00
log5b90fa05a4e5b155f25319713acfc67ad9516c69
tree6aaffe4ec16f7f6a18539bf2397176e001bf71cf
parent0b744d7d670d00fa865ebd17847cbdc1a909ba70

extract ThreadPool and WaitGroup from compiler to std lib

11 files changed, 211 insertions(+), 209 deletions(-)

CMakeLists.txt+2-2
......@@ -506,7 +506,9 @@ set(ZIG_STAGE2_SOURCES
506506 "${CMAKE_SOURCE_DIR}/lib/std/Thread.zig"
507507 "${CMAKE_SOURCE_DIR}/lib/std/Thread/Futex.zig"
508508 "${CMAKE_SOURCE_DIR}/lib/std/Thread/Mutex.zig"
509 "${CMAKE_SOURCE_DIR}/lib/std/Thread/Pool.zig"
509510 "${CMAKE_SOURCE_DIR}/lib/std/Thread/ResetEvent.zig"
511 "${CMAKE_SOURCE_DIR}/lib/std/Thread/WaitGroup.zig"
510512 "${CMAKE_SOURCE_DIR}/lib/std/time.zig"
511513 "${CMAKE_SOURCE_DIR}/lib/std/treap.zig"
512514 "${CMAKE_SOURCE_DIR}/lib/std/unicode.zig"
......@@ -530,9 +532,7 @@ set(ZIG_STAGE2_SOURCES
530532 "${CMAKE_SOURCE_DIR}/src/Package.zig"
531533 "${CMAKE_SOURCE_DIR}/src/RangeSet.zig"
532534 "${CMAKE_SOURCE_DIR}/src/Sema.zig"
533 "${CMAKE_SOURCE_DIR}/src/ThreadPool.zig"
534535 "${CMAKE_SOURCE_DIR}/src/TypedValue.zig"
535 "${CMAKE_SOURCE_DIR}/src/WaitGroup.zig"
536536 "${CMAKE_SOURCE_DIR}/src/Zir.zig"
537537 "${CMAKE_SOURCE_DIR}/src/arch/aarch64/CodeGen.zig"
538538 "${CMAKE_SOURCE_DIR}/src/arch/aarch64/Emit.zig"
lib/std/Thread.zig+2
......@@ -16,6 +16,8 @@ pub const Mutex = @import("Thread/Mutex.zig");
1616pub const Semaphore = @import("Thread/Semaphore.zig");
1717pub const Condition = @import("Thread/Condition.zig");
1818pub const RwLock = @import("Thread/RwLock.zig");
19pub const Pool = @import("Thread/Pool.zig");
20pub const WaitGroup = @import("Thread/WaitGroup.zig");
1921
2022pub const use_pthreads = target.os.tag != .windows and target.os.tag != .wasi and builtin.link_libc;
2123const is_gnu = target.abi.isGnu();
lib/std/Thread/Pool.zig created+152
......@@ -0,0 +1,152 @@
1const std = @import("std");
2const builtin = @import("builtin");
3const Pool = @This();
4const WaitGroup = @import("WaitGroup.zig");
5
6mutex: std.Thread.Mutex = .{},
7cond: std.Thread.Condition = .{},
8run_queue: RunQueue = .{},
9is_running: bool = true,
10allocator: std.mem.Allocator,
11threads: []std.Thread,
12
13const RunQueue = std.SinglyLinkedList(Runnable);
14const Runnable = struct {
15 runFn: RunProto,
16};
17
18const RunProto = *const fn (*Runnable) void;
19
20pub fn init(pool: *Pool, allocator: std.mem.Allocator) !void {
21 pool.* = .{
22 .allocator = allocator,
23 .threads = &[_]std.Thread{},
24 };
25
26 if (builtin.single_threaded) {
27 return;
28 }
29
30 const thread_count = std.math.max(1, std.Thread.getCpuCount() catch 1);
31 pool.threads = try allocator.alloc(std.Thread, thread_count);
32 errdefer allocator.free(pool.threads);
33
34 // kill and join any threads we spawned previously on error.
35 var spawned: usize = 0;
36 errdefer pool.join(spawned);
37
38 for (pool.threads) |*thread| {
39 thread.* = try std.Thread.spawn(.{}, worker, .{pool});
40 spawned += 1;
41 }
42}
43
44pub fn deinit(pool: *Pool) void {
45 pool.join(pool.threads.len); // kill and join all threads.
46 pool.* = undefined;
47}
48
49fn join(pool: *Pool, spawned: usize) void {
50 if (builtin.single_threaded) {
51 return;
52 }
53
54 {
55 pool.mutex.lock();
56 defer pool.mutex.unlock();
57
58 // ensure future worker threads exit the dequeue loop
59 pool.is_running = false;
60 }
61
62 // wake up any sleeping threads (this can be done outside the mutex)
63 // then wait for all the threads we know are spawned to complete.
64 pool.cond.broadcast();
65 for (pool.threads[0..spawned]) |thread| {
66 thread.join();
67 }
68
69 pool.allocator.free(pool.threads);
70}
71
72pub fn spawn(pool: *Pool, comptime func: anytype, args: anytype) !void {
73 if (builtin.single_threaded) {
74 @call(.auto, func, args);
75 return;
76 }
77
78 const Args = @TypeOf(args);
79 const Closure = struct {
80 arguments: Args,
81 pool: *Pool,
82 run_node: RunQueue.Node = .{ .data = .{ .runFn = runFn } },
83
84 fn runFn(runnable: *Runnable) void {
85 const run_node = @fieldParentPtr(RunQueue.Node, "data", runnable);
86 const closure = @fieldParentPtr(@This(), "run_node", run_node);
87 @call(.auto, func, closure.arguments);
88
89 // The thread pool's allocator is protected by the mutex.
90 const mutex = &closure.pool.mutex;
91 mutex.lock();
92 defer mutex.unlock();
93
94 closure.pool.allocator.destroy(closure);
95 }
96 };
97
98 {
99 pool.mutex.lock();
100 defer pool.mutex.unlock();
101
102 const closure = try pool.allocator.create(Closure);
103 closure.* = .{
104 .arguments = args,
105 .pool = pool,
106 };
107
108 pool.run_queue.prepend(&closure.run_node);
109 }
110
111 // Notify waiting threads outside the lock to try and keep the critical section small.
112 pool.cond.signal();
113}
114
115fn worker(pool: *Pool) void {
116 pool.mutex.lock();
117 defer pool.mutex.unlock();
118
119 while (true) {
120 while (pool.run_queue.popFirst()) |run_node| {
121 // Temporarily unlock the mutex in order to execute the run_node
122 pool.mutex.unlock();
123 defer pool.mutex.lock();
124
125 const runFn = run_node.data.runFn;
126 runFn(&run_node.data);
127 }
128
129 // Stop executing instead of waiting if the thread pool is no longer running.
130 if (pool.is_running) {
131 pool.cond.wait(&pool.mutex);
132 } else {
133 break;
134 }
135 }
136}
137
138pub fn waitAndWork(pool: *Pool, wait_group: *WaitGroup) void {
139 while (!wait_group.isDone()) {
140 if (blk: {
141 pool.mutex.lock();
142 defer pool.mutex.unlock();
143 break :blk pool.run_queue.popFirst();
144 }) |run_node| {
145 run_node.data.runFn(&run_node.data);
146 continue;
147 }
148
149 wait_group.wait();
150 return;
151 }
152}
lib/std/Thread/WaitGroup.zig created+46
......@@ -0,0 +1,46 @@
1const std = @import("std");
2const Atomic = std.atomic.Atomic;
3const assert = std.debug.assert;
4const WaitGroup = @This();
5
6const is_waiting: usize = 1 << 0;
7const one_pending: usize = 1 << 1;
8
9state: Atomic(usize) = Atomic(usize).init(0),
10event: std.Thread.ResetEvent = .{},
11
12pub fn start(self: *WaitGroup) void {
13 const state = self.state.fetchAdd(one_pending, .Monotonic);
14 assert((state / one_pending) < (std.math.maxInt(usize) / one_pending));
15}
16
17pub fn finish(self: *WaitGroup) void {
18 const state = self.state.fetchSub(one_pending, .Release);
19 assert((state / one_pending) > 0);
20
21 if (state == (one_pending | is_waiting)) {
22 self.state.fence(.Acquire);
23 self.event.set();
24 }
25}
26
27pub fn wait(self: *WaitGroup) void {
28 var state = self.state.fetchAdd(is_waiting, .Acquire);
29 assert(state & is_waiting == 0);
30
31 if ((state / one_pending) > 0) {
32 self.event.wait();
33 }
34}
35
36pub fn reset(self: *WaitGroup) void {
37 self.state.store(0, .Monotonic);
38 self.event.reset();
39}
40
41pub fn isDone(wg: *WaitGroup) bool {
42 const state = wg.state.load(.Acquire);
43 assert(state & is_waiting == 0);
44
45 return (state / one_pending) == 0;
46}
src/Compilation.zig+2-2
......@@ -7,6 +7,8 @@ const Allocator = std.mem.Allocator;
77const assert = std.debug.assert;
88const log = std.log.scoped(.compilation);
99const Target = std.Target;
10const ThreadPool = std.Thread.Pool;
11const WaitGroup = std.Thread.WaitGroup;
1012
1113const Value = @import("value.zig").Value;
1214const Type = @import("type.zig").Type;
......@@ -30,8 +32,6 @@ const Cache = std.Build.Cache;
3032const translate_c = @import("translate_c.zig");
3133const clang = @import("clang.zig");
3234const c_codegen = @import("codegen/c.zig");
33const ThreadPool = @import("ThreadPool.zig");
34const WaitGroup = @import("WaitGroup.zig");
3535const libtsan = @import("libtsan.zig");
3636const Zir = @import("Zir.zig");
3737const Autodoc = @import("Autodoc.zig");
src/Package.zig+2-2
......@@ -8,11 +8,11 @@ const Allocator = mem.Allocator;
88const assert = std.debug.assert;
99const log = std.log.scoped(.package);
1010const main = @import("main.zig");
11const ThreadPool = std.Thread.Pool;
12const WaitGroup = std.Thread.WaitGroup;
1113
1214const Compilation = @import("Compilation.zig");
1315const Module = @import("Module.zig");
14const ThreadPool = @import("ThreadPool.zig");
15const WaitGroup = @import("WaitGroup.zig");
1616const Cache = std.Build.Cache;
1717const build_options = @import("build_options");
1818const Manifest = @import("Manifest.zig");
src/ThreadPool.zig deleted-152
......@@ -1,152 +0,0 @@
1const std = @import("std");
2const builtin = @import("builtin");
3const ThreadPool = @This();
4const WaitGroup = @import("WaitGroup.zig");
5
6mutex: std.Thread.Mutex = .{},
7cond: std.Thread.Condition = .{},
8run_queue: RunQueue = .{},
9is_running: bool = true,
10allocator: std.mem.Allocator,
11threads: []std.Thread,
12
13const RunQueue = std.SinglyLinkedList(Runnable);
14const Runnable = struct {
15 runFn: RunProto,
16};
17
18const RunProto = *const fn (*Runnable) void;
19
20pub fn init(pool: *ThreadPool, allocator: std.mem.Allocator) !void {
21 pool.* = .{
22 .allocator = allocator,
23 .threads = &[_]std.Thread{},
24 };
25
26 if (builtin.single_threaded) {
27 return;
28 }
29
30 const thread_count = std.math.max(1, std.Thread.getCpuCount() catch 1);
31 pool.threads = try allocator.alloc(std.Thread, thread_count);
32 errdefer allocator.free(pool.threads);
33
34 // kill and join any threads we spawned previously on error.
35 var spawned: usize = 0;
36 errdefer pool.join(spawned);
37
38 for (pool.threads) |*thread| {
39 thread.* = try std.Thread.spawn(.{}, worker, .{pool});
40 spawned += 1;
41 }
42}
43
44pub fn deinit(pool: *ThreadPool) void {
45 pool.join(pool.threads.len); // kill and join all threads.
46 pool.* = undefined;
47}
48
49fn join(pool: *ThreadPool, spawned: usize) void {
50 if (builtin.single_threaded) {
51 return;
52 }
53
54 {
55 pool.mutex.lock();
56 defer pool.mutex.unlock();
57
58 // ensure future worker threads exit the dequeue loop
59 pool.is_running = false;
60 }
61
62 // wake up any sleeping threads (this can be done outside the mutex)
63 // then wait for all the threads we know are spawned to complete.
64 pool.cond.broadcast();
65 for (pool.threads[0..spawned]) |thread| {
66 thread.join();
67 }
68
69 pool.allocator.free(pool.threads);
70}
71
72pub fn spawn(pool: *ThreadPool, comptime func: anytype, args: anytype) !void {
73 if (builtin.single_threaded) {
74 @call(.auto, func, args);
75 return;
76 }
77
78 const Args = @TypeOf(args);
79 const Closure = struct {
80 arguments: Args,
81 pool: *ThreadPool,
82 run_node: RunQueue.Node = .{ .data = .{ .runFn = runFn } },
83
84 fn runFn(runnable: *Runnable) void {
85 const run_node = @fieldParentPtr(RunQueue.Node, "data", runnable);
86 const closure = @fieldParentPtr(@This(), "run_node", run_node);
87 @call(.auto, func, closure.arguments);
88
89 // The thread pool's allocator is protected by the mutex.
90 const mutex = &closure.pool.mutex;
91 mutex.lock();
92 defer mutex.unlock();
93
94 closure.pool.allocator.destroy(closure);
95 }
96 };
97
98 {
99 pool.mutex.lock();
100 defer pool.mutex.unlock();
101
102 const closure = try pool.allocator.create(Closure);
103 closure.* = .{
104 .arguments = args,
105 .pool = pool,
106 };
107
108 pool.run_queue.prepend(&closure.run_node);
109 }
110
111 // Notify waiting threads outside the lock to try and keep the critical section small.
112 pool.cond.signal();
113}
114
115fn worker(pool: *ThreadPool) void {
116 pool.mutex.lock();
117 defer pool.mutex.unlock();
118
119 while (true) {
120 while (pool.run_queue.popFirst()) |run_node| {
121 // Temporarily unlock the mutex in order to execute the run_node
122 pool.mutex.unlock();
123 defer pool.mutex.lock();
124
125 const runFn = run_node.data.runFn;
126 runFn(&run_node.data);
127 }
128
129 // Stop executing instead of waiting if the thread pool is no longer running.
130 if (pool.is_running) {
131 pool.cond.wait(&pool.mutex);
132 } else {
133 break;
134 }
135 }
136}
137
138pub fn waitAndWork(pool: *ThreadPool, wait_group: *WaitGroup) void {
139 while (!wait_group.isDone()) {
140 if (blk: {
141 pool.mutex.lock();
142 defer pool.mutex.unlock();
143 break :blk pool.run_queue.popFirst();
144 }) |run_node| {
145 run_node.data.runFn(&run_node.data);
146 continue;
147 }
148
149 wait_group.wait();
150 return;
151 }
152}
src/WaitGroup.zig deleted-46
......@@ -1,46 +0,0 @@
1const std = @import("std");
2const Atomic = std.atomic.Atomic;
3const assert = std.debug.assert;
4const WaitGroup = @This();
5
6const is_waiting: usize = 1 << 0;
7const one_pending: usize = 1 << 1;
8
9state: Atomic(usize) = Atomic(usize).init(0),
10event: std.Thread.ResetEvent = .{},
11
12pub fn start(self: *WaitGroup) void {
13 const state = self.state.fetchAdd(one_pending, .Monotonic);
14 assert((state / one_pending) < (std.math.maxInt(usize) / one_pending));
15}
16
17pub fn finish(self: *WaitGroup) void {
18 const state = self.state.fetchSub(one_pending, .Release);
19 assert((state / one_pending) > 0);
20
21 if (state == (one_pending | is_waiting)) {
22 self.state.fence(.Acquire);
23 self.event.set();
24 }
25}
26
27pub fn wait(self: *WaitGroup) void {
28 var state = self.state.fetchAdd(is_waiting, .Acquire);
29 assert(state & is_waiting == 0);
30
31 if ((state / one_pending) > 0) {
32 self.event.wait();
33 }
34}
35
36pub fn reset(self: *WaitGroup) void {
37 self.state.store(0, .Monotonic);
38 self.event.reset();
39}
40
41pub fn isDone(wg: *WaitGroup) bool {
42 const state = wg.state.load(.Acquire);
43 assert(state & is_waiting == 0);
44
45 return (state / one_pending) == 0;
46}
src/link/MachO/CodeSignature.zig+2-2
......@@ -7,12 +7,12 @@ const log = std.log.scoped(.link);
77const macho = std.macho;
88const mem = std.mem;
99const testing = std.testing;
10const ThreadPool = std.Thread.Pool;
11const WaitGroup = std.Thread.WaitGroup;
1012
1113const Allocator = mem.Allocator;
1214const Compilation = @import("../../Compilation.zig");
1315const Sha256 = std.crypto.hash.sha2.Sha256;
14const ThreadPool = @import("../../ThreadPool.zig");
15const WaitGroup = @import("../../WaitGroup.zig");
1616
1717const hash_size = Sha256.digest_length;
1818
src/main.zig+1-1
......@@ -9,6 +9,7 @@ const Allocator = mem.Allocator;
99const ArrayList = std.ArrayList;
1010const Ast = std.zig.Ast;
1111const warn = std.log.warn;
12const ThreadPool = std.Thread.Pool;
1213
1314const tracy = @import("tracy.zig");
1415const Compilation = @import("Compilation.zig");
......@@ -22,7 +23,6 @@ const translate_c = @import("translate_c.zig");
2223const clang = @import("clang.zig");
2324const Cache = std.Build.Cache;
2425const target_util = @import("target.zig");
25const ThreadPool = @import("ThreadPool.zig");
2626const crash_report = @import("crash_report.zig");
2727
2828pub const std_options = struct {
src/test.zig+2-2
......@@ -4,14 +4,14 @@ const Allocator = std.mem.Allocator;
44const CrossTarget = std.zig.CrossTarget;
55const print = std.debug.print;
66const assert = std.debug.assert;
7const ThreadPool = std.Thread.Pool;
8const WaitGroup = std.Thread.WaitGroup;
79
810const link = @import("link.zig");
911const Compilation = @import("Compilation.zig");
1012const Package = @import("Package.zig");
1113const introspect = @import("introspect.zig");
1214const build_options = @import("build_options");
13const ThreadPool = @import("ThreadPool.zig");
14const WaitGroup = @import("WaitGroup.zig");
1515const zig_h = link.File.C.zig_h;
1616
1717const enable_qemu: bool = build_options.enable_qemu;