1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
|
const std = @import("../std.zig");
const builtin = @import("builtin");
const assert = std.debug.assert;
const expect = std.testing.expect;
/// Many reader, many writer, non-allocating, thread-safe
/// Uses a spinlock to protect push() and pop()
/// When building in single threaded mode, this is a simple linked list.
pub fn Stack(comptime T: type) type {
return struct {
root: ?*Node,
lock: @TypeOf(lock_init),
const lock_init = if (builtin.single_threaded) {} else false;
pub const Self = @This();
pub const Node = struct {
next: ?*Node,
data: T,
};
pub fn init() Self {
return Self{
.root = null,
.lock = lock_init,
};
}
/// push operation, but only if you are the first item in the stack. if you did not succeed in
/// being the first item in the stack, returns the other item that was there.
pub fn pushFirst(self: *Self, node: *Node) ?*Node {
node.next = null;
return @cmpxchgStrong(?*Node, &self.root, null, node, .SeqCst, .SeqCst);
}
pub fn push(self: *Self, node: *Node) void {
if (builtin.single_threaded) {
node.next = self.root;
self.root = node;
} else {
while (@atomicRmw(bool, &self.lock, .Xchg, true, .SeqCst)) {}
defer assert(@atomicRmw(bool, &self.lock, .Xchg, false, .SeqCst));
node.next = self.root;
self.root = node;
}
}
pub fn pop(self: *Self) ?*Node {
if (builtin.single_threaded) {
const root = self.root orelse return null;
self.root = root.next;
return root;
} else {
while (@atomicRmw(bool, &self.lock, .Xchg, true, .SeqCst)) {}
defer assert(@atomicRmw(bool, &self.lock, .Xchg, false, .SeqCst));
const root = self.root orelse return null;
self.root = root.next;
return root;
}
}
pub fn isEmpty(self: *Self) bool {
return @atomicLoad(?*Node, &self.root, .SeqCst) == null;
}
};
}
const Context = struct {
allocator: *std.mem.Allocator,
stack: *Stack(i32),
put_sum: isize,
get_sum: isize,
get_count: usize,
puts_done: bool,
};
// TODO add lazy evaluated build options and then put puts_per_thread behind
// some option such as: "AggressiveMultithreadedFuzzTest". In the AppVeyor
// CI we would use a less aggressive setting since at 1 core, while we still
// want this test to pass, we need a smaller value since there is so much thrashing
// we would also use a less aggressive setting when running in valgrind
const puts_per_thread = 500;
const put_thread_count = 3;
test "std.atomic.stack" {
var plenty_of_memory = try std.heap.page_allocator.alloc(u8, 300 * 1024);
defer std.heap.page_allocator.free(plenty_of_memory);
var fixed_buffer_allocator = std.heap.ThreadSafeFixedBufferAllocator.init(plenty_of_memory);
var a = &fixed_buffer_allocator.allocator;
var stack = Stack(i32).init();
var context = Context{
.allocator = a,
.stack = &stack,
.put_sum = 0,
.get_sum = 0,
.puts_done = false,
.get_count = 0,
};
if (builtin.single_threaded) {
{
var i: usize = 0;
while (i < put_thread_count) : (i += 1) {
try expect(startPuts(&context) == 0);
}
}
context.puts_done = true;
{
var i: usize = 0;
while (i < put_thread_count) : (i += 1) {
try expect(startGets(&context) == 0);
}
}
} else {
var putters: [put_thread_count]std.Thread = undefined;
for (putters) |*t| {
t.* = try std.Thread.spawn(.{}, startPuts, .{&context});
}
var getters: [put_thread_count]std.Thread = undefined;
for (getters) |*t| {
t.* = try std.Thread.spawn(.{}, startGets, .{&context});
}
for (putters) |t|
t.join();
@atomicStore(bool, &context.puts_done, true, .SeqCst);
for (getters) |t|
t.join();
}
if (context.put_sum != context.get_sum) {
std.debug.panic("failure\nput_sum:{} != get_sum:{}", .{ context.put_sum, context.get_sum });
}
if (context.get_count != puts_per_thread * put_thread_count) {
std.debug.panic("failure\nget_count:{} != puts_per_thread:{} * put_thread_count:{}", .{
context.get_count,
@as(u32, puts_per_thread),
@as(u32, put_thread_count),
});
}
}
fn startPuts(ctx: *Context) u8 {
var put_count: usize = puts_per_thread;
var r = std.rand.DefaultPrng.init(0xdeadbeef);
while (put_count != 0) : (put_count -= 1) {
std.time.sleep(1); // let the os scheduler be our fuzz
const x = @bitCast(i32, r.random.int(u32));
const node = ctx.allocator.create(Stack(i32).Node) catch unreachable;
node.* = Stack(i32).Node{
.next = undefined,
.data = x,
};
ctx.stack.push(node);
_ = @atomicRmw(isize, &ctx.put_sum, .Add, x, .SeqCst);
}
return 0;
}
fn startGets(ctx: *Context) u8 {
while (true) {
const last = @atomicLoad(bool, &ctx.puts_done, .SeqCst);
while (ctx.stack.pop()) |node| {
std.time.sleep(1); // let the os scheduler be our fuzz
_ = @atomicRmw(isize, &ctx.get_sum, .Add, node.data, .SeqCst);
_ = @atomicRmw(usize, &ctx.get_count, .Add, 1, .SeqCst);
}
if (last) return 0;
}
}
|
