authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2018-06-29 14:45:01-04:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2018-06-29 14:45:42-04:00
log0874a5ba77a1d049a0e9e7f9f249605c109a731c
tree7bd78eb0b0f313877c32de5b828f0206580de060
parent4a35d7eeebec3f345e2482bc189f07c19dcf6f8b

std.atomic.queue - document limitation and add MPSC queue

5 files changed, 364 insertions(+), 143 deletions(-)

CMakeLists.txt+2-1
......@@ -431,7 +431,8 @@ set(ZIG_CPP_SOURCES
431431set(ZIG_STD_FILES
432432 "array_list.zig"
433433 "atomic/index.zig"
434 "atomic/queue.zig"
434 "atomic/queue_mpmc.zig"
435 "atomic/queue_mpsc.zig"
435436 "atomic/stack.zig"
436437 "base64.zig"
437438 "buf_map.zig"
std/atomic/index.zig+5-3
......@@ -1,7 +1,9 @@
11pub const Stack = @import("stack.zig").Stack;
2pub const Queue = @import("queue.zig").Queue;
2pub const QueueMpsc = @import("queue_mpsc.zig").QueueMpsc;
3pub const QueueMpmc = @import("queue_mpmc.zig").QueueMpmc;
34
45test "std.atomic" {
5 _ = @import("stack.zig").Stack;
6 _ = @import("queue.zig").Queue;
6 _ = @import("stack.zig");
7 _ = @import("queue_mpsc.zig");
8 _ = @import("queue_mpmc.zig");
79}
std/atomic/queue.zig deleted-139
......@@ -1,139 +0,0 @@
1const builtin = @import("builtin");
2const AtomicOrder = builtin.AtomicOrder;
3const AtomicRmwOp = builtin.AtomicRmwOp;
4
5/// Many reader, many writer, non-allocating, thread-safe, lock-free
6pub fn Queue(comptime T: type) type {
7 return struct {
8 head: *Node,
9 tail: *Node,
10 root: Node,
11
12 pub const Self = this;
13
14 pub const Node = struct {
15 next: ?*Node,
16 data: T,
17 };
18
19 // TODO: well defined copy elision: https://github.com/ziglang/zig/issues/287
20 pub fn init(self: *Self) void {
21 self.root.next = null;
22 self.head = &self.root;
23 self.tail = &self.root;
24 }
25
26 pub fn put(self: *Self, node: *Node) void {
27 node.next = null;
28
29 const tail = @atomicRmw(*Node, &self.tail, AtomicRmwOp.Xchg, node, AtomicOrder.SeqCst);
30 _ = @atomicRmw(?*Node, &tail.next, AtomicRmwOp.Xchg, node, AtomicOrder.SeqCst);
31 }
32
33 pub fn get(self: *Self) ?*Node {
34 var head = @atomicLoad(*Node, &self.head, AtomicOrder.SeqCst);
35 while (true) {
36 const node = head.next orelse return null;
37 head = @cmpxchgWeak(*Node, &self.head, head, node, AtomicOrder.SeqCst, AtomicOrder.SeqCst) orelse return node;
38 }
39 }
40 };
41}
42
43const std = @import("std");
44const Context = struct {
45 allocator: *std.mem.Allocator,
46 queue: *Queue(i32),
47 put_sum: isize,
48 get_sum: isize,
49 get_count: usize,
50 puts_done: u8, // TODO make this a bool
51};
52
53// TODO add lazy evaluated build options and then put puts_per_thread behind
54// some option such as: "AggressiveMultithreadedFuzzTest". In the AppVeyor
55// CI we would use a less aggressive setting since at 1 core, while we still
56// want this test to pass, we need a smaller value since there is so much thrashing
57// we would also use a less aggressive setting when running in valgrind
58const puts_per_thread = 500;
59const put_thread_count = 3;
60
61test "std.atomic.queue" {
62 var direct_allocator = std.heap.DirectAllocator.init();
63 defer direct_allocator.deinit();
64
65 var plenty_of_memory = try direct_allocator.allocator.alloc(u8, 300 * 1024);
66 defer direct_allocator.allocator.free(plenty_of_memory);
67
68 var fixed_buffer_allocator = std.heap.ThreadSafeFixedBufferAllocator.init(plenty_of_memory);
69 var a = &fixed_buffer_allocator.allocator;
70
71 var queue: Queue(i32) = undefined;
72 queue.init();
73 var context = Context{
74 .allocator = a,
75 .queue = &queue,
76 .put_sum = 0,
77 .get_sum = 0,
78 .puts_done = 0,
79 .get_count = 0,
80 };
81
82 var putters: [put_thread_count]*std.os.Thread = undefined;
83 for (putters) |*t| {
84 t.* = try std.os.spawnThread(&context, startPuts);
85 }
86 var getters: [put_thread_count]*std.os.Thread = undefined;
87 for (getters) |*t| {
88 t.* = try std.os.spawnThread(&context, startGets);
89 }
90
91 for (putters) |t|
92 t.wait();
93 _ = @atomicRmw(u8, &context.puts_done, builtin.AtomicRmwOp.Xchg, 1, AtomicOrder.SeqCst);
94 for (getters) |t|
95 t.wait();
96
97 if (context.put_sum != context.get_sum) {
98 std.debug.panic("failure\nput_sum:{} != get_sum:{}", context.put_sum, context.get_sum);
99 }
100
101 if (context.get_count != puts_per_thread * put_thread_count) {
102 std.debug.panic(
103 "failure\nget_count:{} != puts_per_thread:{} * put_thread_count:{}",
104 context.get_count,
105 u32(puts_per_thread),
106 u32(put_thread_count),
107 );
108 }
109}
110
111fn startPuts(ctx: *Context) u8 {
112 var put_count: usize = puts_per_thread;
113 var r = std.rand.DefaultPrng.init(0xdeadbeef);
114 while (put_count != 0) : (put_count -= 1) {
115 std.os.time.sleep(0, 1); // let the os scheduler be our fuzz
116 const x = @bitCast(i32, r.random.scalar(u32));
117 const node = ctx.allocator.create(Queue(i32).Node{
118 .next = undefined,
119 .data = x,
120 }) catch unreachable;
121 ctx.queue.put(node);
122 _ = @atomicRmw(isize, &ctx.put_sum, builtin.AtomicRmwOp.Add, x, AtomicOrder.SeqCst);
123 }
124 return 0;
125}
126
127fn startGets(ctx: *Context) u8 {
128 while (true) {
129 const last = @atomicLoad(u8, &ctx.puts_done, builtin.AtomicOrder.SeqCst) == 1;
130
131 while (ctx.queue.get()) |node| {
132 std.os.time.sleep(0, 1); // let the os scheduler be our fuzz
133 _ = @atomicRmw(isize, &ctx.get_sum, builtin.AtomicRmwOp.Add, node.data, builtin.AtomicOrder.SeqCst);
134 _ = @atomicRmw(usize, &ctx.get_count, builtin.AtomicRmwOp.Add, 1, builtin.AtomicOrder.SeqCst);
135 }
136
137 if (last) return 0;
138 }
139}
std/atomic/queue_mpmc.zig created+214
......@@ -0,0 +1,214 @@
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 created+143
......@@ -0,0 +1,143 @@
1const std = @import("std");
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 pub fn init() Self {
19 return Self{
20 .inboxes = []std.atomic.Stack(T){
21 std.atomic.Stack(T).init(),
22 std.atomic.Stack(T).init(),
23 },
24 .outbox = std.atomic.Stack(T).init(),
25 .inbox_index = 0,
26 };
27 }
28
29 pub fn put(self: *Self, node: *Node) void {
30 const inbox_index = @atomicLoad(usize, &self.inbox_index, AtomicOrder.SeqCst);
31 const inbox = &self.inboxes[inbox_index];
32 inbox.push(node);
33 }
34
35 pub fn get(self: *Self) ?*Node {
36 if (self.outbox.pop()) |node| {
37 return node;
38 }
39 const prev_inbox_index = @atomicRmw(usize, &self.inbox_index, AtomicRmwOp.Xor, 0x1, AtomicOrder.SeqCst);
40 const prev_inbox = &self.inboxes[prev_inbox_index];
41 while (prev_inbox.pop()) |node| {
42 self.outbox.push(node);
43 }
44 return self.outbox.pop();
45 }
46 };
47}
48
49const Context = struct {
50 allocator: *std.mem.Allocator,
51 queue: *QueueMpsc(i32),
52 put_sum: isize,
53 get_sum: isize,
54 get_count: usize,
55 puts_done: u8, // TODO make this a bool
56};
57
58// TODO add lazy evaluated build options and then put puts_per_thread behind
59// some option such as: "AggressiveMultithreadedFuzzTest". In the AppVeyor
60// CI we would use a less aggressive setting since at 1 core, while we still
61// want this test to pass, we need a smaller value since there is so much thrashing
62// we would also use a less aggressive setting when running in valgrind
63const puts_per_thread = 500;
64const put_thread_count = 3;
65
66test "std.atomic.queue_mpsc" {
67 var direct_allocator = std.heap.DirectAllocator.init();
68 defer direct_allocator.deinit();
69
70 var plenty_of_memory = try direct_allocator.allocator.alloc(u8, 300 * 1024);
71 defer direct_allocator.allocator.free(plenty_of_memory);
72
73 var fixed_buffer_allocator = std.heap.ThreadSafeFixedBufferAllocator.init(plenty_of_memory);
74 var a = &fixed_buffer_allocator.allocator;
75
76 var queue = QueueMpsc(i32).init();
77 var context = Context{
78 .allocator = a,
79 .queue = &queue,
80 .put_sum = 0,
81 .get_sum = 0,
82 .puts_done = 0,
83 .get_count = 0,
84 };
85
86 var putters: [put_thread_count]*std.os.Thread = undefined;
87 for (putters) |*t| {
88 t.* = try std.os.spawnThread(&context, startPuts);
89 }
90 var getters: [1]*std.os.Thread = undefined;
91 for (getters) |*t| {
92 t.* = try std.os.spawnThread(&context, startGets);
93 }
94
95 for (putters) |t|
96 t.wait();
97 _ = @atomicRmw(u8, &context.puts_done, builtin.AtomicRmwOp.Xchg, 1, AtomicOrder.SeqCst);
98 for (getters) |t|
99 t.wait();
100
101 if (context.put_sum != context.get_sum) {
102 std.debug.panic("failure\nput_sum:{} != get_sum:{}", context.put_sum, context.get_sum);
103 }
104
105 if (context.get_count != puts_per_thread * put_thread_count) {
106 std.debug.panic(
107 "failure\nget_count:{} != puts_per_thread:{} * put_thread_count:{}",
108 context.get_count,
109 u32(puts_per_thread),
110 u32(put_thread_count),
111 );
112 }
113}
114
115fn startPuts(ctx: *Context) u8 {
116 var put_count: usize = puts_per_thread;
117 var r = std.rand.DefaultPrng.init(0xdeadbeef);
118 while (put_count != 0) : (put_count -= 1) {
119 std.os.time.sleep(0, 1); // let the os scheduler be our fuzz
120 const x = @bitCast(i32, r.random.scalar(u32));
121 const node = ctx.allocator.create(QueueMpsc(i32).Node{
122 .next = undefined,
123 .data = x,
124 }) catch unreachable;
125 ctx.queue.put(node);
126 _ = @atomicRmw(isize, &ctx.put_sum, builtin.AtomicRmwOp.Add, x, AtomicOrder.SeqCst);
127 }
128 return 0;
129}
130
131fn startGets(ctx: *Context) u8 {
132 while (true) {
133 const last = @atomicLoad(u8, &ctx.puts_done, builtin.AtomicOrder.SeqCst) == 1;
134
135 while (ctx.queue.get()) |node| {
136 std.os.time.sleep(0, 1); // let the os scheduler be our fuzz
137 _ = @atomicRmw(isize, &ctx.get_sum, builtin.AtomicRmwOp.Add, node.data, builtin.AtomicOrder.SeqCst);
138 _ = @atomicRmw(usize, &ctx.get_count, builtin.AtomicRmwOp.Add, 1, builtin.AtomicOrder.SeqCst);
139 }
140
141 if (last) return 0;
142 }
143}