authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2019-08-16 21:29:29-04:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2019-08-16 21:29:29-04:00
log3dce41b61a20c23c494a485b1d3092e7c67dd97f
treec5aee6a9c66bdfd672134885fb3001c9aee5c7c6
parentbf7b6fbbdb7d28c0d7dba3e17c46ce156712cfc8
signaturelock-open Commit is signed but in an unrecognized format.

improvements to std lib for event-based I/O

8 files changed, 219 insertions(+), 50 deletions(-)

std/c.zig+1-1
...@@ -64,6 +64,7 @@ pub extern "c" fn fstat(fd: fd_t, buf: *Stat) c_int;...@@ -64,6 +64,7 @@ pub extern "c" fn fstat(fd: fd_t, buf: *Stat) c_int;
64pub extern "c" fn @"fstat$INODE64"(fd: fd_t, buf: *Stat) c_int;64pub extern "c" fn @"fstat$INODE64"(fd: fd_t, buf: *Stat) c_int;
65pub extern "c" fn lseek(fd: fd_t, offset: isize, whence: c_int) isize;65pub extern "c" fn lseek(fd: fd_t, offset: isize, whence: c_int) isize;
66pub extern "c" fn open(path: [*]const u8, oflag: c_uint, ...) c_int;66pub extern "c" fn open(path: [*]const u8, oflag: c_uint, ...) c_int;
67pub extern "c" fn openat(fd: c_int, path: [*]const u8, oflag: c_uint, ...) c_int;
67pub extern "c" fn raise(sig: c_int) c_int;68pub extern "c" fn raise(sig: c_int) c_int;
68pub extern "c" fn read(fd: fd_t, buf: [*]u8, nbyte: usize) isize;69pub extern "c" fn read(fd: fd_t, buf: [*]u8, nbyte: usize) isize;
69pub extern "c" fn pread(fd: fd_t, buf: [*]u8, nbyte: usize, offset: u64) isize;70pub extern "c" fn pread(fd: fd_t, buf: [*]u8, nbyte: usize, offset: u64) isize;
...@@ -112,7 +113,6 @@ pub extern "c" fn accept4(sockfd: fd_t, addr: *sockaddr, addrlen: *socklen_t, fl...@@ -112,7 +113,6 @@ pub extern "c" fn accept4(sockfd: fd_t, addr: *sockaddr, addrlen: *socklen_t, fl
112pub extern "c" fn getsockopt(sockfd: fd_t, level: c_int, optname: c_int, optval: *c_void, optlen: *socklen_t) c_int;113pub extern "c" fn getsockopt(sockfd: fd_t, level: c_int, optname: c_int, optval: *c_void, optlen: *socklen_t) c_int;
113pub extern "c" fn kill(pid: pid_t, sig: c_int) c_int;114pub extern "c" fn kill(pid: pid_t, sig: c_int) c_int;
114pub extern "c" fn getdirentries(fd: fd_t, buf_ptr: [*]u8, nbytes: usize, basep: *i64) isize;115pub extern "c" fn getdirentries(fd: fd_t, buf_ptr: [*]u8, nbytes: usize, basep: *i64) isize;
115pub extern "c" fn openat(fd: c_int, path: [*]const u8, flags: c_int) c_int;
116pub extern "c" fn setgid(ruid: c_uint, euid: c_uint) c_int;116pub extern "c" fn setgid(ruid: c_uint, euid: c_uint) c_int;
117pub extern "c" fn setuid(uid: c_uint) c_int;117pub extern "c" fn setuid(uid: c_uint) c_int;
118pub extern "c" fn clock_gettime(clk_id: c_int, tp: *timespec) c_int;118pub extern "c" fn clock_gettime(clk_id: c_int, tp: *timespec) c_int;
std/event/fs.zig+89-46
...@@ -23,6 +23,7 @@ pub const Request = struct {...@@ -23,6 +23,7 @@ pub const Request = struct {
23 };23 };
2424
25 pub const Msg = union(enum) {25 pub const Msg = union(enum) {
26 WriteV: WriteV,
26 PWriteV: PWriteV,27 PWriteV: PWriteV,
27 PReadV: PReadV,28 PReadV: PReadV,
28 Open: Open,29 Open: Open,
...@@ -30,6 +31,14 @@ pub const Request = struct {...@@ -30,6 +31,14 @@ pub const Request = struct {
30 WriteFile: WriteFile,31 WriteFile: WriteFile,
31 End, // special - means the fs thread should exit32 End, // special - means the fs thread should exit
3233
34 pub const WriteV = struct {
35 fd: fd_t,
36 iov: []const os.iovec_const,
37 result: Error!void,
38
39 pub const Error = os.WriteError;
40 };
41
33 pub const PWriteV = struct {42 pub const PWriteV = struct {
34 fd: fd_t,43 fd: fd_t,
35 iov: []const os.iovec_const,44 iov: []const os.iovec_const,
...@@ -77,7 +86,7 @@ pub const Request = struct {...@@ -77,7 +86,7 @@ pub const Request = struct {
77pub const PWriteVError = error{OutOfMemory} || File.WriteError;86pub const PWriteVError = error{OutOfMemory} || File.WriteError;
7887
79/// data - just the inner references - must live until pwritev frame completes.88/// data - just the inner references - must live until pwritev frame completes.
80pub async fn pwritev(loop: *Loop, fd: fd_t, data: []const []const u8, offset: usize) PWriteVError!void {89pub fn pwritev(loop: *Loop, fd: fd_t, data: []const []const u8, offset: usize) PWriteVError!void {
81 switch (builtin.os) {90 switch (builtin.os) {
82 .macosx,91 .macosx,
83 .linux,92 .linux,
...@@ -94,31 +103,31 @@ pub async fn pwritev(loop: *Loop, fd: fd_t, data: []const []const u8, offset: us...@@ -94,31 +103,31 @@ pub async fn pwritev(loop: *Loop, fd: fd_t, data: []const []const u8, offset: us
94 };103 };
95 }104 }
96105
97 return await (async pwritevPosix(loop, fd, iovecs, offset) catch unreachable);106 return pwritevPosix(loop, fd, iovecs, offset);
98 },107 },
99 .windows => {108 .windows => {
100 const data_copy = try std.mem.dupe(loop.allocator, []const u8, data);109 const data_copy = try std.mem.dupe(loop.allocator, []const u8, data);
101 defer loop.allocator.free(data_copy);110 defer loop.allocator.free(data_copy);
102 return await (async pwritevWindows(loop, fd, data, offset) catch unreachable);111 return pwritevWindows(loop, fd, data, offset);
103 },112 },
104 else => @compileError("Unsupported OS"),113 else => @compileError("Unsupported OS"),
105 }114 }
106}115}
107116
108/// data must outlive the returned frame117/// data must outlive the returned frame
109pub async fn pwritevWindows(loop: *Loop, fd: fd_t, data: []const []const u8, offset: usize) os.WindowsWriteError!void {118pub fn pwritevWindows(loop: *Loop, fd: fd_t, data: []const []const u8, offset: usize) os.WindowsWriteError!void {
110 if (data.len == 0) return;119 if (data.len == 0) return;
111 if (data.len == 1) return await (async pwriteWindows(loop, fd, data[0], offset) catch unreachable);120 if (data.len == 1) return pwriteWindows(loop, fd, data[0], offset);
112121
113 // TODO do these in parallel122 // TODO do these in parallel
114 var off = offset;123 var off = offset;
115 for (data) |buf| {124 for (data) |buf| {
116 try await (async pwriteWindows(loop, fd, buf, off) catch unreachable);125 try pwriteWindows(loop, fd, buf, off);
117 off += buf.len;126 off += buf.len;
118 }127 }
119}128}
120129
121pub async fn pwriteWindows(loop: *Loop, fd: fd_t, data: []const u8, offset: u64) os.WindowsWriteError!void {130pub fn pwriteWindows(loop: *Loop, fd: fd_t, data: []const u8, offset: u64) os.WindowsWriteError!void {
122 var resume_node = Loop.ResumeNode.Basic{131 var resume_node = Loop.ResumeNode.Basic{
123 .base = Loop.ResumeNode{132 .base = Loop.ResumeNode{
124 .id = Loop.ResumeNode.Id.Basic,133 .id = Loop.ResumeNode.Id.Basic,
...@@ -158,7 +167,7 @@ pub async fn pwriteWindows(loop: *Loop, fd: fd_t, data: []const u8, offset: u64)...@@ -158,7 +167,7 @@ pub async fn pwriteWindows(loop: *Loop, fd: fd_t, data: []const u8, offset: u64)
158}167}
159168
160/// iovecs must live until pwritev frame completes.169/// iovecs must live until pwritev frame completes.
161pub async fn pwritevPosix(170pub fn pwritevPosix(
162 loop: *Loop,171 loop: *Loop,
163 fd: fd_t,172 fd: fd_t,
164 iovecs: []const os.iovec_const,173 iovecs: []const os.iovec_const,
...@@ -195,10 +204,44 @@ pub async fn pwritevPosix(...@@ -195,10 +204,44 @@ pub async fn pwritevPosix(
195 return req_node.data.msg.PWriteV.result;204 return req_node.data.msg.PWriteV.result;
196}205}
197206
207/// iovecs must live until pwritev frame completes.
208pub fn writevPosix(
209 loop: *Loop,
210 fd: fd_t,
211 iovecs: []const os.iovec_const,
212) os.WriteError!void {
213 var req_node = RequestNode{
214 .prev = null,
215 .next = null,
216 .data = Request{
217 .msg = Request.Msg{
218 .WriteV = Request.Msg.WriteV{
219 .fd = fd,
220 .iov = iovecs,
221 .result = undefined,
222 },
223 },
224 .finish = Request.Finish{
225 .TickNode = Loop.NextTickNode{
226 .prev = null,
227 .next = null,
228 .data = @frame(),
229 },
230 },
231 },
232 };
233
234 suspend {
235 loop.posixFsRequest(&req_node);
236 }
237
238 return req_node.data.msg.WriteV.result;
239}
240
198pub const PReadVError = error{OutOfMemory} || File.ReadError;241pub const PReadVError = error{OutOfMemory} || File.ReadError;
199242
200/// data - just the inner references - must live until preadv frame completes.243/// data - just the inner references - must live until preadv frame completes.
201pub async fn preadv(loop: *Loop, fd: fd_t, data: []const []u8, offset: usize) PReadVError!usize {244pub fn preadv(loop: *Loop, fd: fd_t, data: []const []u8, offset: usize) PReadVError!usize {
202 assert(data.len != 0);245 assert(data.len != 0);
203 switch (builtin.os) {246 switch (builtin.os) {
204 .macosx,247 .macosx,
...@@ -216,21 +259,21 @@ pub async fn preadv(loop: *Loop, fd: fd_t, data: []const []u8, offset: usize) PR...@@ -216,21 +259,21 @@ pub async fn preadv(loop: *Loop, fd: fd_t, data: []const []u8, offset: usize) PR
216 };259 };
217 }260 }
218261
219 return await (async preadvPosix(loop, fd, iovecs, offset) catch unreachable);262 return preadvPosix(loop, fd, iovecs, offset);
220 },263 },
221 .windows => {264 .windows => {
222 const data_copy = try std.mem.dupe(loop.allocator, []u8, data);265 const data_copy = try std.mem.dupe(loop.allocator, []u8, data);
223 defer loop.allocator.free(data_copy);266 defer loop.allocator.free(data_copy);
224 return await (async preadvWindows(loop, fd, data_copy, offset) catch unreachable);267 return preadvWindows(loop, fd, data_copy, offset);
225 },268 },
226 else => @compileError("Unsupported OS"),269 else => @compileError("Unsupported OS"),
227 }270 }
228}271}
229272
230/// data must outlive the returned frame273/// data must outlive the returned frame
231pub async fn preadvWindows(loop: *Loop, fd: fd_t, data: []const []u8, offset: u64) !usize {274pub fn preadvWindows(loop: *Loop, fd: fd_t, data: []const []u8, offset: u64) !usize {
232 assert(data.len != 0);275 assert(data.len != 0);
233 if (data.len == 1) return await (async preadWindows(loop, fd, data[0], offset) catch unreachable);276 if (data.len == 1) return preadWindows(loop, fd, data[0], offset);
234277
235 // TODO do these in parallel?278 // TODO do these in parallel?
236 var off: usize = 0;279 var off: usize = 0;
...@@ -238,7 +281,7 @@ pub async fn preadvWindows(loop: *Loop, fd: fd_t, data: []const []u8, offset: u6...@@ -238,7 +281,7 @@ pub async fn preadvWindows(loop: *Loop, fd: fd_t, data: []const []u8, offset: u6
238 var inner_off: usize = 0;281 var inner_off: usize = 0;
239 while (true) {282 while (true) {
240 const v = data[iov_i];283 const v = data[iov_i];
241 const amt_read = try await (async preadWindows(loop, fd, v[inner_off .. v.len - inner_off], offset + off) catch unreachable);284 const amt_read = try preadWindows(loop, fd, v[inner_off .. v.len - inner_off], offset + off);
242 off += amt_read;285 off += amt_read;
243 inner_off += amt_read;286 inner_off += amt_read;
244 if (inner_off == v.len) {287 if (inner_off == v.len) {
...@@ -252,7 +295,7 @@ pub async fn preadvWindows(loop: *Loop, fd: fd_t, data: []const []u8, offset: u6...@@ -252,7 +295,7 @@ pub async fn preadvWindows(loop: *Loop, fd: fd_t, data: []const []u8, offset: u6
252 }295 }
253}296}
254297
255pub async fn preadWindows(loop: *Loop, fd: fd_t, data: []u8, offset: u64) !usize {298pub fn preadWindows(loop: *Loop, fd: fd_t, data: []u8, offset: u64) !usize {
256 var resume_node = Loop.ResumeNode.Basic{299 var resume_node = Loop.ResumeNode.Basic{
257 .base = Loop.ResumeNode{300 .base = Loop.ResumeNode{
258 .id = Loop.ResumeNode.Id.Basic,301 .id = Loop.ResumeNode.Id.Basic,
...@@ -291,7 +334,7 @@ pub async fn preadWindows(loop: *Loop, fd: fd_t, data: []u8, offset: u64) !usize...@@ -291,7 +334,7 @@ pub async fn preadWindows(loop: *Loop, fd: fd_t, data: []u8, offset: u64) !usize
291}334}
292335
293/// iovecs must live until preadv frame completes336/// iovecs must live until preadv frame completes
294pub async fn preadvPosix(337pub fn preadvPosix(
295 loop: *Loop,338 loop: *Loop,
296 fd: fd_t,339 fd: fd_t,
297 iovecs: []const os.iovec,340 iovecs: []const os.iovec,
...@@ -328,7 +371,7 @@ pub async fn preadvPosix(...@@ -328,7 +371,7 @@ pub async fn preadvPosix(
328 return req_node.data.msg.PReadV.result;371 return req_node.data.msg.PReadV.result;
329}372}
330373
331pub async fn openPosix(374pub fn openPosix(
332 loop: *Loop,375 loop: *Loop,
333 path: []const u8,376 path: []const u8,
334 flags: u32,377 flags: u32,
...@@ -367,11 +410,11 @@ pub async fn openPosix(...@@ -367,11 +410,11 @@ pub async fn openPosix(
367 return req_node.data.msg.Open.result;410 return req_node.data.msg.Open.result;
368}411}
369412
370pub async fn openRead(loop: *Loop, path: []const u8) File.OpenError!fd_t {413pub fn openRead(loop: *Loop, path: []const u8) File.OpenError!fd_t {
371 switch (builtin.os) {414 switch (builtin.os) {
372 .macosx, .linux, .freebsd, .netbsd => {415 .macosx, .linux, .freebsd, .netbsd => {
373 const flags = os.O_LARGEFILE | os.O_RDONLY | os.O_CLOEXEC;416 const flags = os.O_LARGEFILE | os.O_RDONLY | os.O_CLOEXEC;
374 return await (async openPosix(loop, path, flags, File.default_mode) catch unreachable);417 return openPosix(loop, path, flags, File.default_mode);
375 },418 },
376419
377 .windows => return windows.CreateFile(420 .windows => return windows.CreateFile(
...@@ -390,12 +433,12 @@ pub async fn openRead(loop: *Loop, path: []const u8) File.OpenError!fd_t {...@@ -390,12 +433,12 @@ pub async fn openRead(loop: *Loop, path: []const u8) File.OpenError!fd_t {
390433
391/// Creates if does not exist. Truncates the file if it exists.434/// Creates if does not exist. Truncates the file if it exists.
392/// Uses the default mode.435/// Uses the default mode.
393pub async fn openWrite(loop: *Loop, path: []const u8) File.OpenError!fd_t {436pub fn openWrite(loop: *Loop, path: []const u8) File.OpenError!fd_t {
394 return await (async openWriteMode(loop, path, File.default_mode) catch unreachable);437 return openWriteMode(loop, path, File.default_mode);
395}438}
396439
397/// Creates if does not exist. Truncates the file if it exists.440/// Creates if does not exist. Truncates the file if it exists.
398pub async fn openWriteMode(loop: *Loop, path: []const u8, mode: File.Mode) File.OpenError!fd_t {441pub fn openWriteMode(loop: *Loop, path: []const u8, mode: File.Mode) File.OpenError!fd_t {
399 switch (builtin.os) {442 switch (builtin.os) {
400 .macosx,443 .macosx,
401 .linux,444 .linux,
...@@ -403,7 +446,7 @@ pub async fn openWriteMode(loop: *Loop, path: []const u8, mode: File.Mode) File....@@ -403,7 +446,7 @@ pub async fn openWriteMode(loop: *Loop, path: []const u8, mode: File.Mode) File.
403 .netbsd,446 .netbsd,
404 => {447 => {
405 const flags = os.O_LARGEFILE | os.O_WRONLY | os.O_CREAT | os.O_CLOEXEC | os.O_TRUNC;448 const flags = os.O_LARGEFILE | os.O_WRONLY | os.O_CREAT | os.O_CLOEXEC | os.O_TRUNC;
406 return await (async openPosix(loop, path, flags, File.default_mode) catch unreachable);449 return openPosix(loop, path, flags, File.default_mode);
407 },450 },
408 .windows => return windows.CreateFile(451 .windows => return windows.CreateFile(
409 path,452 path,
...@@ -419,7 +462,7 @@ pub async fn openWriteMode(loop: *Loop, path: []const u8, mode: File.Mode) File....@@ -419,7 +462,7 @@ pub async fn openWriteMode(loop: *Loop, path: []const u8, mode: File.Mode) File.
419}462}
420463
421/// Creates if does not exist. Does not truncate.464/// Creates if does not exist. Does not truncate.
422pub async fn openReadWrite(465pub fn openReadWrite(
423 loop: *Loop,466 loop: *Loop,
424 path: []const u8,467 path: []const u8,
425 mode: File.Mode,468 mode: File.Mode,
...@@ -427,7 +470,7 @@ pub async fn openReadWrite(...@@ -427,7 +470,7 @@ pub async fn openReadWrite(
427 switch (builtin.os) {470 switch (builtin.os) {
428 .macosx, .linux, .freebsd, .netbsd => {471 .macosx, .linux, .freebsd, .netbsd => {
429 const flags = os.O_LARGEFILE | os.O_RDWR | os.O_CREAT | os.O_CLOEXEC;472 const flags = os.O_LARGEFILE | os.O_RDWR | os.O_CREAT | os.O_CLOEXEC;
430 return await (async openPosix(loop, path, flags, mode) catch unreachable);473 return openPosix(loop, path, flags, mode);
431 },474 },
432475
433 .windows => return windows.CreateFile(476 .windows => return windows.CreateFile(
...@@ -576,24 +619,24 @@ pub const CloseOperation = struct {...@@ -576,24 +619,24 @@ pub const CloseOperation = struct {
576619
577/// contents must remain alive until writeFile completes.620/// contents must remain alive until writeFile completes.
578/// TODO make this atomic or provide writeFileAtomic and rename this one to writeFileTruncate621/// TODO make this atomic or provide writeFileAtomic and rename this one to writeFileTruncate
579pub async fn writeFile(loop: *Loop, path: []const u8, contents: []const u8) !void {622pub fn writeFile(loop: *Loop, path: []const u8, contents: []const u8) !void {
580 return await (async writeFileMode(loop, path, contents, File.default_mode) catch unreachable);623 return writeFileMode(loop, path, contents, File.default_mode);
581}624}
582625
583/// contents must remain alive until writeFile completes.626/// contents must remain alive until writeFile completes.
584pub async fn writeFileMode(loop: *Loop, path: []const u8, contents: []const u8, mode: File.Mode) !void {627pub fn writeFileMode(loop: *Loop, path: []const u8, contents: []const u8, mode: File.Mode) !void {
585 switch (builtin.os) {628 switch (builtin.os) {
586 .linux,629 .linux,
587 .macosx,630 .macosx,
588 .freebsd,631 .freebsd,
589 .netbsd,632 .netbsd,
590 => return await (async writeFileModeThread(loop, path, contents, mode) catch unreachable),633 => return writeFileModeThread(loop, path, contents, mode),
591 .windows => return await (async writeFileWindows(loop, path, contents) catch unreachable),634 .windows => return writeFileWindows(loop, path, contents),
592 else => @compileError("Unsupported OS"),635 else => @compileError("Unsupported OS"),
593 }636 }
594}637}
595638
596async fn writeFileWindows(loop: *Loop, path: []const u8, contents: []const u8) !void {639fn writeFileWindows(loop: *Loop, path: []const u8, contents: []const u8) !void {
597 const handle = try windows.CreateFile(640 const handle = try windows.CreateFile(
598 path,641 path,
599 windows.GENERIC_WRITE,642 windows.GENERIC_WRITE,
...@@ -605,10 +648,10 @@ async fn writeFileWindows(loop: *Loop, path: []const u8, contents: []const u8) !...@@ -605,10 +648,10 @@ async fn writeFileWindows(loop: *Loop, path: []const u8, contents: []const u8) !
605 );648 );
606 defer os.close(handle);649 defer os.close(handle);
607650
608 try await (async pwriteWindows(loop, handle, contents, 0) catch unreachable);651 try pwriteWindows(loop, handle, contents, 0);
609}652}
610653
611async fn writeFileModeThread(loop: *Loop, path: []const u8, contents: []const u8, mode: File.Mode) !void {654fn writeFileModeThread(loop: *Loop, path: []const u8, contents: []const u8, mode: File.Mode) !void {
612 const path_with_null = try std.cstr.addNullByte(loop.allocator, path);655 const path_with_null = try std.cstr.addNullByte(loop.allocator, path);
613 defer loop.allocator.free(path_with_null);656 defer loop.allocator.free(path_with_null);
614657
...@@ -646,11 +689,11 @@ async fn writeFileModeThread(loop: *Loop, path: []const u8, contents: []const u8...@@ -646,11 +689,11 @@ async fn writeFileModeThread(loop: *Loop, path: []const u8, contents: []const u8
646/// The frame resumes when the last data has been confirmed written, but before the file handle689/// The frame resumes when the last data has been confirmed written, but before the file handle
647/// is closed.690/// is closed.
648/// Caller owns returned memory.691/// Caller owns returned memory.
649pub async fn readFile(loop: *Loop, file_path: []const u8, max_size: usize) ![]u8 {692pub fn readFile(loop: *Loop, file_path: []const u8, max_size: usize) ![]u8 {
650 var close_op = try CloseOperation.start(loop);693 var close_op = try CloseOperation.start(loop);
651 defer close_op.finish();694 defer close_op.finish();
652695
653 const fd = try await (async openRead(loop, file_path) catch unreachable);696 const fd = try openRead(loop, file_path);
654 close_op.setHandle(fd);697 close_op.setHandle(fd);
655698
656 var list = std.ArrayList(u8).init(loop.allocator);699 var list = std.ArrayList(u8).init(loop.allocator);
...@@ -660,7 +703,7 @@ pub async fn readFile(loop: *Loop, file_path: []const u8, max_size: usize) ![]u8...@@ -660,7 +703,7 @@ pub async fn readFile(loop: *Loop, file_path: []const u8, max_size: usize) ![]u8
660 try list.ensureCapacity(list.len + mem.page_size);703 try list.ensureCapacity(list.len + mem.page_size);
661 const buf = list.items[list.len..];704 const buf = list.items[list.len..];
662 const buf_array = [_][]u8{buf};705 const buf_array = [_][]u8{buf};
663 const amt = try await (async preadv(loop, fd, buf_array, list.len) catch unreachable);706 const amt = try preadv(loop, fd, buf_array, list.len);
664 list.len += amt;707 list.len += amt;
665 if (list.len > max_size) {708 if (list.len > max_size) {
666 return error.FileTooBig;709 return error.FileTooBig;
...@@ -1273,11 +1316,11 @@ const test_tmp_dir = "std_event_fs_test";...@@ -1273,11 +1316,11 @@ const test_tmp_dir = "std_event_fs_test";
1273// return result;1316// return result;
1274//}1317//}
12751318
1276async fn testFsWatchCantFail(loop: *Loop, result: *(anyerror!void)) void {1319fn testFsWatchCantFail(loop: *Loop, result: *(anyerror!void)) void {
1277 result.* = await (async testFsWatch(loop) catch unreachable);1320 result.* = testFsWatch(loop);
1278}1321}
12791322
1280async fn testFsWatch(loop: *Loop) !void {1323fn testFsWatch(loop: *Loop) !void {
1281 const file_path = try std.fs.path.join(loop.allocator, [][]const u8{ test_tmp_dir, "file.txt" });1324 const file_path = try std.fs.path.join(loop.allocator, [][]const u8{ test_tmp_dir, "file.txt" });
1282 defer loop.allocator.free(file_path);1325 defer loop.allocator.free(file_path);
12831326
...@@ -1288,27 +1331,27 @@ async fn testFsWatch(loop: *Loop) !void {...@@ -1288,27 +1331,27 @@ async fn testFsWatch(loop: *Loop) !void {
1288 const line2_offset = 7;1331 const line2_offset = 7;
12891332
1290 // first just write then read the file1333 // first just write then read the file
1291 try await try async writeFile(loop, file_path, contents);1334 try writeFile(loop, file_path, contents);
12921335
1293 const read_contents = try await try async readFile(loop, file_path, 1024 * 1024);1336 const read_contents = try readFile(loop, file_path, 1024 * 1024);
1294 testing.expectEqualSlices(u8, contents, read_contents);1337 testing.expectEqualSlices(u8, contents, read_contents);
12951338
1296 // now watch the file1339 // now watch the file
1297 var watch = try Watch(void).create(loop, 0);1340 var watch = try Watch(void).create(loop, 0);
1298 defer watch.destroy();1341 defer watch.destroy();
12991342
1300 testing.expect((try await try async watch.addFile(file_path, {})) == null);1343 testing.expect((try watch.addFile(file_path, {})) == null);
13011344
1302 const ev = try async watch.channel.get();1345 const ev = async watch.channel.get();
1303 var ev_consumed = false;1346 var ev_consumed = false;
1304 defer if (!ev_consumed) await ev;1347 defer if (!ev_consumed) await ev;
13051348
1306 // overwrite line 21349 // overwrite line 2
1307 const fd = try await try async openReadWrite(loop, file_path, File.default_mode);1350 const fd = try await openReadWrite(loop, file_path, File.default_mode);
1308 {1351 {
1309 defer os.close(fd);1352 defer os.close(fd);
13101353
1311 try await try async pwritev(loop, fd, []const []const u8{"lorem ipsum"}, line2_offset);1354 try pwritev(loop, fd, []const []const u8{"lorem ipsum"}, line2_offset);
1312 }1355 }
13131356
1314 ev_consumed = true;1357 ev_consumed = true;
...@@ -1316,7 +1359,7 @@ async fn testFsWatch(loop: *Loop) !void {...@@ -1316,7 +1359,7 @@ async fn testFsWatch(loop: *Loop) !void {
1316 WatchEventId.CloseWrite => {},1359 WatchEventId.CloseWrite => {},
1317 WatchEventId.Delete => @panic("wrong event"),1360 WatchEventId.Delete => @panic("wrong event"),
1318 }1361 }
1319 const contents_updated = try await try async readFile(loop, file_path, 1024 * 1024);1362 const contents_updated = try readFile(loop, file_path, 1024 * 1024);
1320 testing.expectEqualSlices(u8,1363 testing.expectEqualSlices(u8,
1321 \\line 11364 \\line 1
1322 \\lorem ipsum1365 \\lorem ipsum
std/event/loop.zig+6
...@@ -89,12 +89,15 @@ pub const Loop = struct {...@@ -89,12 +89,15 @@ pub const Loop = struct {
89 pub const IoMode = enum {89 pub const IoMode = enum {
90 blocking,90 blocking,
91 evented,91 evented,
92 mixed,
92 };93 };
93 pub const io_mode: IoMode = if (@hasDecl(root, "io_mode")) root.io_mode else IoMode.blocking;94 pub const io_mode: IoMode = if (@hasDecl(root, "io_mode")) root.io_mode else IoMode.blocking;
94 var global_instance_state: Loop = undefined;95 var global_instance_state: Loop = undefined;
96 threadlocal var per_thread_instance: ?*Loop = null;
95 const default_instance: ?*Loop = switch (io_mode) {97 const default_instance: ?*Loop = switch (io_mode) {
96 .blocking => null,98 .blocking => null,
97 .evented => &global_instance_state,99 .evented => &global_instance_state,
100 .mixed => per_thread_instance,
98 };101 };
99 pub const instance: ?*Loop = if (@hasDecl(root, "event_loop")) root.event_loop else default_instance;102 pub const instance: ?*Loop = if (@hasDecl(root, "event_loop")) root.event_loop else default_instance;
100103
...@@ -796,6 +799,9 @@ pub const Loop = struct {...@@ -796,6 +799,9 @@ pub const Loop = struct {
796 while (self.os_data.fs_queue.get()) |node| {799 while (self.os_data.fs_queue.get()) |node| {
797 switch (node.data.msg) {800 switch (node.data.msg) {
798 .End => return,801 .End => return,
802 .WriteV => |*msg| {
803 msg.result = os.writev(msg.fd, msg.iov);
804 },
799 .PWriteV => |*msg| {805 .PWriteV => |*msg| {
800 msg.result = os.pwritev(msg.fd, msg.iov, msg.offset);806 msg.result = os.pwritev(msg.fd, msg.iov, msg.offset);
801 },807 },
std/fs.zig+13
...@@ -442,6 +442,8 @@ pub fn deleteTree(allocator: *Allocator, full_path: []const u8) DeleteTreeError!...@@ -442,6 +442,8 @@ pub fn deleteTree(allocator: *Allocator, full_path: []const u8) DeleteTreeError!
442 }442 }
443}443}
444444
445/// TODO: separate this API into the one that opens directory handles to then subsequently open
446/// files, and into the one that reads files from an open directory handle.
445pub const Dir = struct {447pub const Dir = struct {
446 handle: Handle,448 handle: Handle,
447 allocator: *Allocator,449 allocator: *Allocator,
...@@ -564,6 +566,17 @@ pub const Dir = struct {...@@ -564,6 +566,17 @@ pub const Dir = struct {
564 }566 }
565 }567 }
566568
569 pub fn openRead(self: Dir, file_path: []const u8) os.OpenError!File {
570 const path_c = try os.toPosixPath(file_path);
571 return self.openReadC(&path_c);
572 }
573
574 pub fn openReadC(self: Dir, file_path: [*]const u8) OpenError!File {
575 const flags = os.O_LARGEFILE | os.O_RDONLY;
576 const fd = try os.openatC(self.handle.fd, file_path, flags, 0);
577 return File.openHandle(fd);
578 }
579
567 fn nextDarwin(self: *Dir) !?Entry {580 fn nextDarwin(self: *Dir) !?Entry {
568 start_over: while (true) {581 start_over: while (true) {
569 if (self.handle.index >= self.handle.end_index) {582 if (self.handle.index >= self.handle.end_index) {
std/fs/file.zig+8
...@@ -302,6 +302,14 @@ pub const File = struct {...@@ -302,6 +302,14 @@ pub const File = struct {
302 return os.write(self.handle, bytes);302 return os.write(self.handle, bytes);
303 }303 }
304304
305 pub fn writev_iovec(self: File, iovecs: []const os.iovec_const) WriteError!void {
306 if (std.event.Loop.instance) |loop| {
307 return std.event.fs.writevPosix(loop, self.handle, iovecs);
308 } else {
309 return os.writev(self.handle, iovecs);
310 }
311 }
312
305 pub fn inStream(file: File) InStream {313 pub fn inStream(file: File) InStream {
306 return InStream{314 return InStream{
307 .file = file,315 .file = file,
std/net.zig+30
...@@ -215,3 +215,33 @@ test "std.net.parseIp6" {...@@ -215,3 +215,33 @@ test "std.net.parseIp6" {
215 assert(addr.addr[1] == 0x01);215 assert(addr.addr[1] == 0x01);
216 assert(addr.addr[2] == 0x00);216 assert(addr.addr[2] == 0x00);
217}217}
218
219pub fn connectUnixSocket(path: []const u8) !std.fs.File {
220 const opt_non_block = if (std.event.Loop.instance != null) os.SOCK_NONBLOCK else 0;
221 const sockfd = try os.socket(
222 os.AF_UNIX,
223 os.SOCK_STREAM | os.SOCK_CLOEXEC | opt_non_block,
224 0,
225 );
226 errdefer os.close(sockfd);
227
228 var sock_addr = os.sockaddr{
229 .un = os.sockaddr_un{
230 .family = os.AF_UNIX,
231 .path = undefined,
232 },
233 };
234
235 if (path.len > @typeOf(sock_addr.un.path).len) return error.NameTooLong;
236 mem.copy(u8, sock_addr.un.path[0..], path);
237 const size = @intCast(u32, @sizeOf(os.sa_family_t) + path.len);
238 if (std.event.Loop.instance) |loop| {
239 try os.connect_async(sockfd, &sock_addr, size);
240 try loop.linuxWaitFd(sockfd, os.EPOLLIN | os.EPOLLOUT | os.EPOLLET);
241 try os.getsockoptError(sockfd);
242 } else {
243 try os.connect(sockfd, &sock_addr, size);
244 }
245
246 return std.fs.File.openHandle(sockfd);
247}
std/os.zig+71-3
...@@ -440,6 +440,33 @@ pub fn write(fd: fd_t, bytes: []const u8) WriteError!void {...@@ -440,6 +440,33 @@ pub fn write(fd: fd_t, bytes: []const u8) WriteError!void {
440440
441/// Write multiple buffers to a file descriptor. Keeps trying if it gets interrupted.441/// Write multiple buffers to a file descriptor. Keeps trying if it gets interrupted.
442/// This function is for blocking file descriptors only. For non-blocking, see442/// This function is for blocking file descriptors only. For non-blocking, see
443/// `writevAsync`.
444pub fn writev(fd: fd_t, iov: []const iovec_const) WriteError!void {
445 while (true) {
446 // TODO handle the case when iov_len is too large and get rid of this @intCast
447 const rc = system.writev(fd, iov.ptr, @intCast(u32, iov.len));
448 switch (errno(rc)) {
449 0 => return,
450 EINTR => continue,
451 EINVAL => unreachable,
452 EFAULT => unreachable,
453 EAGAIN => unreachable, // This function is for blocking writes.
454 EBADF => unreachable, // Always a race condition.
455 EDESTADDRREQ => unreachable, // `connect` was never called.
456 EDQUOT => return error.DiskQuota,
457 EFBIG => return error.FileTooBig,
458 EIO => return error.InputOutput,
459 ENOSPC => return error.NoSpaceLeft,
460 EPERM => return error.AccessDenied,
461 EPIPE => return error.BrokenPipe,
462 else => |err| return unexpectedErrno(err),
463 }
464 }
465}
466
467/// Write multiple buffers to a file descriptor, with a position offset.
468/// Keeps trying if it gets interrupted.
469/// This function is for blocking file descriptors only. For non-blocking, see
443/// `pwritevAsync`.470/// `pwritevAsync`.
444pub fn pwritev(fd: fd_t, iov: []const iovec_const, offset: u64) WriteError!void {471pub fn pwritev(fd: fd_t, iov: []const iovec_const, offset: u64) WriteError!void {
445 if (darwin.is_the_target) {472 if (darwin.is_the_target) {
...@@ -524,7 +551,6 @@ pub const OpenError = error{...@@ -524,7 +551,6 @@ pub const OpenError = error{
524};551};
525552
526/// Open and possibly create a file. Keeps trying if it gets interrupted.553/// Open and possibly create a file. Keeps trying if it gets interrupted.
527/// `file_path` needs to be copied in memory to add a null terminating byte.
528/// See also `openC`.554/// See also `openC`.
529pub fn open(file_path: []const u8, flags: u32, perm: usize) OpenError!fd_t {555pub fn open(file_path: []const u8, flags: u32, perm: usize) OpenError!fd_t {
530 const file_path_c = try toPosixPath(file_path);556 const file_path_c = try toPosixPath(file_path);
...@@ -564,6 +590,47 @@ pub fn openC(file_path: [*]const u8, flags: u32, perm: usize) OpenError!fd_t {...@@ -564,6 +590,47 @@ pub fn openC(file_path: [*]const u8, flags: u32, perm: usize) OpenError!fd_t {
564 }590 }
565}591}
566592
593/// Open and possibly create a file. Keeps trying if it gets interrupted.
594/// `file_path` is relative to the open directory handle `dir_fd`.
595/// See also `openatC`.
596pub fn openat(dir_fd: fd_t, file_path: []const u8, flags: u32, mode: usize) OpenError!fd_t {
597 const file_path_c = try toPosixPath(file_path);
598 return openatC(dir_fd, &file_path_c, flags, mode);
599}
600
601/// Open and possibly create a file. Keeps trying if it gets interrupted.
602/// `file_path` is relative to the open directory handle `dir_fd`.
603/// See also `openat`.
604pub fn openatC(dir_fd: fd_t, file_path: [*]const u8, flags: u32, mode: usize) OpenError!fd_t {
605 while (true) {
606 const rc = system.openat(dir_fd, file_path, flags, mode);
607 switch (errno(rc)) {
608 0 => return @intCast(fd_t, rc),
609 EINTR => continue,
610
611 EFAULT => unreachable,
612 EINVAL => unreachable,
613 EACCES => return error.AccessDenied,
614 EFBIG => return error.FileTooBig,
615 EOVERFLOW => return error.FileTooBig,
616 EISDIR => return error.IsDir,
617 ELOOP => return error.SymLinkLoop,
618 EMFILE => return error.ProcessFdQuotaExceeded,
619 ENAMETOOLONG => return error.NameTooLong,
620 ENFILE => return error.SystemFdQuotaExceeded,
621 ENODEV => return error.NoDevice,
622 ENOENT => return error.FileNotFound,
623 ENOMEM => return error.SystemResources,
624 ENOSPC => return error.NoSpaceLeft,
625 ENOTDIR => return error.NotDir,
626 EPERM => return error.AccessDenied,
627 EEXIST => return error.PathAlreadyExists,
628 EBUSY => return error.DeviceBusy,
629 else => |err| return unexpectedErrno(err),
630 }
631 }
632}
633
567pub fn dup2(old_fd: fd_t, new_fd: fd_t) !void {634pub fn dup2(old_fd: fd_t, new_fd: fd_t) !void {
568 while (true) {635 while (true) {
569 switch (errno(system.dup2(old_fd, new_fd))) {636 switch (errno(system.dup2(old_fd, new_fd))) {
...@@ -1655,7 +1722,7 @@ pub const ConnectError = error{...@@ -1655,7 +1722,7 @@ pub const ConnectError = error{
1655/// For non-blocking, see `connect_async`.1722/// For non-blocking, see `connect_async`.
1656pub fn connect(sockfd: i32, sock_addr: *sockaddr, len: socklen_t) ConnectError!void {1723pub fn connect(sockfd: i32, sock_addr: *sockaddr, len: socklen_t) ConnectError!void {
1657 while (true) {1724 while (true) {
1658 switch (errno(system.connect(sockfd, sock_addr, @sizeOf(sockaddr)))) {1725 switch (errno(system.connect(sockfd, sock_addr, len))) {
1659 0 => return,1726 0 => return,
1660 EACCES => return error.PermissionDenied,1727 EACCES => return error.PermissionDenied,
1661 EPERM => return error.PermissionDenied,1728 EPERM => return error.PermissionDenied,
...@@ -1683,7 +1750,8 @@ pub fn connect(sockfd: i32, sock_addr: *sockaddr, len: socklen_t) ConnectError!v...@@ -1683,7 +1750,8 @@ pub fn connect(sockfd: i32, sock_addr: *sockaddr, len: socklen_t) ConnectError!v
1683/// It expects to receive EINPROGRESS`.1750/// It expects to receive EINPROGRESS`.
1684pub fn connect_async(sockfd: i32, sock_addr: *sockaddr, len: socklen_t) ConnectError!void {1751pub fn connect_async(sockfd: i32, sock_addr: *sockaddr, len: socklen_t) ConnectError!void {
1685 while (true) {1752 while (true) {
1686 switch (errno(system.connect(sockfd, sock_addr, @sizeOf(sockaddr)))) {1753 switch (errno(system.connect(sockfd, sock_addr, len))) {
1754 EINVAL => unreachable,
1687 EINTR => continue,1755 EINTR => continue,
1688 0, EINPROGRESS => return,1756 0, EINPROGRESS => return,
1689 EACCES => return error.PermissionDenied,1757 EACCES => return error.PermissionDenied,
std/os/bits/linux.zig+1
...@@ -783,6 +783,7 @@ pub const socklen_t = u32;...@@ -783,6 +783,7 @@ pub const socklen_t = u32;
783pub const sockaddr = extern union {783pub const sockaddr = extern union {
784 in: sockaddr_in,784 in: sockaddr_in,
785 in6: sockaddr_in6,785 in6: sockaddr_in6,
786 un: sockaddr_un,
786};787};
787788
788pub const sockaddr_in = extern struct {789pub const sockaddr_in = extern struct {