zigcoro added, p2p stab

Author: rsepassi

deps/libuv/build.wren

@@ -28,6 +28,19 @@ var libuv = Fn.new { |b, args|
   b.install("include/uv", headers)
   b.installLibConfig(zig.libConfig(b, "uv", {
     "ldflags": os["ldflags"] || [],
+    "libc": true,
+  }))
+}
+
+var zig = Fn.new { |b, args|
+  var zig = b.deptool("//toolchains/zig")
+  b.srcGlob("zig/*.zig")
+  b.install("zig", zig.moduleConfig(b, {
+    "root": b.src("zig/uv.zig"),
+    "modules": {
+      "zigcoro": b.dep("//deps/zigcoro"),
+    },
+    "c_deps": [zig.cDep(b.dep(":libuv"), "uv")],
   }))
 }
 

deps/libuv/zig/cimport.zig

@@ -0,0 +1,17 @@
+const std = @import("std");
+const uv = @import("uv.zig");
+const coro = @import("zigcoro");
+
+pub const uv_close_cb = struct {
+    frame: coro.Frame,
+
+    pub fn init(self: *@This(), handle: *uv.uv_handle_t) void {
+        uv.setHandleData(handle, self);
+        self.frame = coro.xframe();
+    }
+
+    pub fn cb(handle: [*c]uv.uv_handle_t) callconv(.C) void {
+        const self = uv.getHandleData(handle, @This());
+        coro.xresume(self.frame);
+    }
+};

deps/libuv/zig/common.zig

@@ -0,0 +1,201 @@
+const std = @import("std");
+const uv = @import("uv.zig");
+const coro = @import("zigcoro");
+
+const log = std.log.scoped(.uvzig);
+
+pub const File = struct {
+    loop: *uv.uv_loop_t,
+    fd: c_int,
+
+    const Data = struct {
+        frame: coro.Frame,
+        done: bool = false,
+        fn init() @This() {
+            return .{ .frame = coro.xframe() };
+        }
+    };
+
+    pub fn open(loop: *uv.uv_loop_t, path: [:0]const u8, flags: c_int, mode: c_int) !@This() {
+        log.debug("open {s} flags={b} mode={o}", .{ path, flags, mode });
+        var data = Data.init();
+        var req = uv.uv_fs_t{};
+        defer uv.uv_fs_req_cleanup(&req);
+        uv.setReqData(&req, &data);
+        try uv.check(uv.uv_fs_open(loop, &req, path, flags, mode, xresume));
+        while (!data.done) coro.xsuspend();
+        try uv.check(req.result);
+
+        return .{
+            .loop = loop,
+            .fd = @intCast(req.result),
+        };
+    }
+
+    pub fn close(self: @This()) void {
+        var req = uv.uv_fs_t{};
+        defer uv.uv_fs_req_cleanup(&req);
+        var data = Data.init();
+        uv.setReqData(&req, &data);
+        uv.check(uv.uv_fs_close(self.loop, &req, self.fd, xresume)) catch {};
+        coro.xsuspend();
+    }
+
+    fn xresume(req: [*c]uv.uv_fs_t) callconv(.C) void {
+        const data = uv.getReqData(req, Data);
+        data.done = true;
+        coro.xresume(data.frame);
+    }
+
+    const Error = error{
+        Read,
+        Write,
+    } || uv.Error;
+    const Reader = std.io.Reader(@This(), Error, read);
+
+    fn read(self: @This(), buf: []u8) Error!usize {
+        log.debug("read {d}", .{buf.len});
+        var req = uv.uv_fs_t{};
+        defer uv.uv_fs_req_cleanup(&req);
+        var data = Data.init();
+        uv.setReqData(&req, &data);
+        var uvbuf = uv.newbuf(buf);
+        try uv.check(uv.uv_fs_read(self.loop, &req, self.fd, &uvbuf, 1, -1, xresume));
+        coro.xsuspend();
+        if (req.result < 0) {
+            return Error.Read;
+        } else {
+            return @intCast(req.result);
+        }
+    }
+
+    const Writer = std.io.Writer(@This(), Error, write);
+
+    fn write(self: @This(), buf: []const u8) Error!usize {
+        log.debug("write", .{});
+        var req = uv.uv_fs_t{};
+        defer uv.uv_fs_req_cleanup(&req);
+        var data = Data.init();
+        uv.setReqData(&req, &data);
+        var uvbuf = uv.newbuf(@constCast(buf));
+        try uv.check(uv.uv_fs_write(self.loop, &req, self.fd, &uvbuf, 1, -1, xresume));
+        coro.xsuspend();
+        if (req.result < 0) {
+            return Error.Write;
+        } else {
+            return @intCast(req.result);
+        }
+    }
+
+    pub fn writer(self: @This()) Writer {
+        return .{ .context = self };
+    }
+
+    pub fn reader(self: @This()) Reader {
+        return .{ .context = self };
+    }
+
+    pub fn stat(self: @This()) !uv.uv_stat_t {
+        var req = uv.uv_fs_t{};
+        defer uv.uv_fs_req_cleanup(&req);
+        var data = Data.init();
+        uv.setReqData(&req, &data);
+        try uv.check(uv.uv_fs_fstat(self.loop, &req, self.fd, xresume));
+        coro.xsuspend();
+        try uv.check(req.result);
+        return uv.uv_fs_get_statbuf(&req).*;
+    }
+
+    pub fn fsync(self: @This()) !void {
+        var req = uv.uv_fs_t{};
+        defer uv.uv_fs_req_cleanup(&req);
+        var data = Data.init();
+        uv.setReqData(&req, &data);
+        try uv.check(uv.uv_fs_fsync(self.loop, &req, self.fd, xresume));
+        coro.xsuspend();
+        try uv.check(req.result);
+    }
+
+    pub fn sendfile(self: @This(), dst: c_int, src_offset: ?usize, n: usize) !usize {
+        log.debug("send {d}", .{n});
+        var req = uv.uv_fs_t{};
+        defer uv.uv_fs_req_cleanup(&req);
+        var data = Data.init();
+        uv.setReqData(&req, &data);
+
+        try uv.check(uv.uv_fs_sendfile(self.loop, &req, dst, self.fd, @intCast(src_offset orelse 0), n, xresume));
+        coro.xsuspend();
+
+        try uv.check(req.result);
+        return @intCast(req.result);
+    }
+
+    // TODO:
+    // chown
+    // utime
+    // chmod
+    // datasync
+};
+
+pub const fs = struct {
+    pub fn mkdir(loop: *uv.uv_loop_t, path: [:0]const u8, mode: c_int) !void {
+        var req: uv_fs_cb = undefined;
+        req.init();
+        defer req.deinit();
+        try uv.check(uv.uv_fs_mkdir(loop, &req.req, path.ptr, mode, uv_fs_cb.cb));
+        coro.xsuspend();
+    }
+
+    pub fn mkdtemp(loop: *uv.uv_loop_t, tpl: [:0]const u8, out: []u8) ![:0]const u8 {
+        var req: uv_fs_cb = undefined;
+        req.init();
+        defer req.deinit();
+        try uv.check(uv.uv_fs_mkdtemp(loop, &req.req, tpl.ptr, uv_fs_cb.cb));
+        coro.xsuspend();
+        const n = std.mem.len(req.req.path);
+        if (n >= out.len) return error.BufTooSmall;
+        std.mem.copyForwards(u8, out, req.req.path[0..n]);
+        out[n] = 0;
+        return out[0..n :0];
+    }
+
+    const TmpFileInfo = struct {
+        path: ?[:0]const u8,
+        fd: uv.uv_file,
+    };
+    pub fn mkstemp(loop: *uv.uv_loop_t, tpl: [:0]const u8, out: ?[]u8) !TmpFileInfo {
+        var req: uv_fs_cb = undefined;
+        req.init();
+        defer req.deinit();
+        try uv.check(uv.uv_fs_mkstemp(loop, &req.req, tpl.ptr, uv_fs_cb.cb));
+        coro.xsuspend();
+        var info = TmpFileInfo{ .path = null, .fd = @intCast(req.req.result) };
+        if (out) |buf| {
+            const n = std.mem.len(req.req.path);
+            if (n >= buf.len) return error.BufTooSmall;
+            std.mem.copyForwards(u8, buf, req.req.path[0..n]);
+            buf[n] = 0;
+            info.path = buf[0..n :0];
+        }
+        return info;
+    }
+};
+
+const uv_fs_cb = struct {
+    frame: coro.Frame,
+    req: uv.uv_fs_t,
+
+    fn init(self: *@This()) void {
+        uv.setReqData(&self.req, self);
+        self.frame = coro.xframe();
+    }
+
+    fn deinit(self: @This()) void {
+        uv.uv_fs_req_cleanup(@constCast(&self.req));
+    }
+
+    fn cb(req: [*c]uv.uv_fs_t) callconv(.C) void {
+        const self = uv.getReqData(req, @This());
+        coro.xresume(self.frame);
+    }
+};

deps/libuv/zig/file.zig

@@ -0,0 +1,136 @@
+const std = @import("std");
+const uv = @import("uv.zig");
+const coro = @import("zigcoro");
+
+const log = std.log.scoped(.uvzig);
+
+pub const Process = struct {
+    pub const StdioOpts = union(enum) {
+        inherit_fd: uv.uv_file,
+        inherit_stream: *uv.uv_stream_t,
+        create_pipe: struct {
+            pipe: *uv.uv_stream_t,
+            flow: enum { RO, WO, RW },
+            nonblock: bool = false,
+        },
+    };
+    const RunOpts = struct {
+        env: ?[][:0]const u8 = null,
+        cwd: ?[:0]const u8 = null,
+        stdio: [3]?StdioOpts = .{ null, null, null },
+    };
+
+    pub const Status = struct {
+        exit_status: i64,
+        term_signal: c_int,
+    };
+
+    pub fn run(loop: *uv.uv_loop_t, alloc: std.mem.Allocator, args: [][:0]const u8, opts: RunOpts) !Status {
+        log.debug("process run", .{});
+        var handle = uv.uv_process_t{};
+        defer {
+            var closer = Closer.init();
+            closer.close(@ptrCast(&handle));
+        }
+        var data = Data.init();
+        uv.setHandleData(&handle, &data);
+
+        const cargs = try alloc.alloc([*c]const u8, args.len + 1);
+        defer alloc.free(cargs);
+        cargs[args.len] = null;
+        for (args, 0..) |a, i| {
+            log.debug("- {s}", .{a});
+            cargs[i] = a;
+        }
+
+        var cenv: ?[][*c]const u8 = null;
+        if (opts.env) |env| {
+            cenv = try alloc.alloc([*c]const u8, env.len + 1);
+            cenv.?[env.len] = null;
+            for (env, 0..) |e, i| cenv.?[i] = e;
+        }
+        defer if (cenv) |e| alloc.free(e);
+
+        var stdio: [3]uv.uv_stdio_container_t = undefined;
+        for (0..3) |i| {
+            if (opts.stdio[i]) |sopt| {
+                switch (sopt) {
+                    .inherit_fd => |fd| {
+                        log.debug("stdio[{d}]=.inherit_fd={d}", .{ i, fd });
+                        stdio[i].flags = uv.UV_INHERIT_FD;
+                        stdio[i].data.fd = fd;
+                    },
+                    .inherit_stream => |s| {
+                        log.debug("stdio[{d}]=.inherit_stream={*}", .{ i, s });
+                        stdio[i].flags = uv.UV_INHERIT_STREAM;
+                        stdio[i].data.stream = s;
+                    },
+                    .create_pipe => |opt| {
+                        log.debug("stdio[{d}]=.create_pipe {s}", .{ i, @tagName(opt.flow) });
+                        var flags = uv.UV_CREATE_PIPE;
+                        if (opt.nonblock) flags |= uv.UV_NONBLOCK_PIPE;
+                        flags |= switch (opt.flow) {
+                            .RO => uv.UV_READABLE_PIPE,
+                            .WO => uv.UV_WRITABLE_PIPE,
+                            .RW => uv.UV_READABLE_PIPE | uv.UV_WRITABLE_PIPE,
+                        };
+
+                        stdio[i].flags = @intCast(flags);
+                        stdio[i].data.stream = opt.pipe;
+                    },
+                }
+            } else {
+                stdio[i].flags = uv.UV_IGNORE;
+            }
+        }
+
+        var o = uv.uv_process_options_t{};
+        o.file = cargs[0];
+        o.args = @constCast(@ptrCast(cargs.ptr));
+        o.env = if (cenv) |env| @constCast(@ptrCast(env.ptr)) else null;
+        o.cwd = if (opts.cwd) |cwd| cwd.ptr else null;
+        o.stdio_count = 3;
+        o.stdio = &stdio;
+        o.exit_cb = Data.onExit;
+
+        try uv.check(uv.uv_spawn(loop, &handle, &o));
+        log.debug("process spawned pid={d}", .{handle.pid});
+        coro.xsuspend();
+        log.debug("process exited pid={d} code={d}", .{ handle.pid, data.out.?.exit_status });
+
+        return data.out.?;
+    }
+
+    const Data = struct {
+        frame: coro.Frame,
+        out: ?Status = null,
+        fn init() @This() {
+            return .{ .frame = coro.xframe() };
+        }
+
+        fn onExit(process: [*c]uv.uv_process_t, exit_status: i64, term_signal: c_int) callconv(.C) void {
+            const data = uv.getHandleData(process, Data);
+            data.out = .{
+                .exit_status = exit_status,
+                .term_signal = term_signal,
+            };
+            coro.xresume(data.frame);
+        }
+    };
+};
+
+const Closer = struct {
+    frame: coro.Frame,
+    fn init() @This() {
+        return .{ .frame = coro.xframe() };
+    }
+    fn close(self: *@This(), handle: [*c]uv.uv_handle_t) void {
+        uv.setHandleData(handle, self);
+        uv.uv_close(handle, onClose);
+        coro.xsuspend();
+    }
+    fn onClose(handle: [*c]uv.uv_handle_t) callconv(.C) void {
+        const data = uv.getHandleData(handle, @This());
+        coro.xresume(data.frame);
+    }
+};