authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2018-07-09 16:49:46-04:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2018-07-09 16:49:46-04:00
log9462852433a815496e0edf5d5b2e00726f5ea072
tree33e3b4930b65aed55d46dd866bfed24b10151c7a
parentcaa008505729f9511f6f0b070636013e9597b3f7

std.event.Loop multithreading for windows using IOCP

5 files changed, 191 insertions(+), 12 deletions(-)

std/event.zig+116-6
......@@ -4,6 +4,7 @@ const assert = std.debug.assert;
44const event = this;
55const mem = std.mem;
66const posix = std.os.posix;
7const windows = std.os.windows;
78const AtomicRmwOp = builtin.AtomicRmwOp;
89const AtomicOrder = builtin.AtomicOrder;
910
......@@ -113,10 +114,10 @@ pub const Loop = struct {
113114 allocator: *mem.Allocator,
114115 next_tick_queue: std.atomic.QueueMpsc(promise),
115116 os_data: OsData,
117 final_resume_node: ResumeNode,
116118 dispatch_lock: u8, // TODO make this a bool
117119 pending_event_count: usize,
118120 extra_threads: []*std.os.Thread,
119 final_resume_node: ResumeNode,
120121
121122 // pre-allocated eventfds. all permanently active.
122123 // this is how we send promises to be resumed on other threads.
......@@ -144,6 +145,7 @@ pub const Loop = struct {
144145 },
145146 builtin.Os.windows => struct {
146147 base: ResumeNode,
148 completion_key: usize,
147149 },
148150 else => @compileError("unsupported OS"),
149151 };
......@@ -181,12 +183,12 @@ pub const Loop = struct {
181183 .next_tick_queue = std.atomic.QueueMpsc(promise).init(),
182184 .dispatch_lock = 1, // start locked so threads go directly into epoll wait
183185 .extra_threads = undefined,
186 .available_eventfd_resume_nodes = std.atomic.Stack(ResumeNode.EventFd).init(),
187 .eventfd_resume_nodes = undefined,
184188 .final_resume_node = ResumeNode{
185189 .id = ResumeNode.Id.Stop,
186190 .handle = undefined,
187191 },
188 .available_eventfd_resume_nodes = std.atomic.Stack(ResumeNode.EventFd).init(),
189 .eventfd_resume_nodes = undefined,
190192 };
191193 const extra_thread_count = thread_count - 1;
192194 self.eventfd_resume_nodes = try self.allocator.alloc(
......@@ -209,7 +211,8 @@ pub const Loop = struct {
209211 }
210212
211213 const InitOsDataError = std.os.LinuxEpollCreateError || mem.Allocator.Error || std.os.LinuxEventFdError ||
212 std.os.SpawnThreadError || std.os.LinuxEpollCtlError || std.os.BsdKEventError;
214 std.os.SpawnThreadError || std.os.LinuxEpollCtlError || std.os.BsdKEventError ||
215 std.os.WindowsCreateIoCompletionPortError;
213216
214217 const wakeup_bytes = []u8{0x1} ** 8;
215218
......@@ -335,6 +338,51 @@ pub const Loop = struct {
335338 self.extra_threads[extra_thread_index] = try std.os.spawnThread(self, workerRun);
336339 }
337340 },
341 builtin.Os.windows => {
342 self.os_data.extra_thread_count = extra_thread_count;
343
344 self.os_data.io_port = try std.os.windowsCreateIoCompletionPort(
345 windows.INVALID_HANDLE_VALUE,
346 null,
347 undefined,
348 undefined,
349 );
350 errdefer std.os.close(self.os_data.io_port);
351
352 for (self.eventfd_resume_nodes) |*eventfd_node, i| {
353 eventfd_node.* = std.atomic.Stack(ResumeNode.EventFd).Node{
354 .data = ResumeNode.EventFd{
355 .base = ResumeNode{
356 .id = ResumeNode.Id.EventFd,
357 .handle = undefined,
358 },
359 // this one is for sending events
360 .completion_key = @ptrToInt(&eventfd_node.data.base),
361 },
362 .next = undefined,
363 };
364 self.available_eventfd_resume_nodes.push(eventfd_node);
365 }
366
367 var extra_thread_index: usize = 0;
368 errdefer {
369 var i: usize = 0;
370 while (i < extra_thread_index) : (i += 1) {
371 while (true) {
372 const overlapped = @intToPtr(?*windows.OVERLAPPED, 0x1);
373 std.os.windowsPostQueuedCompletionStatus(self.os_data.io_port, undefined, @ptrToInt(&self.final_resume_node), overlapped) catch continue;
374 break;
375 }
376 }
377 while (extra_thread_index != 0) {
378 extra_thread_index -= 1;
379 self.extra_threads[extra_thread_index].wait();
380 }
381 }
382 while (extra_thread_index < extra_thread_count) : (extra_thread_index += 1) {
383 self.extra_threads[extra_thread_index] = try std.os.spawnThread(self, workerRun);
384 }
385 },
338386 else => {},
339387 }
340388 }
......@@ -349,6 +397,10 @@ pub const Loop = struct {
349397 },
350398 builtin.Os.macosx => {
351399 self.allocator.free(self.os_data.kevents);
400 std.os.close(self.os_data.kqfd);
401 },
402 builtin.Os.windows => {
403 std.os.close(self.os_data.io_port);
352404 },
353405 else => {},
354406 }
......@@ -434,7 +486,7 @@ pub const Loop = struct {
434486 builtin.Os.macosx => {
435487 const kevent_array = (*[1]posix.Kevent)(&eventfd_node.kevent);
436488 const eventlist = ([*]posix.Kevent)(undefined)[0..0];
437 _ = std.os.bsdKEvent(self.os_data.kqfd, kevent_array, eventlist, null) catch |_| {
489 _ = std.os.bsdKEvent(self.os_data.kqfd, kevent_array, eventlist, null) catch {
438490 // fine, we didn't need it anyway
439491 _ = @atomicRmw(u8, &self.dispatch_lock, AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst);
440492 self.available_eventfd_resume_nodes.push(resume_stack_node);
......@@ -446,7 +498,21 @@ pub const Loop = struct {
446498 builtin.Os.linux => {
447499 // the pending count is already accounted for
448500 const epoll_events = posix.EPOLLONESHOT | std.os.linux.EPOLLIN | std.os.linux.EPOLLOUT | std.os.linux.EPOLLET;
449 self.modFd(eventfd_node.eventfd, eventfd_node.epoll_op, epoll_events, &eventfd_node.base) catch |_| {
501 self.modFd(eventfd_node.eventfd, eventfd_node.epoll_op, epoll_events, &eventfd_node.base) catch {
502 // fine, we didn't need it anyway
503 _ = @atomicRmw(u8, &self.dispatch_lock, AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst);
504 self.available_eventfd_resume_nodes.push(resume_stack_node);
505 resume handle;
506 _ = @atomicRmw(usize, &self.pending_event_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst);
507 continue :start_over;
508 };
509 },
510 builtin.Os.windows => {
511 // this value is never dereferenced but we need it to be non-null so that
512 // the consumer code can decide whether to read the completion key.
513 // it has to do this for normal I/O, so we match that behavior here.
514 const overlapped = @intToPtr(?*windows.OVERLAPPED, 0x1);
515 std.os.windowsPostQueuedCompletionStatus(self.os_data.io_port, undefined, eventfd_node.completion_key, overlapped) catch {
450516 // fine, we didn't need it anyway
451517 _ = @atomicRmw(u8, &self.dispatch_lock, AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst);
452518 self.available_eventfd_resume_nodes.push(resume_stack_node);
......@@ -482,6 +548,17 @@ pub const Loop = struct {
482548 _ = std.os.bsdKEvent(self.os_data.kqfd, final_kevent, eventlist, null) catch unreachable;
483549 return;
484550 },
551 builtin.Os.windows => {
552 var i: usize = 0;
553 while (i < self.os_data.extra_thread_count) : (i += 1) {
554 while (true) {
555 const overlapped = @intToPtr(?*windows.OVERLAPPED, 0x1);
556 std.os.windowsPostQueuedCompletionStatus(self.os_data.io_port, undefined, @ptrToInt(&self.final_resume_node), overlapped) catch continue;
557 break;
558 }
559 }
560 return;
561 },
485562 else => @compileError("unsupported OS"),
486563 }
487564 }
......@@ -536,6 +613,35 @@ pub const Loop = struct {
536613 }
537614 }
538615 },
616 builtin.Os.windows => {
617 var completion_key: usize = undefined;
618 while (true) {
619 var nbytes: windows.DWORD = undefined;
620 var overlapped: ?*windows.OVERLAPPED = undefined;
621 switch (std.os.windowsGetQueuedCompletionStatus(self.os_data.io_port, &nbytes, &completion_key,
622 &overlapped, windows.INFINITE)) {
623 std.os.WindowsWaitResult.Aborted => return,
624 std.os.WindowsWaitResult.Normal => {},
625 }
626 if (overlapped != null) break;
627 }
628 const resume_node = @intToPtr(*ResumeNode, completion_key);
629 const handle = resume_node.handle;
630 const resume_node_id = resume_node.id;
631 switch (resume_node_id) {
632 ResumeNode.Id.Basic => {},
633 ResumeNode.Id.Stop => return,
634 ResumeNode.Id.EventFd => {
635 const event_fd_node = @fieldParentPtr(ResumeNode.EventFd, "base", resume_node);
636 const stack_node = @fieldParentPtr(std.atomic.Stack(ResumeNode.EventFd).Node, "data", event_fd_node);
637 self.available_eventfd_resume_nodes.push(stack_node);
638 },
639 }
640 resume handle;
641 if (resume_node_id == ResumeNode.Id.EventFd) {
642 _ = @atomicRmw(usize, &self.pending_event_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst);
643 }
644 },
539645 else => @compileError("unsupported OS"),
540646 }
541647 }
......@@ -548,6 +654,10 @@ pub const Loop = struct {
548654 final_eventfd_event: std.os.linux.epoll_event,
549655 },
550656 builtin.Os.macosx => MacOsData,
657 builtin.Os.windows => struct {
658 io_port: windows.HANDLE,
659 extra_thread_count: usize,
660 },
551661 else => struct {},
552662 };
553663
std/heap.zig+1-1
......@@ -98,7 +98,7 @@ pub const DirectAllocator = struct {
9898 const amt = n + alignment + @sizeOf(usize);
9999 const optional_heap_handle = @atomicLoad(?HeapHandle, &self.heap_handle, builtin.AtomicOrder.SeqCst);
100100 const heap_handle = optional_heap_handle orelse blk: {
101 const hh = os.windows.HeapCreate(os.windows.HEAP_NO_SERIALIZE, amt, 0) orelse return error.OutOfMemory;
101 const hh = os.windows.HeapCreate(0, amt, 0) orelse return error.OutOfMemory;
102102 const other_hh = @cmpxchgStrong(?HeapHandle, &self.heap_handle, null, hh, builtin.AtomicOrder.SeqCst, builtin.AtomicOrder.SeqCst) orelse break :blk hh;
103103 _ = os.windows.HeapDestroy(hh);
104104 break :blk other_hh.?; // can't be null because of the cmpxchg
std/os/index.zig+20-5
......@@ -61,6 +61,15 @@ pub const windowsLoadDll = windows_util.windowsLoadDll;
6161pub const windowsUnloadDll = windows_util.windowsUnloadDll;
6262pub const createWindowsEnvBlock = windows_util.createWindowsEnvBlock;
6363
64pub const WindowsCreateIoCompletionPortError = windows_util.WindowsCreateIoCompletionPortError;
65pub const windowsCreateIoCompletionPort = windows_util.windowsCreateIoCompletionPort;
66
67pub const WindowsPostQueuedCompletionStatusError = windows_util.WindowsPostQueuedCompletionStatusError;
68pub const windowsPostQueuedCompletionStatus = windows_util.windowsPostQueuedCompletionStatus;
69
70pub const WindowsWaitResult = windows_util.WindowsWaitResult;
71pub const windowsGetQueuedCompletionStatus = windows_util.windowsGetQueuedCompletionStatus;
72
6473pub const WindowsWaitError = windows_util.WaitError;
6574pub const WindowsOpenError = windows_util.OpenError;
6675pub const WindowsWriteError = windows_util.WriteError;
......@@ -2592,11 +2601,17 @@ pub fn spawnThread(context: var, comptime startFn: var) SpawnThreadError!*Thread
25922601 thread: Thread,
25932602 inner: Context,
25942603 };
2595 extern fn threadMain(arg: windows.LPVOID) windows.DWORD {
2596 if (@sizeOf(Context) == 0) {
2597 return startFn({});
2598 } else {
2599 return startFn(@ptrCast(*Context, @alignCast(@alignOf(Context), arg)).*);
2604 extern fn threadMain(raw_arg: windows.LPVOID) windows.DWORD {
2605 const arg = if (@sizeOf(Context) == 0) {} else @ptrCast(*Context, @alignCast(@alignOf(Context), raw_arg)).*;
2606 switch (@typeId(@typeOf(startFn).ReturnType)) {
2607 builtin.TypeId.Int => {
2608 return startFn(arg);
2609 },
2610 builtin.TypeId.Void => {
2611 startFn(arg);
2612 return 0;
2613 },
2614 else => @compileError("expected return type of startFn to be 'u8', 'noreturn', 'void', or '!void'"),
26002615 }
26012616 }
26022617 };
std/os/windows/index.zig+7
......@@ -59,6 +59,9 @@ pub extern "kernel32" stdcallcc fn CreateSymbolicLinkA(
5959 dwFlags: DWORD,
6060) BOOLEAN;
6161
62
63pub extern "kernel32" stdcallcc fn CreateIoCompletionPort(FileHandle: HANDLE, ExistingCompletionPort: ?HANDLE, CompletionKey: ULONG_PTR, NumberOfConcurrentThreads: DWORD) ?HANDLE;
64
6265pub extern "kernel32" stdcallcc fn CreateThread(lpThreadAttributes: ?LPSECURITY_ATTRIBUTES, dwStackSize: SIZE_T, lpStartAddress: LPTHREAD_START_ROUTINE, lpParameter: ?LPVOID, dwCreationFlags: DWORD, lpThreadId: ?LPDWORD) ?HANDLE;
6366
6467pub extern "kernel32" stdcallcc fn DeleteFileA(lpFileName: LPCSTR) BOOL;
......@@ -106,6 +109,7 @@ pub extern "kernel32" stdcallcc fn GetFinalPathNameByHandleA(
106109) DWORD;
107110
108111pub extern "kernel32" stdcallcc fn GetProcessHeap() ?HANDLE;
112pub extern "kernel32" stdcallcc fn GetQueuedCompletionStatus(CompletionPort: HANDLE, lpNumberOfBytesTransferred: LPDWORD, lpCompletionKey: *ULONG_PTR, lpOverlapped: *?*OVERLAPPED, dwMilliseconds: DWORD) BOOL;
109113
110114pub extern "kernel32" stdcallcc fn GetSystemInfo(lpSystemInfo: *SYSTEM_INFO) void;
111115pub extern "kernel32" stdcallcc fn GetSystemTimeAsFileTime(*FILETIME) void;
......@@ -130,6 +134,9 @@ pub extern "kernel32" stdcallcc fn MoveFileExA(
130134 dwFlags: DWORD,
131135) BOOL;
132136
137
138pub extern "kernel32" stdcallcc fn PostQueuedCompletionStatus(CompletionPort: HANDLE, dwNumberOfBytesTransferred: DWORD, dwCompletionKey: ULONG_PTR, lpOverlapped: ?*OVERLAPPED) BOOL;
139
133140pub extern "kernel32" stdcallcc fn QueryPerformanceCounter(lpPerformanceCount: *LARGE_INTEGER) BOOL;
134141
135142pub extern "kernel32" stdcallcc fn QueryPerformanceFrequency(lpFrequency: *LARGE_INTEGER) BOOL;
std/os/windows/util.zig+47
......@@ -214,3 +214,50 @@ pub fn windowsFindNextFile(handle: windows.HANDLE, find_file_data: *windows.WIN3
214214 }
215215 return true;
216216}
217
218
219pub const WindowsCreateIoCompletionPortError = error {
220 Unexpected,
221};
222
223pub fn windowsCreateIoCompletionPort(file_handle: windows.HANDLE, existing_completion_port: ?windows.HANDLE, completion_key: usize, concurrent_thread_count: windows.DWORD) !windows.HANDLE {
224 const handle = windows.CreateIoCompletionPort(file_handle, existing_completion_port, completion_key, concurrent_thread_count) orelse {
225 const err = windows.GetLastError();
226 switch (err) {
227 else => return os.unexpectedErrorWindows(err),
228 }
229 };
230 return handle;
231}
232
233pub const WindowsPostQueuedCompletionStatusError = error {
234 Unexpected,
235};
236
237pub fn windowsPostQueuedCompletionStatus(completion_port: windows.HANDLE, bytes_transferred_count: windows.DWORD, completion_key: usize, lpOverlapped: ?*windows.OVERLAPPED) WindowsPostQueuedCompletionStatusError!void {
238 if (windows.PostQueuedCompletionStatus(completion_port, bytes_transferred_count, completion_key, lpOverlapped) == 0) {
239 const err = windows.GetLastError();
240 switch (err) {
241 else => return os.unexpectedErrorWindows(err),
242 }
243 }
244}
245
246pub const WindowsWaitResult = error {
247 Normal,
248 Aborted,
249};
250
251pub fn windowsGetQueuedCompletionStatus(completion_port: windows.HANDLE, bytes_transferred_count: *windows.DWORD, lpCompletionKey: *usize, lpOverlapped: *?*windows.OVERLAPPED, dwMilliseconds: windows.DWORD) WindowsWaitResult {
252 if (windows.GetQueuedCompletionStatus(completion_port, bytes_transferred_count, lpCompletionKey, lpOverlapped, dwMilliseconds) == windows.FALSE) {
253 if (std.debug.runtime_safety) {
254 const err = windows.GetLastError();
255 if (err != windows.ERROR.ABANDONED_WAIT_0) {
256 std.debug.warn("err: {}\n", err);
257 }
258 assert(err == windows.ERROR.ABANDONED_WAIT_0);
259 }
260 return WindowsWaitResult.Aborted;
261 }
262 return WindowsWaitResult.Normal;
263}