authorgravatar for git@vexu.euVeikka Tuominen <git@vexu.eu> 2019-11-24 13:16:09+02:00
committergravatar for git@vexu.euVeikka Tuominen <git@vexu.eu> 2019-11-24 17:39:08+02:00
logab534cc9f16697da4846820b038e17fd1bc27f35
tree9b6fa7801fb98ffecaacc260fc665f2290e32b13
parent29d7b5a80c9faf640c3db0de14cd229e90b2d8c3
signaturelock-open Commit is signed but in an unrecognized format.

update event.fs to use global event loop

1 files changed, 80 insertions(+), 102 deletions(-)

lib/std/event/fs.zig+80-102
......@@ -10,6 +10,9 @@ const Loop = event.Loop;
1010const fd_t = os.fd_t;
1111const File = std.fs.File;
1212
13const global_event_loop = Loop.instance orelse
14 @compileError("std.event.fs currently only works with event-based I/O");
15
1316pub const RequestNode = std.atomic.Queue(Request).Node;
1417
1518pub const Request = struct {
......@@ -86,7 +89,7 @@ pub const Request = struct {
8689pub const PWriteVError = error{OutOfMemory} || File.WriteError;
8790
8891/// data - just the inner references - must live until pwritev frame completes.
89pub fn pwritev(loop: *Loop, fd: fd_t, data: []const []const u8, offset: usize) PWriteVError!void {
92pub fn pwritev(allocator: *Allocator, fd: fd_t, data: []const []const u8, offset: usize) PWriteVError!void {
9093 switch (builtin.os) {
9194 .macosx,
9295 .linux,
......@@ -94,8 +97,8 @@ pub fn pwritev(loop: *Loop, fd: fd_t, data: []const []const u8, offset: usize) P
9497 .netbsd,
9598 .dragonfly,
9699 => {
97 const iovecs = try loop.allocator.alloc(os.iovec_const, data.len);
98 defer loop.allocator.free(iovecs);
100 const iovecs = try allocator.alloc(os.iovec_const, data.len);
101 defer allocator.free(iovecs);
99102
100103 for (data) |buf, i| {
101104 iovecs[i] = os.iovec_const{
......@@ -104,31 +107,31 @@ pub fn pwritev(loop: *Loop, fd: fd_t, data: []const []const u8, offset: usize) P
104107 };
105108 }
106109
107 return pwritevPosix(loop, fd, iovecs, offset);
110 return pwritevPosix(fd, iovecs, offset);
108111 },
109112 .windows => {
110 const data_copy = try std.mem.dupe(loop.allocator, []const u8, data);
111 defer loop.allocator.free(data_copy);
112 return pwritevWindows(loop, fd, data, offset);
113 const data_copy = try std.mem.dupe(allocator, []const u8, data);
114 defer allocator.free(data_copy);
115 return pwritevWindows(fd, data, offset);
113116 },
114117 else => @compileError("Unsupported OS"),
115118 }
116119}
117120
118121/// data must outlive the returned frame
119pub fn pwritevWindows(loop: *Loop, fd: fd_t, data: []const []const u8, offset: usize) os.WindowsWriteError!void {
122pub fn pwritevWindows(fd: fd_t, data: []const []const u8, offset: usize) os.WindowsWriteError!void {
120123 if (data.len == 0) return;
121 if (data.len == 1) return pwriteWindows(loop, fd, data[0], offset);
124 if (data.len == 1) return pwriteWindows(fd, data[0], offset);
122125
123126 // TODO do these in parallel
124127 var off = offset;
125128 for (data) |buf| {
126 try pwriteWindows(loop, fd, buf, off);
129 try pwriteWindows(fd, buf, off);
127130 off += buf.len;
128131 }
129132}
130133
131pub fn pwriteWindows(loop: *Loop, fd: fd_t, data: []const u8, offset: u64) os.WindowsWriteError!void {
134pub fn pwriteWindows(fd: fd_t, data: []const u8, offset: u64) os.WindowsWriteError!void {
132135 var resume_node = Loop.ResumeNode.Basic{
133136 .base = Loop.ResumeNode{
134137 .id = Loop.ResumeNode.Id.Basic,
......@@ -143,9 +146,9 @@ pub fn pwriteWindows(loop: *Loop, fd: fd_t, data: []const u8, offset: u64) os.Wi
143146 },
144147 };
145148 // TODO only call create io completion port once per fd
146 _ = windows.CreateIoCompletionPort(fd, loop.os_data.io_port, undefined, undefined);
147 loop.beginOneEvent();
148 errdefer loop.finishOneEvent();
149 _ = windows.CreateIoCompletionPort(fd, global_event_loop.os_data.io_port, undefined, undefined);
150 global_event_loop.beginOneEvent();
151 errdefer global_event_loop.finishOneEvent();
149152
150153 errdefer {
151154 _ = windows.kernel32.CancelIoEx(fd, &resume_node.base.overlapped);
......@@ -168,12 +171,7 @@ pub fn pwriteWindows(loop: *Loop, fd: fd_t, data: []const u8, offset: u64) os.Wi
168171}
169172
170173/// iovecs must live until pwritev frame completes.
171pub fn pwritevPosix(
172 loop: *Loop,
173 fd: fd_t,
174 iovecs: []const os.iovec_const,
175 offset: usize,
176) os.WriteError!void {
174pub fn pwritevPosix(fd: fd_t, iovecs: []const os.iovec_const, offset: usize) os.WriteError!void {
177175 var req_node = RequestNode{
178176 .prev = null,
179177 .next = null,
......@@ -196,21 +194,17 @@ pub fn pwritevPosix(
196194 },
197195 };
198196
199 errdefer loop.posixFsCancel(&req_node);
197 errdefer global_event_loop.posixFsCancel(&req_node);
200198
201199 suspend {
202 loop.posixFsRequest(&req_node);
200 global_event_loop.posixFsRequest(&req_node);
203201 }
204202
205203 return req_node.data.msg.PWriteV.result;
206204}
207205
208206/// iovecs must live until pwritev frame completes.
209pub fn writevPosix(
210 loop: *Loop,
211 fd: fd_t,
212 iovecs: []const os.iovec_const,
213) os.WriteError!void {
207pub fn writevPosix(fd: fd_t, iovecs: []const os.iovec_const) os.WriteError!void {
214208 var req_node = RequestNode{
215209 .prev = null,
216210 .next = null,
......@@ -233,7 +227,7 @@ pub fn writevPosix(
233227 };
234228
235229 suspend {
236 loop.posixFsRequest(&req_node);
230 global_event_loop.posixFsRequest(&req_node);
237231 }
238232
239233 return req_node.data.msg.WriteV.result;
......@@ -242,7 +236,7 @@ pub fn writevPosix(
242236pub const PReadVError = error{OutOfMemory} || File.ReadError;
243237
244238/// data - just the inner references - must live until preadv frame completes.
245pub fn preadv(loop: *Loop, fd: fd_t, data: []const []u8, offset: usize) PReadVError!usize {
239pub fn preadv(allocator: *Allocator, fd: fd_t, data: []const []u8, offset: usize) PReadVError!usize {
246240 assert(data.len != 0);
247241 switch (builtin.os) {
248242 .macosx,
......@@ -251,8 +245,8 @@ pub fn preadv(loop: *Loop, fd: fd_t, data: []const []u8, offset: usize) PReadVEr
251245 .netbsd,
252246 .dragonfly,
253247 => {
254 const iovecs = try loop.allocator.alloc(os.iovec, data.len);
255 defer loop.allocator.free(iovecs);
248 const iovecs = try allocator.alloc(os.iovec, data.len);
249 defer allocator.free(iovecs);
256250
257251 for (data) |buf, i| {
258252 iovecs[i] = os.iovec{
......@@ -261,21 +255,21 @@ pub fn preadv(loop: *Loop, fd: fd_t, data: []const []u8, offset: usize) PReadVEr
261255 };
262256 }
263257
264 return preadvPosix(loop, fd, iovecs, offset);
258 return preadvPosix(fd, iovecs, offset);
265259 },
266260 .windows => {
267 const data_copy = try std.mem.dupe(loop.allocator, []u8, data);
268 defer loop.allocator.free(data_copy);
269 return preadvWindows(loop, fd, data_copy, offset);
261 const data_copy = try std.mem.dupe(allocator, []u8, data);
262 defer allocator.free(data_copy);
263 return preadvWindows(fd, data_copy, offset);
270264 },
271265 else => @compileError("Unsupported OS"),
272266 }
273267}
274268
275269/// data must outlive the returned frame
276pub fn preadvWindows(loop: *Loop, fd: fd_t, data: []const []u8, offset: u64) !usize {
270pub fn preadvWindows(fd: fd_t, data: []const []u8, offset: u64) !usize {
277271 assert(data.len != 0);
278 if (data.len == 1) return preadWindows(loop, fd, data[0], offset);
272 if (data.len == 1) return preadWindows(fd, data[0], offset);
279273
280274 // TODO do these in parallel?
281275 var off: usize = 0;
......@@ -283,7 +277,7 @@ pub fn preadvWindows(loop: *Loop, fd: fd_t, data: []const []u8, offset: u64) !us
283277 var inner_off: usize = 0;
284278 while (true) {
285279 const v = data[iov_i];
286 const amt_read = try preadWindows(loop, fd, v[inner_off .. v.len - inner_off], offset + off);
280 const amt_read = try preadWindows(fd, v[inner_off .. v.len - inner_off], offset + off);
287281 off += amt_read;
288282 inner_off += amt_read;
289283 if (inner_off == v.len) {
......@@ -297,7 +291,7 @@ pub fn preadvWindows(loop: *Loop, fd: fd_t, data: []const []u8, offset: u64) !us
297291 }
298292}
299293
300pub fn preadWindows(loop: *Loop, fd: fd_t, data: []u8, offset: u64) !usize {
294pub fn preadWindows(fd: fd_t, data: []u8, offset: u64) !usize {
301295 var resume_node = Loop.ResumeNode.Basic{
302296 .base = Loop.ResumeNode{
303297 .id = Loop.ResumeNode.Id.Basic,
......@@ -312,9 +306,9 @@ pub fn preadWindows(loop: *Loop, fd: fd_t, data: []u8, offset: u64) !usize {
312306 },
313307 };
314308 // TODO only call create io completion port once per fd
315 _ = windows.CreateIoCompletionPort(fd, loop.os_data.io_port, undefined, undefined) catch undefined;
316 loop.beginOneEvent();
317 errdefer loop.finishOneEvent();
309 _ = windows.CreateIoCompletionPort(fd, global_event_loop.os_data.io_port, undefined, undefined) catch undefined;
310 global_event_loop.beginOneEvent();
311 errdefer global_event_loop.finishOneEvent();
318312
319313 errdefer {
320314 _ = windows.kernel32.CancelIoEx(fd, &resume_node.base.overlapped);
......@@ -336,12 +330,7 @@ pub fn preadWindows(loop: *Loop, fd: fd_t, data: []u8, offset: u64) !usize {
336330}
337331
338332/// iovecs must live until preadv frame completes
339pub fn preadvPosix(
340 loop: *Loop,
341 fd: fd_t,
342 iovecs: []const os.iovec,
343 offset: usize,
344) os.ReadError!usize {
333pub fn preadvPosix(fd: fd_t, iovecs: []const os.iovec, offset: usize) os.ReadError!usize {
345334 var req_node = RequestNode{
346335 .prev = null,
347336 .next = null,
......@@ -364,21 +353,16 @@ pub fn preadvPosix(
364353 },
365354 };
366355
367 errdefer loop.posixFsCancel(&req_node);
356 errdefer global_event_loop.posixFsCancel(&req_node);
368357
369358 suspend {
370 loop.posixFsRequest(&req_node);
359 global_event_loop.posixFsRequest(&req_node);
371360 }
372361
373362 return req_node.data.msg.PReadV.result;
374363}
375364
376pub fn openPosix(
377 loop: *Loop,
378 path: []const u8,
379 flags: u32,
380 mode: File.Mode,
381) File.OpenError!fd_t {
365pub fn openPosix(path: []const u8, flags: u32, mode: File.Mode) File.OpenError!fd_t {
382366 const path_c = try std.os.toPosixPath(path);
383367
384368 var req_node = RequestNode{
......@@ -403,21 +387,21 @@ pub fn openPosix(
403387 },
404388 };
405389
406 errdefer loop.posixFsCancel(&req_node);
390 errdefer global_event_loop.posixFsCancel(&req_node);
407391
408392 suspend {
409 loop.posixFsRequest(&req_node);
393 global_event_loop.posixFsRequest(&req_node);
410394 }
411395
412396 return req_node.data.msg.Open.result;
413397}
414398
415pub fn openRead(loop: *Loop, path: []const u8) File.OpenError!fd_t {
399pub fn openRead(path: []const u8) File.OpenError!fd_t {
416400 switch (builtin.os) {
417401 .macosx, .linux, .freebsd, .netbsd, .dragonfly => {
418402 const O_LARGEFILE = if (@hasDecl(os, "O_LARGEFILE")) os.O_LARGEFILE else 0;
419403 const flags = O_LARGEFILE | os.O_RDONLY | os.O_CLOEXEC;
420 return openPosix(loop, path, flags, File.default_mode);
404 return openPosix(path, flags, File.default_mode);
421405 },
422406
423407 .windows => return windows.CreateFile(
......@@ -436,12 +420,12 @@ pub fn openRead(loop: *Loop, path: []const u8) File.OpenError!fd_t {
436420
437421/// Creates if does not exist. Truncates the file if it exists.
438422/// Uses the default mode.
439pub fn openWrite(loop: *Loop, path: []const u8) File.OpenError!fd_t {
440 return openWriteMode(loop, path, File.default_mode);
423pub fn openWrite(path: []const u8) File.OpenError!fd_t {
424 return openWriteMode(path, File.default_mode);
441425}
442426
443427/// Creates if does not exist. Truncates the file if it exists.
444pub fn openWriteMode(loop: *Loop, path: []const u8, mode: File.Mode) File.OpenError!fd_t {
428pub fn openWriteMode(path: []const u8, mode: File.Mode) File.OpenError!fd_t {
445429 switch (builtin.os) {
446430 .macosx,
447431 .linux,
......@@ -451,7 +435,7 @@ pub fn openWriteMode(loop: *Loop, path: []const u8, mode: File.Mode) File.OpenEr
451435 => {
452436 const O_LARGEFILE = if (@hasDecl(os, "O_LARGEFILE")) os.O_LARGEFILE else 0;
453437 const flags = O_LARGEFILE | os.O_WRONLY | os.O_CREAT | os.O_CLOEXEC | os.O_TRUNC;
454 return openPosix(loop, path, flags, File.default_mode);
438 return openPosix(path, flags, File.default_mode);
455439 },
456440 .windows => return windows.CreateFile(
457441 path,
......@@ -467,16 +451,12 @@ pub fn openWriteMode(loop: *Loop, path: []const u8, mode: File.Mode) File.OpenEr
467451}
468452
469453/// Creates if does not exist. Does not truncate.
470pub fn openReadWrite(
471 loop: *Loop,
472 path: []const u8,
473 mode: File.Mode,
474) File.OpenError!fd_t {
454pub fn openReadWrite(path: []const u8, mode: File.Mode) File.OpenError!fd_t {
475455 switch (builtin.os) {
476456 .macosx, .linux, .freebsd, .netbsd, .dragonfly => {
477457 const O_LARGEFILE = if (@hasDecl(os, "O_LARGEFILE")) os.O_LARGEFILE else 0;
478458 const flags = O_LARGEFILE | os.O_RDWR | os.O_CREAT | os.O_CLOEXEC;
479 return openPosix(loop, path, flags, mode);
459 return openPosix(path, flags, mode);
480460 },
481461
482462 .windows => return windows.CreateFile(
......@@ -500,7 +480,7 @@ pub fn openReadWrite(
500480/// If you call `setHandle` then finishing will close the fd; otherwise finishing
501481/// will deallocate the `CloseOperation`.
502482pub const CloseOperation = struct {
503 loop: *Loop,
483 allocator: *Allocator,
504484 os_data: OsData,
505485
506486 const OsData = switch (builtin.os) {
......@@ -518,10 +498,10 @@ pub const CloseOperation = struct {
518498 close_req_node: RequestNode,
519499 };
520500
521 pub fn start(loop: *Loop) (error{OutOfMemory}!*CloseOperation) {
522 const self = try loop.allocator.create(CloseOperation);
501 pub fn start(allocator: *Allocator) (error{OutOfMemory}!*CloseOperation) {
502 const self = try allocator.create(CloseOperation);
523503 self.* = CloseOperation{
524 .loop = loop,
504 .allocator = allocator,
525505 .os_data = switch (builtin.os) {
526506 .linux, .macosx, .freebsd, .netbsd, .dragonfly => initOsDataPosix(self),
527507 .windows => OsData{ .handle = null },
......@@ -557,16 +537,16 @@ pub const CloseOperation = struct {
557537 .dragonfly,
558538 => {
559539 if (self.os_data.have_fd) {
560 self.loop.posixFsRequest(&self.os_data.close_req_node);
540 global_event_loop.posixFsRequest(&self.os_data.close_req_node);
561541 } else {
562 self.loop.allocator.destroy(self);
542 self.allocator.destroy(self);
563543 }
564544 },
565545 .windows => {
566546 if (self.os_data.handle) |handle| {
567547 os.close(handle);
568548 }
569 self.loop.allocator.destroy(self);
549 self.allocator.destroy(self);
570550 },
571551 else => @compileError("Unsupported OS"),
572552 }
......@@ -629,25 +609,25 @@ pub const CloseOperation = struct {
629609
630610/// contents must remain alive until writeFile completes.
631611/// TODO make this atomic or provide writeFileAtomic and rename this one to writeFileTruncate
632pub fn writeFile(loop: *Loop, path: []const u8, contents: []const u8) !void {
633 return writeFileMode(loop, path, contents, File.default_mode);
612pub fn writeFile(allocator: *Allocator, path: []const u8, contents: []const u8) !void {
613 return writeFileMode(allocator, path, contents, File.default_mode);
634614}
635615
636616/// contents must remain alive until writeFile completes.
637pub fn writeFileMode(loop: *Loop, path: []const u8, contents: []const u8, mode: File.Mode) !void {
617pub fn writeFileMode(allocator: *Allocator, path: []const u8, contents: []const u8, mode: File.Mode) !void {
638618 switch (builtin.os) {
639619 .linux,
640620 .macosx,
641621 .freebsd,
642622 .netbsd,
643623 .dragonfly,
644 => return writeFileModeThread(loop, path, contents, mode),
645 .windows => return writeFileWindows(loop, path, contents),
624 => return writeFileModeThread(allocator, path, contents, mode),
625 .windows => return writeFileWindows(path, contents),
646626 else => @compileError("Unsupported OS"),
647627 }
648628}
649629
650fn writeFileWindows(loop: *Loop, path: []const u8, contents: []const u8) !void {
630fn writeFileWindows(path: []const u8, contents: []const u8) !void {
651631 const handle = try windows.CreateFile(
652632 path,
653633 windows.GENERIC_WRITE,
......@@ -659,12 +639,12 @@ fn writeFileWindows(loop: *Loop, path: []const u8, contents: []const u8) !void {
659639 );
660640 defer os.close(handle);
661641
662 try pwriteWindows(loop, handle, contents, 0);
642 try pwriteWindows(handle, contents, 0);
663643}
664644
665fn writeFileModeThread(loop: *Loop, path: []const u8, contents: []const u8, mode: File.Mode) !void {
666 const path_with_null = try std.cstr.addNullByte(loop.allocator, path);
667 defer loop.allocator.free(path_with_null);
645fn writeFileModeThread(allocator: *Allocator, path: []const u8, contents: []const u8, mode: File.Mode) !void {
646 const path_with_null = try std.cstr.addNullByte(allocator, path);
647 defer allocator.free(path_with_null);
668648
669649 var req_node = RequestNode{
670650 .prev = null,
......@@ -688,10 +668,10 @@ fn writeFileModeThread(loop: *Loop, path: []const u8, contents: []const u8, mode
688668 },
689669 };
690670
691 errdefer loop.posixFsCancel(&req_node);
671 errdefer global_event_loop.posixFsCancel(&req_node);
692672
693673 suspend {
694 loop.posixFsRequest(&req_node);
674 global_event_loop.posixFsRequest(&req_node);
695675 }
696676
697677 return req_node.data.msg.WriteFile.result;
......@@ -700,21 +680,21 @@ fn writeFileModeThread(loop: *Loop, path: []const u8, contents: []const u8, mode
700680/// The frame resumes when the last data has been confirmed written, but before the file handle
701681/// is closed.
702682/// Caller owns returned memory.
703pub fn readFile(loop: *Loop, file_path: []const u8, max_size: usize) ![]u8 {
704 var close_op = try CloseOperation.start(loop);
683pub fn readFile(allocator: *Allocator, file_path: []const u8, max_size: usize) ![]u8 {
684 var close_op = try CloseOperation.start();
705685 defer close_op.finish();
706686
707 const fd = try openRead(loop, file_path);
687 const fd = try openRead(file_path);
708688 close_op.setHandle(fd);
709689
710 var list = std.ArrayList(u8).init(loop.allocator);
690 var list = std.ArrayList(u8).init(allocator);
711691 defer list.deinit();
712692
713693 while (true) {
714694 try list.ensureCapacity(list.len + mem.page_size);
715695 const buf = list.items[list.len..];
716696 const buf_array = [_][]u8{buf};
717 const amt = try preadv(loop, fd, buf_array, list.len);
697 const amt = try preadv(fd, buf_array, list.len);
718698 list.len += amt;
719699 if (list.len > max_size) {
720700 return error.FileTooBig;
......@@ -1392,16 +1372,15 @@ fn testFsWatch(loop: *Loop) !void {
13921372pub const OutStream = struct {
13931373 fd: fd_t,
13941374 stream: Stream,
1395 loop: *Loop,
1375 allocator: *Allocator,
13961376 offset: usize,
13971377
13981378 pub const Error = File.WriteError;
13991379 pub const Stream = event.io.OutStream(Error);
14001380
1401 pub fn init(loop: *Loop, fd: fd_t, offset: usize) OutStream {
1381 pub fn init(allocator: *Allocator, fd: fd_t, offset: usize) OutStream {
14021382 return OutStream{
14031383 .fd = fd,
1404 .loop = loop,
14051384 .offset = offset,
14061385 .stream = Stream{ .writeFn = writeFn },
14071386 };
......@@ -1411,23 +1390,22 @@ pub const OutStream = struct {
14111390 const self = @fieldParentPtr(OutStream, "stream", out_stream);
14121391 const offset = self.offset;
14131392 self.offset += bytes.len;
1414 return pwritev(self.loop, self.fd, [][]const u8{bytes}, offset);
1393 return pwritev(self.allocator, self.fd, [_][]const u8{bytes}, offset);
14151394 }
14161395};
14171396
14181397pub const InStream = struct {
14191398 fd: fd_t,
14201399 stream: Stream,
1421 loop: *Loop,
1400 allocator: *Allocator,
14221401 offset: usize,
14231402
14241403 pub const Error = PReadVError; // TODO make this not have OutOfMemory
14251404 pub const Stream = event.io.InStream(Error);
14261405
1427 pub fn init(loop: *Loop, fd: fd_t, offset: usize) InStream {
1406 pub fn init(allocator: *Allocator, fd: fd_t, offset: usize) InStream {
14281407 return InStream{
14291408 .fd = fd,
1430 .loop = loop,
14311409 .offset = offset,
14321410 .stream = Stream{ .readFn = readFn },
14331411 };
......@@ -1435,7 +1413,7 @@ pub const InStream = struct {
14351413
14361414 fn readFn(in_stream: *Stream, bytes: []u8) Error!usize {
14371415 const self = @fieldParentPtr(InStream, "stream", in_stream);
1438 const amt = try preadv(self.loop, self.fd, [][]u8{bytes}, self.offset);
1416 const amt = try preadv(self.allocator, self.fd, [_][]u8{bytes}, self.offset);
14391417 self.offset += amt;
14401418 return amt;
14411419 }