authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2020-12-21 14:21:51-07:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2020-12-23 13:36:21-08:00
log829c00a77fd2d6b7576c6d2b724f69ba9cfe10f2
tree879be7bfbb6b021489f1d45d63af9732dba4593a
parent4eb4d26fa14524652bed69325eb491f39701d995

kprotty ThreadPool and WaitGroup patch

5 files changed, 57 insertions(+), 104 deletions(-)

CMakeLists.txt-1
...@@ -512,7 +512,6 @@ set(ZIG_STAGE2_SOURCES...@@ -512,7 +512,6 @@ set(ZIG_STAGE2_SOURCES
512 "${CMAKE_SOURCE_DIR}/src/Cache.zig"512 "${CMAKE_SOURCE_DIR}/src/Cache.zig"
513 "${CMAKE_SOURCE_DIR}/src/Compilation.zig"513 "${CMAKE_SOURCE_DIR}/src/Compilation.zig"
514 "${CMAKE_SOURCE_DIR}/src/DepTokenizer.zig"514 "${CMAKE_SOURCE_DIR}/src/DepTokenizer.zig"
515 "${CMAKE_SOURCE_DIR}/src/Event.zig"
516 "${CMAKE_SOURCE_DIR}/src/Module.zig"515 "${CMAKE_SOURCE_DIR}/src/Module.zig"
517 "${CMAKE_SOURCE_DIR}/src/Package.zig"516 "${CMAKE_SOURCE_DIR}/src/Package.zig"
518 "${CMAKE_SOURCE_DIR}/src/RangeSet.zig"517 "${CMAKE_SOURCE_DIR}/src/RangeSet.zig"
ci/drone/linux_script+1-2
...@@ -17,8 +17,7 @@ git config core.abbrev 9...@@ -17,8 +17,7 @@ git config core.abbrev 9
1717
18mkdir build18mkdir build
19cd build19cd build
20# TODO figure out why Drone CI is deadlocking and stop passing -DZIG_SINGLE_THREADED=ON20cmake .. -DCMAKE_BUILD_TYPE=Release "-DCMAKE_INSTALL_PREFIX=$DISTDIR" -DZIG_STATIC=ON -DCMAKE_PREFIX_PATH=/deps/local -GNinja
21cmake .. -DCMAKE_BUILD_TYPE=Release "-DCMAKE_INSTALL_PREFIX=$DISTDIR" -DZIG_STATIC=ON -DCMAKE_PREFIX_PATH=/deps/local -GNinja -DZIG_SINGLE_THREADED=ON
2221
23samu install22samu install
24./zig build test -Dskip-release -Dskip-non-native23./zig build test -Dskip-release -Dskip-non-native
src/Event.zig deleted-43
...@@ -1,43 +0,0 @@
1// SPDX-License-Identifier: MIT
2// Copyright (c) 2015-2020 Zig Contributors
3// This file is part of [zig](https://ziglang.org/), which is MIT licensed.
4// The MIT license requires this copyright notice to be included in all copies
5// and substantial portions of the software.
6const std = @import("std");
7const Event = @This();
8
9lock: std.Mutex = .{},
10event: std.ResetEvent = undefined,
11state: enum { empty, waiting, notified } = .empty,
12
13pub fn wait(self: *Event) void {
14 const held = self.lock.acquire();
15
16 switch (self.state) {
17 .empty => {
18 self.state = .waiting;
19 self.event = @TypeOf(self.event).init();
20 held.release();
21 self.event.wait();
22 self.event.deinit();
23 },
24 .waiting => unreachable,
25 .notified => held.release(),
26 }
27}
28
29pub fn set(self: *Event) void {
30 const held = self.lock.acquire();
31
32 switch (self.state) {
33 .empty => {
34 self.state = .notified;
35 held.release();
36 },
37 .waiting => {
38 held.release();
39 self.event.set();
40 },
41 .notified => unreachable,
42 }
43}
src/ThreadPool.zig+35-41
...@@ -9,12 +9,12 @@ const ThreadPool = @This();...@@ -9,12 +9,12 @@ const ThreadPool = @This();
9lock: std.Mutex = .{},9lock: std.Mutex = .{},
10is_running: bool = true,10is_running: bool = true,
11allocator: *std.mem.Allocator,11allocator: *std.mem.Allocator,
12running: usize = 0,12spawned: usize = 0,
13threads: []*std.Thread,13threads: []*std.Thread,
14run_queue: RunQueue = .{},14run_queue: RunQueue = .{},
15idle_queue: IdleQueue = .{},15idle_queue: IdleQueue = .{},
1616
17const IdleQueue = std.SinglyLinkedList(std.AutoResetEvent);17const IdleQueue = std.SinglyLinkedList(std.ResetEvent);
18const RunQueue = std.SinglyLinkedList(Runnable);18const RunQueue = std.SinglyLinkedList(Runnable);
19const Runnable = struct {19const Runnable = struct {
20 runFn: fn (*Runnable) void,20 runFn: fn (*Runnable) void,
...@@ -30,49 +30,37 @@ pub fn init(self: *ThreadPool, allocator: *std.mem.Allocator) !void {...@@ -30,49 +30,37 @@ pub fn init(self: *ThreadPool, allocator: *std.mem.Allocator) !void {
3030
31 errdefer self.deinit();31 errdefer self.deinit();
3232
33 var num_threads = std.Thread.cpuCount() catch 1;33 var num_threads = std.math.max(1, std.Thread.cpuCount() catch 1);
34 if (num_threads > 0)34 self.threads = try allocator.alloc(*std.Thread, num_threads);
35 self.threads = try allocator.alloc(*std.Thread, num_threads);
3635
37 while (num_threads > 0) : (num_threads -= 1) {36 while (num_threads > 0) : (num_threads -= 1) {
38 const thread = try std.Thread.spawn(self, runWorker);37 const thread = try std.Thread.spawn(self, runWorker);
39 self.threads[self.running] = thread;38 self.threads[self.spawned] = thread;
40 self.running += 1;39 self.spawned += 1;
41 }40 }
42}41}
4342
44pub fn deinit(self: *ThreadPool) void {43pub fn deinit(self: *ThreadPool) void {
45 self.shutdown();44 {
4645 const held = self.lock.acquire();
47 std.debug.assert(!self.is_running);46 defer held.release();
48 for (self.threads[0..self.running]) |thread|
49 thread.wait();
50
51 defer self.threads = &[_]*std.Thread{};
52 if (self.running > 0)
53 self.allocator.free(self.threads);
54}
55
56pub fn shutdown(self: *ThreadPool) void {
57 const held = self.lock.acquire();
58
59 if (!self.is_running)
60 return held.release();
6147
62 var idle_queue = self.idle_queue;48 self.is_running = false;
63 self.idle_queue = .{};49 while (self.idle_queue.popFirst()) |idle_node|
64 self.is_running = false;50 idle_node.data.set();
65 held.release();51 }
6652
67 while (idle_queue.popFirst()) |idle_node|53 defer self.allocator.free(self.threads);
68 idle_node.data.set();54 for (self.threads[0..self.spawned]) |thread|
55 thread.wait();
69}56}
7057
71pub fn spawn(self: *ThreadPool, comptime func: anytype, args: anytype) !void {58pub fn spawn(self: *ThreadPool, comptime func: anytype, args: anytype) !void {
72 if (std.builtin.single_threaded) {59 if (std.builtin.single_threaded) {
73 @call(.{}, func, args);60 const result = @call(.{}, func, args);
74 return;61 return;
75 }62 }
63
76 const Args = @TypeOf(args);64 const Args = @TypeOf(args);
77 const Closure = struct {65 const Closure = struct {
78 arguments: Args,66 arguments: Args,
...@@ -83,24 +71,26 @@ pub fn spawn(self: *ThreadPool, comptime func: anytype, args: anytype) !void {...@@ -83,24 +71,26 @@ pub fn spawn(self: *ThreadPool, comptime func: anytype, args: anytype) !void {
83 const run_node = @fieldParentPtr(RunQueue.Node, "data", runnable);71 const run_node = @fieldParentPtr(RunQueue.Node, "data", runnable);
84 const closure = @fieldParentPtr(@This(), "run_node", run_node);72 const closure = @fieldParentPtr(@This(), "run_node", run_node);
85 const result = @call(.{}, func, closure.arguments);73 const result = @call(.{}, func, closure.arguments);
74
75 const held = closure.pool.lock.acquire();
76 defer held.release();
86 closure.pool.allocator.destroy(closure);77 closure.pool.allocator.destroy(closure);
87 }78 }
88 };79 };
8980
81 const held = self.lock.acquire();
82 defer held.release();
83
90 const closure = try self.allocator.create(Closure);84 const closure = try self.allocator.create(Closure);
91 closure.* = .{85 closure.* = .{
92 .arguments = args,86 .arguments = args,
93 .pool = self,87 .pool = self,
94 };88 };
9589
96 const held = self.lock.acquire();
97 self.run_queue.prepend(&closure.run_node);90 self.run_queue.prepend(&closure.run_node);
9891
99 const idle_node = self.idle_queue.popFirst();92 if (self.idle_queue.popFirst()) |idle_node|
100 held.release();93 idle_node.data.set();
101
102 if (idle_node) |node|
103 node.data.set();
104}94}
10595
106fn runWorker(self: *ThreadPool) void {96fn runWorker(self: *ThreadPool) void {
...@@ -113,14 +103,18 @@ fn runWorker(self: *ThreadPool) void {...@@ -113,14 +103,18 @@ fn runWorker(self: *ThreadPool) void {
113 continue;103 continue;
114 }104 }
115105
116 if (!self.is_running) {106 if (self.is_running) {
107 var idle_node = IdleQueue.Node{ .data = std.ResetEvent.init() };
108 defer idle_node.data.deinit();
109
110 self.idle_queue.prepend(&idle_node);
117 held.release();111 held.release();
118 return;112
113 idle_node.data.wait();
114 continue;
119 }115 }
120116
121 var idle_node = IdleQueue.Node{ .data = .{} };
122 self.idle_queue.prepend(&idle_node);
123 held.release();117 held.release();
124 idle_node.data.wait();118 return;
125 }119 }
126}120}
src/WaitGroup.zig+21-17
...@@ -5,11 +5,10 @@...@@ -5,11 +5,10 @@
5// and substantial portions of the software.5// and substantial portions of the software.
6const std = @import("std");6const std = @import("std");
7const WaitGroup = @This();7const WaitGroup = @This();
8const Event = @import("Event.zig");
98
10lock: std.Mutex = .{},9lock: std.Mutex = .{},
11counter: usize = 0,10counter: usize = 0,
12event: ?*Event = null,11event: ?*std.ResetEvent = null,
1312
14pub fn start(self: *WaitGroup) void {13pub fn start(self: *WaitGroup) void {
15 const held = self.lock.acquire();14 const held = self.lock.acquire();
...@@ -19,28 +18,33 @@ pub fn start(self: *WaitGroup) void {...@@ -19,28 +18,33 @@ pub fn start(self: *WaitGroup) void {
19}18}
2019
21pub fn stop(self: *WaitGroup) void {20pub fn stop(self: *WaitGroup) void {
22 var event: ?*Event = null;
23 defer if (event) |waiter|
24 waiter.set();
25
26 const held = self.lock.acquire();21 const held = self.lock.acquire();
27 defer held.release();22 defer held.release();
2823
29 self.counter -= 1;24 self.counter -= 1;
30 if (self.counter == 0)25
31 std.mem.swap(?*Event, &self.event, &event);26 if (self.counter == 0) {
27 if (self.event) |event| {
28 self.event = null;
29 event.set();
30 }
31 }
32}32}
3333
34pub fn wait(self: *WaitGroup) void {34pub fn wait(self: *WaitGroup) void {
35 var event = Event{};
36 var has_event = false;
37 defer if (has_event)
38 event.wait();
39
40 const held = self.lock.acquire();35 const held = self.lock.acquire();
41 defer held.release();
4236
43 has_event = self.counter != 0;37 if (self.counter == 0) {
44 if (has_event)38 held.release();
45 self.event = &event;39 return;
40 }
41
42 var event = std.ResetEvent.init();
43 defer event.deinit();
44
45 std.debug.assert(self.event == null);
46 self.event = &event;
47
48 held.release();
49 event.wait();
46}50}