authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2019-11-25 12:31:23-05:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2019-11-25 12:31:23-05:00
loga061ef42c15082b52ba027ac84f0c24b4d1b4a99
tree1d538cdd8f7028bacba5c79d1b32db6d16d3c297
parent5a98dd42b38b9188cfb96c9ab57dc91af923029f
parent7dba5ea9cfc197cb3a2bbc84d0bce20d8df5886a
signaturebadge-question-mark Signed by PGP key 4AEE18F83AFDEB23

Merge pull request #3761 from Vexu/event.fs

Update event.fs to new event loop

1 files changed, 683 insertions(+), 709 deletions(-)

lib/std/event/fs.zig+683-709
...@@ -9,6 +9,12 @@ const windows = os.windows;...@@ -9,6 +9,12 @@ const windows = os.windows;
9const Loop = event.Loop;9const Loop = event.Loop;
10const fd_t = os.fd_t;10const fd_t = os.fd_t;
11const File = std.fs.File;11const File = std.fs.File;
12const Allocator = mem.Allocator;
13
14//! TODO mege this with `std.fs`
15
16const global_event_loop = Loop.instance orelse
17 @compileError("std.event.fs currently only works with event-based I/O");
1218
13pub const RequestNode = std.atomic.Queue(Request).Node;19pub const RequestNode = std.atomic.Queue(Request).Node;
1420
...@@ -84,7 +90,7 @@ pub const Request = struct {...@@ -84,7 +90,7 @@ pub const Request = struct {
84pub const PWriteVError = error{OutOfMemory} || File.WriteError;90pub const PWriteVError = error{OutOfMemory} || File.WriteError;
8591
86/// data - just the inner references - must live until pwritev frame completes.92/// data - just the inner references - must live until pwritev frame completes.
87pub fn pwritev(loop: *Loop, fd: fd_t, data: []const []const u8, offset: usize) PWriteVError!void {93pub fn pwritev(allocator: *Allocator, fd: fd_t, data: []const []const u8, offset: usize) PWriteVError!void {
88 switch (builtin.os) {94 switch (builtin.os) {
89 .macosx,95 .macosx,
90 .linux,96 .linux,
...@@ -92,8 +98,8 @@ pub fn pwritev(loop: *Loop, fd: fd_t, data: []const []const u8, offset: usize) P...@@ -92,8 +98,8 @@ pub fn pwritev(loop: *Loop, fd: fd_t, data: []const []const u8, offset: usize) P
92 .netbsd,98 .netbsd,
93 .dragonfly,99 .dragonfly,
94 => {100 => {
95 const iovecs = try loop.allocator.alloc(os.iovec_const, data.len);101 const iovecs = try allocator.alloc(os.iovec_const, data.len);
96 defer loop.allocator.free(iovecs);102 defer allocator.free(iovecs);
97103
98 for (data) |buf, i| {104 for (data) |buf, i| {
99 iovecs[i] = os.iovec_const{105 iovecs[i] = os.iovec_const{
...@@ -102,31 +108,31 @@ pub fn pwritev(loop: *Loop, fd: fd_t, data: []const []const u8, offset: usize) P...@@ -102,31 +108,31 @@ pub fn pwritev(loop: *Loop, fd: fd_t, data: []const []const u8, offset: usize) P
102 };108 };
103 }109 }
104110
105 return pwritevPosix(loop, fd, iovecs, offset);111 return pwritevPosix(fd, iovecs, offset);
106 },112 },
107 .windows => {113 .windows => {
108 const data_copy = try std.mem.dupe(loop.allocator, []const u8, data);114 const data_copy = try std.mem.dupe(allocator, []const u8, data);
109 defer loop.allocator.free(data_copy);115 defer allocator.free(data_copy);
110 return pwritevWindows(loop, fd, data, offset);116 return pwritevWindows(fd, data, offset);
111 },117 },
112 else => @compileError("Unsupported OS"),118 else => @compileError("Unsupported OS"),
113 }119 }
114}120}
115121
116/// data must outlive the returned frame122/// data must outlive the returned frame
117pub fn pwritevWindows(loop: *Loop, fd: fd_t, data: []const []const u8, offset: usize) os.WindowsWriteError!void {123pub fn pwritevWindows(fd: fd_t, data: []const []const u8, offset: usize) os.WindowsWriteError!void {
118 if (data.len == 0) return;124 if (data.len == 0) return;
119 if (data.len == 1) return pwriteWindows(loop, fd, data[0], offset);125 if (data.len == 1) return pwriteWindows(fd, data[0], offset);
120126
121 // TODO do these in parallel127 // TODO do these in parallel
122 var off = offset;128 var off = offset;
123 for (data) |buf| {129 for (data) |buf| {
124 try pwriteWindows(loop, fd, buf, off);130 try pwriteWindows(fd, buf, off);
125 off += buf.len;131 off += buf.len;
126 }132 }
127}133}
128134
129pub fn pwriteWindows(loop: *Loop, fd: fd_t, data: []const u8, offset: u64) os.WindowsWriteError!void {135pub fn pwriteWindows(fd: fd_t, data: []const u8, offset: u64) os.WindowsWriteError!void {
130 var resume_node = Loop.ResumeNode.Basic{136 var resume_node = Loop.ResumeNode.Basic{
131 .base = Loop.ResumeNode{137 .base = Loop.ResumeNode{
132 .id = Loop.ResumeNode.Id.Basic,138 .id = Loop.ResumeNode.Id.Basic,
...@@ -141,9 +147,9 @@ pub fn pwriteWindows(loop: *Loop, fd: fd_t, data: []const u8, offset: u64) os.Wi...@@ -141,9 +147,9 @@ pub fn pwriteWindows(loop: *Loop, fd: fd_t, data: []const u8, offset: u64) os.Wi
141 },147 },
142 };148 };
143 // TODO only call create io completion port once per fd149 // TODO only call create io completion port once per fd
144 _ = windows.CreateIoCompletionPort(fd, loop.os_data.io_port, undefined, undefined);150 _ = windows.CreateIoCompletionPort(fd, global_event_loop.os_data.io_port, undefined, undefined);
145 loop.beginOneEvent();151 global_event_loop.beginOneEvent();
146 errdefer loop.finishOneEvent();152 errdefer global_event_loop.finishOneEvent();
147153
148 errdefer {154 errdefer {
149 _ = windows.kernel32.CancelIoEx(fd, &resume_node.base.overlapped);155 _ = windows.kernel32.CancelIoEx(fd, &resume_node.base.overlapped);
...@@ -166,12 +172,7 @@ pub fn pwriteWindows(loop: *Loop, fd: fd_t, data: []const u8, offset: u64) os.Wi...@@ -166,12 +172,7 @@ pub fn pwriteWindows(loop: *Loop, fd: fd_t, data: []const u8, offset: u64) os.Wi
166}172}
167173
168/// iovecs must live until pwritev frame completes.174/// iovecs must live until pwritev frame completes.
169pub fn pwritevPosix(175pub fn pwritevPosix(fd: fd_t, iovecs: []const os.iovec_const, offset: usize) os.WriteError!void {
170 loop: *Loop,
171 fd: fd_t,
172 iovecs: []const os.iovec_const,
173 offset: usize,
174) os.WriteError!void {
175 var req_node = RequestNode{176 var req_node = RequestNode{
176 .prev = null,177 .prev = null,
177 .next = null,178 .next = null,
...@@ -194,21 +195,17 @@ pub fn pwritevPosix(...@@ -194,21 +195,17 @@ pub fn pwritevPosix(
194 },195 },
195 };196 };
196197
197 errdefer loop.posixFsCancel(&req_node);198 errdefer global_event_loop.posixFsCancel(&req_node);
198199
199 suspend {200 suspend {
200 loop.posixFsRequest(&req_node);201 global_event_loop.posixFsRequest(&req_node);
201 }202 }
202203
203 return req_node.data.msg.PWriteV.result;204 return req_node.data.msg.PWriteV.result;
204}205}
205206
206/// iovecs must live until pwritev frame completes.207/// iovecs must live until pwritev frame completes.
207pub fn writevPosix(208pub fn writevPosix(fd: fd_t, iovecs: []const os.iovec_const) os.WriteError!void {
208 loop: *Loop,
209 fd: fd_t,
210 iovecs: []const os.iovec_const,
211) os.WriteError!void {
212 var req_node = RequestNode{209 var req_node = RequestNode{
213 .prev = null,210 .prev = null,
214 .next = null,211 .next = null,
...@@ -231,7 +228,7 @@ pub fn writevPosix(...@@ -231,7 +228,7 @@ pub fn writevPosix(
231 };228 };
232229
233 suspend {230 suspend {
234 loop.posixFsRequest(&req_node);231 global_event_loop.posixFsRequest(&req_node);
235 }232 }
236233
237 return req_node.data.msg.WriteV.result;234 return req_node.data.msg.WriteV.result;
...@@ -240,7 +237,7 @@ pub fn writevPosix(...@@ -240,7 +237,7 @@ pub fn writevPosix(
240pub const PReadVError = error{OutOfMemory} || File.ReadError;237pub const PReadVError = error{OutOfMemory} || File.ReadError;
241238
242/// data - just the inner references - must live until preadv frame completes.239/// data - just the inner references - must live until preadv frame completes.
243pub fn preadv(loop: *Loop, fd: fd_t, data: []const []u8, offset: usize) PReadVError!usize {240pub fn preadv(allocator: *Allocator, fd: fd_t, data: []const []u8, offset: usize) PReadVError!usize {
244 assert(data.len != 0);241 assert(data.len != 0);
245 switch (builtin.os) {242 switch (builtin.os) {
246 .macosx,243 .macosx,
...@@ -249,8 +246,8 @@ pub fn preadv(loop: *Loop, fd: fd_t, data: []const []u8, offset: usize) PReadVEr...@@ -249,8 +246,8 @@ pub fn preadv(loop: *Loop, fd: fd_t, data: []const []u8, offset: usize) PReadVEr
249 .netbsd,246 .netbsd,
250 .dragonfly,247 .dragonfly,
251 => {248 => {
252 const iovecs = try loop.allocator.alloc(os.iovec, data.len);249 const iovecs = try allocator.alloc(os.iovec, data.len);
253 defer loop.allocator.free(iovecs);250 defer allocator.free(iovecs);
254251
255 for (data) |buf, i| {252 for (data) |buf, i| {
256 iovecs[i] = os.iovec{253 iovecs[i] = os.iovec{
...@@ -259,21 +256,21 @@ pub fn preadv(loop: *Loop, fd: fd_t, data: []const []u8, offset: usize) PReadVEr...@@ -259,21 +256,21 @@ pub fn preadv(loop: *Loop, fd: fd_t, data: []const []u8, offset: usize) PReadVEr
259 };256 };
260 }257 }
261258
262 return preadvPosix(loop, fd, iovecs, offset);259 return preadvPosix(fd, iovecs, offset);
263 },260 },
264 .windows => {261 .windows => {
265 const data_copy = try std.mem.dupe(loop.allocator, []u8, data);262 const data_copy = try std.mem.dupe(allocator, []u8, data);
266 defer loop.allocator.free(data_copy);263 defer allocator.free(data_copy);
267 return preadvWindows(loop, fd, data_copy, offset);264 return preadvWindows(fd, data_copy, offset);
268 },265 },
269 else => @compileError("Unsupported OS"),266 else => @compileError("Unsupported OS"),
270 }267 }
271}268}
272269
273/// data must outlive the returned frame270/// data must outlive the returned frame
274pub fn preadvWindows(loop: *Loop, fd: fd_t, data: []const []u8, offset: u64) !usize {271pub fn preadvWindows(fd: fd_t, data: []const []u8, offset: u64) !usize {
275 assert(data.len != 0);272 assert(data.len != 0);
276 if (data.len == 1) return preadWindows(loop, fd, data[0], offset);273 if (data.len == 1) return preadWindows(fd, data[0], offset);
277274
278 // TODO do these in parallel?275 // TODO do these in parallel?
279 var off: usize = 0;276 var off: usize = 0;
...@@ -281,7 +278,7 @@ pub fn preadvWindows(loop: *Loop, fd: fd_t, data: []const []u8, offset: u64) !us...@@ -281,7 +278,7 @@ pub fn preadvWindows(loop: *Loop, fd: fd_t, data: []const []u8, offset: u64) !us
281 var inner_off: usize = 0;278 var inner_off: usize = 0;
282 while (true) {279 while (true) {
283 const v = data[iov_i];280 const v = data[iov_i];
284 const amt_read = try preadWindows(loop, fd, v[inner_off .. v.len - inner_off], offset + off);281 const amt_read = try preadWindows(fd, v[inner_off .. v.len - inner_off], offset + off);
285 off += amt_read;282 off += amt_read;
286 inner_off += amt_read;283 inner_off += amt_read;
287 if (inner_off == v.len) {284 if (inner_off == v.len) {
...@@ -295,7 +292,7 @@ pub fn preadvWindows(loop: *Loop, fd: fd_t, data: []const []u8, offset: u64) !us...@@ -295,7 +292,7 @@ pub fn preadvWindows(loop: *Loop, fd: fd_t, data: []const []u8, offset: u64) !us
295 }292 }
296}293}
297294
298pub fn preadWindows(loop: *Loop, fd: fd_t, data: []u8, offset: u64) !usize {295pub fn preadWindows(fd: fd_t, data: []u8, offset: u64) !usize {
299 var resume_node = Loop.ResumeNode.Basic{296 var resume_node = Loop.ResumeNode.Basic{
300 .base = Loop.ResumeNode{297 .base = Loop.ResumeNode{
301 .id = Loop.ResumeNode.Id.Basic,298 .id = Loop.ResumeNode.Id.Basic,
...@@ -310,9 +307,9 @@ pub fn preadWindows(loop: *Loop, fd: fd_t, data: []u8, offset: u64) !usize {...@@ -310,9 +307,9 @@ pub fn preadWindows(loop: *Loop, fd: fd_t, data: []u8, offset: u64) !usize {
310 },307 },
311 };308 };
312 // TODO only call create io completion port once per fd309 // TODO only call create io completion port once per fd
313 _ = windows.CreateIoCompletionPort(fd, loop.os_data.io_port, undefined, undefined) catch undefined;310 _ = windows.CreateIoCompletionPort(fd, global_event_loop.os_data.io_port, undefined, undefined) catch undefined;
314 loop.beginOneEvent();311 global_event_loop.beginOneEvent();
315 errdefer loop.finishOneEvent();312 errdefer global_event_loop.finishOneEvent();
316313
317 errdefer {314 errdefer {
318 _ = windows.kernel32.CancelIoEx(fd, &resume_node.base.overlapped);315 _ = windows.kernel32.CancelIoEx(fd, &resume_node.base.overlapped);
...@@ -334,12 +331,7 @@ pub fn preadWindows(loop: *Loop, fd: fd_t, data: []u8, offset: u64) !usize {...@@ -334,12 +331,7 @@ pub fn preadWindows(loop: *Loop, fd: fd_t, data: []u8, offset: u64) !usize {
334}331}
335332
336/// iovecs must live until preadv frame completes333/// iovecs must live until preadv frame completes
337pub fn preadvPosix(334pub fn preadvPosix(fd: fd_t, iovecs: []const os.iovec, offset: usize) os.ReadError!usize {
338 loop: *Loop,
339 fd: fd_t,
340 iovecs: []const os.iovec,
341 offset: usize,
342) os.ReadError!usize {
343 var req_node = RequestNode{335 var req_node = RequestNode{
344 .prev = null,336 .prev = null,
345 .next = null,337 .next = null,
...@@ -362,21 +354,16 @@ pub fn preadvPosix(...@@ -362,21 +354,16 @@ pub fn preadvPosix(
362 },354 },
363 };355 };
364356
365 errdefer loop.posixFsCancel(&req_node);357 errdefer global_event_loop.posixFsCancel(&req_node);
366358
367 suspend {359 suspend {
368 loop.posixFsRequest(&req_node);360 global_event_loop.posixFsRequest(&req_node);
369 }361 }
370362
371 return req_node.data.msg.PReadV.result;363 return req_node.data.msg.PReadV.result;
372}364}
373365
374pub fn openPosix(366pub fn openPosix(path: []const u8, flags: u32, mode: File.Mode) File.OpenError!fd_t {
375 loop: *Loop,
376 path: []const u8,
377 flags: u32,
378 mode: File.Mode,
379) File.OpenError!fd_t {
380 const path_c = try std.os.toPosixPath(path);367 const path_c = try std.os.toPosixPath(path);
381368
382 var req_node = RequestNode{369 var req_node = RequestNode{
...@@ -401,21 +388,21 @@ pub fn openPosix(...@@ -401,21 +388,21 @@ pub fn openPosix(
401 },388 },
402 };389 };
403390
404 errdefer loop.posixFsCancel(&req_node);391 errdefer global_event_loop.posixFsCancel(&req_node);
405392
406 suspend {393 suspend {
407 loop.posixFsRequest(&req_node);394 global_event_loop.posixFsRequest(&req_node);
408 }395 }
409396
410 return req_node.data.msg.Open.result;397 return req_node.data.msg.Open.result;
411}398}
412399
413pub fn openRead(loop: *Loop, path: []const u8) File.OpenError!fd_t {400pub fn openRead(path: []const u8) File.OpenError!fd_t {
414 switch (builtin.os) {401 switch (builtin.os) {
415 .macosx, .linux, .freebsd, .netbsd, .dragonfly => {402 .macosx, .linux, .freebsd, .netbsd, .dragonfly => {
416 const O_LARGEFILE = if (@hasDecl(os, "O_LARGEFILE")) os.O_LARGEFILE else 0;403 const O_LARGEFILE = if (@hasDecl(os, "O_LARGEFILE")) os.O_LARGEFILE else 0;
417 const flags = O_LARGEFILE | os.O_RDONLY | os.O_CLOEXEC;404 const flags = O_LARGEFILE | os.O_RDONLY | os.O_CLOEXEC;
418 return openPosix(loop, path, flags, File.default_mode);405 return openPosix(path, flags, File.default_mode);
419 },406 },
420407
421 .windows => return windows.CreateFile(408 .windows => return windows.CreateFile(
...@@ -434,12 +421,12 @@ pub fn openRead(loop: *Loop, path: []const u8) File.OpenError!fd_t {...@@ -434,12 +421,12 @@ pub fn openRead(loop: *Loop, path: []const u8) File.OpenError!fd_t {
434421
435/// Creates if does not exist. Truncates the file if it exists.422/// Creates if does not exist. Truncates the file if it exists.
436/// Uses the default mode.423/// Uses the default mode.
437pub fn openWrite(loop: *Loop, path: []const u8) File.OpenError!fd_t {424pub fn openWrite(path: []const u8) File.OpenError!fd_t {
438 return openWriteMode(loop, path, File.default_mode);425 return openWriteMode(path, File.default_mode);
439}426}
440427
441/// Creates if does not exist. Truncates the file if it exists.428/// Creates if does not exist. Truncates the file if it exists.
442pub fn openWriteMode(loop: *Loop, path: []const u8, mode: File.Mode) File.OpenError!fd_t {429pub fn openWriteMode(path: []const u8, mode: File.Mode) File.OpenError!fd_t {
443 switch (builtin.os) {430 switch (builtin.os) {
444 .macosx,431 .macosx,
445 .linux,432 .linux,
...@@ -449,7 +436,7 @@ pub fn openWriteMode(loop: *Loop, path: []const u8, mode: File.Mode) File.OpenEr...@@ -449,7 +436,7 @@ pub fn openWriteMode(loop: *Loop, path: []const u8, mode: File.Mode) File.OpenEr
449 => {436 => {
450 const O_LARGEFILE = if (@hasDecl(os, "O_LARGEFILE")) os.O_LARGEFILE else 0;437 const O_LARGEFILE = if (@hasDecl(os, "O_LARGEFILE")) os.O_LARGEFILE else 0;
451 const flags = O_LARGEFILE | os.O_WRONLY | os.O_CREAT | os.O_CLOEXEC | os.O_TRUNC;438 const flags = O_LARGEFILE | os.O_WRONLY | os.O_CREAT | os.O_CLOEXEC | os.O_TRUNC;
452 return openPosix(loop, path, flags, File.default_mode);439 return openPosix(path, flags, File.default_mode);
453 },440 },
454 .windows => return windows.CreateFile(441 .windows => return windows.CreateFile(
455 path,442 path,
...@@ -465,16 +452,12 @@ pub fn openWriteMode(loop: *Loop, path: []const u8, mode: File.Mode) File.OpenEr...@@ -465,16 +452,12 @@ pub fn openWriteMode(loop: *Loop, path: []const u8, mode: File.Mode) File.OpenEr
465}452}
466453
467/// Creates if does not exist. Does not truncate.454/// Creates if does not exist. Does not truncate.
468pub fn openReadWrite(455pub fn openReadWrite(path: []const u8, mode: File.Mode) File.OpenError!fd_t {
469 loop: *Loop,
470 path: []const u8,
471 mode: File.Mode,
472) File.OpenError!fd_t {
473 switch (builtin.os) {456 switch (builtin.os) {
474 .macosx, .linux, .freebsd, .netbsd, .dragonfly => {457 .macosx, .linux, .freebsd, .netbsd, .dragonfly => {
475 const O_LARGEFILE = if (@hasDecl(os, "O_LARGEFILE")) os.O_LARGEFILE else 0;458 const O_LARGEFILE = if (@hasDecl(os, "O_LARGEFILE")) os.O_LARGEFILE else 0;
476 const flags = O_LARGEFILE | os.O_RDWR | os.O_CREAT | os.O_CLOEXEC;459 const flags = O_LARGEFILE | os.O_RDWR | os.O_CREAT | os.O_CLOEXEC;
477 return openPosix(loop, path, flags, mode);460 return openPosix(path, flags, mode);
478 },461 },
479462
480 .windows => return windows.CreateFile(463 .windows => return windows.CreateFile(
...@@ -498,7 +481,7 @@ pub fn openReadWrite(...@@ -498,7 +481,7 @@ pub fn openReadWrite(
498/// If you call `setHandle` then finishing will close the fd; otherwise finishing481/// If you call `setHandle` then finishing will close the fd; otherwise finishing
499/// will deallocate the `CloseOperation`.482/// will deallocate the `CloseOperation`.
500pub const CloseOperation = struct {483pub const CloseOperation = struct {
501 loop: *Loop,484 allocator: *Allocator,
502 os_data: OsData,485 os_data: OsData,
503486
504 const OsData = switch (builtin.os) {487 const OsData = switch (builtin.os) {
...@@ -516,10 +499,10 @@ pub const CloseOperation = struct {...@@ -516,10 +499,10 @@ pub const CloseOperation = struct {
516 close_req_node: RequestNode,499 close_req_node: RequestNode,
517 };500 };
518501
519 pub fn start(loop: *Loop) (error{OutOfMemory}!*CloseOperation) {502 pub fn start(allocator: *Allocator) (error{OutOfMemory}!*CloseOperation) {
520 const self = try loop.allocator.create(CloseOperation);503 const self = try allocator.create(CloseOperation);
521 self.* = CloseOperation{504 self.* = CloseOperation{
522 .loop = loop,505 .allocator = allocator,
523 .os_data = switch (builtin.os) {506 .os_data = switch (builtin.os) {
524 .linux, .macosx, .freebsd, .netbsd, .dragonfly => initOsDataPosix(self),507 .linux, .macosx, .freebsd, .netbsd, .dragonfly => initOsDataPosix(self),
525 .windows => OsData{ .handle = null },508 .windows => OsData{ .handle = null },
...@@ -555,16 +538,16 @@ pub const CloseOperation = struct {...@@ -555,16 +538,16 @@ pub const CloseOperation = struct {
555 .dragonfly,538 .dragonfly,
556 => {539 => {
557 if (self.os_data.have_fd) {540 if (self.os_data.have_fd) {
558 self.loop.posixFsRequest(&self.os_data.close_req_node);541 global_event_loop.posixFsRequest(&self.os_data.close_req_node);
559 } else {542 } else {
560 self.loop.allocator.destroy(self);543 self.allocator.destroy(self);
561 }544 }
562 },545 },
563 .windows => {546 .windows => {
564 if (self.os_data.handle) |handle| {547 if (self.os_data.handle) |handle| {
565 os.close(handle);548 os.close(handle);
566 }549 }
567 self.loop.allocator.destroy(self);550 self.allocator.destroy(self);
568 },551 },
569 else => @compileError("Unsupported OS"),552 else => @compileError("Unsupported OS"),
570 }553 }
...@@ -627,25 +610,25 @@ pub const CloseOperation = struct {...@@ -627,25 +610,25 @@ pub const CloseOperation = struct {
627610
628/// contents must remain alive until writeFile completes.611/// contents must remain alive until writeFile completes.
629/// TODO make this atomic or provide writeFileAtomic and rename this one to writeFileTruncate612/// TODO make this atomic or provide writeFileAtomic and rename this one to writeFileTruncate
630pub fn writeFile(loop: *Loop, path: []const u8, contents: []const u8) !void {613pub fn writeFile(allocator: *Allocator, path: []const u8, contents: []const u8) !void {
631 return writeFileMode(loop, path, contents, File.default_mode);614 return writeFileMode(allocator, path, contents, File.default_mode);
632}615}
633616
634/// contents must remain alive until writeFile completes.617/// contents must remain alive until writeFile completes.
635pub fn writeFileMode(loop: *Loop, path: []const u8, contents: []const u8, mode: File.Mode) !void {618pub fn writeFileMode(allocator: *Allocator, path: []const u8, contents: []const u8, mode: File.Mode) !void {
636 switch (builtin.os) {619 switch (builtin.os) {
637 .linux,620 .linux,
638 .macosx,621 .macosx,
639 .freebsd,622 .freebsd,
640 .netbsd,623 .netbsd,
641 .dragonfly,624 .dragonfly,
642 => return writeFileModeThread(loop, path, contents, mode),625 => return writeFileModeThread(allocator, path, contents, mode),
643 .windows => return writeFileWindows(loop, path, contents),626 .windows => return writeFileWindows(path, contents),
644 else => @compileError("Unsupported OS"),627 else => @compileError("Unsupported OS"),
645 }628 }
646}629}
647630
648fn writeFileWindows(loop: *Loop, path: []const u8, contents: []const u8) !void {631fn writeFileWindows(path: []const u8, contents: []const u8) !void {
649 const handle = try windows.CreateFile(632 const handle = try windows.CreateFile(
650 path,633 path,
651 windows.GENERIC_WRITE,634 windows.GENERIC_WRITE,
...@@ -657,12 +640,12 @@ fn writeFileWindows(loop: *Loop, path: []const u8, contents: []const u8) !void {...@@ -657,12 +640,12 @@ fn writeFileWindows(loop: *Loop, path: []const u8, contents: []const u8) !void {
657 );640 );
658 defer os.close(handle);641 defer os.close(handle);
659642
660 try pwriteWindows(loop, handle, contents, 0);643 try pwriteWindows(handle, contents, 0);
661}644}
662645
663fn writeFileModeThread(loop: *Loop, path: []const u8, contents: []const u8, mode: File.Mode) !void {646fn writeFileModeThread(allocator: *Allocator, path: []const u8, contents: []const u8, mode: File.Mode) !void {
664 const path_with_null = try std.cstr.addNullByte(loop.allocator, path);647 const path_with_null = try std.cstr.addNullByte(allocator, path);
665 defer loop.allocator.free(path_with_null);648 defer allocator.free(path_with_null);
666649
667 var req_node = RequestNode{650 var req_node = RequestNode{
668 .prev = null,651 .prev = null,
...@@ -686,10 +669,10 @@ fn writeFileModeThread(loop: *Loop, path: []const u8, contents: []const u8, mode...@@ -686,10 +669,10 @@ fn writeFileModeThread(loop: *Loop, path: []const u8, contents: []const u8, mode
686 },669 },
687 };670 };
688671
689 errdefer loop.posixFsCancel(&req_node);672 errdefer global_event_loop.posixFsCancel(&req_node);
690673
691 suspend {674 suspend {
692 loop.posixFsRequest(&req_node);675 global_event_loop.posixFsRequest(&req_node);
693 }676 }
694677
695 return req_node.data.msg.WriteFile.result;678 return req_node.data.msg.WriteFile.result;
...@@ -698,21 +681,21 @@ fn writeFileModeThread(loop: *Loop, path: []const u8, contents: []const u8, mode...@@ -698,21 +681,21 @@ fn writeFileModeThread(loop: *Loop, path: []const u8, contents: []const u8, mode
698/// The frame resumes when the last data has been confirmed written, but before the file handle681/// The frame resumes when the last data has been confirmed written, but before the file handle
699/// is closed.682/// is closed.
700/// Caller owns returned memory.683/// Caller owns returned memory.
701pub fn readFile(loop: *Loop, file_path: []const u8, max_size: usize) ![]u8 {684pub fn readFile(allocator: *Allocator, file_path: []const u8, max_size: usize) ![]u8 {
702 var close_op = try CloseOperation.start(loop);685 var close_op = try CloseOperation.start(allocator);
703 defer close_op.finish();686 defer close_op.finish();
704687
705 const fd = try openRead(loop, file_path);688 const fd = try openRead(file_path);
706 close_op.setHandle(fd);689 close_op.setHandle(fd);
707690
708 var list = std.ArrayList(u8).init(loop.allocator);691 var list = std.ArrayList(u8).init(allocator);
709 defer list.deinit();692 defer list.deinit();
710693
711 while (true) {694 while (true) {
712 try list.ensureCapacity(list.len + mem.page_size);695 try list.ensureCapacity(list.len + mem.page_size);
713 const buf = list.items[list.len..];696 const buf = list.items[list.len..];
714 const buf_array = [_][]u8{buf};697 const buf_array = [_][]u8{buf};
715 const amt = try preadv(loop, fd, buf_array, list.len);698 const amt = try preadv(allocator, fd, buf_array, list.len);
716 list.len += amt;699 list.len += amt;
717 if (list.len > max_size) {700 if (list.len > max_size) {
718 return error.FileTooBig;701 return error.FileTooBig;
...@@ -738,610 +721,603 @@ fn hashString(s: []const u16) u32 {...@@ -738,610 +721,603 @@ fn hashString(s: []const u16) u32 {
738 return @truncate(u32, std.hash.Wyhash.hash(0, @sliceToBytes(s)));721 return @truncate(u32, std.hash.Wyhash.hash(0, @sliceToBytes(s)));
739}722}
740723
741//pub const WatchEventError = error{724pub const WatchEventError = error{
742// UserResourceLimitReached,725 UserResourceLimitReached,
743// SystemResources,726 SystemResources,
744// AccessDenied,727 AccessDenied,
745// Unexpected, // TODO remove this possibility728 Unexpected, // TODO remove this possibility
746//};729};
747//730
748//pub fn Watch(comptime V: type) type {731pub fn Watch(comptime V: type) type {
749// return struct {732 return struct {
750// channel: *event.Channel(Event.Error!Event),733 channel: *event.Channel(Event.Error!Event),
751// os_data: OsData,734 os_data: OsData,
752//735 allocator: *Allocator,
753// const OsData = switch (builtin.os) {736
754// .macosx, .freebsd, .netbsd, .dragonfly => struct {737 const OsData = switch (builtin.os) {
755// file_table: FileTable,738 .macosx, .freebsd, .netbsd, .dragonfly => struct {
756// table_lock: event.Lock,739 file_table: FileTable,
757//740 table_lock: event.Lock,
758// const FileTable = std.StringHashmap(*Put);741
759// const Put = struct {742 const FileTable = std.StringHashMap(*Put);
760// putter: anyframe,743 const Put = struct {
761// value_ptr: *V,744 putter_frame: @Frame(kqPutEvents),
762// };745 cancelled: bool = false,
763// },746 value: V,
764//747 };
765// .linux => LinuxOsData,748 },
766// .windows => WindowsOsData,749
767//750 .linux => LinuxOsData,
768// else => @compileError("Unsupported OS"),751 .windows => WindowsOsData,
769// };752
770//753 else => @compileError("Unsupported OS"),
771// const WindowsOsData = struct {754 };
772// table_lock: event.Lock,755
773// dir_table: DirTable,756 const WindowsOsData = struct {
774// all_putters: std.atomic.Queue(anyframe),757 table_lock: event.Lock,
775// ref_count: std.atomic.Int(usize),758 dir_table: DirTable,
776//759 all_putters: std.atomic.Queue(Put),
777// const DirTable = std.StringHashMap(*Dir);760 ref_count: std.atomic.Int(usize),
778// const FileTable = std.HashMap([]const u16, V, hashString, eqlString);761
779//762 const Put = struct {
780// const Dir = struct {763 putter: anyframe,
781// putter: anyframe,764 cancelled: bool = false,
782// file_table: FileTable,765 };
783// table_lock: event.Lock,766
784// };767 const DirTable = std.StringHashMap(*Dir);
785// };768 const FileTable = std.HashMap([]const u16, V, hashString, eqlString);
786//769
787// const LinuxOsData = struct {770 const Dir = struct {
788// putter: anyframe,771 putter_frame: @Frame(windowsDirReader),
789// inotify_fd: i32,772 file_table: FileTable,
790// wd_table: WdTable,773 table_lock: event.Lock,
791// table_lock: event.Lock,774 };
792//775 };
793// const WdTable = std.AutoHashMap(i32, Dir);776
794// const FileTable = std.StringHashMap(V);777 const LinuxOsData = struct {
795//778 putter_frame: @Frame(linuxEventPutter),
796// const Dir = struct {779 inotify_fd: i32,
797// dirname: []const u8,780 wd_table: WdTable,
798// file_table: FileTable,781 table_lock: event.Lock,
799// };782 cancelled: bool = false,
800// };783
801//784 const WdTable = std.AutoHashMap(i32, Dir);
802// const FileToHandle = std.StringHashMap(anyframe);785 const FileTable = std.StringHashMap(V);
803//786
804// const Self = @This();787 const Dir = struct {
805//788 dirname: []const u8,
806// pub const Event = struct {789 file_table: FileTable,
807// id: Id,790 };
808// data: V,791 };
809//792
810// pub const Id = WatchEventId;793 const Self = @This();
811// pub const Error = WatchEventError;794
812// };795 pub const Event = struct {
813//796 id: Id,
814// pub fn create(loop: *Loop, event_buf_count: usize) !*Self {797 data: V,
815// const channel = try event.Channel(Self.Event.Error!Self.Event).create(loop, event_buf_count);798
816// errdefer channel.destroy();799 pub const Id = WatchEventId;
817//800 pub const Error = WatchEventError;
818// switch (builtin.os) {801 };
819// .linux => {802
820// const inotify_fd = try os.inotify_init1(os.linux.IN_NONBLOCK | os.linux.IN_CLOEXEC);803 pub fn init(allocator: *Allocator, event_buf_count: usize) !*Self {
821// errdefer os.close(inotify_fd);804 const channel = try allocator.create(event.Channel(Event.Error!Event));
822//805 errdefer allocator.destroy(channel);
823// var result: *Self = undefined;806 var buf = try allocator.alloc(Event.Error!Event, event_buf_count);
824// _ = try async<loop.allocator> linuxEventPutter(inotify_fd, channel, &result);807 errdefer allocator.free(buf);
825// return result;808 channel.init(buf);
826// },809 errdefer channel.deinit();
827//810
828// .windows => {811 const self = try allocator.create(Self);
829// const self = try loop.allocator.create(Self);812 errdefer allocator.destroy(self);
830// errdefer loop.allocator.destroy(self);813
831// self.* = Self{814 switch (builtin.os) {
832// .channel = channel,815 .linux => {
833// .os_data = OsData{816 const inotify_fd = try os.inotify_init1(os.linux.IN_NONBLOCK | os.linux.IN_CLOEXEC);
834// .table_lock = event.Lock.init(loop),817 errdefer os.close(inotify_fd);
835// .dir_table = OsData.DirTable.init(loop.allocator),818
836// .ref_count = std.atomic.Int(usize).init(1),819 self.* = Self{
837// .all_putters = std.atomic.Queue(anyframe).init(),820 .allocator = allocator,
838// },821 .channel = channel,
839// };822 .os_data = OsData{
840// return self;823 .putter_frame = undefined,
841// },824 .inotify_fd = inotify_fd,
842//825 .wd_table = OsData.WdTable.init(allocator),
843// .macosx, .freebsd, .netbsd, .dragonfly => {826 .table_lock = event.Lock.init(),
844// const self = try loop.allocator.create(Self);827 },
845// errdefer loop.allocator.destroy(self);828 };
846//829
847// self.* = Self{830 self.os_data.putter_frame = async self.linuxEventPutter();
848// .channel = channel,831 return self;
849// .os_data = OsData{832 },
850// .table_lock = event.Lock.init(loop),833
851// .file_table = OsData.FileTable.init(loop.allocator),834 .windows => {
852// },835 self.* = Self{
853// };836 .allocator = allocator,
854// return self;837 .channel = channel,
855// },838 .os_data = OsData{
856// else => @compileError("Unsupported OS"),839 .table_lock = event.Lock.init(),
857// }840 .dir_table = OsData.DirTable.init(allocator),
858// }841 .ref_count = std.atomic.Int(usize).init(1),
859//842 .all_putters = std.atomic.Queue(anyframe).init(),
860// /// All addFile calls and removeFile calls must have completed.843 },
861// pub fn destroy(self: *Self) void {844 };
862// switch (builtin.os) {845 return self;
863// .macosx, .freebsd, .netbsd, .dragonfly => {846 },
864// // TODO we need to cancel the frames before destroying the lock847
865// self.os_data.table_lock.deinit();848 .macosx, .freebsd, .netbsd, .dragonfly => {
866// var it = self.os_data.file_table.iterator();849 self.* = Self{
867// while (it.next()) |entry| {850 .allocator = allocator,
868// cancel entry.value.putter;851 .channel = channel,
869// self.channel.loop.allocator.free(entry.key);852 .os_data = OsData{
870// }853 .table_lock = event.Lock.init(),
871// self.channel.destroy();854 .file_table = OsData.FileTable.init(allocator),
872// },855 },
873// .linux => cancel self.os_data.putter,856 };
874// .windows => {857 return self;
875// while (self.os_data.all_putters.get()) |putter_node| {858 },
876// cancel putter_node.data;859 else => @compileError("Unsupported OS"),
877// }860 }
878// self.deref();861 }
879// },862
880// else => @compileError("Unsupported OS"),863 /// All addFile calls and removeFile calls must have completed.
881// }864 pub fn deinit(self: *Self) void {
882// }865 switch (builtin.os) {
883//866 .macosx, .freebsd, .netbsd, .dragonfly => {
884// fn ref(self: *Self) void {867 // TODO we need to cancel the frames before destroying the lock
885// _ = self.os_data.ref_count.incr();868 self.os_data.table_lock.deinit();
886// }869 var it = self.os_data.file_table.iterator();
887//870 while (it.next()) |entry| {
888// fn deref(self: *Self) void {871 entry.cancelled = true;
889// if (self.os_data.ref_count.decr() == 1) {872 await entry.value.putter;
890// const allocator = self.channel.loop.allocator;873 self.allocator.free(entry.key);
891// self.os_data.table_lock.deinit();874 self.allocator.free(entry.value);
892// var it = self.os_data.dir_table.iterator();875 }
893// while (it.next()) |entry| {876 self.channel.deinit();
894// allocator.free(entry.key);877 self.allocator.destroy(self.channel.buffer_nodes);
895// allocator.destroy(entry.value);878 self.allocator.destroy(self);
896// }879 },
897// self.os_data.dir_table.deinit();880 .linux => {
898// self.channel.destroy();881 self.os_data.cancelled = true;
899// allocator.destroy(self);882 await self.os_data.putter_frame;
900// }883 self.allocator.destroy(self);
901// }884 },
902//885 .windows => {
903// pub async fn addFile(self: *Self, file_path: []const u8, value: V) !?V {886 while (self.os_data.all_putters.get()) |putter_node| {
904// switch (builtin.os) {887 putter_node.cancelled = true;
905// .macosx, .freebsd, .netbsd, .dragonfly => return await (async addFileKEvent(self, file_path, value) catch unreachable),888 await putter_node.frame;
906// .linux => return await (async addFileLinux(self, file_path, value) catch unreachable),889 }
907// .windows => return await (async addFileWindows(self, file_path, value) catch unreachable),890 self.deref();
908// else => @compileError("Unsupported OS"),891 },
909// }892 else => @compileError("Unsupported OS"),
910// }893 }
911//894 }
912// async fn addFileKEvent(self: *Self, file_path: []const u8, value: V) !?V {895
913// const resolved_path = try std.fs.path.resolve(self.channel.loop.allocator, [_][]const u8{file_path});896 fn ref(self: *Self) void {
914// var resolved_path_consumed = false;897 _ = self.os_data.ref_count.incr();
915// defer if (!resolved_path_consumed) self.channel.loop.allocator.free(resolved_path);898 }
916//899
917// var close_op = try CloseOperation.start(self.channel.loop);900 fn deref(self: *Self) void {
918// var close_op_consumed = false;901 if (self.os_data.ref_count.decr() == 1) {
919// defer if (!close_op_consumed) close_op.finish();902 self.os_data.table_lock.deinit();
920//903 var it = self.os_data.dir_table.iterator();
921// const flags = if (comptime std.Target.current.isDarwin()) os.O_SYMLINK | os.O_EVTONLY else 0;904 while (it.next()) |entry| {
922// const mode = 0;905 self.allocator.free(entry.key);
923// const fd = try await (async openPosix(self.channel.loop, resolved_path, flags, mode) catch unreachable);906 self.allocator.destroy(entry.value);
924// close_op.setHandle(fd);907 }
925//908 self.os_data.dir_table.deinit();
926// var put_data: *OsData.Put = undefined;909 self.channel.deinit();
927// const putter = try async self.kqPutEvents(close_op, value, &put_data);910 self.allocator.destroy(self.channel.buffer_nodes);
928// close_op_consumed = true;911 self.allocator.destroy(self);
929// errdefer cancel putter;912 }
930//913 }
931// const result = blk: {914
932// const held = await (async self.os_data.table_lock.acquire() catch unreachable);915 pub fn addFile(self: *Self, file_path: []const u8, value: V) !?V {
933// defer held.release();916 switch (builtin.os) {
934//917 .macosx, .freebsd, .netbsd, .dragonfly => return addFileKEvent(self, file_path, value),
935// const gop = try self.os_data.file_table.getOrPut(resolved_path);918 .linux => return addFileLinux(self, file_path, value),
936// if (gop.found_existing) {919 .windows => return addFileWindows(self, file_path, value),
937// const prev_value = gop.kv.value.value_ptr.*;920 else => @compileError("Unsupported OS"),
938// cancel gop.kv.value.putter;921 }
939// gop.kv.value = put_data;922 }
940// break :blk prev_value;923
941// } else {924 fn addFileKEvent(self: *Self, file_path: []const u8, value: V) !?V {
942// resolved_path_consumed = true;925 const resolved_path = try std.fs.path.resolve(self.allocator, [_][]const u8{file_path});
943// gop.kv.value = put_data;926 var resolved_path_consumed = false;
944// break :blk null;927 defer if (!resolved_path_consumed) self.allocator.free(resolved_path);
945// }928
946// };929 var close_op = try CloseOperation.start(self.allocator);
947//930 var close_op_consumed = false;
948// return result;931 defer if (!close_op_consumed) close_op.finish();
949// }932
950//933 const flags = if (comptime std.Target.current.isDarwin()) os.O_SYMLINK | os.O_EVTONLY else 0;
951// async fn kqPutEvents(self: *Self, close_op: *CloseOperation, value: V, out_put: **OsData.Put) void {934 const mode = 0;
952// var value_copy = value;935 const fd = try openPosix(self.allocator, resolved_path, flags, mode);
953// var put = OsData.Put{936 close_op.setHandle(fd);
954// .putter = @frame(),937
955// .value_ptr = &value_copy,938 var put = try self.allocator.create(OsData.Put);
956// };939 errdefer self.allocator.destroy(put);
957// out_put.* = &put;940 put.* = OsData.Put{
958// self.channel.loop.beginOneEvent();941 .value = value,
959//942 .putter_frame = undefined,
960// defer {943 };
961// close_op.finish();944 put.putter_frame = async self.kqPutEvents(close_op, put);
962// self.channel.loop.finishOneEvent();945 close_op_consumed = true;
963// }946 errdefer {
964//947 put.cancelled = true;
965// while (true) {948 await put.putter_frame;
966// if (await (async self.channel.loop.bsdWaitKev(949 }
967// @intCast(usize, close_op.getHandle()),950
968// os.EVFILT_VNODE,951 const result = blk: {
969// os.NOTE_WRITE | os.NOTE_DELETE,952 const held = self.os_data.table_lock.acquire();
970// ) catch unreachable)) |kev| {953 defer held.release();
971// // TODO handle EV_ERROR954
972// if (kev.fflags & os.NOTE_DELETE != 0) {955 const gop = try self.os_data.file_table.getOrPut(resolved_path);
973// await (async self.channel.put(Self.Event{956 if (gop.found_existing) {
974// .id = Event.Id.Delete,957 const prev_value = gop.kv.value.value;
975// .data = value_copy,958 await gop.kv.value.putter_frame;
976// }) catch unreachable);959 gop.kv.value = put;
977// } else if (kev.fflags & os.NOTE_WRITE != 0) {960 break :blk prev_value;
978// await (async self.channel.put(Self.Event{961 } else {
979// .id = Event.Id.CloseWrite,962 resolved_path_consumed = true;
980// .data = value_copy,963 gop.kv.value = put;
981// }) catch unreachable);964 break :blk null;
982// }965 }
983// } else |err| switch (err) {966 };
984// error.EventNotFound => unreachable,967
985// error.ProcessNotFound => unreachable,968 return result;
986// error.Overflow => unreachable,969 }
987// error.AccessDenied, error.SystemResources => |casted_err| {970
988// await (async self.channel.put(casted_err) catch unreachable);971 fn kqPutEvents(self: *Self, close_op: *CloseOperation, put: *OsData.Put) void {
989// },972 global_event_loop.beginOneEvent();
990// }973
991// }974 defer {
992// }975 close_op.finish();
993//976 global_event_loop.finishOneEvent();
994// async fn addFileLinux(self: *Self, file_path: []const u8, value: V) !?V {977 }
995// const value_copy = value;978
996//979 while (!put.cancelled) {
997// const dirname = std.fs.path.dirname(file_path) orelse ".";980 if (global_event_loop.bsdWaitKev(
998// const dirname_with_null = try std.cstr.addNullByte(self.channel.loop.allocator, dirname);981 @intCast(usize, close_op.getHandle()),
999// var dirname_with_null_consumed = false;982 os.EVFILT_VNODE,
1000// defer if (!dirname_with_null_consumed) self.channel.loop.allocator.free(dirname_with_null);983 os.NOTE_WRITE | os.NOTE_DELETE,
1001//984 )) |kev| {
1002// const basename = std.fs.path.basename(file_path);985 // TODO handle EV_ERROR
1003// const basename_with_null = try std.cstr.addNullByte(self.channel.loop.allocator, basename);986 if (kev.fflags & os.NOTE_DELETE != 0) {
1004// var basename_with_null_consumed = false;987 self.channel.put(Self.Event{
1005// defer if (!basename_with_null_consumed) self.channel.loop.allocator.free(basename_with_null);988 .id = Event.Id.Delete,
1006//989 .data = put.value,
1007// const wd = try os.inotify_add_watchC(990 });
1008// self.os_data.inotify_fd,991 } else if (kev.fflags & os.NOTE_WRITE != 0) {
1009// dirname_with_null.ptr,992 self.channel.put(Self.Event{
1010// os.linux.IN_CLOSE_WRITE | os.linux.IN_ONLYDIR | os.linux.IN_EXCL_UNLINK,993 .id = Event.Id.CloseWrite,
1011// );994 .data = put.value,
1012// // wd is either a newly created watch or an existing one.995 });
1013//996 }
1014// const held = await (async self.os_data.table_lock.acquire() catch unreachable);997 } else |err| switch (err) {
1015// defer held.release();998 error.EventNotFound => unreachable,
1016//999 error.ProcessNotFound => unreachable,
1017// const gop = try self.os_data.wd_table.getOrPut(wd);1000 error.Overflow => unreachable,
1018// if (!gop.found_existing) {1001 error.AccessDenied, error.SystemResources => |casted_err| {
1019// gop.kv.value = OsData.Dir{1002 self.channel.put(casted_err);
1020// .dirname = dirname_with_null,1003 },
1021// .file_table = OsData.FileTable.init(self.channel.loop.allocator),1004 }
1022// };1005 }
1023// dirname_with_null_consumed = true;1006 }
1024// }1007
1025// const dir = &gop.kv.value;1008 fn addFileLinux(self: *Self, file_path: []const u8, value: V) !?V {
1026//1009 const dirname = std.fs.path.dirname(file_path) orelse ".";
1027// const file_table_gop = try dir.file_table.getOrPut(basename_with_null);1010 const dirname_with_null = try std.cstr.addNullByte(self.allocator, dirname);
1028// if (file_table_gop.found_existing) {1011 var dirname_with_null_consumed = false;
1029// const prev_value = file_table_gop.kv.value;1012 defer if (!dirname_with_null_consumed) self.channel.free(dirname_with_null);
1030// file_table_gop.kv.value = value_copy;1013
1031// return prev_value;1014 const basename = std.fs.path.basename(file_path);
1032// } else {1015 const basename_with_null = try std.cstr.addNullByte(self.allocator, basename);
1033// file_table_gop.kv.value = value_copy;1016 var basename_with_null_consumed = false;
1034// basename_with_null_consumed = true;1017 defer if (!basename_with_null_consumed) self.allocator.free(basename_with_null);
1035// return null;1018
1036// }1019 const wd = try os.inotify_add_watchC(
1037// }1020 self.os_data.inotify_fd,
1038//1021 dirname_with_null.ptr,
1039// async fn addFileWindows(self: *Self, file_path: []const u8, value: V) !?V {1022 os.linux.IN_CLOSE_WRITE | os.linux.IN_ONLYDIR | os.linux.IN_EXCL_UNLINK,
1040// const value_copy = value;1023 );
1041// // TODO we might need to convert dirname and basename to canonical file paths ("short"?)1024 // wd is either a newly created watch or an existing one.
1042//1025
1043// const dirname = try std.mem.dupe(self.channel.loop.allocator, u8, std.fs.path.dirname(file_path) orelse ".");1026 const held = self.os_data.table_lock.acquire();
1044// var dirname_consumed = false;1027 defer held.release();
1045// defer if (!dirname_consumed) self.channel.loop.allocator.free(dirname);1028
1046//1029 const gop = try self.os_data.wd_table.getOrPut(wd);
1047// const dirname_utf16le = try std.unicode.utf8ToUtf16LeWithNull(self.channel.loop.allocator, dirname);1030 if (!gop.found_existing) {
1048// defer self.channel.loop.allocator.free(dirname_utf16le);1031 gop.kv.value = OsData.Dir{
1049//1032 .dirname = dirname_with_null,
1050// // TODO https://github.com/ziglang/zig/issues/2651033 .file_table = OsData.FileTable.init(self.allocator),
1051// const basename = std.fs.path.basename(file_path);1034 };
1052// const basename_utf16le_null = try std.unicode.utf8ToUtf16LeWithNull(self.channel.loop.allocator, basename);1035 dirname_with_null_consumed = true;
1053// var basename_utf16le_null_consumed = false;1036 }
1054// defer if (!basename_utf16le_null_consumed) self.channel.loop.allocator.free(basename_utf16le_null);1037 const dir = &gop.kv.value;
1055// const basename_utf16le_no_null = basename_utf16le_null[0 .. basename_utf16le_null.len - 1];1038
1056//1039 const file_table_gop = try dir.file_table.getOrPut(basename_with_null);
1057// const dir_handle = try windows.CreateFileW(1040 if (file_table_gop.found_existing) {
1058// dirname_utf16le.ptr,1041 const prev_value = file_table_gop.kv.value;
1059// windows.FILE_LIST_DIRECTORY,1042 file_table_gop.kv.value = value;
1060// windows.FILE_SHARE_READ | windows.FILE_SHARE_DELETE | windows.FILE_SHARE_WRITE,1043 return prev_value;
1061// null,1044 } else {
1062// windows.OPEN_EXISTING,1045 file_table_gop.kv.value = value;
1063// windows.FILE_FLAG_BACKUP_SEMANTICS | windows.FILE_FLAG_OVERLAPPED,1046 basename_with_null_consumed = true;
1064// null,1047 return null;
1065// );1048 }
1066// var dir_handle_consumed = false;1049 }
1067// defer if (!dir_handle_consumed) windows.CloseHandle(dir_handle);1050
1068//1051 fn addFileWindows(self: *Self, file_path: []const u8, value: V) !?V {
1069// const held = await (async self.os_data.table_lock.acquire() catch unreachable);1052 // TODO we might need to convert dirname and basename to canonical file paths ("short"?)
1070// defer held.release();1053 const dirname = try std.mem.dupe(self.allocator, u8, std.fs.path.dirname(file_path) orelse ".");
1071//1054 var dirname_consumed = false;
1072// const gop = try self.os_data.dir_table.getOrPut(dirname);1055 defer if (!dirname_consumed) self.allocator.free(dirname);
1073// if (gop.found_existing) {1056
1074// const dir = gop.kv.value;1057 const dirname_utf16le = try std.unicode.utf8ToUtf16LeWithNull(self.allocator, dirname);
1075// const held_dir_lock = await (async dir.table_lock.acquire() catch unreachable);1058 defer self.allocator.free(dirname_utf16le);
1076// defer held_dir_lock.release();1059
1077//1060 // TODO https://github.com/ziglang/zig/issues/265
1078// const file_gop = try dir.file_table.getOrPut(basename_utf16le_no_null);1061 const basename = std.fs.path.basename(file_path);
1079// if (file_gop.found_existing) {1062 const basename_utf16le_null = try std.unicode.utf8ToUtf16LeWithNull(self.allocator, basename);
1080// const prev_value = file_gop.kv.value;1063 var basename_utf16le_null_consumed = false;
1081// file_gop.kv.value = value_copy;1064 defer if (!basename_utf16le_null_consumed) self.allocator.free(basename_utf16le_null);
1082// return prev_value;1065 const basename_utf16le_no_null = basename_utf16le_null[0 .. basename_utf16le_null.len - 1];
1083// } else {1066
1084// file_gop.kv.value = value_copy;1067 const dir_handle = try windows.CreateFileW(
1085// basename_utf16le_null_consumed = true;1068 dirname_utf16le.ptr,
1086// return null;1069 windows.FILE_LIST_DIRECTORY,
1087// }1070 windows.FILE_SHARE_READ | windows.FILE_SHARE_DELETE | windows.FILE_SHARE_WRITE,
1088// } else {1071 null,
1089// errdefer _ = self.os_data.dir_table.remove(dirname);1072 windows.OPEN_EXISTING,
1090// const dir = try self.channel.loop.allocator.create(OsData.Dir);1073 windows.FILE_FLAG_BACKUP_SEMANTICS | windows.FILE_FLAG_OVERLAPPED,
1091// errdefer self.channel.loop.allocator.destroy(dir);1074 null,
1092//1075 );
1093// dir.* = OsData.Dir{1076 var dir_handle_consumed = false;
1094// .file_table = OsData.FileTable.init(self.channel.loop.allocator),1077 defer if (!dir_handle_consumed) windows.CloseHandle(dir_handle);
1095// .table_lock = event.Lock.init(self.channel.loop),1078
1096// .putter = undefined,1079 const held = self.os_data.table_lock.acquire();
1097// };1080 defer held.release();
1098// gop.kv.value = dir;1081
1099// assert((try dir.file_table.put(basename_utf16le_no_null, value_copy)) == null);1082 const gop = try self.os_data.dir_table.getOrPut(dirname);
1100// basename_utf16le_null_consumed = true;1083 if (gop.found_existing) {
1101//1084 const dir = gop.kv.value;
1102// dir.putter = try async self.windowsDirReader(dir_handle, dir);1085 const held_dir_lock = dir.table_lock.acquire();
1103// dir_handle_consumed = true;1086 defer held_dir_lock.release();
1104//1087
1105// dirname_consumed = true;1088 const file_gop = try dir.file_table.getOrPut(basename_utf16le_no_null);
1106//1089 if (file_gop.found_existing) {
1107// return null;1090 const prev_value = file_gop.kv.value;
1108// }1091 file_gop.kv.value = value;
1109// }1092 return prev_value;
1110//1093 } else {
1111// async fn windowsDirReader(self: *Self, dir_handle: windows.HANDLE, dir: *OsData.Dir) void {1094 file_gop.kv.value = value;
1112// self.ref();1095 basename_utf16le_null_consumed = true;
1113// defer self.deref();1096 return null;
1114//1097 }
1115// defer os.close(dir_handle);1098 } else {
1116//1099 errdefer _ = self.os_data.dir_table.remove(dirname);
1117// var putter_node = std.atomic.Queue(anyframe).Node{1100 const dir = try self.allocator.create(OsData.Dir);
1118// .data = @frame(),1101 errdefer self.allocator.destroy(dir);
1119// .prev = null,1102
1120// .next = null,1103 dir.* = OsData.Dir{
1121// };1104 .file_table = OsData.FileTable.init(self.allocator),
1122// self.os_data.all_putters.put(&putter_node);1105 .table_lock = event.Lock.init(),
1123// defer _ = self.os_data.all_putters.remove(&putter_node);1106 .putter_frame = undefined,
1124//1107 };
1125// var resume_node = Loop.ResumeNode.Basic{1108 gop.kv.value = dir;
1126// .base = Loop.ResumeNode{1109 assert((try dir.file_table.put(basename_utf16le_no_null, value)) == null);
1127// .id = Loop.ResumeNode.Id.Basic,1110 basename_utf16le_null_consumed = true;
1128// .handle = @frame(),1111
1129// .overlapped = windows.OVERLAPPED{1112 dir.putter_frame = async self.windowsDirReader(dir_handle, dir);
1130// .Internal = 0,1113 dir_handle_consumed = true;
1131// .InternalHigh = 0,1114
1132// .Offset = 0,1115 dirname_consumed = true;
1133// .OffsetHigh = 0,1116
1134// .hEvent = null,1117 return null;
1135// },1118 }
1136// },1119 }
1137// };1120
1138// var event_buf: [4096]u8 align(@alignOf(windows.FILE_NOTIFY_INFORMATION)) = undefined;1121 fn windowsDirReader(self: *Self, dir_handle: windows.HANDLE, dir: *OsData.Dir) void {
1139//1122 self.ref();
1140// // TODO handle this error not in the channel but in the setup1123 defer self.deref();
1141// _ = windows.CreateIoCompletionPort(1124
1142// dir_handle,1125 defer os.close(dir_handle);
1143// self.channel.loop.os_data.io_port,1126
1144// undefined,1127 var putter_node = std.atomic.Queue(anyframe).Node{
1145// undefined,1128 .data = .{ .putter = @frame() },
1146// ) catch |err| {1129 .prev = null,
1147// await (async self.channel.put(err) catch unreachable);1130 .next = null,
1148// return;1131 };
1149// };1132 self.os_data.all_putters.put(&putter_node);
1150//1133 defer _ = self.os_data.all_putters.remove(&putter_node);
1151// while (true) {1134
1152// {1135 var resume_node = Loop.ResumeNode.Basic{
1153// // TODO only 1 beginOneEvent for the whole function1136 .base = Loop.ResumeNode{
1154// self.channel.loop.beginOneEvent();1137 .id = Loop.ResumeNode.Id.Basic,
1155// errdefer self.channel.loop.finishOneEvent();1138 .handle = @frame(),
1156// errdefer {1139 .overlapped = windows.OVERLAPPED{
1157// _ = windows.kernel32.CancelIoEx(dir_handle, &resume_node.base.overlapped);1140 .Internal = 0,
1158// }1141 .InternalHigh = 0,
1159// suspend {1142 .Offset = 0,
1160// _ = windows.kernel32.ReadDirectoryChangesW(1143 .OffsetHigh = 0,
1161// dir_handle,1144 .hEvent = null,
1162// &event_buf,1145 },
1163// @intCast(windows.DWORD, event_buf.len),1146 },
1164// windows.FALSE, // watch subtree1147 };
1165// windows.FILE_NOTIFY_CHANGE_FILE_NAME | windows.FILE_NOTIFY_CHANGE_DIR_NAME |1148 var event_buf: [4096]u8 align(@alignOf(windows.FILE_NOTIFY_INFORMATION)) = undefined;
1166// windows.FILE_NOTIFY_CHANGE_ATTRIBUTES | windows.FILE_NOTIFY_CHANGE_SIZE |1149
1167// windows.FILE_NOTIFY_CHANGE_LAST_WRITE | windows.FILE_NOTIFY_CHANGE_LAST_ACCESS |1150 // TODO handle this error not in the channel but in the setup
1168// windows.FILE_NOTIFY_CHANGE_CREATION | windows.FILE_NOTIFY_CHANGE_SECURITY,1151 _ = windows.CreateIoCompletionPort(
1169// null, // number of bytes transferred (unused for async)1152 dir_handle,
1170// &resume_node.base.overlapped,1153 global_event_loop.os_data.io_port,
1171// null, // completion routine - unused because we use IOCP1154 undefined,
1172// );1155 undefined,
1173// }1156 ) catch |err| {
1174// }1157 self.channel.put(err);
1175// var bytes_transferred: windows.DWORD = undefined;1158 return;
1176// if (windows.kernel32.GetOverlappedResult(dir_handle, &resume_node.base.overlapped, &bytes_transferred, windows.FALSE) == 0) {1159 };
1177// const err = switch (windows.kernel32.GetLastError()) {1160
1178// else => |err| windows.unexpectedError(err),1161 while (!putter_node.data.cancelled) {
1179// };1162 {
1180// await (async self.channel.put(err) catch unreachable);1163 // TODO only 1 beginOneEvent for the whole function
1181// } else {1164 global_event_loop.beginOneEvent();
1182// // can't use @bytesToSlice because of the special variable length name field1165 errdefer global_event_loop.finishOneEvent();
1183// var ptr = event_buf[0..].ptr;1166 errdefer {
1184// const end_ptr = ptr + bytes_transferred;1167 _ = windows.kernel32.CancelIoEx(dir_handle, &resume_node.base.overlapped);
1185// var ev: *windows.FILE_NOTIFY_INFORMATION = undefined;1168 }
1186// while (@ptrToInt(ptr) < @ptrToInt(end_ptr)) : (ptr += ev.NextEntryOffset) {1169 suspend {
1187// ev = @ptrCast(*windows.FILE_NOTIFY_INFORMATION, ptr);1170 _ = windows.kernel32.ReadDirectoryChangesW(
1188// const emit = switch (ev.Action) {1171 dir_handle,
1189// windows.FILE_ACTION_REMOVED => WatchEventId.Delete,1172 &event_buf,
1190// windows.FILE_ACTION_MODIFIED => WatchEventId.CloseWrite,1173 @intCast(windows.DWORD, event_buf.len),
1191// else => null,1174 windows.FALSE, // watch subtree
1192// };1175 windows.FILE_NOTIFY_CHANGE_FILE_NAME | windows.FILE_NOTIFY_CHANGE_DIR_NAME |
1193// if (emit) |id| {1176 windows.FILE_NOTIFY_CHANGE_ATTRIBUTES | windows.FILE_NOTIFY_CHANGE_SIZE |
1194// const basename_utf16le = ([*]u16)(&ev.FileName)[0 .. ev.FileNameLength / 2];1177 windows.FILE_NOTIFY_CHANGE_LAST_WRITE | windows.FILE_NOTIFY_CHANGE_LAST_ACCESS |
1195// const user_value = blk: {1178 windows.FILE_NOTIFY_CHANGE_CREATION | windows.FILE_NOTIFY_CHANGE_SECURITY,
1196// const held = await (async dir.table_lock.acquire() catch unreachable);1179 null, // number of bytes transferred (unused for async)
1197// defer held.release();1180 &resume_node.base.overlapped,
1198//1181 null, // completion routine - unused because we use IOCP
1199// if (dir.file_table.get(basename_utf16le)) |entry| {1182 );
1200// break :blk entry.value;1183 }
1201// } else {1184 }
1202// break :blk null;1185 var bytes_transferred: windows.DWORD = undefined;
1203// }1186 if (windows.kernel32.GetOverlappedResult(dir_handle, &resume_node.base.overlapped, &bytes_transferred, windows.FALSE) == 0) {
1204// };1187 const err = switch (windows.kernel32.GetLastError()) {
1205// if (user_value) |v| {1188 else => |err| windows.unexpectedError(err),
1206// await (async self.channel.put(Event{1189 };
1207// .id = id,1190 self.channel.put(err);
1208// .data = v,1191 } else {
1209// }) catch unreachable);1192 // can't use @bytesToSlice because of the special variable length name field
1210// }1193 var ptr = event_buf[0..].ptr;
1211// }1194 const end_ptr = ptr + bytes_transferred;
1212// if (ev.NextEntryOffset == 0) break;1195 var ev: *windows.FILE_NOTIFY_INFORMATION = undefined;
1213// }1196 while (@ptrToInt(ptr) < @ptrToInt(end_ptr)) : (ptr += ev.NextEntryOffset) {
1214// }1197 ev = @ptrCast(*windows.FILE_NOTIFY_INFORMATION, ptr);
1215// }1198 const emit = switch (ev.Action) {
1216// }1199 windows.FILE_ACTION_REMOVED => WatchEventId.Delete,
1217//1200 windows.FILE_ACTION_MODIFIED => WatchEventId.CloseWrite,
1218// pub async fn removeFile(self: *Self, file_path: []const u8) ?V {1201 else => null,
1219// @panic("TODO");1202 };
1220// }1203 if (emit) |id| {
1221//1204 const basename_utf16le = ([*]u16)(&ev.FileName)[0 .. ev.FileNameLength / 2];
1222// async fn linuxEventPutter(inotify_fd: i32, channel: *event.Channel(Event.Error!Event), out_watch: **Self) void {1205 const user_value = blk: {
1223// const loop = channel.loop;1206 const held = dir.table_lock.acquire();
1224//1207 defer held.release();
1225// var watch = Self{1208
1226// .channel = channel,1209 if (dir.file_table.get(basename_utf16le)) |entry| {
1227// .os_data = OsData{1210 break :blk entry.value;
1228// .putter = @frame(),1211 } else {
1229// .inotify_fd = inotify_fd,1212 break :blk null;
1230// .wd_table = OsData.WdTable.init(loop.allocator),1213 }
1231// .table_lock = event.Lock.init(loop),1214 };
1232// },1215 if (user_value) |v| {
1233// };1216 self.channel.put(Event{
1234// out_watch.* = &watch;1217 .id = id,
1235//1218 .data = v,
1236// loop.beginOneEvent();1219 });
1237//1220 }
1238// defer {1221 }
1239// watch.os_data.table_lock.deinit();1222 if (ev.NextEntryOffset == 0) break;
1240// var wd_it = watch.os_data.wd_table.iterator();1223 }
1241// while (wd_it.next()) |wd_entry| {1224 }
1242// var file_it = wd_entry.value.file_table.iterator();1225 }
1243// while (file_it.next()) |file_entry| {1226 }
1244// loop.allocator.free(file_entry.key);1227
1245// }1228 pub fn removeFile(self: *Self, file_path: []const u8) ?V {
1246// loop.allocator.free(wd_entry.value.dirname);1229 @panic("TODO");
1247// }1230 }
1248// loop.finishOneEvent();1231
1249// os.close(inotify_fd);1232 fn linuxEventPutter(self: *Self) void {
1250// channel.destroy();1233 global_event_loop.beginOneEvent();
1251// }1234
1252//1235 defer {
1253// var event_buf: [4096]u8 align(@alignOf(os.linux.inotify_event)) = undefined;1236 self.os_data.table_lock.deinit();
1254//1237 var wd_it = self.os_data.wd_table.iterator();
1255// while (true) {1238 while (wd_it.next()) |wd_entry| {
1256// const rc = os.linux.read(inotify_fd, &event_buf, event_buf.len);1239 var file_it = wd_entry.value.file_table.iterator();
1257// const errno = os.linux.getErrno(rc);1240 while (file_it.next()) |file_entry| {
1258// switch (errno) {1241 self.allocator.free(file_entry.key);
1259// 0 => {1242 }
1260// // can't use @bytesToSlice because of the special variable length name field1243 self.allocator.free(wd_entry.value.dirname);
1261// var ptr = event_buf[0..].ptr;1244 wd_entry.value.file_table.deinit();
1262// const end_ptr = ptr + event_buf.len;1245 }
1263// var ev: *os.linux.inotify_event = undefined;1246 self.os_data.wd_table.deinit();
1264// while (@ptrToInt(ptr) < @ptrToInt(end_ptr)) : (ptr += @sizeOf(os.linux.inotify_event) + ev.len) {1247 global_event_loop.finishOneEvent();
1265// ev = @ptrCast(*os.linux.inotify_event, ptr);1248 os.close(self.os_data.inotify_fd);
1266// if (ev.mask & os.linux.IN_CLOSE_WRITE == os.linux.IN_CLOSE_WRITE) {1249 self.channel.deinit();
1267// const basename_ptr = ptr + @sizeOf(os.linux.inotify_event);1250 self.allocator.free(self.channel.buffer_nodes);
1268// const basename_with_null = basename_ptr[0 .. std.mem.len(u8, basename_ptr) + 1];1251 }
1269// const user_value = blk: {1252
1270// const held = await (async watch.os_data.table_lock.acquire() catch unreachable);1253 var event_buf: [4096]u8 align(@alignOf(os.linux.inotify_event)) = undefined;
1271// defer held.release();1254
1272//1255 while (!self.os_data.cancelled) {
1273// const dir = &watch.os_data.wd_table.get(ev.wd).?.value;1256 const rc = os.linux.read(self.os_data.inotify_fd, &event_buf, event_buf.len);
1274// if (dir.file_table.get(basename_with_null)) |entry| {1257 const errno = os.linux.getErrno(rc);
1275// break :blk entry.value;1258 switch (errno) {
1276// } else {1259 0 => {
1277// break :blk null;1260 // can't use @bytesToSlice because of the special variable length name field
1278// }1261 var ptr = event_buf[0..].ptr;
1279// };1262 const end_ptr = ptr + event_buf.len;
1280// if (user_value) |v| {1263 var ev: *os.linux.inotify_event = undefined;
1281// await (async channel.put(Event{1264 while (@ptrToInt(ptr) < @ptrToInt(end_ptr)) : (ptr += @sizeOf(os.linux.inotify_event) + ev.len) {
1282// .id = WatchEventId.CloseWrite,1265 ev = @ptrCast(*os.linux.inotify_event, ptr);
1283// .data = v,1266 if (ev.mask & os.linux.IN_CLOSE_WRITE == os.linux.IN_CLOSE_WRITE) {
1284// }) catch unreachable);1267 const basename_ptr = ptr + @sizeOf(os.linux.inotify_event);
1285// }1268 // `ev.len` counts all bytes in `ev.name` including terminating null byte.
1286// }1269 const basename_with_null = basename_ptr[0 .. ev.len];
1287// }1270 const user_value = blk: {
1288// },1271 const held = self.os_data.table_lock.acquire();
1289// os.linux.EINTR => continue,1272 defer held.release();
1290// os.linux.EINVAL => unreachable,1273
1291// os.linux.EFAULT => unreachable,1274 const dir = &self.os_data.wd_table.get(ev.wd).?.value;
1292// os.linux.EAGAIN => {1275 if (dir.file_table.get(basename_with_null)) |entry| {
1293// (await (async loop.linuxWaitFd(1276 break :blk entry.value;
1294// inotify_fd,1277 } else {
1295// os.linux.EPOLLET | os.linux.EPOLLIN,1278 break :blk null;
1296// ) catch unreachable)) catch |err| {1279 }
1297// const transformed_err = switch (err) {1280 };
1298// error.FileDescriptorAlreadyPresentInSet => unreachable,1281 if (user_value) |v| {
1299// error.OperationCausesCircularLoop => unreachable,1282 self.channel.put(Event{
1300// error.FileDescriptorNotRegistered => unreachable,1283 .id = WatchEventId.CloseWrite,
1301// error.FileDescriptorIncompatibleWithEpoll => unreachable,1284 .data = v,
1302// error.Unexpected => unreachable,1285 });
1303// else => |e| e,1286 }
1304// };1287 }
1305// await (async channel.put(transformed_err) catch unreachable);1288 }
1306// };1289 },
1307// },1290 os.linux.EINTR => continue,
1308// else => unreachable,1291 os.linux.EINVAL => unreachable,
1309// }1292 os.linux.EFAULT => unreachable,
1310// }1293 os.linux.EAGAIN => {
1311// }1294 global_event_loop.linuxWaitFd(self.os_data.inotify_fd, os.linux.EPOLLET | os.linux.EPOLLIN);
1312// };1295 },
1313//}1296 else => unreachable,
1297 }
1298 }
1299 }
1300 };
1301}
13141302
1315const test_tmp_dir = "std_event_fs_test";1303const test_tmp_dir = "std_event_fs_test";
13161304
1317// TODO this test is disabled until the async function rewrite is finished.1305test "write a file, watch it, write it again" {
1318//test "write a file, watch it, write it again" {1306 // TODO provide a way to run tests in evented I/O mode
1319// return error.SkipZigTest;1307 if (!std.io.is_async) return error.SkipZigTest;
1320// const allocator = std.heap.direct_allocator;1308
1321//1309 const allocator = std.heap.direct_allocator;
1322// // TODO move this into event loop too1310
1323// try os.makePath(allocator, test_tmp_dir);1311 // TODO move this into event loop too
1324// defer os.deleteTree(test_tmp_dir) catch {};1312 try os.makePath(allocator, test_tmp_dir);
1325//1313 defer os.deleteTree(test_tmp_dir) catch {};
1326// var loop: Loop = undefined;1314
1327// try loop.initMultiThreaded(allocator);1315 return testFsWatch(&allocator);
1328// defer loop.deinit();
1329//
1330// var result: anyerror!void = error.ResultNeverWritten;
1331// const handle = try async<allocator> testFsWatchCantFail(&loop, &result);
1332// defer cancel handle;
1333//
1334// loop.run();
1335// return result;
1336//}
1337
1338fn testFsWatchCantFail(loop: *Loop, result: *(anyerror!void)) void {
1339 result.* = testFsWatch(loop);
1340}1316}
13411317
1342fn testFsWatch(loop: *Loop) !void {1318fn testFsWatch(allocator: *Allocator) !void {
1343 const file_path = try std.fs.path.join(loop.allocator, [][]const u8{ test_tmp_dir, "file.txt" });1319 const file_path = try std.fs.path.join(allocator, [_][]const u8{ test_tmp_dir, "file.txt" });
1344 defer loop.allocator.free(file_path);1320 defer allocator.free(file_path);
13451321
1346 const contents =1322 const contents =
1347 \\line 11323 \\line 1
...@@ -1350,27 +1326,27 @@ fn testFsWatch(loop: *Loop) !void {...@@ -1350,27 +1326,27 @@ fn testFsWatch(loop: *Loop) !void {
1350 const line2_offset = 7;1326 const line2_offset = 7;
13511327
1352 // first just write then read the file1328 // first just write then read the file
1353 try writeFile(loop, file_path, contents);1329 try writeFile(allocator, file_path, contents);
13541330
1355 const read_contents = try readFile(loop, file_path, 1024 * 1024);1331 const read_contents = try readFile(allocator, file_path, 1024 * 1024);
1356 testing.expectEqualSlices(u8, contents, read_contents);1332 testing.expectEqualSlices(u8, contents, read_contents);
13571333
1358 // now watch the file1334 // now watch the file
1359 var watch = try Watch(void).create(loop, 0);1335 var watch = try Watch(void).init(allocator, 0);
1360 defer watch.destroy();1336 defer watch.deinit();
13611337
1362 testing.expect((try watch.addFile(file_path, {})) == null);1338 testing.expect((try watch.addFile(file_path, {})) == null);
13631339
1364 const ev = async watch.channel.get();1340 const ev = watch.channel.get();
1365 var ev_consumed = false;1341 var ev_consumed = false;
1366 defer if (!ev_consumed) await ev;1342 defer if (!ev_consumed) await ev;
13671343
1368 // overwrite line 21344 // overwrite line 2
1369 const fd = try await openReadWrite(loop, file_path, File.default_mode);1345 const fd = try await openReadWrite(file_path, File.default_mode);
1370 {1346 {
1371 defer os.close(fd);1347 defer os.close(fd);
13721348
1373 try pwritev(loop, fd, []const []const u8{"lorem ipsum"}, line2_offset);1349 try pwritev(allocator, fd, []const []const u8{"lorem ipsum"}, line2_offset);
1374 }1350 }
13751351
1376 ev_consumed = true;1352 ev_consumed = true;
...@@ -1378,7 +1354,7 @@ fn testFsWatch(loop: *Loop) !void {...@@ -1378,7 +1354,7 @@ fn testFsWatch(loop: *Loop) !void {
1378 WatchEventId.CloseWrite => {},1354 WatchEventId.CloseWrite => {},
1379 WatchEventId.Delete => @panic("wrong event"),1355 WatchEventId.Delete => @panic("wrong event"),
1380 }1356 }
1381 const contents_updated = try readFile(loop, file_path, 1024 * 1024);1357 const contents_updated = try readFile(allocator, file_path, 1024 * 1024);
1382 testing.expectEqualSlices(u8,1358 testing.expectEqualSlices(u8,
1383 \\line 11359 \\line 1
1384 \\lorem ipsum1360 \\lorem ipsum
...@@ -1390,16 +1366,15 @@ fn testFsWatch(loop: *Loop) !void {...@@ -1390,16 +1366,15 @@ fn testFsWatch(loop: *Loop) !void {
1390pub const OutStream = struct {1366pub const OutStream = struct {
1391 fd: fd_t,1367 fd: fd_t,
1392 stream: Stream,1368 stream: Stream,
1393 loop: *Loop,1369 allocator: *Allocator,
1394 offset: usize,1370 offset: usize,
13951371
1396 pub const Error = File.WriteError;1372 pub const Error = File.WriteError;
1397 pub const Stream = event.io.OutStream(Error);1373 pub const Stream = event.io.OutStream(Error);
13981374
1399 pub fn init(loop: *Loop, fd: fd_t, offset: usize) OutStream {1375 pub fn init(allocator: *Allocator, fd: fd_t, offset: usize) OutStream {
1400 return OutStream{1376 return OutStream{
1401 .fd = fd,1377 .fd = fd,
1402 .loop = loop,
1403 .offset = offset,1378 .offset = offset,
1404 .stream = Stream{ .writeFn = writeFn },1379 .stream = Stream{ .writeFn = writeFn },
1405 };1380 };
...@@ -1409,23 +1384,22 @@ pub const OutStream = struct {...@@ -1409,23 +1384,22 @@ pub const OutStream = struct {
1409 const self = @fieldParentPtr(OutStream, "stream", out_stream);1384 const self = @fieldParentPtr(OutStream, "stream", out_stream);
1410 const offset = self.offset;1385 const offset = self.offset;
1411 self.offset += bytes.len;1386 self.offset += bytes.len;
1412 return pwritev(self.loop, self.fd, [][]const u8{bytes}, offset);1387 return pwritev(self.allocator, self.fd, [_][]const u8{bytes}, offset);
1413 }1388 }
1414};1389};
14151390
1416pub const InStream = struct {1391pub const InStream = struct {
1417 fd: fd_t,1392 fd: fd_t,
1418 stream: Stream,1393 stream: Stream,
1419 loop: *Loop,1394 allocator: *Allocator,
1420 offset: usize,1395 offset: usize,
14211396
1422 pub const Error = PReadVError; // TODO make this not have OutOfMemory1397 pub const Error = PReadVError; // TODO make this not have OutOfMemory
1423 pub const Stream = event.io.InStream(Error);1398 pub const Stream = event.io.InStream(Error);
14241399
1425 pub fn init(loop: *Loop, fd: fd_t, offset: usize) InStream {1400 pub fn init(allocator: *Allocator, fd: fd_t, offset: usize) InStream {
1426 return InStream{1401 return InStream{
1427 .fd = fd,1402 .fd = fd,
1428 .loop = loop,
1429 .offset = offset,1403 .offset = offset,
1430 .stream = Stream{ .readFn = readFn },1404 .stream = Stream{ .readFn = readFn },
1431 };1405 };
...@@ -1433,7 +1407,7 @@ pub const InStream = struct {...@@ -1433,7 +1407,7 @@ pub const InStream = struct {
14331407
1434 fn readFn(in_stream: *Stream, bytes: []u8) Error!usize {1408 fn readFn(in_stream: *Stream, bytes: []u8) Error!usize {
1435 const self = @fieldParentPtr(InStream, "stream", in_stream);1409 const self = @fieldParentPtr(InStream, "stream", in_stream);
1436 const amt = try preadv(self.loop, self.fd, [][]u8{bytes}, self.offset);1410 const amt = try preadv(self.allocator, self.fd, [_][]u8{bytes}, self.offset);
1437 self.offset += amt;1411 self.offset += amt;
1438 return amt;1412 return amt;
1439 }1413 }