authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2018-07-11 19:38:01-04:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2018-07-11 19:38:01-04:00
log9751a0ae045110fb615c866b94ad47680b9c48c7
treeee995b6ee80e52de40ef5961589b8e3275832328
parent9bdcd2a495d4189d6536d43f1294dffb38daa9a5

std.atomic: use spinlocks

the lock-free data structures all had ABA problems and std.atomic.Stack had a possibility to load an unmapped memory address.

12 files changed, 286 insertions(+), 455 deletions(-)

CMakeLists.txt+1-2
......@@ -432,8 +432,7 @@ set(ZIG_STD_FILES
432432 "array_list.zig"
433433 "atomic/index.zig"
434434 "atomic/int.zig"
435 "atomic/queue_mpmc.zig"
436 "atomic/queue_mpsc.zig"
435 "atomic/queue.zig"
437436 "atomic/stack.zig"
438437 "base64.zig"
439438 "buf_map.zig"
build.zig+4-4
......@@ -91,11 +91,11 @@ pub fn build(b: *Builder) !void {
9191
9292 test_step.dependOn(tests.addPkgTests(b, test_filter, "std/special/compiler_rt/index.zig", "compiler-rt", "Run the compiler_rt tests", modes));
9393
94 test_step.dependOn(tests.addCompareOutputTests(b, test_filter));
94 test_step.dependOn(tests.addCompareOutputTests(b, test_filter, modes));
9595 test_step.dependOn(tests.addBuildExampleTests(b, test_filter));
96 test_step.dependOn(tests.addCompileErrorTests(b, test_filter));
97 test_step.dependOn(tests.addAssembleAndLinkTests(b, test_filter));
98 test_step.dependOn(tests.addRuntimeSafetyTests(b, test_filter));
96 test_step.dependOn(tests.addCompileErrorTests(b, test_filter, modes));
97 test_step.dependOn(tests.addAssembleAndLinkTests(b, test_filter, modes));
98 test_step.dependOn(tests.addRuntimeSafetyTests(b, test_filter, modes));
9999 test_step.dependOn(tests.addTranslateCTests(b, test_filter));
100100 test_step.dependOn(tests.addGenHTests(b, test_filter));
101101 test_step.dependOn(docs_step);
std/atomic/index.zig+2-4
......@@ -1,11 +1,9 @@
11pub const Stack = @import("stack.zig").Stack;
2pub const QueueMpsc = @import("queue_mpsc.zig").QueueMpsc;
3pub const QueueMpmc = @import("queue_mpmc.zig").QueueMpmc;
2pub const Queue = @import("queue.zig").Queue;
43pub const Int = @import("int.zig").Int;
54
65test "std.atomic" {
76 _ = @import("stack.zig");
8 _ = @import("queue_mpsc.zig");
9 _ = @import("queue_mpmc.zig");
7 _ = @import("queue.zig");
108 _ = @import("int.zig");
119}
std/atomic/queue.zig created+226
......@@ -0,0 +1,226 @@
1const builtin = @import("builtin");
2const AtomicOrder = builtin.AtomicOrder;
3const AtomicRmwOp = builtin.AtomicRmwOp;
4
5/// Many producer, many consumer, non-allocating, thread-safe.
6/// Uses a spinlock to protect get() and put().
7pub fn Queue(comptime T: type) type {
8 return struct {
9 head: ?*Node,
10 tail: ?*Node,
11 lock: u8,
12
13 pub const Self = this;
14
15 pub const Node = struct {
16 next: ?*Node,
17 data: T,
18 };
19
20 pub fn init() Self {
21 return Self{
22 .head = null,
23 .tail = null,
24 .lock = 0,
25 };
26 }
27
28 pub fn put(self: *Self, node: *Node) void {
29 node.next = null;
30
31 while (@atomicRmw(u8, &self.lock, builtin.AtomicRmwOp.Xchg, 1, AtomicOrder.SeqCst) != 0) {}
32 defer assert(@atomicRmw(u8, &self.lock, builtin.AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst) == 1);
33
34 const opt_tail = self.tail;
35 self.tail = node;
36 if (opt_tail) |tail| {
37 tail.next = node;
38 } else {
39 assert(self.head == null);
40 self.head = node;
41 }
42 }
43
44 pub fn get(self: *Self) ?*Node {
45 while (@atomicRmw(u8, &self.lock, builtin.AtomicRmwOp.Xchg, 1, AtomicOrder.SeqCst) != 0) {}
46 defer assert(@atomicRmw(u8, &self.lock, builtin.AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst) == 1);
47
48 const head = self.head orelse return null;
49 self.head = head.next;
50 if (head.next == null) self.tail = null;
51 return head;
52 }
53
54 pub fn isEmpty(self: *Self) bool {
55 return @atomicLoad(?*Node, &self.head, builtin.AtomicOrder.SeqCst) != null;
56 }
57
58 pub fn dump(self: *Self) void {
59 while (@atomicRmw(u8, &self.lock, builtin.AtomicRmwOp.Xchg, 1, AtomicOrder.SeqCst) != 0) {}
60 defer assert(@atomicRmw(u8, &self.lock, builtin.AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst) == 1);
61
62 std.debug.warn("head: ");
63 dumpRecursive(self.head, 0);
64 std.debug.warn("tail: ");
65 dumpRecursive(self.tail, 0);
66 }
67
68 fn dumpRecursive(optional_node: ?*Node, indent: usize) void {
69 var stderr_file = std.io.getStdErr() catch return;
70 const stderr = &std.io.FileOutStream.init(&stderr_file).stream;
71 stderr.writeByteNTimes(' ', indent) catch return;
72 if (optional_node) |node| {
73 std.debug.warn("0x{x}={}\n", @ptrToInt(node), node.data);
74 dumpRecursive(node.next, indent + 1);
75 } else {
76 std.debug.warn("(null)\n");
77 }
78 }
79 };
80}
81
82const std = @import("../index.zig");
83const assert = std.debug.assert;
84
85const Context = struct {
86 allocator: *std.mem.Allocator,
87 queue: *Queue(i32),
88 put_sum: isize,
89 get_sum: isize,
90 get_count: usize,
91 puts_done: u8, // TODO make this a bool
92};
93
94// TODO add lazy evaluated build options and then put puts_per_thread behind
95// some option such as: "AggressiveMultithreadedFuzzTest". In the AppVeyor
96// CI we would use a less aggressive setting since at 1 core, while we still
97// want this test to pass, we need a smaller value since there is so much thrashing
98// we would also use a less aggressive setting when running in valgrind
99const puts_per_thread = 500;
100const put_thread_count = 3;
101
102test "std.atomic.Queue" {
103 var direct_allocator = std.heap.DirectAllocator.init();
104 defer direct_allocator.deinit();
105
106 var plenty_of_memory = try direct_allocator.allocator.alloc(u8, 300 * 1024);
107 defer direct_allocator.allocator.free(plenty_of_memory);
108
109 var fixed_buffer_allocator = std.heap.ThreadSafeFixedBufferAllocator.init(plenty_of_memory);
110 var a = &fixed_buffer_allocator.allocator;
111
112 var queue = Queue(i32).init();
113 var context = Context{
114 .allocator = a,
115 .queue = &queue,
116 .put_sum = 0,
117 .get_sum = 0,
118 .puts_done = 0,
119 .get_count = 0,
120 };
121
122 var putters: [put_thread_count]*std.os.Thread = undefined;
123 for (putters) |*t| {
124 t.* = try std.os.spawnThread(&context, startPuts);
125 }
126 var getters: [put_thread_count]*std.os.Thread = undefined;
127 for (getters) |*t| {
128 t.* = try std.os.spawnThread(&context, startGets);
129 }
130
131 for (putters) |t|
132 t.wait();
133 _ = @atomicRmw(u8, &context.puts_done, builtin.AtomicRmwOp.Xchg, 1, AtomicOrder.SeqCst);
134 for (getters) |t|
135 t.wait();
136
137 if (context.put_sum != context.get_sum) {
138 std.debug.panic("failure\nput_sum:{} != get_sum:{}", context.put_sum, context.get_sum);
139 }
140
141 if (context.get_count != puts_per_thread * put_thread_count) {
142 std.debug.panic(
143 "failure\nget_count:{} != puts_per_thread:{} * put_thread_count:{}",
144 context.get_count,
145 u32(puts_per_thread),
146 u32(put_thread_count),
147 );
148 }
149}
150
151fn startPuts(ctx: *Context) u8 {
152 var put_count: usize = puts_per_thread;
153 var r = std.rand.DefaultPrng.init(0xdeadbeef);
154 while (put_count != 0) : (put_count -= 1) {
155 std.os.time.sleep(0, 1); // let the os scheduler be our fuzz
156 const x = @bitCast(i32, r.random.scalar(u32));
157 const node = ctx.allocator.create(Queue(i32).Node{
158 .next = undefined,
159 .data = x,
160 }) catch unreachable;
161 ctx.queue.put(node);
162 _ = @atomicRmw(isize, &ctx.put_sum, builtin.AtomicRmwOp.Add, x, AtomicOrder.SeqCst);
163 }
164 return 0;
165}
166
167fn startGets(ctx: *Context) u8 {
168 while (true) {
169 const last = @atomicLoad(u8, &ctx.puts_done, builtin.AtomicOrder.SeqCst) == 1;
170
171 while (ctx.queue.get()) |node| {
172 std.os.time.sleep(0, 1); // let the os scheduler be our fuzz
173 _ = @atomicRmw(isize, &ctx.get_sum, builtin.AtomicRmwOp.Add, node.data, builtin.AtomicOrder.SeqCst);
174 _ = @atomicRmw(usize, &ctx.get_count, builtin.AtomicRmwOp.Add, 1, builtin.AtomicOrder.SeqCst);
175 }
176
177 if (last) return 0;
178 }
179}
180
181test "std.atomic.Queue single-threaded" {
182 var queue = Queue(i32).init();
183
184 var node_0 = Queue(i32).Node{
185 .data = 0,
186 .next = undefined,
187 };
188 queue.put(&node_0);
189
190 var node_1 = Queue(i32).Node{
191 .data = 1,
192 .next = undefined,
193 };
194 queue.put(&node_1);
195
196 assert(queue.get().?.data == 0);
197
198 var node_2 = Queue(i32).Node{
199 .data = 2,
200 .next = undefined,
201 };
202 queue.put(&node_2);
203
204 var node_3 = Queue(i32).Node{
205 .data = 3,
206 .next = undefined,
207 };
208 queue.put(&node_3);
209
210 assert(queue.get().?.data == 1);
211
212 assert(queue.get().?.data == 2);
213
214 var node_4 = Queue(i32).Node{
215 .data = 4,
216 .next = undefined,
217 };
218 queue.put(&node_4);
219
220 assert(queue.get().?.data == 3);
221 node_3.next = null;
222
223 assert(queue.get().?.data == 4);
224
225 assert(queue.get() == null);
226}
std/atomic/queue_mpmc.zig deleted-214
......@@ -1,214 +0,0 @@
1const builtin = @import("builtin");
2const AtomicOrder = builtin.AtomicOrder;
3const AtomicRmwOp = builtin.AtomicRmwOp;
4
5/// Many producer, many consumer, non-allocating, thread-safe, lock-free
6/// This implementation has a crippling limitation - it hangs onto node
7/// memory for 1 extra get() and 1 extra put() operation - when get() returns a node, that
8/// node must not be freed until both the next get() and the next put() completes.
9pub fn QueueMpmc(comptime T: type) type {
10 return struct {
11 head: *Node,
12 tail: *Node,
13 root: Node,
14
15 pub const Self = this;
16
17 pub const Node = struct {
18 next: ?*Node,
19 data: T,
20 };
21
22 /// TODO: well defined copy elision: https://github.com/ziglang/zig/issues/287
23 pub fn init(self: *Self) void {
24 self.root.next = null;
25 self.head = &self.root;
26 self.tail = &self.root;
27 }
28
29 pub fn put(self: *Self, node: *Node) void {
30 node.next = null;
31
32 const tail = @atomicRmw(*Node, &self.tail, AtomicRmwOp.Xchg, node, AtomicOrder.SeqCst);
33 _ = @atomicRmw(?*Node, &tail.next, AtomicRmwOp.Xchg, node, AtomicOrder.SeqCst);
34 }
35
36 /// node must not be freed until both the next get() and the next put() complete
37 pub fn get(self: *Self) ?*Node {
38 var head = @atomicLoad(*Node, &self.head, AtomicOrder.SeqCst);
39 while (true) {
40 const node = head.next orelse return null;
41 head = @cmpxchgWeak(*Node, &self.head, head, node, AtomicOrder.SeqCst, AtomicOrder.SeqCst) orelse return node;
42 }
43 }
44
45 ///// This is a debug function that is not thread-safe.
46 pub fn dump(self: *Self) void {
47 std.debug.warn("head: ");
48 dumpRecursive(self.head, 0);
49 std.debug.warn("tail: ");
50 dumpRecursive(self.tail, 0);
51 }
52
53 fn dumpRecursive(optional_node: ?*Node, indent: usize) void {
54 var stderr_file = std.io.getStdErr() catch return;
55 const stderr = &std.io.FileOutStream.init(&stderr_file).stream;
56 stderr.writeByteNTimes(' ', indent) catch return;
57 if (optional_node) |node| {
58 std.debug.warn("0x{x}={}\n", @ptrToInt(node), node.data);
59 dumpRecursive(node.next, indent + 1);
60 } else {
61 std.debug.warn("(null)\n");
62 }
63 }
64 };
65}
66
67const std = @import("std");
68const assert = std.debug.assert;
69
70const Context = struct {
71 allocator: *std.mem.Allocator,
72 queue: *QueueMpmc(i32),
73 put_sum: isize,
74 get_sum: isize,
75 get_count: usize,
76 puts_done: u8, // TODO make this a bool
77};
78
79// TODO add lazy evaluated build options and then put puts_per_thread behind
80// some option such as: "AggressiveMultithreadedFuzzTest". In the AppVeyor
81// CI we would use a less aggressive setting since at 1 core, while we still
82// want this test to pass, we need a smaller value since there is so much thrashing
83// we would also use a less aggressive setting when running in valgrind
84const puts_per_thread = 500;
85const put_thread_count = 3;
86
87test "std.atomic.queue_mpmc" {
88 var direct_allocator = std.heap.DirectAllocator.init();
89 defer direct_allocator.deinit();
90
91 var plenty_of_memory = try direct_allocator.allocator.alloc(u8, 300 * 1024);
92 defer direct_allocator.allocator.free(plenty_of_memory);
93
94 var fixed_buffer_allocator = std.heap.ThreadSafeFixedBufferAllocator.init(plenty_of_memory);
95 var a = &fixed_buffer_allocator.allocator;
96
97 var queue: QueueMpmc(i32) = undefined;
98 queue.init();
99 var context = Context{
100 .allocator = a,
101 .queue = &queue,
102 .put_sum = 0,
103 .get_sum = 0,
104 .puts_done = 0,
105 .get_count = 0,
106 };
107
108 var putters: [put_thread_count]*std.os.Thread = undefined;
109 for (putters) |*t| {
110 t.* = try std.os.spawnThread(&context, startPuts);
111 }
112 var getters: [put_thread_count]*std.os.Thread = undefined;
113 for (getters) |*t| {
114 t.* = try std.os.spawnThread(&context, startGets);
115 }
116
117 for (putters) |t|
118 t.wait();
119 _ = @atomicRmw(u8, &context.puts_done, builtin.AtomicRmwOp.Xchg, 1, AtomicOrder.SeqCst);
120 for (getters) |t|
121 t.wait();
122
123 if (context.put_sum != context.get_sum) {
124 std.debug.panic("failure\nput_sum:{} != get_sum:{}", context.put_sum, context.get_sum);
125 }
126
127 if (context.get_count != puts_per_thread * put_thread_count) {
128 std.debug.panic(
129 "failure\nget_count:{} != puts_per_thread:{} * put_thread_count:{}",
130 context.get_count,
131 u32(puts_per_thread),
132 u32(put_thread_count),
133 );
134 }
135}
136
137fn startPuts(ctx: *Context) u8 {
138 var put_count: usize = puts_per_thread;
139 var r = std.rand.DefaultPrng.init(0xdeadbeef);
140 while (put_count != 0) : (put_count -= 1) {
141 std.os.time.sleep(0, 1); // let the os scheduler be our fuzz
142 const x = @bitCast(i32, r.random.scalar(u32));
143 const node = ctx.allocator.create(QueueMpmc(i32).Node{
144 .next = undefined,
145 .data = x,
146 }) catch unreachable;
147 ctx.queue.put(node);
148 _ = @atomicRmw(isize, &ctx.put_sum, builtin.AtomicRmwOp.Add, x, AtomicOrder.SeqCst);
149 }
150 return 0;
151}
152
153fn startGets(ctx: *Context) u8 {
154 while (true) {
155 const last = @atomicLoad(u8, &ctx.puts_done, builtin.AtomicOrder.SeqCst) == 1;
156
157 while (ctx.queue.get()) |node| {
158 std.os.time.sleep(0, 1); // let the os scheduler be our fuzz
159 _ = @atomicRmw(isize, &ctx.get_sum, builtin.AtomicRmwOp.Add, node.data, builtin.AtomicOrder.SeqCst);
160 _ = @atomicRmw(usize, &ctx.get_count, builtin.AtomicRmwOp.Add, 1, builtin.AtomicOrder.SeqCst);
161 }
162
163 if (last) return 0;
164 }
165}
166
167test "std.atomic.queue_mpmc single-threaded" {
168 var queue: QueueMpmc(i32) = undefined;
169 queue.init();
170
171 var node_0 = QueueMpmc(i32).Node{
172 .data = 0,
173 .next = undefined,
174 };
175 queue.put(&node_0);
176
177 var node_1 = QueueMpmc(i32).Node{
178 .data = 1,
179 .next = undefined,
180 };
181 queue.put(&node_1);
182
183 assert(queue.get().?.data == 0);
184
185 var node_2 = QueueMpmc(i32).Node{
186 .data = 2,
187 .next = undefined,
188 };
189 queue.put(&node_2);
190
191 var node_3 = QueueMpmc(i32).Node{
192 .data = 3,
193 .next = undefined,
194 };
195 queue.put(&node_3);
196
197 assert(queue.get().?.data == 1);
198
199 assert(queue.get().?.data == 2);
200
201 var node_4 = QueueMpmc(i32).Node{
202 .data = 4,
203 .next = undefined,
204 };
205 queue.put(&node_4);
206
207 assert(queue.get().?.data == 3);
208 // if we were to set node_3.next to null here, it would cause this test
209 // to fail. this demonstrates the limitation of hanging on to extra memory.
210
211 assert(queue.get().?.data == 4);
212
213 assert(queue.get() == null);
214}
std/atomic/queue_mpsc.zig deleted-185
......@@ -1,185 +0,0 @@
1const std = @import("../index.zig");
2const assert = std.debug.assert;
3const builtin = @import("builtin");
4const AtomicOrder = builtin.AtomicOrder;
5const AtomicRmwOp = builtin.AtomicRmwOp;
6
7/// Many producer, single consumer, non-allocating, thread-safe, lock-free
8pub fn QueueMpsc(comptime T: type) type {
9 return struct {
10 inboxes: [2]std.atomic.Stack(T),
11 outbox: std.atomic.Stack(T),
12 inbox_index: usize,
13
14 pub const Self = this;
15
16 pub const Node = std.atomic.Stack(T).Node;
17
18 /// Not thread-safe. The call to init() must complete before any other functions are called.
19 /// No deinitialization required.
20 pub fn init() Self {
21 return Self{
22 .inboxes = []std.atomic.Stack(T){
23 std.atomic.Stack(T).init(),
24 std.atomic.Stack(T).init(),
25 },
26 .outbox = std.atomic.Stack(T).init(),
27 .inbox_index = 0,
28 };
29 }
30
31 /// Fully thread-safe. put() may be called from any thread at any time.
32 pub fn put(self: *Self, node: *Node) void {
33 const inbox_index = @atomicLoad(usize, &self.inbox_index, AtomicOrder.SeqCst);
34 const inbox = &self.inboxes[inbox_index];
35 inbox.push(node);
36 }
37
38 /// Must be called by only 1 consumer at a time. Every call to get() and isEmpty() must complete before
39 /// the next call to get().
40 pub fn get(self: *Self) ?*Node {
41 if (self.outbox.pop()) |node| {
42 return node;
43 }
44 const prev_inbox_index = @atomicRmw(usize, &self.inbox_index, AtomicRmwOp.Xor, 0x1, AtomicOrder.SeqCst);
45 const prev_inbox = &self.inboxes[prev_inbox_index];
46 while (prev_inbox.pop()) |node| {
47 self.outbox.push(node);
48 }
49 return self.outbox.pop();
50 }
51
52 /// Must be called by only 1 consumer at a time. Every call to get() and isEmpty() must complete before
53 /// the next call to isEmpty().
54 pub fn isEmpty(self: *Self) bool {
55 if (!self.outbox.isEmpty()) return false;
56 const prev_inbox_index = @atomicRmw(usize, &self.inbox_index, AtomicRmwOp.Xor, 0x1, AtomicOrder.SeqCst);
57 const prev_inbox = &self.inboxes[prev_inbox_index];
58 while (prev_inbox.pop()) |node| {
59 self.outbox.push(node);
60 }
61 return self.outbox.isEmpty();
62 }
63
64 /// For debugging only. No API guarantees about what this does.
65 pub fn dump(self: *Self) void {
66 {
67 var it = self.outbox.root;
68 while (it) |node| {
69 std.debug.warn("0x{x} -> ", @ptrToInt(node));
70 it = node.next;
71 }
72 }
73 const inbox_index = self.inbox_index;
74 const inboxes = []*std.atomic.Stack(T){
75 &self.inboxes[self.inbox_index],
76 &self.inboxes[1 - self.inbox_index],
77 };
78 for (inboxes) |inbox| {
79 var it = inbox.root;
80 while (it) |node| {
81 std.debug.warn("0x{x} -> ", @ptrToInt(node));
82 it = node.next;
83 }
84 }
85
86 std.debug.warn("null\n");
87 }
88 };
89}
90
91const Context = struct {
92 allocator: *std.mem.Allocator,
93 queue: *QueueMpsc(i32),
94 put_sum: isize,
95 get_sum: isize,
96 get_count: usize,
97 puts_done: u8, // TODO make this a bool
98};
99
100// TODO add lazy evaluated build options and then put puts_per_thread behind
101// some option such as: "AggressiveMultithreadedFuzzTest". In the AppVeyor
102// CI we would use a less aggressive setting since at 1 core, while we still
103// want this test to pass, we need a smaller value since there is so much thrashing
104// we would also use a less aggressive setting when running in valgrind
105const puts_per_thread = 500;
106const put_thread_count = 3;
107
108test "std.atomic.queue_mpsc" {
109 var direct_allocator = std.heap.DirectAllocator.init();
110 defer direct_allocator.deinit();
111
112 var plenty_of_memory = try direct_allocator.allocator.alloc(u8, 300 * 1024);
113 defer direct_allocator.allocator.free(plenty_of_memory);
114
115 var fixed_buffer_allocator = std.heap.ThreadSafeFixedBufferAllocator.init(plenty_of_memory);
116 var a = &fixed_buffer_allocator.allocator;
117
118 var queue = QueueMpsc(i32).init();
119 var context = Context{
120 .allocator = a,
121 .queue = &queue,
122 .put_sum = 0,
123 .get_sum = 0,
124 .puts_done = 0,
125 .get_count = 0,
126 };
127
128 var putters: [put_thread_count]*std.os.Thread = undefined;
129 for (putters) |*t| {
130 t.* = try std.os.spawnThread(&context, startPuts);
131 }
132 var getters: [1]*std.os.Thread = undefined;
133 for (getters) |*t| {
134 t.* = try std.os.spawnThread(&context, startGets);
135 }
136
137 for (putters) |t|
138 t.wait();
139 _ = @atomicRmw(u8, &context.puts_done, builtin.AtomicRmwOp.Xchg, 1, AtomicOrder.SeqCst);
140 for (getters) |t|
141 t.wait();
142
143 if (context.put_sum != context.get_sum) {
144 std.debug.panic("failure\nput_sum:{} != get_sum:{}", context.put_sum, context.get_sum);
145 }
146
147 if (context.get_count != puts_per_thread * put_thread_count) {
148 std.debug.panic(
149 "failure\nget_count:{} != puts_per_thread:{} * put_thread_count:{}",
150 context.get_count,
151 u32(puts_per_thread),
152 u32(put_thread_count),
153 );
154 }
155}
156
157fn startPuts(ctx: *Context) u8 {
158 var put_count: usize = puts_per_thread;
159 var r = std.rand.DefaultPrng.init(0xdeadbeef);
160 while (put_count != 0) : (put_count -= 1) {
161 std.os.time.sleep(0, 1); // let the os scheduler be our fuzz
162 const x = @bitCast(i32, r.random.scalar(u32));
163 const node = ctx.allocator.create(QueueMpsc(i32).Node{
164 .next = undefined,
165 .data = x,
166 }) catch unreachable;
167 ctx.queue.put(node);
168 _ = @atomicRmw(isize, &ctx.put_sum, builtin.AtomicRmwOp.Add, x, AtomicOrder.SeqCst);
169 }
170 return 0;
171}
172
173fn startGets(ctx: *Context) u8 {
174 while (true) {
175 const last = @atomicLoad(u8, &ctx.puts_done, builtin.AtomicOrder.SeqCst) == 1;
176
177 while (ctx.queue.get()) |node| {
178 std.os.time.sleep(0, 1); // let the os scheduler be our fuzz
179 _ = @atomicRmw(isize, &ctx.get_sum, builtin.AtomicRmwOp.Add, node.data, builtin.AtomicOrder.SeqCst);
180 _ = @atomicRmw(usize, &ctx.get_count, builtin.AtomicRmwOp.Add, 1, builtin.AtomicOrder.SeqCst);
181 }
182
183 if (last) return 0;
184 }
185}
std/atomic/stack.zig+20-12
......@@ -1,10 +1,13 @@
1const assert = std.debug.assert;
12const builtin = @import("builtin");
23const AtomicOrder = builtin.AtomicOrder;
34
4/// Many reader, many writer, non-allocating, thread-safe, lock-free
5/// Many reader, many writer, non-allocating, thread-safe
6/// Uses a spinlock to protect push() and pop()
57pub fn Stack(comptime T: type) type {
68 return struct {
79 root: ?*Node,
10 lock: u8,
811
912 pub const Self = this;
1013
......@@ -14,7 +17,10 @@ pub fn Stack(comptime T: type) type {
1417 };
1518
1619 pub fn init() Self {
17 return Self{ .root = null };
20 return Self{
21 .root = null,
22 .lock = 0,
23 };
1824 }
1925
2026 /// push operation, but only if you are the first item in the stack. if you did not succeed in
......@@ -25,18 +31,20 @@ pub fn Stack(comptime T: type) type {
2531 }
2632
2733 pub fn push(self: *Self, node: *Node) void {
28 var root = @atomicLoad(?*Node, &self.root, AtomicOrder.SeqCst);
29 while (true) {
30 node.next = root;
31 root = @cmpxchgWeak(?*Node, &self.root, root, node, AtomicOrder.SeqCst, AtomicOrder.SeqCst) orelse break;
32 }
34 while (@atomicRmw(u8, &self.lock, builtin.AtomicRmwOp.Xchg, 1, AtomicOrder.SeqCst) != 0) {}
35 defer assert(@atomicRmw(u8, &self.lock, builtin.AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst) == 1);
36
37 node.next = self.root;
38 self.root = node;
3339 }
3440
3541 pub fn pop(self: *Self) ?*Node {
36 var root = @atomicLoad(?*Node, &self.root, AtomicOrder.SeqCst);
37 while (true) {
38 root = @cmpxchgWeak(?*Node, &self.root, root, (root orelse return null).next, AtomicOrder.SeqCst, AtomicOrder.SeqCst) orelse return root;
39 }
42 while (@atomicRmw(u8, &self.lock, builtin.AtomicRmwOp.Xchg, 1, AtomicOrder.SeqCst) != 0) {}
43 defer assert(@atomicRmw(u8, &self.lock, builtin.AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst) == 1);
44
45 const root = self.root orelse return null;
46 self.root = root.next;
47 return root;
4048 }
4149
4250 pub fn isEmpty(self: *Self) bool {
......@@ -45,7 +53,7 @@ pub fn Stack(comptime T: type) type {
4553 };
4654}
4755
48const std = @import("std");
56const std = @import("../index.zig");
4957const Context = struct {
5058 allocator: *std.mem.Allocator,
5159 stack: *Stack(i32),
std/event/channel.zig+6-6
......@@ -12,8 +12,8 @@ pub fn Channel(comptime T: type) type {
1212 return struct {
1313 loop: *Loop,
1414
15 getters: std.atomic.QueueMpsc(GetNode),
16 putters: std.atomic.QueueMpsc(PutNode),
15 getters: std.atomic.Queue(GetNode),
16 putters: std.atomic.Queue(PutNode),
1717 get_count: usize,
1818 put_count: usize,
1919 dispatch_lock: u8, // TODO make this a bool
......@@ -46,8 +46,8 @@ pub fn Channel(comptime T: type) type {
4646 .buffer_index = 0,
4747 .dispatch_lock = 0,
4848 .need_dispatch = 0,
49 .getters = std.atomic.QueueMpsc(GetNode).init(),
50 .putters = std.atomic.QueueMpsc(PutNode).init(),
49 .getters = std.atomic.Queue(GetNode).init(),
50 .putters = std.atomic.Queue(PutNode).init(),
5151 .get_count = 0,
5252 .put_count = 0,
5353 });
......@@ -81,7 +81,7 @@ pub fn Channel(comptime T: type) type {
8181 .next = undefined,
8282 .data = handle,
8383 };
84 var queue_node = std.atomic.QueueMpsc(PutNode).Node{
84 var queue_node = std.atomic.Queue(PutNode).Node{
8585 .data = PutNode{
8686 .tick_node = &my_tick_node,
8787 .data = data,
......@@ -111,7 +111,7 @@ pub fn Channel(comptime T: type) type {
111111 .next = undefined,
112112 .data = handle,
113113 };
114 var queue_node = std.atomic.QueueMpsc(GetNode).Node{
114 var queue_node = std.atomic.Queue(GetNode).Node{
115115 .data = GetNode{
116116 .ptr = &result,
117117 .tick_node = &my_tick_node,
std/event/future.zig+11-10
......@@ -17,7 +17,7 @@ pub fn Future(comptime T: type) type {
1717 available: u8, // TODO make this a bool
1818
1919 const Self = this;
20 const Queue = std.atomic.QueueMpsc(promise);
20 const Queue = std.atomic.Queue(promise);
2121
2222 pub fn init(loop: *Loop) Self {
2323 return Self{
......@@ -30,19 +30,19 @@ pub fn Future(comptime T: type) type {
3030 /// Obtain the value. If it's not available, wait until it becomes
3131 /// available.
3232 /// Thread-safe.
33 pub async fn get(self: *Self) T {
33 pub async fn get(self: *Self) *T {
3434 if (@atomicLoad(u8, &self.available, AtomicOrder.SeqCst) == 1) {
35 return self.data;
35 return &self.data;
3636 }
3737 const held = await (async self.lock.acquire() catch unreachable);
38 defer held.release();
38 held.release();
3939
40 return self.data;
40 return &self.data;
4141 }
4242
4343 /// Make the data become available. May be called only once.
44 pub fn put(self: *Self, value: T) void {
45 self.data = value;
44 /// Before calling this, modify the `data` property.
45 pub fn resolve(self: *Self) void {
4646 const prev = @atomicRmw(u8, &self.available, AtomicRmwOp.Xchg, 1, AtomicOrder.SeqCst);
4747 assert(prev == 0); // put() called twice
4848 Lock.Held.release(Lock.Held{ .lock = &self.lock });
......@@ -57,7 +57,7 @@ test "std.event.Future" {
5757 const allocator = &da.allocator;
5858
5959 var loop: Loop = undefined;
60 try loop.initMultiThreaded(allocator);
60 try loop.initSingleThreaded(allocator);
6161 defer loop.deinit();
6262
6363 const handle = try async<allocator> testFuture(&loop);
......@@ -79,9 +79,10 @@ async fn testFuture(loop: *Loop) void {
7979}
8080
8181async fn waitOnFuture(future: *Future(i32)) i32 {
82 return await (async future.get() catch @panic("memory"));
82 return (await (async future.get() catch @panic("memory"))).*;
8383}
8484
8585async fn resolveFuture(future: *Future(i32)) void {
86 future.put(6);
86 future.data = 6;
87 future.resolve();
8788}
std/event/lock.zig+1-1
......@@ -15,7 +15,7 @@ pub const Lock = struct {
1515 queue: Queue,
1616 queue_empty_bit: u8, // TODO make this a bool
1717
18 const Queue = std.atomic.QueueMpsc(promise);
18 const Queue = std.atomic.Queue(promise);
1919
2020 pub const Held = struct {
2121 lock: *Lock,
std/event/loop.zig+3-3
......@@ -9,7 +9,7 @@ const AtomicOrder = builtin.AtomicOrder;
99
1010pub const Loop = struct {
1111 allocator: *mem.Allocator,
12 next_tick_queue: std.atomic.QueueMpsc(promise),
12 next_tick_queue: std.atomic.Queue(promise),
1313 os_data: OsData,
1414 final_resume_node: ResumeNode,
1515 dispatch_lock: u8, // TODO make this a bool
......@@ -21,7 +21,7 @@ pub const Loop = struct {
2121 available_eventfd_resume_nodes: std.atomic.Stack(ResumeNode.EventFd),
2222 eventfd_resume_nodes: []std.atomic.Stack(ResumeNode.EventFd).Node,
2323
24 pub const NextTickNode = std.atomic.QueueMpsc(promise).Node;
24 pub const NextTickNode = std.atomic.Queue(promise).Node;
2525
2626 pub const ResumeNode = struct {
2727 id: Id,
......@@ -77,7 +77,7 @@ pub const Loop = struct {
7777 .pending_event_count = 0,
7878 .allocator = allocator,
7979 .os_data = undefined,
80 .next_tick_queue = std.atomic.QueueMpsc(promise).init(),
80 .next_tick_queue = std.atomic.Queue(promise).init(),
8181 .dispatch_lock = 1, // start locked so threads go directly into epoll wait
8282 .extra_threads = undefined,
8383 .available_eventfd_resume_nodes = std.atomic.Stack(ResumeNode.EventFd).init(),
test/tests.zig+12-14
......@@ -47,12 +47,13 @@ const test_targets = []TestTarget{
4747
4848const max_stdout_size = 1 * 1024 * 1024; // 1 MB
4949
50pub fn addCompareOutputTests(b: *build.Builder, test_filter: ?[]const u8) *build.Step {
50pub fn addCompareOutputTests(b: *build.Builder, test_filter: ?[]const u8, modes: []const Mode) *build.Step {
5151 const cases = b.allocator.create(CompareOutputContext{
5252 .b = b,
5353 .step = b.step("test-compare-output", "Run the compare output tests"),
5454 .test_index = 0,
5555 .test_filter = test_filter,
56 .modes = modes,
5657 }) catch unreachable;
5758
5859 compare_output.addCases(cases);
......@@ -60,12 +61,13 @@ pub fn addCompareOutputTests(b: *build.Builder, test_filter: ?[]const u8) *build
6061 return cases.step;
6162}
6263
63pub fn addRuntimeSafetyTests(b: *build.Builder, test_filter: ?[]const u8) *build.Step {
64pub fn addRuntimeSafetyTests(b: *build.Builder, test_filter: ?[]const u8, modes: []const Mode) *build.Step {
6465 const cases = b.allocator.create(CompareOutputContext{
6566 .b = b,
6667 .step = b.step("test-runtime-safety", "Run the runtime safety tests"),
6768 .test_index = 0,
6869 .test_filter = test_filter,
70 .modes = modes,
6971 }) catch unreachable;
7072
7173 runtime_safety.addCases(cases);
......@@ -73,12 +75,13 @@ pub fn addRuntimeSafetyTests(b: *build.Builder, test_filter: ?[]const u8) *build
7375 return cases.step;
7476}
7577
76pub fn addCompileErrorTests(b: *build.Builder, test_filter: ?[]const u8) *build.Step {
78pub fn addCompileErrorTests(b: *build.Builder, test_filter: ?[]const u8, modes: []const Mode) *build.Step {
7779 const cases = b.allocator.create(CompileErrorContext{
7880 .b = b,
7981 .step = b.step("test-compile-errors", "Run the compile error tests"),
8082 .test_index = 0,
8183 .test_filter = test_filter,
84 .modes = modes,
8285 }) catch unreachable;
8386
8487 compile_errors.addCases(cases);
......@@ -99,12 +102,13 @@ pub fn addBuildExampleTests(b: *build.Builder, test_filter: ?[]const u8) *build.
99102 return cases.step;
100103}
101104
102pub fn addAssembleAndLinkTests(b: *build.Builder, test_filter: ?[]const u8) *build.Step {
105pub fn addAssembleAndLinkTests(b: *build.Builder, test_filter: ?[]const u8, modes: []const Mode) *build.Step {
103106 const cases = b.allocator.create(CompareOutputContext{
104107 .b = b,
105108 .step = b.step("test-asm-link", "Run the assemble and link tests"),
106109 .test_index = 0,
107110 .test_filter = test_filter,
111 .modes = modes,
108112 }) catch unreachable;
109113
110114 assemble_and_link.addCases(cases);
......@@ -173,6 +177,7 @@ pub const CompareOutputContext = struct {
173177 step: *build.Step,
174178 test_index: usize,
175179 test_filter: ?[]const u8,
180 modes: []const Mode,
176181
177182 const Special = enum {
178183 None,
......@@ -423,12 +428,7 @@ pub const CompareOutputContext = struct {
423428 self.step.dependOn(&run_and_cmp_output.step);
424429 },
425430 Special.None => {
426 for ([]Mode{
427 Mode.Debug,
428 Mode.ReleaseSafe,
429 Mode.ReleaseFast,
430 Mode.ReleaseSmall,
431 }) |mode| {
431 for (self.modes) |mode| {
432432 const annotated_case_name = fmt.allocPrint(self.b.allocator, "{} {} ({})", "compare-output", case.name, @tagName(mode)) catch unreachable;
433433 if (self.test_filter) |filter| {
434434 if (mem.indexOf(u8, annotated_case_name, filter) == null) continue;
......@@ -483,6 +483,7 @@ pub const CompileErrorContext = struct {
483483 step: *build.Step,
484484 test_index: usize,
485485 test_filter: ?[]const u8,
486 modes: []const Mode,
486487
487488 const TestCase = struct {
488489 name: []const u8,
......@@ -673,10 +674,7 @@ pub const CompileErrorContext = struct {
673674 pub fn addCase(self: *CompileErrorContext, case: *const TestCase) void {
674675 const b = self.b;
675676
676 for ([]Mode{
677 Mode.Debug,
678 Mode.ReleaseFast,
679 }) |mode| {
677 for (self.modes) |mode| {
680678 const annotated_case_name = fmt.allocPrint(self.b.allocator, "compile-error {} ({})", case.name, @tagName(mode)) catch unreachable;
681679 if (self.test_filter) |filter| {
682680 if (mem.indexOf(u8, annotated_case_name, filter) == null) continue;