authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2018-06-29 15:39:55-04:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2018-07-02 14:38:11-04:00
loga3f55aaf34f0a459c8aec4b35e55ad4534eaca30
tree5799189e210d53271de654a2f713e7d5f9056fed
parent2759c7951da050d825cf765c4b660f5562fb01a4

add event loop Channel abstraction

This is akin to channels in Go, except: * implemented in userland * they are lock-free and thread-safe * they integrate with the userland event loop The self hosted compiler is changed to use a channel for events, and made to stay alive, watching files and performing builds when things change, however the main.zig file exits after 1 build. Note that nothing is actually built yet, it just parses the input and then declares that the build succeeded. Next items to do: * add windows and macos support for std.event.Loop * improve the event loop stop() operation * make the event loop multiplex coroutines onto kernel threads * watch source file for updates, and provide AST diffs (at least list the top level declaration changes) * top level declaration analysis

6 files changed, 416 insertions(+), 41 deletions(-)

src-self-hosted/main.zig+34-3
...@@ -1,6 +1,7 @@...@@ -1,6 +1,7 @@
1const std = @import("std");1const std = @import("std");
2const builtin = @import("builtin");2const builtin = @import("builtin");
33
4const event = std.event;
4const os = std.os;5const os = std.os;
5const io = std.io;6const io = std.io;
6const mem = std.mem;7const mem = std.mem;
...@@ -43,6 +44,9 @@ const Command = struct {...@@ -43,6 +44,9 @@ const Command = struct {
43};44};
4445
45pub fn main() !void {46pub fn main() !void {
47 // This allocator needs to be thread-safe because we use it for the event.Loop
48 // which multiplexes coroutines onto kernel threads.
49 // libc allocator is guaranteed to have this property.
46 const allocator = std.heap.c_allocator;50 const allocator = std.heap.c_allocator;
4751
48 var stdout_file = try std.io.getStdOut();52 var stdout_file = try std.io.getStdOut();
...@@ -380,8 +384,10 @@ fn buildOutputType(allocator: *Allocator, args: []const []const u8, out_type: Mo...@@ -380,8 +384,10 @@ fn buildOutputType(allocator: *Allocator, args: []const []const u8, out_type: Mo
380 const zig_lib_dir = introspect.resolveZigLibDir(allocator) catch os.exit(1);384 const zig_lib_dir = introspect.resolveZigLibDir(allocator) catch os.exit(1);
381 defer allocator.free(zig_lib_dir);385 defer allocator.free(zig_lib_dir);
382386
387 var loop = try event.Loop.init(allocator);
388
383 var module = try Module.create(389 var module = try Module.create(
384 allocator,390 &loop,
385 root_name,391 root_name,
386 root_source_file,392 root_source_file,
387 Target.Native,393 Target.Native,
...@@ -471,9 +477,35 @@ fn buildOutputType(allocator: *Allocator, args: []const []const u8, out_type: Mo...@@ -471,9 +477,35 @@ fn buildOutputType(allocator: *Allocator, args: []const []const u8, out_type: Mo
471 module.emit_file_type = emit_type;477 module.emit_file_type = emit_type;
472 module.link_objects = link_objects;478 module.link_objects = link_objects;
473 module.assembly_files = assembly_files;479 module.assembly_files = assembly_files;
480 module.link_out_file = flags.single("out-file");
474481
475 try module.build();482 try module.build();
476 try module.link(flags.single("out-file"));483 const process_build_events_handle = try async<loop.allocator> processBuildEvents(module, true);
484 defer cancel process_build_events_handle;
485 loop.run();
486}
487
488async fn processBuildEvents(module: *Module, watch: bool) void {
489 while (watch) {
490 // TODO directly awaiting async should guarantee memory allocation elision
491 const build_event = await (async module.events.get() catch unreachable);
492
493 switch (build_event) {
494 Module.Event.Ok => {
495 std.debug.warn("Build succeeded\n");
496 // for now we stop after 1
497 module.loop.stop();
498 return;
499 },
500 Module.Event.Error => |err| {
501 std.debug.warn("build failed: {}\n", @errorName(err));
502 @panic("TODO error return trace");
503 },
504 Module.Event.Fail => |errs| {
505 @panic("TODO print compile error messages");
506 },
507 }
508 }
477}509}
478510
479fn cmdBuildExe(allocator: *Allocator, args: []const []const u8) !void {511fn cmdBuildExe(allocator: *Allocator, args: []const []const u8) !void {
...@@ -780,4 +812,3 @@ const CliPkg = struct {...@@ -780,4 +812,3 @@ const CliPkg = struct {
780 self.children.deinit();812 self.children.deinit();
781 }813 }
782};814};
783
src-self-hosted/module.zig+100-35
...@@ -11,9 +11,11 @@ const warn = std.debug.warn;...@@ -11,9 +11,11 @@ const warn = std.debug.warn;
11const Token = std.zig.Token;11const Token = std.zig.Token;
12const ArrayList = std.ArrayList;12const ArrayList = std.ArrayList;
13const errmsg = @import("errmsg.zig");13const errmsg = @import("errmsg.zig");
14const ast = std.zig.ast;
15const event = std.event;
1416
15pub const Module = struct {17pub const Module = struct {
16 allocator: *mem.Allocator,18 loop: *event.Loop,
17 name: Buffer,19 name: Buffer,
18 root_src_path: ?[]const u8,20 root_src_path: ?[]const u8,
19 module: llvm.ModuleRef,21 module: llvm.ModuleRef,
...@@ -76,6 +78,50 @@ pub const Module = struct {...@@ -76,6 +78,50 @@ pub const Module = struct {
7678
77 kind: Kind,79 kind: Kind,
7880
81 link_out_file: ?[]const u8,
82 events: *event.Channel(Event),
83
84 // TODO handle some of these earlier and report them in a way other than error codes
85 pub const BuildError = error{
86 OutOfMemory,
87 EndOfStream,
88 BadFd,
89 Io,
90 IsDir,
91 Unexpected,
92 SystemResources,
93 SharingViolation,
94 PathAlreadyExists,
95 FileNotFound,
96 AccessDenied,
97 PipeBusy,
98 FileTooBig,
99 SymLinkLoop,
100 ProcessFdQuotaExceeded,
101 NameTooLong,
102 SystemFdQuotaExceeded,
103 NoDevice,
104 PathNotFound,
105 NoSpaceLeft,
106 NotDir,
107 FileSystem,
108 OperationAborted,
109 IoPending,
110 BrokenPipe,
111 WouldBlock,
112 FileClosed,
113 DestinationAddressRequired,
114 DiskQuota,
115 InputOutput,
116 NoStdHandles,
117 };
118
119 pub const Event = union(enum) {
120 Ok,
121 Fail: []errmsg.Msg,
122 Error: BuildError,
123 };
124
79 pub const DarwinVersionMin = union(enum) {125 pub const DarwinVersionMin = union(enum) {
80 None,126 None,
81 MacOS: []const u8,127 MacOS: []const u8,
...@@ -104,7 +150,7 @@ pub const Module = struct {...@@ -104,7 +150,7 @@ pub const Module = struct {
104 };150 };
105151
106 pub fn create(152 pub fn create(
107 allocator: *mem.Allocator,153 loop: *event.Loop,
108 name: []const u8,154 name: []const u8,
109 root_src_path: ?[]const u8,155 root_src_path: ?[]const u8,
110 target: *const Target,156 target: *const Target,
...@@ -113,7 +159,7 @@ pub const Module = struct {...@@ -113,7 +159,7 @@ pub const Module = struct {
113 zig_lib_dir: []const u8,159 zig_lib_dir: []const u8,
114 cache_dir: []const u8,160 cache_dir: []const u8,
115 ) !*Module {161 ) !*Module {
116 var name_buffer = try Buffer.init(allocator, name);162 var name_buffer = try Buffer.init(loop.allocator, name);
117 errdefer name_buffer.deinit();163 errdefer name_buffer.deinit();
118164
119 const context = c.LLVMContextCreate() orelse return error.OutOfMemory;165 const context = c.LLVMContextCreate() orelse return error.OutOfMemory;
...@@ -125,8 +171,12 @@ pub const Module = struct {...@@ -125,8 +171,12 @@ pub const Module = struct {
125 const builder = c.LLVMCreateBuilderInContext(context) orelse return error.OutOfMemory;171 const builder = c.LLVMCreateBuilderInContext(context) orelse return error.OutOfMemory;
126 errdefer c.LLVMDisposeBuilder(builder);172 errdefer c.LLVMDisposeBuilder(builder);
127173
128 const module_ptr = try allocator.create(Module{174 const events = try event.Channel(Event).create(loop, 0);
129 .allocator = allocator,175 errdefer events.destroy();
176
177 return loop.allocator.create(Module{
178 .loop = loop,
179 .events = events,
130 .name = name_buffer,180 .name = name_buffer,
131 .root_src_path = root_src_path,181 .root_src_path = root_src_path,
132 .module = module,182 .module = module,
...@@ -171,7 +221,7 @@ pub const Module = struct {...@@ -171,7 +221,7 @@ pub const Module = struct {
171 .link_objects = [][]const u8{},221 .link_objects = [][]const u8{},
172 .windows_subsystem_windows = false,222 .windows_subsystem_windows = false,
173 .windows_subsystem_console = false,223 .windows_subsystem_console = false,
174 .link_libs_list = ArrayList(*LinkLib).init(allocator),224 .link_libs_list = ArrayList(*LinkLib).init(loop.allocator),
175 .libc_link_lib = null,225 .libc_link_lib = null,
176 .err_color = errmsg.Color.Auto,226 .err_color = errmsg.Color.Auto,
177 .darwin_frameworks = [][]const u8{},227 .darwin_frameworks = [][]const u8{},
...@@ -179,9 +229,8 @@ pub const Module = struct {...@@ -179,9 +229,8 @@ pub const Module = struct {
179 .test_filters = [][]const u8{},229 .test_filters = [][]const u8{},
180 .test_name_prefix = null,230 .test_name_prefix = null,
181 .emit_file_type = Emit.Binary,231 .emit_file_type = Emit.Binary,
232 .link_out_file = null,
182 });233 });
183 errdefer allocator.destroy(module_ptr);
184 return module_ptr;
185 }234 }
186235
187 fn dump(self: *Module) void {236 fn dump(self: *Module) void {
...@@ -189,58 +238,70 @@ pub const Module = struct {...@@ -189,58 +238,70 @@ pub const Module = struct {
189 }238 }
190239
191 pub fn destroy(self: *Module) void {240 pub fn destroy(self: *Module) void {
241 self.events.destroy();
192 c.LLVMDisposeBuilder(self.builder);242 c.LLVMDisposeBuilder(self.builder);
193 c.LLVMDisposeModule(self.module);243 c.LLVMDisposeModule(self.module);
194 c.LLVMContextDispose(self.context);244 c.LLVMContextDispose(self.context);
195 self.name.deinit();245 self.name.deinit();
196246
197 self.allocator.destroy(self);247 self.a().destroy(self);
198 }248 }
199249
200 pub fn build(self: *Module) !void {250 pub fn build(self: *Module) !void {
201 if (self.llvm_argv.len != 0) {251 if (self.llvm_argv.len != 0) {
202 var c_compatible_args = try std.cstr.NullTerminated2DArray.fromSlices(self.allocator, [][]const []const u8{252 var c_compatible_args = try std.cstr.NullTerminated2DArray.fromSlices(self.a(), [][]const []const u8{
203 [][]const u8{"zig (LLVM option parsing)"},253 [][]const u8{"zig (LLVM option parsing)"},
204 self.llvm_argv,254 self.llvm_argv,
205 });255 });
206 defer c_compatible_args.deinit();256 defer c_compatible_args.deinit();
257 // TODO this sets global state
207 c.ZigLLVMParseCommandLineOptions(self.llvm_argv.len + 1, c_compatible_args.ptr);258 c.ZigLLVMParseCommandLineOptions(self.llvm_argv.len + 1, c_compatible_args.ptr);
208 }259 }
209260
261 _ = try async<self.a()> self.buildAsync();
262 }
263
264 async fn buildAsync(self: *Module) void {
265 while (true) {
266 // TODO directly awaiting async should guarantee memory allocation elision
267 // TODO also async before suspending should guarantee memory allocation elision
268 (await (async self.addRootSrc() catch unreachable)) catch |err| {
269 await (async self.events.put(Event{ .Error = err }) catch unreachable);
270 return;
271 };
272 await (async self.events.put(Event.Ok) catch unreachable);
273 }
274 }
275
276 async fn addRootSrc(self: *Module) !void {
210 const root_src_path = self.root_src_path orelse @panic("TODO handle null root src path");277 const root_src_path = self.root_src_path orelse @panic("TODO handle null root src path");
211 const root_src_real_path = os.path.real(self.allocator, root_src_path) catch |err| {278 const root_src_real_path = os.path.real(self.a(), root_src_path) catch |err| {
212 try printError("unable to get real path '{}': {}", root_src_path, err);279 try printError("unable to get real path '{}': {}", root_src_path, err);
213 return err;280 return err;
214 };281 };
215 errdefer self.allocator.free(root_src_real_path);282 errdefer self.a().free(root_src_real_path);
216283
217 const source_code = io.readFileAlloc(self.allocator, root_src_real_path) catch |err| {284 const source_code = io.readFileAlloc(self.a(), root_src_real_path) catch |err| {
218 try printError("unable to open '{}': {}", root_src_real_path, err);285 try printError("unable to open '{}': {}", root_src_real_path, err);
219 return err;286 return err;
220 };287 };
221 errdefer self.allocator.free(source_code);288 errdefer self.a().free(source_code);
222
223 warn("====input:====\n");
224
225 warn("{}", source_code);
226289
227 warn("====parse:====\n");290 var tree = try std.zig.parse(self.a(), source_code);
228
229 var tree = try std.zig.parse(self.allocator, source_code);
230 defer tree.deinit();291 defer tree.deinit();
231292
232 var stderr_file = try std.io.getStdErr();293 //var it = tree.root_node.decls.iterator();
233 var stderr_file_out_stream = std.io.FileOutStream.init(&stderr_file);294 //while (it.next()) |decl_ptr| {
234 const out_stream = &stderr_file_out_stream.stream;295 // const decl = decl_ptr.*;
235296 // switch (decl.id) {
236 warn("====fmt:====\n");297 // ast.Node.Comptime => @panic("TODO"),
237 _ = try std.zig.render(self.allocator, out_stream, &tree);298 // ast.Node.VarDecl => @panic("TODO"),
238299 // ast.Node.UseDecl => @panic("TODO"),
239 warn("====ir:====\n");300 // ast.Node.FnDef => @panic("TODO"),
240 warn("TODO\n\n");301 // ast.Node.TestDecl => @panic("TODO"),
241302 // else => unreachable,
242 warn("====llvm ir:====\n");303 // }
243 self.dump();304 //}
244 }305 }
245306
246 pub fn link(self: *Module, out_file: ?[]const u8) !void {307 pub fn link(self: *Module, out_file: ?[]const u8) !void {
...@@ -263,11 +324,11 @@ pub const Module = struct {...@@ -263,11 +324,11 @@ pub const Module = struct {
263 }324 }
264 }325 }
265326
266 const link_lib = try self.allocator.create(LinkLib{327 const link_lib = try self.a().create(LinkLib{
267 .name = name,328 .name = name,
268 .path = null,329 .path = null,
269 .provided_explicitly = provided_explicitly,330 .provided_explicitly = provided_explicitly,
270 .symbols = ArrayList([]u8).init(self.allocator),331 .symbols = ArrayList([]u8).init(self.a()),
271 });332 });
272 try self.link_libs_list.append(link_lib);333 try self.link_libs_list.append(link_lib);
273 if (is_libc) {334 if (is_libc) {
...@@ -275,6 +336,10 @@ pub const Module = struct {...@@ -275,6 +336,10 @@ pub const Module = struct {
275 }336 }
276 return link_lib;337 return link_lib;
277 }338 }
339
340 fn a(self: Module) *mem.Allocator {
341 return self.loop.allocator;
342 }
278};343};
279344
280fn printError(comptime format: []const u8, args: ...) !void {345fn printError(comptime format: []const u8, args: ...) !void {
std/atomic/queue_mpsc.zig+1-1
...@@ -1,4 +1,4 @@...@@ -1,4 +1,4 @@
1const std = @import("std");1const std = @import("../index.zig");
2const assert = std.debug.assert;2const assert = std.debug.assert;
3const builtin = @import("builtin");3const builtin = @import("builtin");
4const AtomicOrder = builtin.AtomicOrder;4const AtomicOrder = builtin.AtomicOrder;
std/event.zig+277-2
...@@ -4,6 +4,8 @@ const assert = std.debug.assert;...@@ -4,6 +4,8 @@ const assert = std.debug.assert;
4const event = this;4const event = this;
5const mem = std.mem;5const mem = std.mem;
6const posix = std.os.posix;6const posix = std.os.posix;
7const AtomicRmwOp = builtin.AtomicRmwOp;
8const AtomicOrder = builtin.AtomicOrder;
79
8pub const TcpServer = struct {10pub const TcpServer = struct {
9 handleRequestFn: async<*mem.Allocator> fn (*TcpServer, *const std.net.Address, *const std.os.File) void,11 handleRequestFn: async<*mem.Allocator> fn (*TcpServer, *const std.net.Address, *const std.os.File) void,
...@@ -95,16 +97,29 @@ pub const Loop = struct {...@@ -95,16 +97,29 @@ pub const Loop = struct {
95 allocator: *mem.Allocator,97 allocator: *mem.Allocator,
96 epollfd: i32,98 epollfd: i32,
97 keep_running: bool,99 keep_running: bool,
100 next_tick_queue: std.atomic.QueueMpsc(promise),
98101
99 fn init(allocator: *mem.Allocator) !Loop {102 pub const NextTickNode = std.atomic.QueueMpsc(promise).Node;
103
104 /// The allocator must be thread-safe because we use it for multiplexing
105 /// coroutines onto kernel threads.
106 pub fn init(allocator: *mem.Allocator) !Loop {
100 const epollfd = try std.os.linuxEpollCreate(std.os.linux.EPOLL_CLOEXEC);107 const epollfd = try std.os.linuxEpollCreate(std.os.linux.EPOLL_CLOEXEC);
108 errdefer std.os.close(epollfd);
109
101 return Loop{110 return Loop{
102 .keep_running = true,111 .keep_running = true,
103 .allocator = allocator,112 .allocator = allocator,
104 .epollfd = epollfd,113 .epollfd = epollfd,
114 .next_tick_queue = std.atomic.QueueMpsc(promise).init(),
105 };115 };
106 }116 }
107117
118 /// must call stop before deinit
119 pub fn deinit(self: *Loop) void {
120 std.os.close(self.epollfd);
121 }
122
108 pub fn addFd(self: *Loop, fd: i32, prom: promise) !void {123 pub fn addFd(self: *Loop, fd: i32, prom: promise) !void {
109 var ev = std.os.linux.epoll_event{124 var ev = std.os.linux.epoll_event{
110 .events = std.os.linux.EPOLLIN | std.os.linux.EPOLLOUT | std.os.linux.EPOLLET,125 .events = std.os.linux.EPOLLIN | std.os.linux.EPOLLOUT | std.os.linux.EPOLLET,
...@@ -126,11 +141,21 @@ pub const Loop = struct {...@@ -126,11 +141,21 @@ pub const Loop = struct {
126 pub fn stop(self: *Loop) void {141 pub fn stop(self: *Loop) void {
127 // TODO make atomic142 // TODO make atomic
128 self.keep_running = false;143 self.keep_running = false;
129 // TODO activate an fd in the epoll set144 // TODO activate an fd in the epoll set which should cancel all the promises
145 }
146
147 /// bring your own linked list node. this means it can't fail.
148 pub fn onNextTick(self: *Loop, node: *NextTickNode) void {
149 self.next_tick_queue.put(node);
130 }150 }
131151
132 pub fn run(self: *Loop) void {152 pub fn run(self: *Loop) void {
133 while (self.keep_running) {153 while (self.keep_running) {
154 // TODO multiplex the next tick queue and the epoll event results onto a thread pool
155 while (self.next_tick_queue.get()) |node| {
156 resume node.data;
157 }
158 if (!self.keep_running) break;
134 var events: [16]std.os.linux.epoll_event = undefined;159 var events: [16]std.os.linux.epoll_event = undefined;
135 const count = std.os.linuxEpollWait(self.epollfd, events[0..], -1);160 const count = std.os.linuxEpollWait(self.epollfd, events[0..], -1);
136 for (events[0..count]) |ev| {161 for (events[0..count]) |ev| {
...@@ -141,6 +166,215 @@ pub const Loop = struct {...@@ -141,6 +166,215 @@ pub const Loop = struct {
141 }166 }
142};167};
143168
169/// many producer, many consumer, thread-safe, lock-free, runtime configurable buffer size
170/// when buffer is empty, consumers suspend and are resumed by producers
171/// when buffer is full, producers suspend and are resumed by consumers
172pub fn Channel(comptime T: type) type {
173 return struct {
174 loop: *Loop,
175
176 getters: std.atomic.QueueMpsc(GetNode),
177 putters: std.atomic.QueueMpsc(PutNode),
178 get_count: usize,
179 put_count: usize,
180 dispatch_lock: u8, // TODO make this a bool
181 need_dispatch: u8, // TODO make this a bool
182
183 // simple fixed size ring buffer
184 buffer_nodes: []T,
185 buffer_index: usize,
186 buffer_len: usize,
187
188 const SelfChannel = this;
189 const GetNode = struct {
190 ptr: *T,
191 tick_node: *Loop.NextTickNode,
192 };
193 const PutNode = struct {
194 data: T,
195 tick_node: *Loop.NextTickNode,
196 };
197
198 /// call destroy when done
199 pub fn create(loop: *Loop, capacity: usize) !*SelfChannel {
200 const buffer_nodes = try loop.allocator.alloc(T, capacity);
201 errdefer loop.allocator.free(buffer_nodes);
202
203 const self = try loop.allocator.create(SelfChannel{
204 .loop = loop,
205 .buffer_len = 0,
206 .buffer_nodes = buffer_nodes,
207 .buffer_index = 0,
208 .dispatch_lock = 0,
209 .need_dispatch = 0,
210 .getters = std.atomic.QueueMpsc(GetNode).init(),
211 .putters = std.atomic.QueueMpsc(PutNode).init(),
212 .get_count = 0,
213 .put_count = 0,
214 });
215 errdefer loop.allocator.destroy(self);
216
217 return self;
218 }
219
220 /// must be called when all calls to put and get have suspended and no more calls occur
221 pub fn destroy(self: *SelfChannel) void {
222 while (self.getters.get()) |get_node| {
223 cancel get_node.data.tick_node.data;
224 }
225 while (self.putters.get()) |put_node| {
226 cancel put_node.data.tick_node.data;
227 }
228 self.loop.allocator.free(self.buffer_nodes);
229 self.loop.allocator.destroy(self);
230 }
231
232 /// puts a data item in the channel. The promise completes when the value has been added to the
233 /// buffer, or in the case of a zero size buffer, when the item has been retrieved by a getter.
234 pub async fn put(self: *SelfChannel, data: T) void {
235 // TODO should be able to group memory allocation failure before first suspend point
236 // so that the async invocation catches it
237 var dispatch_tick_node_ptr: *Loop.NextTickNode = undefined;
238 _ = async self.dispatch(&dispatch_tick_node_ptr) catch unreachable;
239
240 suspend |handle| {
241 var my_tick_node = Loop.NextTickNode{
242 .next = undefined,
243 .data = handle,
244 };
245 var queue_node = std.atomic.QueueMpsc(PutNode).Node{
246 .data = PutNode{
247 .tick_node = &my_tick_node,
248 .data = data,
249 },
250 .next = undefined,
251 };
252 self.putters.put(&queue_node);
253 _ = @atomicRmw(usize, &self.put_count, AtomicRmwOp.Add, 1, AtomicOrder.SeqCst);
254
255 self.loop.onNextTick(dispatch_tick_node_ptr);
256 }
257 }
258
259 /// await this function to get an item from the channel. If the buffer is empty, the promise will
260 /// complete when the next item is put in the channel.
261 pub async fn get(self: *SelfChannel) T {
262 // TODO should be able to group memory allocation failure before first suspend point
263 // so that the async invocation catches it
264 var dispatch_tick_node_ptr: *Loop.NextTickNode = undefined;
265 _ = async self.dispatch(&dispatch_tick_node_ptr) catch unreachable;
266
267 // TODO integrate this function with named return values
268 // so we can get rid of this extra result copy
269 var result: T = undefined;
270 var debug_handle: usize = undefined;
271 suspend |handle| {
272 debug_handle = @ptrToInt(handle);
273 var my_tick_node = Loop.NextTickNode{
274 .next = undefined,
275 .data = handle,
276 };
277 var queue_node = std.atomic.QueueMpsc(GetNode).Node{
278 .data = GetNode{
279 .ptr = &result,
280 .tick_node = &my_tick_node,
281 },
282 .next = undefined,
283 };
284 self.getters.put(&queue_node);
285 _ = @atomicRmw(usize, &self.get_count, AtomicRmwOp.Add, 1, AtomicOrder.SeqCst);
286
287 self.loop.onNextTick(dispatch_tick_node_ptr);
288 }
289 return result;
290 }
291
292 async fn dispatch(self: *SelfChannel, tick_node_ptr: **Loop.NextTickNode) void {
293 // resumed by onNextTick
294 suspend |handle| {
295 var tick_node = Loop.NextTickNode{
296 .data = handle,
297 .next = undefined,
298 };
299 tick_node_ptr.* = &tick_node;
300 }
301
302 // set the "need dispatch" flag
303 _ = @atomicRmw(u8, &self.need_dispatch, AtomicRmwOp.Xchg, 1, AtomicOrder.SeqCst);
304
305 lock: while (true) {
306 // set the lock flag
307 const prev_lock = @atomicRmw(u8, &self.dispatch_lock, AtomicRmwOp.Xchg, 1, AtomicOrder.SeqCst);
308 if (prev_lock != 0) return;
309
310 // clear the need_dispatch flag since we're about to do it
311 _ = @atomicRmw(u8, &self.need_dispatch, AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst);
312
313 while (true) {
314 one_dispatch: {
315 // later we correct these extra subtractions
316 var get_count = @atomicRmw(usize, &self.get_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst);
317 var put_count = @atomicRmw(usize, &self.put_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst);
318
319 // transfer self.buffer to self.getters
320 while (self.buffer_len != 0) {
321 if (get_count == 0) break :one_dispatch;
322
323 const get_node = &self.getters.get().?.data;
324 get_node.ptr.* = self.buffer_nodes[self.buffer_index -% self.buffer_len];
325 self.loop.onNextTick(get_node.tick_node);
326 self.buffer_len -= 1;
327
328 get_count = @atomicRmw(usize, &self.get_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst);
329 }
330
331 // direct transfer self.putters to self.getters
332 while (get_count != 0 and put_count != 0) {
333 const get_node = &self.getters.get().?.data;
334 const put_node = &self.putters.get().?.data;
335
336 get_node.ptr.* = put_node.data;
337 self.loop.onNextTick(get_node.tick_node);
338 self.loop.onNextTick(put_node.tick_node);
339
340 get_count = @atomicRmw(usize, &self.get_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst);
341 put_count = @atomicRmw(usize, &self.put_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst);
342 }
343
344 // transfer self.putters to self.buffer
345 while (self.buffer_len != self.buffer_nodes.len and put_count != 0) {
346 const put_node = &self.putters.get().?.data;
347
348 self.buffer_nodes[self.buffer_index] = put_node.data;
349 self.loop.onNextTick(put_node.tick_node);
350 self.buffer_index +%= 1;
351 self.buffer_len += 1;
352
353 put_count = @atomicRmw(usize, &self.put_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst);
354 }
355 }
356
357 // undo the extra subtractions
358 _ = @atomicRmw(usize, &self.get_count, AtomicRmwOp.Add, 1, AtomicOrder.SeqCst);
359 _ = @atomicRmw(usize, &self.put_count, AtomicRmwOp.Add, 1, AtomicOrder.SeqCst);
360
361 // clear need-dispatch flag
362 const need_dispatch = @atomicRmw(u8, &self.need_dispatch, AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst);
363 if (need_dispatch != 0) continue;
364
365 const my_lock = @atomicRmw(u8, &self.dispatch_lock, AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst);
366 assert(my_lock != 0);
367
368 // we have to check again now that we unlocked
369 if (@atomicLoad(u8, &self.need_dispatch, AtomicOrder.SeqCst) != 0) continue :lock;
370
371 return;
372 }
373 }
374 }
375 };
376}
377
144pub async fn connect(loop: *Loop, _address: *const std.net.Address) !std.os.File {378pub async fn connect(loop: *Loop, _address: *const std.net.Address) !std.os.File {
145 var address = _address.*; // TODO https://github.com/ziglang/zig/issues/733379 var address = _address.*; // TODO https://github.com/ziglang/zig/issues/733
146380
...@@ -199,6 +433,7 @@ test "listen on a port, send bytes, receive bytes" {...@@ -199,6 +433,7 @@ test "listen on a port, send bytes, receive bytes" {
199 defer cancel p;433 defer cancel p;
200 loop.run();434 loop.run();
201}435}
436
202async fn doAsyncTest(loop: *Loop, address: *const std.net.Address) void {437async fn doAsyncTest(loop: *Loop, address: *const std.net.Address) void {
203 errdefer @panic("test failure");438 errdefer @panic("test failure");
204439
...@@ -211,3 +446,43 @@ async fn doAsyncTest(loop: *Loop, address: *const std.net.Address) void {...@@ -211,3 +446,43 @@ async fn doAsyncTest(loop: *Loop, address: *const std.net.Address) void {
211 assert(mem.eql(u8, msg, "hello from server\n"));446 assert(mem.eql(u8, msg, "hello from server\n"));
212 loop.stop();447 loop.stop();
213}448}
449
450test "std.event.Channel" {
451 var da = std.heap.DirectAllocator.init();
452 defer da.deinit();
453
454 const allocator = &da.allocator;
455
456 var loop = try Loop.init(allocator);
457 defer loop.deinit();
458
459 const channel = try Channel(i32).create(&loop, 0);
460 defer channel.destroy();
461
462 const handle = try async<allocator> testChannelGetter(&loop, channel);
463 defer cancel handle;
464
465 const putter = try async<allocator> testChannelPutter(channel);
466 defer cancel putter;
467
468 loop.run();
469}
470
471async fn testChannelGetter(loop: *Loop, channel: *Channel(i32)) void {
472 errdefer @panic("test failed");
473
474 const value1_promise = try async channel.get();
475 const value1 = await value1_promise;
476 assert(value1 == 1234);
477
478 const value2_promise = try async channel.get();
479 const value2 = await value2_promise;
480 assert(value2 == 4567);
481
482 loop.stop();
483}
484
485async fn testChannelPutter(channel: *Channel(i32)) void {
486 await (async channel.put(1234) catch @panic("out of memory"));
487 await (async channel.put(4567) catch @panic("out of memory"));
488}
std/fmt/index.zig+3
...@@ -130,6 +130,9 @@ pub fn formatType(...@@ -130,6 +130,9 @@ pub fn formatType(
130 try output(context, "error.");130 try output(context, "error.");
131 return output(context, @errorName(value));131 return output(context, @errorName(value));
132 },132 },
133 builtin.TypeId.Promise => {
134 return format(context, Errors, output, "promise@{x}", @ptrToInt(value));
135 },
133 builtin.TypeId.Pointer => |ptr_info| switch (ptr_info.size) {136 builtin.TypeId.Pointer => |ptr_info| switch (ptr_info.size) {
134 builtin.TypeInfo.Pointer.Size.One => switch (@typeInfo(ptr_info.child)) {137 builtin.TypeInfo.Pointer.Size.One => switch (@typeInfo(ptr_info.child)) {
135 builtin.TypeId.Array => |info| {138 builtin.TypeId.Array => |info| {
std/heap.zig+1
...@@ -38,6 +38,7 @@ fn cFree(self: *Allocator, old_mem: []u8) void {...@@ -38,6 +38,7 @@ fn cFree(self: *Allocator, old_mem: []u8) void {
38}38}
3939
40/// This allocator makes a syscall directly for every allocation and free.40/// This allocator makes a syscall directly for every allocation and free.
41/// TODO make this thread-safe. The windows implementation will need some atomics.
41pub const DirectAllocator = struct {42pub const DirectAllocator = struct {
42 allocator: Allocator,43 allocator: Allocator,
43 heap_handle: ?HeapHandle,44 heap_handle: ?HeapHandle,