authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2018-04-29 12:29:40-04:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2018-04-29 12:29:40-04:00
logf37e79e720215a3a41c603fe41d05d6910c79de2
tree551877ba83cb7f5abc731efd25a21ab7fe55b170
parent0bb054e5e7ccb164ea1649608d2f6e4195519cb2
parentc76b0a845fb4176479c8bbf915e57dbdfdb7a594
signaturebadge-question-mark Signed by PGP key 4AEE18F83AFDEB23

Merge pull request #963 from zig-lang/atomic-stack-and-queue

Atomic stack and queue

16 files changed, 527 insertions(+), 90 deletions(-)

CMakeLists.txt+3
......@@ -415,6 +415,9 @@ set(ZIG_CPP_SOURCES
415415
416416set(ZIG_STD_FILES
417417 "array_list.zig"
418 "atomic/index.zig"
419 "atomic/stack.zig"
420 "atomic/queue.zig"
418421 "base64.zig"
419422 "buf_map.zig"
420423 "buf_set.zig"
src/ir.cpp+5
......@@ -18184,6 +18184,11 @@ static TypeTableEntry *ir_analyze_instruction_atomic_rmw(IrAnalyze *ira, IrInstr
1818418184 } else {
1818518185 if (!ir_resolve_atomic_order(ira, instruction->ordering->other, &ordering))
1818618186 return ira->codegen->builtin_types.entry_invalid;
18187 if (ordering == AtomicOrderUnordered) {
18188 ir_add_error(ira, instruction->ordering,
18189 buf_sprintf("@atomicRmw atomic ordering must not be Unordered"));
18190 return ira->codegen->builtin_types.entry_invalid;
18191 }
1818718192 }
1818818193
1818918194 if (instr_is_comptime(casted_operand) && instr_is_comptime(casted_ptr) && casted_ptr->value.data.x_ptr.mut == ConstPtrMutComptimeVar)
std/atomic/index.zig created+7
......@@ -0,0 +1,7 @@
1pub const Stack = @import("stack.zig").Stack;
2pub const Queue = @import("queue.zig").Queue;
3
4test "std.atomic" {
5 _ = @import("stack.zig").Stack;
6 _ = @import("queue.zig").Queue;
7}
std/atomic/queue.zig created+120
......@@ -0,0 +1,120 @@
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/zig-lang/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.Acquire);
35 while (true) {
36 const node = head.next ?? return null;
37 head = @cmpxchgWeak(&Node, &self.head, head, node, AtomicOrder.Release, AtomicOrder.Acquire) ?? 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};
52const puts_per_thread = 10000;
53const put_thread_count = 3;
54
55test "std.atomic.queue" {
56 var direct_allocator = std.heap.DirectAllocator.init();
57 defer direct_allocator.deinit();
58
59 var plenty_of_memory = try direct_allocator.allocator.alloc(u8, 64 * 1024 * 1024);
60 defer direct_allocator.allocator.free(plenty_of_memory);
61
62 var fixed_buffer_allocator = std.heap.ThreadSafeFixedBufferAllocator.init(plenty_of_memory);
63 var a = &fixed_buffer_allocator.allocator;
64
65 var queue: Queue(i32) = undefined;
66 queue.init();
67 var context = Context {
68 .allocator = a,
69 .queue = &queue,
70 .put_sum = 0,
71 .get_sum = 0,
72 .puts_done = 0,
73 .get_count = 0,
74 };
75
76 var putters: [put_thread_count]&std.os.Thread = undefined;
77 for (putters) |*t| {
78 *t = try std.os.spawnThread(&context, startPuts);
79 }
80 var getters: [put_thread_count]&std.os.Thread = undefined;
81 for (getters) |*t| {
82 *t = try std.os.spawnThread(&context, startGets);
83 }
84
85 for (putters) |t| t.wait();
86 _ = @atomicRmw(u8, &context.puts_done, builtin.AtomicRmwOp.Xchg, 1, AtomicOrder.SeqCst);
87 for (getters) |t| t.wait();
88
89 std.debug.assert(context.put_sum == context.get_sum);
90 std.debug.assert(context.get_count == puts_per_thread * put_thread_count);
91}
92
93fn startPuts(ctx: &Context) u8 {
94 var put_count: usize = puts_per_thread;
95 var r = std.rand.DefaultPrng.init(0xdeadbeef);
96 while (put_count != 0) : (put_count -= 1) {
97 std.os.time.sleep(0, 1); // let the os scheduler be our fuzz
98 const x = @bitCast(i32, r.random.scalar(u32));
99 const node = ctx.allocator.create(Queue(i32).Node) catch unreachable;
100 node.data = x;
101 ctx.queue.put(node);
102 _ = @atomicRmw(isize, &ctx.put_sum, builtin.AtomicRmwOp.Add, x, AtomicOrder.SeqCst);
103 }
104 return 0;
105}
106
107fn startGets(ctx: &Context) u8 {
108 while (true) {
109 while (ctx.queue.get()) |node| {
110 std.os.time.sleep(0, 1); // let the os scheduler be our fuzz
111 _ = @atomicRmw(isize, &ctx.get_sum, builtin.AtomicRmwOp.Add, node.data, builtin.AtomicOrder.SeqCst);
112 _ = @atomicRmw(usize, &ctx.get_count, builtin.AtomicRmwOp.Add, 1, builtin.AtomicOrder.SeqCst);
113 }
114
115 if (@atomicLoad(u8, &ctx.puts_done, builtin.AtomicOrder.SeqCst) == 1) {
116 break;
117 }
118 }
119 return 0;
120}
std/atomic/stack.zig created+126
......@@ -0,0 +1,126 @@
1const builtin = @import("builtin");
2const AtomicOrder = builtin.AtomicOrder;
3
4/// Many reader, many writer, non-allocating, thread-safe, lock-free
5pub fn Stack(comptime T: type) type {
6 return struct {
7 root: ?&Node,
8
9 pub const Self = this;
10
11 pub const Node = struct {
12 next: ?&Node,
13 data: T,
14 };
15
16 pub fn init() Self {
17 return Self {
18 .root = null,
19 };
20 }
21
22 /// push operation, but only if you are the first item in the stack. if you did not succeed in
23 /// being the first item in the stack, returns the other item that was there.
24 pub fn pushFirst(self: &Self, node: &Node) ?&Node {
25 node.next = null;
26 return @cmpxchgStrong(?&Node, &self.root, null, node, AtomicOrder.SeqCst, AtomicOrder.SeqCst);
27 }
28
29 pub fn push(self: &Self, node: &Node) void {
30 var root = @atomicLoad(?&Node, &self.root, AtomicOrder.SeqCst);
31 while (true) {
32 node.next = root;
33 root = @cmpxchgWeak(?&Node, &self.root, root, node, AtomicOrder.SeqCst, AtomicOrder.SeqCst) ?? break;
34 }
35 }
36
37 pub fn pop(self: &Self) ?&Node {
38 var root = @atomicLoad(?&Node, &self.root, AtomicOrder.Acquire);
39 while (true) {
40 root = @cmpxchgWeak(?&Node, &self.root, root, (root ?? return null).next, AtomicOrder.SeqCst, AtomicOrder.SeqCst) ?? return root;
41 }
42 }
43
44 pub fn isEmpty(self: &Self) bool {
45 return @atomicLoad(?&Node, &self.root, AtomicOrder.SeqCst) == null;
46 }
47 };
48}
49
50const std = @import("std");
51const Context = struct {
52 allocator: &std.mem.Allocator,
53 stack: &Stack(i32),
54 put_sum: isize,
55 get_sum: isize,
56 get_count: usize,
57 puts_done: u8, // TODO make this a bool
58};
59const puts_per_thread = 1000;
60const put_thread_count = 3;
61
62test "std.atomic.stack" {
63 var direct_allocator = std.heap.DirectAllocator.init();
64 defer direct_allocator.deinit();
65
66 var plenty_of_memory = try direct_allocator.allocator.alloc(u8, 64 * 1024 * 1024);
67 defer direct_allocator.allocator.free(plenty_of_memory);
68
69 var fixed_buffer_allocator = std.heap.ThreadSafeFixedBufferAllocator.init(plenty_of_memory);
70 var a = &fixed_buffer_allocator.allocator;
71
72 var stack = Stack(i32).init();
73 var context = Context {
74 .allocator = a,
75 .stack = &stack,
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| t.wait();
92 _ = @atomicRmw(u8, &context.puts_done, builtin.AtomicRmwOp.Xchg, 1, AtomicOrder.SeqCst);
93 for (getters) |t| t.wait();
94
95 std.debug.assert(context.put_sum == context.get_sum);
96 std.debug.assert(context.get_count == puts_per_thread * put_thread_count);
97}
98
99fn startPuts(ctx: &Context) u8 {
100 var put_count: usize = puts_per_thread;
101 var r = std.rand.DefaultPrng.init(0xdeadbeef);
102 while (put_count != 0) : (put_count -= 1) {
103 std.os.time.sleep(0, 1); // let the os scheduler be our fuzz
104 const x = @bitCast(i32, r.random.scalar(u32));
105 const node = ctx.allocator.create(Stack(i32).Node) catch unreachable;
106 node.data = x;
107 ctx.stack.push(node);
108 _ = @atomicRmw(isize, &ctx.put_sum, builtin.AtomicRmwOp.Add, x, AtomicOrder.SeqCst);
109 }
110 return 0;
111}
112
113fn startGets(ctx: &Context) u8 {
114 while (true) {
115 while (ctx.stack.pop()) |node| {
116 std.os.time.sleep(0, 1); // let the os scheduler be our fuzz
117 _ = @atomicRmw(isize, &ctx.get_sum, builtin.AtomicRmwOp.Add, node.data, builtin.AtomicOrder.SeqCst);
118 _ = @atomicRmw(usize, &ctx.get_count, builtin.AtomicRmwOp.Add, 1, builtin.AtomicOrder.SeqCst);
119 }
120
121 if (@atomicLoad(u8, &ctx.puts_done, builtin.AtomicOrder.SeqCst) == 1) {
122 break;
123 }
124 }
125 return 0;
126}
std/c/darwin.zig+5
......@@ -81,3 +81,8 @@ pub const sockaddr = extern struct {
8181};
8282
8383pub const sa_family_t = u8;
84
85pub const pthread_attr_t = extern struct {
86 __sig: c_long,
87 __opaque: [56]u8,
88};
std/c/index.zig+10
......@@ -53,3 +53,13 @@ pub extern "c" fn malloc(usize) ?&c_void;
5353pub extern "c" fn realloc(&c_void, usize) ?&c_void;
5454pub extern "c" fn free(&c_void) void;
5555pub extern "c" fn posix_memalign(memptr: &&c_void, alignment: usize, size: usize) c_int;
56
57pub extern "pthread" fn pthread_create(noalias newthread: &pthread_t,
58 noalias attr: ?&const pthread_attr_t, start_routine: extern fn(?&c_void) ?&c_void,
59 noalias arg: ?&c_void) c_int;
60pub extern "pthread" fn pthread_attr_init(attr: &pthread_attr_t) c_int;
61pub extern "pthread" fn pthread_attr_setstack(attr: &pthread_attr_t, stackaddr: &c_void, stacksize: usize) c_int;
62pub extern "pthread" fn pthread_attr_destroy(attr: &pthread_attr_t) c_int;
63pub extern "pthread" fn pthread_join(thread: pthread_t, arg_return: ?&?&c_void) c_int;
64
65pub const pthread_t = &@OpaqueType();
std/c/linux.zig+5
......@@ -3,3 +3,8 @@ pub use @import("../os/linux/errno.zig");
33pub extern "c" fn getrandom(buf_ptr: &u8, buf_len: usize, flags: c_uint) c_int;
44extern "c" fn __errno_location() &c_int;
55pub const _errno = __errno_location;
6
7pub const pthread_attr_t = extern struct {
8 __size: [56]u8,
9 __align: c_long,
10};
std/heap.zig+59-11
......@@ -47,13 +47,6 @@ pub const DirectAllocator = struct {
4747
4848 const HeapHandle = if (builtin.os == Os.windows) os.windows.HANDLE else void;
4949
50 //pub const canary_bytes = []u8 {48, 239, 128, 46, 18, 49, 147, 9, 195, 59, 203, 3, 245, 54, 9, 122};
51 //pub const want_safety = switch (builtin.mode) {
52 // builtin.Mode.Debug => true,
53 // builtin.Mode.ReleaseSafe => true,
54 // else => false,
55 //};
56
5750 pub fn init() DirectAllocator {
5851 return DirectAllocator {
5952 .allocator = Allocator {
......@@ -98,7 +91,7 @@ pub const DirectAllocator = struct {
9891 const unused_start = addr;
9992 const unused_len = aligned_addr - 1 - unused_start;
10093
101 var err = p.munmap(@intToPtr(&u8, unused_start), unused_len);
94 var err = p.munmap(unused_start, unused_len);
10295 debug.assert(p.getErrno(err) == 0);
10396
10497 //It is impossible that there is an unoccupied page at the top of our
......@@ -139,7 +132,7 @@ pub const DirectAllocator = struct {
139132 const rem = @rem(new_addr_end, os.page_size);
140133 const new_addr_end_rounded = new_addr_end + if (rem == 0) 0 else (os.page_size - rem);
141134 if (old_addr_end > new_addr_end_rounded) {
142 _ = os.posix.munmap(@intToPtr(&u8, new_addr_end_rounded), old_addr_end - new_addr_end_rounded);
135 _ = os.posix.munmap(new_addr_end_rounded, old_addr_end - new_addr_end_rounded);
143136 }
144137 return old_mem[0..new_size];
145138 }
......@@ -177,7 +170,7 @@ pub const DirectAllocator = struct {
177170
178171 switch (builtin.os) {
179172 Os.linux, Os.macosx, Os.ios => {
180 _ = os.posix.munmap(bytes.ptr, bytes.len);
173 _ = os.posix.munmap(@ptrToInt(bytes.ptr), bytes.len);
181174 },
182175 Os.windows => {
183176 const record_addr = @ptrToInt(bytes.ptr) + bytes.len;
......@@ -298,7 +291,7 @@ pub const FixedBufferAllocator = struct {
298291
299292 fn alloc(allocator: &Allocator, n: usize, alignment: u29) ![]u8 {
300293 const self = @fieldParentPtr(FixedBufferAllocator, "allocator", allocator);
301 const addr = @ptrToInt(&self.buffer[self.end_index]);
294 const addr = @ptrToInt(self.buffer.ptr) + self.end_index;
302295 const rem = @rem(addr, alignment);
303296 const march_forward_bytes = if (rem == 0) 0 else (alignment - rem);
304297 const adjusted_index = self.end_index + march_forward_bytes;
......@@ -325,6 +318,54 @@ pub const FixedBufferAllocator = struct {
325318 fn free(allocator: &Allocator, bytes: []u8) void { }
326319};
327320
321/// lock free
322pub const ThreadSafeFixedBufferAllocator = struct {
323 allocator: Allocator,
324 end_index: usize,
325 buffer: []u8,
326
327 pub fn init(buffer: []u8) ThreadSafeFixedBufferAllocator {
328 return ThreadSafeFixedBufferAllocator {
329 .allocator = Allocator {
330 .allocFn = alloc,
331 .reallocFn = realloc,
332 .freeFn = free,
333 },
334 .buffer = buffer,
335 .end_index = 0,
336 };
337 }
338
339 fn alloc(allocator: &Allocator, n: usize, alignment: u29) ![]u8 {
340 const self = @fieldParentPtr(ThreadSafeFixedBufferAllocator, "allocator", allocator);
341 var end_index = @atomicLoad(usize, &self.end_index, builtin.AtomicOrder.SeqCst);
342 while (true) {
343 const addr = @ptrToInt(self.buffer.ptr) + end_index;
344 const rem = @rem(addr, alignment);
345 const march_forward_bytes = if (rem == 0) 0 else (alignment - rem);
346 const adjusted_index = end_index + march_forward_bytes;
347 const new_end_index = adjusted_index + n;
348 if (new_end_index > self.buffer.len) {
349 return error.OutOfMemory;
350 }
351 end_index = @cmpxchgWeak(usize, &self.end_index, end_index, new_end_index,
352 builtin.AtomicOrder.SeqCst, builtin.AtomicOrder.SeqCst) ?? return self.buffer[adjusted_index .. new_end_index];
353 }
354 }
355
356 fn realloc(allocator: &Allocator, old_mem: []u8, new_size: usize, alignment: u29) ![]u8 {
357 if (new_size <= old_mem.len) {
358 return old_mem[0..new_size];
359 } else {
360 const result = try alloc(allocator, new_size, alignment);
361 mem.copy(u8, result, old_mem);
362 return result;
363 }
364 }
365
366 fn free(allocator: &Allocator, bytes: []u8) void { }
367};
368
328369
329370
330371test "c_allocator" {
......@@ -363,6 +404,13 @@ test "FixedBufferAllocator" {
363404 try testAllocatorLargeAlignment(&fixed_buffer_allocator.allocator);
364405}
365406
407test "ThreadSafeFixedBufferAllocator" {
408 var fixed_buffer_allocator = ThreadSafeFixedBufferAllocator.init(test_fixed_buffer_allocator_memory[0..]);
409
410 try testAllocator(&fixed_buffer_allocator.allocator);
411 try testAllocatorLargeAlignment(&fixed_buffer_allocator.allocator);
412}
413
366414fn testAllocator(allocator: &mem.Allocator) !void {
367415 var slice = try allocator.alloc(&i32, 100);
368416
std/index.zig+2
......@@ -8,6 +8,7 @@ pub const HashMap = @import("hash_map.zig").HashMap;
88pub const LinkedList = @import("linked_list.zig").LinkedList;
99pub const IntrusiveLinkedList = @import("linked_list.zig").IntrusiveLinkedList;
1010
11pub const atomic = @import("atomic/index.zig");
1112pub const base64 = @import("base64.zig");
1213pub const build = @import("build.zig");
1314pub const c = @import("c/index.zig");
......@@ -34,6 +35,7 @@ pub const zig = @import("zig/index.zig");
3435
3536test "std" {
3637 // run tests from these
38 _ = @import("atomic/index.zig");
3739 _ = @import("array_list.zig");
3840 _ = @import("buf_map.zig");
3941 _ = @import("buf_set.zig");
std/mem.zig+1
......@@ -32,6 +32,7 @@ pub const Allocator = struct {
3232 freeFn: fn (self: &Allocator, old_mem: []u8) void,
3333
3434 fn create(self: &Allocator, comptime T: type) !&T {
35 if (@sizeOf(T) == 0) return &{};
3536 const slice = try self.alloc(T, 1);
3637 return &slice[0];
3738 }
std/os/darwin.zig+4-4
......@@ -184,7 +184,7 @@ pub fn write(fd: i32, buf: &const u8, nbyte: usize) usize {
184184 return errnoWrap(c.write(fd, @ptrCast(&const c_void, buf), nbyte));
185185}
186186
187pub fn mmap(address: ?&u8, length: usize, prot: usize, flags: usize, fd: i32,
187pub fn mmap(address: ?&u8, length: usize, prot: usize, flags: u32, fd: i32,
188188 offset: isize) usize
189189{
190190 const ptr_result = c.mmap(@ptrCast(&c_void, address), length,
......@@ -193,8 +193,8 @@ pub fn mmap(address: ?&u8, length: usize, prot: usize, flags: usize, fd: i32,
193193 return errnoWrap(isize_result);
194194}
195195
196pub fn munmap(address: &u8, length: usize) usize {
197 return errnoWrap(c.munmap(@ptrCast(&c_void, address), length));
196pub fn munmap(address: usize, length: usize) usize {
197 return errnoWrap(c.munmap(@intToPtr(&c_void, address), length));
198198}
199199
200200pub fn unlink(path: &const u8) usize {
......@@ -341,4 +341,4 @@ pub const timeval = c.timeval;
341341pub const mach_timebase_info_data = c.mach_timebase_info_data;
342342
343343pub const mach_absolute_time = c.mach_absolute_time;
344pub const mach_timebase_info = c.mach_timebase_info;
\ No newline at end of file
344pub const mach_timebase_info = c.mach_timebase_info;
std/os/index.zig+167-56
......@@ -2,6 +2,10 @@ const std = @import("../index.zig");
22const builtin = @import("builtin");
33const Os = builtin.Os;
44const is_windows = builtin.os == Os.windows;
5const is_posix = switch (builtin.os) {
6 builtin.Os.linux, builtin.Os.macosx => true,
7 else => false,
8};
59const os = this;
610
711test "std.os" {
......@@ -2343,24 +2347,58 @@ pub fn posixGetSockOptConnectError(sockfd: i32) PosixConnectError!void {
23432347}
23442348
23452349pub const Thread = struct {
2346 pid: i32,
2347 allocator: ?&mem.Allocator,
2348 stack: []u8,
2350 data: Data,
2351
2352 pub const use_pthreads = is_posix and builtin.link_libc;
2353 const Data = if (use_pthreads) struct {
2354 handle: c.pthread_t,
2355 stack_addr: usize,
2356 stack_len: usize,
2357 } else switch (builtin.os) {
2358 builtin.Os.linux => struct {
2359 pid: i32,
2360 stack_addr: usize,
2361 stack_len: usize,
2362 },
2363 builtin.Os.windows => struct {
2364 handle: windows.HANDLE,
2365 alloc_start: &c_void,
2366 heap_handle: windows.HANDLE,
2367 },
2368 else => @compileError("Unsupported OS"),
2369 };
23492370
23502371 pub fn wait(self: &const Thread) void {
2351 while (true) {
2352 const pid_value = @atomicLoad(i32, &self.pid, builtin.AtomicOrder.SeqCst);
2353 if (pid_value == 0) break;
2354 const rc = linux.futex_wait(@ptrToInt(&self.pid), linux.FUTEX_WAIT, pid_value, null);
2355 switch (linux.getErrno(rc)) {
2356 0 => continue,
2357 posix.EINTR => continue,
2358 posix.EAGAIN => continue,
2372 if (use_pthreads) {
2373 const err = c.pthread_join(self.data.handle, null);
2374 switch (err) {
2375 0 => {},
2376 posix.EINVAL => unreachable,
2377 posix.ESRCH => unreachable,
2378 posix.EDEADLK => unreachable,
23592379 else => unreachable,
23602380 }
2361 }
2362 if (self.allocator) |a| {
2363 a.free(self.stack);
2381 assert(posix.munmap(self.data.stack_addr, self.data.stack_len) == 0);
2382 } else switch (builtin.os) {
2383 builtin.Os.linux => {
2384 while (true) {
2385 const pid_value = @atomicLoad(i32, &self.data.pid, builtin.AtomicOrder.SeqCst);
2386 if (pid_value == 0) break;
2387 const rc = linux.futex_wait(@ptrToInt(&self.data.pid), linux.FUTEX_WAIT, pid_value, null);
2388 switch (linux.getErrno(rc)) {
2389 0 => continue,
2390 posix.EINTR => continue,
2391 posix.EAGAIN => continue,
2392 else => unreachable,
2393 }
2394 }
2395 assert(posix.munmap(self.data.stack_addr, self.data.stack_len) == 0);
2396 },
2397 builtin.Os.windows => {
2398 assert(windows.WaitForSingleObject(self.data.handle, windows.INFINITE) == windows.WAIT_OBJECT_0);
2399 assert(windows.HeapFree(self.data.heap_handle, 0, self.data.alloc_start) != 0);
2400 },
2401 else => @compileError("Unsupported OS"),
23642402 }
23652403 }
23662404};
......@@ -2385,38 +2423,94 @@ pub const SpawnThreadError = error {
23852423 /// be copied.
23862424 SystemResources,
23872425
2426 /// Not enough userland memory to spawn the thread.
2427 OutOfMemory,
2428
23882429 Unexpected,
23892430};
23902431
2391pub const SpawnThreadAllocatorError = SpawnThreadError || error{OutOfMemory};
2392
23932432/// caller must call wait on the returned thread
23942433/// fn startFn(@typeOf(context)) T
23952434/// where T is u8, noreturn, void, or !void
2396pub fn spawnThreadAllocator(allocator: &mem.Allocator, context: var, comptime startFn: var) SpawnThreadAllocatorError!&Thread {
2435/// caller must call wait on the returned thread
2436pub fn spawnThread(context: var, comptime startFn: var) SpawnThreadError!&Thread {
23972437 // TODO compile-time call graph analysis to determine stack upper bound
23982438 // https://github.com/zig-lang/zig/issues/157
23992439 const default_stack_size = 8 * 1024 * 1024;
2400 const stack_bytes = try allocator.alloc(u8, default_stack_size);
2401 const thread = try spawnThread(stack_bytes, context, startFn);
2402 thread.allocator = allocator;
2403 return thread;
2404}
24052440
2406/// stack must be big enough to store one Thread and one @typeOf(context), each with default alignment, at the end
2407/// fn startFn(@typeOf(context)) T
2408/// where T is u8, noreturn, void, or !void
2409/// caller must call wait on the returned thread
2410pub fn spawnThread(stack: []u8, context: var, comptime startFn: var) SpawnThreadError!&Thread {
24112441 const Context = @typeOf(context);
24122442 comptime assert(@ArgType(@typeOf(startFn), 0) == Context);
24132443
2414 var stack_end: usize = @ptrToInt(stack.ptr) + stack.len;
2444 if (builtin.os == builtin.Os.windows) {
2445 const WinThread = struct {
2446 const OuterContext = struct {
2447 thread: Thread,
2448 inner: Context,
2449 };
2450 extern fn threadMain(arg: windows.LPVOID) windows.DWORD {
2451 if (@sizeOf(Context) == 0) {
2452 return startFn({});
2453 } else {
2454 return startFn(*@ptrCast(&Context, @alignCast(@alignOf(Context), arg)));
2455 }
2456 }
2457 };
2458
2459 const heap_handle = windows.GetProcessHeap() ?? return SpawnThreadError.OutOfMemory;
2460 const byte_count = @alignOf(WinThread.OuterContext) + @sizeOf(WinThread.OuterContext);
2461 const bytes_ptr = windows.HeapAlloc(heap_handle, 0, byte_count) ?? return SpawnThreadError.OutOfMemory;
2462 errdefer assert(windows.HeapFree(heap_handle, 0, bytes_ptr) != 0);
2463 const bytes = @ptrCast(&u8, bytes_ptr)[0..byte_count];
2464 const outer_context = std.heap.FixedBufferAllocator.init(bytes).allocator.create(WinThread.OuterContext) catch unreachable;
2465 outer_context.inner = context;
2466 outer_context.thread.data.heap_handle = heap_handle;
2467 outer_context.thread.data.alloc_start = bytes_ptr;
2468
2469 const parameter = if (@sizeOf(Context) == 0) null else @ptrCast(&c_void, &outer_context.inner);
2470 outer_context.thread.data.handle = windows.CreateThread(null, default_stack_size, WinThread.threadMain,
2471 parameter, 0, null) ??
2472 {
2473 const err = windows.GetLastError();
2474 return switch (err) {
2475 else => os.unexpectedErrorWindows(err),
2476 };
2477 };
2478 return &outer_context.thread;
2479 }
2480
2481 const MainFuncs = struct {
2482 extern fn linuxThreadMain(ctx_addr: usize) u8 {
2483 if (@sizeOf(Context) == 0) {
2484 return startFn({});
2485 } else {
2486 return startFn(*@intToPtr(&const Context, ctx_addr));
2487 }
2488 }
2489 extern fn posixThreadMain(ctx: ?&c_void) ?&c_void {
2490 if (@sizeOf(Context) == 0) {
2491 _ = startFn({});
2492 return null;
2493 } else {
2494 _ = startFn(*@ptrCast(&const Context, @alignCast(@alignOf(Context), ctx)));
2495 return null;
2496 }
2497 }
2498 };
2499
2500 const MAP_GROWSDOWN = if (builtin.os == builtin.Os.linux) linux.MAP_GROWSDOWN else 0;
2501
2502 const mmap_len = default_stack_size;
2503 const stack_addr = posix.mmap(null, mmap_len, posix.PROT_READ|posix.PROT_WRITE,
2504 posix.MAP_PRIVATE|posix.MAP_ANONYMOUS|MAP_GROWSDOWN, -1, 0);
2505 if (stack_addr == posix.MAP_FAILED) return error.OutOfMemory;
2506 errdefer assert(posix.munmap(stack_addr, mmap_len) == 0);
2507
2508 var stack_end: usize = stack_addr + mmap_len;
24152509 var arg: usize = undefined;
24162510 if (@sizeOf(Context) != 0) {
24172511 stack_end -= @sizeOf(Context);
24182512 stack_end -= stack_end % @alignOf(Context);
2419 assert(stack_end >= @ptrToInt(stack.ptr));
2513 assert(stack_end >= stack_addr);
24202514 const context_ptr = @alignCast(@alignOf(Context), @intToPtr(&Context, stack_end));
24212515 *context_ptr = context;
24222516 arg = stack_end;
......@@ -2424,36 +2518,53 @@ pub fn spawnThread(stack: []u8, context: var, comptime startFn: var) SpawnThread
24242518
24252519 stack_end -= @sizeOf(Thread);
24262520 stack_end -= stack_end % @alignOf(Thread);
2427 assert(stack_end >= @ptrToInt(stack.ptr));
2521 assert(stack_end >= stack_addr);
24282522 const thread_ptr = @alignCast(@alignOf(Thread), @intToPtr(&Thread, stack_end));
2429 thread_ptr.stack = stack;
2430 thread_ptr.allocator = null;
24312523
2432 const threadMain = struct {
2433 extern fn threadMain(ctx_addr: usize) u8 {
2434 if (@sizeOf(Context) == 0) {
2435 return startFn({});
2436 } else {
2437 return startFn(*@intToPtr(&const Context, ctx_addr));
2438 }
2439 }
2440 }.threadMain;
2524 thread_ptr.data.stack_addr = stack_addr;
2525 thread_ptr.data.stack_len = mmap_len;
24412526
2442 const flags = posix.CLONE_VM | posix.CLONE_FS | posix.CLONE_FILES | posix.CLONE_SIGHAND
2443 | posix.CLONE_THREAD | posix.CLONE_SYSVSEM // | posix.CLONE_SETTLS
2444 | posix.CLONE_PARENT_SETTID | posix.CLONE_CHILD_CLEARTID | posix.CLONE_DETACHED;
2445 const newtls: usize = 0;
2446 const rc = posix.clone(threadMain, stack_end, flags, arg, &thread_ptr.pid, newtls, &thread_ptr.pid);
2447 const err = posix.getErrno(rc);
2448 switch (err) {
2449 0 => return thread_ptr,
2450 posix.EAGAIN => return SpawnThreadError.ThreadQuotaExceeded,
2451 posix.EINVAL => unreachable,
2452 posix.ENOMEM => return SpawnThreadError.SystemResources,
2453 posix.ENOSPC => unreachable,
2454 posix.EPERM => unreachable,
2455 posix.EUSERS => unreachable,
2456 else => return unexpectedErrorPosix(err),
2527 if (builtin.os == builtin.Os.windows) {
2528 // use windows API directly
2529 @compileError("TODO support spawnThread for Windows");
2530 } else if (Thread.use_pthreads) {
2531 // use pthreads
2532 var attr: c.pthread_attr_t = undefined;
2533 if (c.pthread_attr_init(&attr) != 0) return SpawnThreadError.SystemResources;
2534 defer assert(c.pthread_attr_destroy(&attr) == 0);
2535
2536 // align to page
2537 stack_end -= stack_end % os.page_size;
2538 assert(c.pthread_attr_setstack(&attr, @intToPtr(&c_void, stack_addr), stack_end - stack_addr) == 0);
2539
2540 const err = c.pthread_create(&thread_ptr.data.handle, &attr, MainFuncs.posixThreadMain, @intToPtr(&c_void, arg));
2541 switch (err) {
2542 0 => return thread_ptr,
2543 posix.EAGAIN => return SpawnThreadError.SystemResources,
2544 posix.EPERM => unreachable,
2545 posix.EINVAL => unreachable,
2546 else => return unexpectedErrorPosix(usize(err)),
2547 }
2548 } else if (builtin.os == builtin.Os.linux) {
2549 // use linux API directly
2550 const flags = posix.CLONE_VM | posix.CLONE_FS | posix.CLONE_FILES | posix.CLONE_SIGHAND
2551 | posix.CLONE_THREAD | posix.CLONE_SYSVSEM // | posix.CLONE_SETTLS
2552 | posix.CLONE_PARENT_SETTID | posix.CLONE_CHILD_CLEARTID | posix.CLONE_DETACHED;
2553 const newtls: usize = 0;
2554 const rc = posix.clone(MainFuncs.linuxThreadMain, stack_end, flags, arg, &thread_ptr.data.pid, newtls, &thread_ptr.data.pid);
2555 const err = posix.getErrno(rc);
2556 switch (err) {
2557 0 => return thread_ptr,
2558 posix.EAGAIN => return SpawnThreadError.ThreadQuotaExceeded,
2559 posix.EINVAL => unreachable,
2560 posix.ENOMEM => return SpawnThreadError.SystemResources,
2561 posix.ENOSPC => unreachable,
2562 posix.EPERM => unreachable,
2563 posix.EUSERS => unreachable,
2564 else => return unexpectedErrorPosix(err),
2565 }
2566 } else {
2567 @compileError("Unsupported OS");
24572568 }
24582569}
24592570
std/os/linux/index.zig+3-3
......@@ -706,13 +706,13 @@ pub fn umount2(special: &const u8, flags: u32) usize {
706706 return syscall2(SYS_umount2, @ptrToInt(special), flags);
707707}
708708
709pub fn mmap(address: ?&u8, length: usize, prot: usize, flags: usize, fd: i32, offset: isize) usize {
709pub fn mmap(address: ?&u8, length: usize, prot: usize, flags: u32, fd: i32, offset: isize) usize {
710710 return syscall6(SYS_mmap, @ptrToInt(address), length, prot, flags, usize(fd),
711711 @bitCast(usize, offset));
712712}
713713
714pub fn munmap(address: &u8, length: usize) usize {
715 return syscall2(SYS_munmap, @ptrToInt(address), length);
714pub fn munmap(address: usize, length: usize) usize {
715 return syscall2(SYS_munmap, address, length);
716716}
717717
718718pub fn read(fd: i32, buf: &u8, count: usize) usize {
std/os/test.zig+4-16
......@@ -44,24 +44,12 @@ test "access file" {
4444}
4545
4646test "spawn threads" {
47 if (builtin.os != builtin.Os.linux) {
48 // TODO implement threads on macos and windows
49 return;
50 }
51
52 var direct_allocator = std.heap.DirectAllocator.init();
53 defer direct_allocator.deinit();
54
5547 var shared_ctx: i32 = 1;
5648
57 const thread1 = try std.os.spawnThreadAllocator(&direct_allocator.allocator, {}, start1);
58 const thread4 = try std.os.spawnThreadAllocator(&direct_allocator.allocator, &shared_ctx, start2);
59
60 var stack1: [1024]u8 = undefined;
61 var stack2: [1024]u8 = undefined;
62
63 const thread2 = try std.os.spawnThread(stack1[0..], &shared_ctx, start2);
64 const thread3 = try std.os.spawnThread(stack2[0..], &shared_ctx, start2);
49 const thread1 = try std.os.spawnThread({}, start1);
50 const thread2 = try std.os.spawnThread(&shared_ctx, start2);
51 const thread3 = try std.os.spawnThread(&shared_ctx, start2);
52 const thread4 = try std.os.spawnThread(&shared_ctx, start2);
6553
6654 thread1.wait();
6755 thread2.wait();
std/os/windows/index.zig+6
......@@ -28,6 +28,9 @@ pub extern "kernel32" stdcallcc fn CreateProcessA(lpApplicationName: ?LPCSTR, lp
2828pub extern "kernel32" stdcallcc fn CreateSymbolicLinkA(lpSymlinkFileName: LPCSTR, lpTargetFileName: LPCSTR,
2929 dwFlags: DWORD) BOOLEAN;
3030
31
32pub extern "kernel32" stdcallcc fn CreateThread(lpThreadAttributes: ?LPSECURITY_ATTRIBUTES, dwStackSize: SIZE_T, lpStartAddress: LPTHREAD_START_ROUTINE, lpParameter: ?LPVOID, dwCreationFlags: DWORD, lpThreadId: ?LPDWORD) ?HANDLE;
33
3134pub extern "kernel32" stdcallcc fn DeleteFileA(lpFileName: LPCSTR) BOOL;
3235
3336pub extern "kernel32" stdcallcc fn ExitProcess(exit_code: UINT) noreturn;
......@@ -318,6 +321,9 @@ pub const HEAP_CREATE_ENABLE_EXECUTE = 0x00040000;
318321pub const HEAP_GENERATE_EXCEPTIONS = 0x00000004;
319322pub const HEAP_NO_SERIALIZE = 0x00000001;
320323
324pub const PTHREAD_START_ROUTINE = extern fn(LPVOID) DWORD;
325pub const LPTHREAD_START_ROUTINE = PTHREAD_START_ROUTINE;
326
321327test "import" {
322328 _ = @import("util.zig");
323329}