Lindenii Project Forge
Add ev::exec, ev::wait, ev::kill Signed-of-by: Drew DeVault <drew@ddevault.org>
use errors; use ev; use fmt; use os; use os::exec; use time; export fn main() void = { const loop = ev::newloop()!; defer ev::finish(&loop); const cmd = exec::cmd(os::args[1], os::args[2..]...)!; const child = ev::exec(&loop, &cmd)!; ev::wait(child, &child_exited); const timer = ev::newtimer(&loop, &expired, time::clock::MONOTONIC)!; ev::timer_configure(timer, 5 * time::SECOND, 0); ev::setuser(timer, child); for (ev::dispatch(&loop, -1)!) void; }; fn expired(timer: *ev::file) void = { fmt::println("Child timed out, sending SIGTERM")!; const child = ev::getuser(timer): *ev::file; ev::kill(child)!; }; fn child_exited(child: *ev::file, r: (exec::status | errors::error)) void = { match (r) { case let st: exec::status => const exit = exec::exit(&st); fmt::printfln("child exited: {}", exec::exitstr(exit))!; case let err: errors::error => fmt::printfln("ev::wait: {}", errors::strerror(err))!; }; const loop = ev::getloop(child); ev::stop(loop); };
use errors; use io; use net; use net::ip; use rt; use unix::signal; export type op = enum u64 { NONE = 0, READV = 1 << 0, WRITEV = 1 << 1, READABLE = 1 << 16, WRITABLE = 2 << 16, ACCEPT = 3 << 16, CONNECT_TCP = 4 << 16, CONNECT_UNIX = 5 << 16, SIGNAL = 6 << 16, TIMER = 7 << 16, SENDTO = 8 << 16, RECVFROM = 9 << 16, SEND = 10 << 16, RECV = 11 << 16,
WAIT = 12 << 16,
}; export type fflags = enum uint { NONE = 0, BLOCKING = 1 << 31, }; export type file = struct { fd: io::file, ev: *loop, flags: fflags, op: op, cb: nullable *opaque, cb2: nullable *opaque, user: nullable *opaque, prio: int, // Operation-specific data union { struct { rvbuf: io::vector, rvec: []io::vector, wvbuf: io::vector, wvec: []io::vector, }, sockflag: net::sockflag, sigmask: signal::sigset, sendrecv: struct { sbuf: []u8, rbuf: []u8, dest: ip::addr, port: u16, }, }, }; // Registers a file descriptor with an event loop. export fn register( loop: *loop, fd: io::file, user: nullable *opaque = null, ) (*file | errors::error | nomem) = { const file = alloc(file { flags = fflags::NONE, fd = fd, ev = loop, op = op::NONE, user = user, ... })?; let ev = rt::epoll_event { events = 0, data = rt::epoll_data { fd = 0, } }; ev.data.ptr = file; match (rt::epoll_ctl(loop.fd, rt::EPOLL_CTL_ADD, fd, &ev)) { case void => yield; case let err: rt::errno => if (err == rt::EPERM) { // epoll(2) does not support regular files, use blocking // I/O instead file.flags = fflags::BLOCKING; return file; }; return errors::errno(err); }; return file; }; // Unregisters a file object with an event loop and frees resources associated // with it. Does not close the underlying file descriptor. export fn unregister(file: *file) void = { const loop = file.ev; if (file.flags & fflags::BLOCKING == 0) { // The only way that this could fail is in the event of a // use-after-free or if the user fucks around and constructs a // custom [[file]] which was never registered, so assert on // error. rt::epoll_ctl(loop.fd, rt::EPOLL_CTL_DEL, file.fd, null)!; }; if (file.op == op::SIGNAL) { signal_restore(file); }; for (let ev &.. loop.events) { if (file != ev.data.ptr) { continue; }; ev.events = 0; ev.data.ptr = null: *opaque; ev.data.fd = 0; break; }; free(file); }; // Unregisters a file object with an event loop, frees resources associated with // it, and closes the underlying file descriptor. export fn close(file: *file) void = { const fd = file.fd; unregister(file); io::close(fd)!; }; // Sets the user data field on this file object to the provided object. export fn setuser(file: *file, user: nullable *opaque) void = { file.user = user; }; // Returns the user data field from this file object. If the field was null, an // assertion is raised. export fn getuser(file: *file) *opaque = { return file.user as *opaque; }; // Returns the file descriptor for a given file. Note that ev assumes that it // will be responsible for all I/O on the file and any user modifications may // cause the event loop to enter an invalid state. export fn getfd(file: *file) io::file = { return file.fd; }; // Returns the event loop for a given file. export fn getloop(file: *file) *loop = { return file.ev; }; // Sets the priority on this file object. Used to priorize concurrent events, // starting on the next dispatch. The lowest run first. export fn setprio(file: *file, prio: int) void = { file.prio = prio; }; fn sort_events(a: const *opaque, b: const *opaque) int = { const a = a: *rt::epoll_event; const b = b: *rt::epoll_event; if (a.data.fd == 0 || b.data.fd == 0) { return 0; }; const a = a.data.ptr: *file; const b = b.data.ptr: *file; return a.prio - b.prio; }; // Updates epoll events for a given file. For internal use. fn file_epoll_ctl(file: *file) void = { let events = rt::EPOLLONESHOT; if (file.op & op::READV != 0 || file.op == op::READABLE) { events |= rt::EPOLLIN | rt::EPOLLHUP; }; if (file.op & op::WRITEV != 0 || file.op == op::WRITABLE) { events |= rt::EPOLLOUT | rt::EPOLLHUP; }; switch (file.op) { case op::ACCEPT => events |= rt::EPOLLIN; case op::CONNECT_TCP, op::CONNECT_UNIX => events |= rt::EPOLLOUT; case op::SIGNAL => events |= rt::EPOLLIN; case op::TIMER => events &= ~rt::EPOLLONESHOT; events |= rt::EPOLLIN; case op::SEND, op::SENDTO => events |= rt::EPOLLOUT; case op::RECV, op::RECVFROM =>
events |= rt::EPOLLIN; case op::WAIT =>
events |= rt::EPOLLIN; case => yield; }; let ev = rt::epoll_event { events = events, data = rt::epoll_data { fd = 0, }, }; ev.data.ptr = file; // This can only fail under conditions associated with EPOLLEXCLUSIVE, // which we do not support. rt::epoll_ctl(file.ev.fd, rt::EPOLL_CTL_MOD, file.fd, &ev)!; };
use errors; use io; use rt; use sort; use time; use types; use unix::signal; // Dispatch callback. See [[ondispatch]]. export type dispatchcb = fn(loop: *loop, user: nullable *opaque) void; export type ondispatch = struct { cb: *dispatchcb, user: nullable *opaque, loop: *loop, }; export type loop = struct { fd: io::file, events: []rt::epoll_event, dispatch: []*ondispatch, stop: bool, }; // Creates a new event loop. The user must pass the return value to [[finish]] // to free associated resources when done using the loop. // // The optional "events" parameter controls how many events may be pending at // once. Most applications should not need to configure this parameter. export fn newloop(events: size = 256) (loop | nomem | errors::error) = { const fd = match (rt::epoll_create1(rt::EPOLL_CLOEXEC)) { case let fd: int => yield fd: io::file; case let err: rt::errno => return errors::errno(err); }; return loop { fd = fd, events = alloc([rt::epoll_event { events = 0, data = rt::epoll_data { fd = 0, } }...], events)?, dispatch = [], stop = false, }; }; // Frees resources associated with an event loop. Must only be called once per // event loop object. Calling finish invalidates all I/O objects associated with // the event loop. export fn finish(loop: *loop) void = { free(loop.events); io::close(loop.fd)!; }; // Returns an [[io::file]] for this event loop which can be polled on when // events are available for processing, for chaining together different event // loops. The exact semantics of this function are platform-specific, and it may // not be available for all implementations. export fn loop_file(loop: *loop) io::file = { return loop.fd; }; // Registers a callback to be invoked before the next call to [[dispatch]] // processes requests. The callback may schedule additional I/O requests to be // processed in this batch. // // Dispatch callbacks are only called once. If you wish to schedule another // dispatch callback after the first, call [[do]] again. export fn do( loop: *loop, cb: *dispatchcb, user: nullable *opaque = null, ) (req | nomem) = { const dispatch = alloc(ondispatch { cb = cb, user = user, loop = loop, })?; append(loop.dispatch, dispatch)?; return mkreq(&do_cancel, dispatch); }; fn do_cancel(req: *req) void = { const dispatch = req.user: *ondispatch; const loop = dispatch.loop; for (let i = 0z; i < len(loop.dispatch); i += 1) { if (loop.dispatch[i] == dispatch) { delete(loop.dispatch[i]); break; }; }; free(dispatch); }; // Dispatches the event loop, waiting for new events and calling their callbacks // as appropriate. // // A timeout of -1 will block indefinitely until the next event occurs. A // timeout of 0 will cause dispatch to return immediately if no events are // available to process. Portable use of the timeout argument supports only // millisecond granularity of up to 24 days ([[types::INT_MAX]] milliseconds). // Negative values other than -1 will cause the program to abort. // // Returns false if the loop has been stopped via [[stop]], or true otherwise. export fn dispatch( loop: *loop, timeout: time::duration, ) (bool | errors::error) = { const millis: int = if (timeout == -1) { yield -1; } else if (timeout < 0) { abort("ev::dispatch: invalid timeout"); } else { yield (timeout / time::MILLISECOND): int; }; if (loop.stop) { return false; }; let todo = loop.dispatch; loop.dispatch = []; for (let dispatch .. todo) { dispatch.cb(loop, dispatch.user); free(dispatch); }; free(todo); if (len(loop.events) == 0) { return true; }; // TODO: Deal with signals const maxev = len(loop.events); assert(maxev <= types::INT_MAX: size, "ev::dispatch: too many events"); const nevent = match(rt::epoll_pwait( loop.fd, &loop.events[0], maxev: int, millis, null)) { case let nevent: int => yield nevent; case let err: rt::errno => switch (err) { case rt::EINTR => // We shallow system suspension error code return true; case => abort("ev::dispatch: epoll_pwait failure"); }; }; if (nevent == 0) { return true; }; sort::sort(loop.events[..nevent], size(rt::epoll_event), &sort_events); for (let ev &.. loop.events[..nevent]) { const file = ev.data.ptr: *file; if (ev.events == 0) { continue; }; const pending = file.op; if (ev.events & (rt::EPOLLIN | rt::EPOLLHUP) != 0 && file.op & op::READV != 0) { readv_ready(file, ev); }; if (ev.events & (rt::EPOLLOUT | rt::EPOLLHUP) != 0 && file.op & op::WRITEV != 0) { writev_ready(file, ev); }; switch (pending) { case op::NONE => abort("No operation pending for ready object"); case op::READABLE => readable_ready(file, ev); case op::WRITABLE => writable_ready(file, ev); case op::ACCEPT => accept_ready(file, ev); case op::CONNECT_TCP => connect_tcp_ready(file, ev); case op::CONNECT_UNIX => connect_unix_ready(file, ev); case op::SIGNAL => signal_ready(file, ev); case op::TIMER => timer_ready(file, ev); case op::SENDTO => sendto_ready(file, ev); case op::RECVFROM => recvfrom_ready(file, ev); case op::SEND => send_ready(file, ev); case op::RECV => recv_ready(file, ev);
case op::WAIT => wait_ready(file, ev);
case => assert(pending & ~(op::READV | op::WRITEV) == 0); }; }; return !loop.stop; }; // Signals the loop to stop processing events. If called during a callback, it // will cause that invocation of [[dispatch]] to return false. Otherwise, false // will be returned only upon the next call to [[dispatch]]. export fn stop(loop: *loop) void = { loop.stop = true; };
use errors; use os::exec; use rt; use time; use unix::signal::{sig, code}; // Starts a child process prepared with [[exec::cmd]] and returns a [[file]] // that can be used to [[wait]] for it to exit. export fn exec( loop: *loop, cmd: *exec::command, ) (*file | errors::error | nomem) = { let pidfd: int = 0; let args = rt::clone_args { flags = rt::CLONE_PIDFD, pidfd = &pidfd: uintptr: u64, exit_signal = 0, ... }; const pid = match (rt::clone3(&args)) { case let pid: rt::pid_t => yield pid; case let errno: rt::errno => return errors::errno(errno); }; if (pid == 0) { exec::exec(cmd); }; return register(loop, pidfd); }; // A callback for a [[wait]] operation. export type waitcb = fn(file: *file, result: (exec::status | errors::error)) void; // Waits for a process to exit. After the callback is run, the [[file]] is // closed and unregistered from the event loop and may not be used again. export fn wait(proc: *file, cb: *waitcb) req = { assert(proc.op & op::WAIT == 0); proc.op |= op::WAIT; proc.cb = cb; file_epoll_ctl(proc); return mkreq(&wait_cancel, proc); }; fn wait_cancel(req: *req) void = { const proc = req.user: *file; assert(proc.op & op::WAIT != 0); proc.op &= ~op::WAIT; file_epoll_ctl(proc); }; fn wait_ready(proc: *file, ev: *rt::epoll_event) void = { assert(proc.op & op::WAIT != 0); assert(proc.cb != null); const cb = proc.cb: *waitcb; proc.op &= ~op::WAIT; file_epoll_ctl(proc); let si = rt::siginfo { ... }; let ru = rt::rusage { ... }; match (rt::waitid(rt::idtype::P_PIDFD, proc.fd: rt::id_t, &si, rt::WEXITED | rt::WSTOPPED | rt::WCONTINUED, &ru)) { case let err: rt::errno => cb(proc, errors::errno(err)); case void => void; }; // Convert the siginfo data into a wait(2)-style exit status let status = 0i; if (si.si_code == code::EXITED) { status = si.si_status << 8; } else { status = si.si_status; }; let st = exec::status { status = status, ... }; rusage(&st, &ru); cb(proc, st); close(proc); }; // Copied from os::exec fn rusage(st: *exec::status, ru: *rt::rusage) void = { st.rusage.utime = time::instant { sec = ru.ru_utime.tv_sec, nsec = ru.ru_utime.tv_usec * time::MICROSECOND: i64, }; st.rusage.stime = time::instant { sec = ru.ru_stime.tv_sec, nsec = ru.ru_stime.tv_usec * time::MICROSECOND: i64, }; st.rusage.maxrss = ru.ru_maxrss; st.rusage.minflt = ru.ru_minflt; st.rusage.majflt = ru.ru_majflt; st.rusage.inblock = ru.ru_inblock; st.rusage.oublock = ru.ru_oublock; st.rusage.nvcsw = ru.ru_nvcsw; st.rusage.nivcsw = ru.ru_nivcsw; }; // Sends a signal to a process started with [[ev::exec]]. export fn kill(child: *file, sig: sig = sig::TERM) (void | errors::error) = { match (rt::pidfd_send_signal(child.fd, sig, null, 0)) { case void => void; case let err: rt::errno => return errors::errno(err); }; };