Lindenii Project Forge
Flesh out dispatch
use errors; use io; use rt; export type op = enum { NONE, READV, WRITEV, }; export type file = struct { fd: io::file, ev: *loop, // Pending operation on this file object op: op,
callback: nullable *void,
cb: nullable *void,
// Operation-specific data vbuf: rt::iovec, union { vec: []rt::iovec, }, }; // Registers a file descriptor with an event loop. export fn register( loop: *loop, fd: io::file, ) (*file | errors::error) = { const file = alloc(file { fd = fd, ev = loop, op = op::NONE, ... }); let ev = rt::epoll_event { events = 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 => 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(loop: *loop, file: *file) void = { // 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)!; free(file); }; // Modifies the epoll events for a given file. For internal use. fn filemod(file: *file, events: u32) void = { let ev = rt::epoll_event { events = events, ... }; 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 io; use rt; // A callback for a [[read]] or [[readv]] operation. export type readcb = *fn(file: *file, result: (size | io::EOF | io::error)) void; // Schedules a read operation on a file object. export fn read( file: *file, cb: *readcb, buf: []u8, ) req = { file.vbuf = io::mkvector(buf); // XXX: Bit of a hack to avoid allocating a slice const vec = (&file.vbuf: *[*]io::vector)[..1]; return readv(file, cb, vec...); }; // Schedules a vectored read operation on a file object. export fn readv( file: *file, cb: *readcb, vec: io::vector... ) req = { assert(file.op == op::NONE); file.op = op::READV;
file.callback = cb;
file.cb = cb;
file.vec = vec; filemod(file, rt::EPOLLOUT); return req { ... };
}; fn readv_finish(file: *file, ev: *rt::epoll_event) void = { assert(file.op == op::READV && ev.events & rt::EPOLLIN != 0); assert(file.cb != null); const r = io::readv(file.fd, file.vec...); const cb = file.cb: *readcb; cb(file, r); file.op = op::NONE;
}; // A callback for a [[write]] or [[writev]] operation. export type writecb = *fn(file: *file, result: (size | io::error)) void; // Schedules a write operation on a file object. export fn write( file: *file, cb: *writecb, buf: []u8, ) req = { file.vbuf = io::mkvector(buf); // XXX: Bit of a hack to avoid allocating a slice const vec = (&file.vbuf: *[*]io::vector)[..1]; return writev(file, cb, vec...); }; // Schedules a vectored read operation on a file object. export fn writev( file: *file, cb: *writecb, vec: io::vector... ) req = { // XXX: Should we support both pending reads and writes at the same // time? (yes) assert(file.op == op::NONE); file.op = op::WRITEV;
file.callback = cb;
file.cb = cb;
file.vec = vec; filemod(file, rt::EPOLLOUT); return req { ... }; };
fn writev_finish(file: *file, ev: *rt::epoll_event) void = { assert(file.op == op::WRITEV && ev.events & rt::EPOLLOUT != 0); assert(file.cb != null); const r = io::writev(file.fd, file.vec...); const cb = file.cb: *writecb; cb(file, r); file.op = op::NONE; };
use errors; use io; use rt; use time; use types; export type loop = struct { fd: io::file, events: []rt::epoll_event, }; // Creates a new event loop. The user must pass the return value to // [[loop_finish]] to free associated resources when done using the loop. export fn newloop() (loop | 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 = [], }; }; // Frees resources associated with an event loop. Must only be called once per // event loop object. export fn loop_finish(loop: *loop) void = { 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; };
// Dispatches the event loop. A duration of -1 specifies an indefinite timeout, // and will cause dispatch to block until the next event is available.
// Dispatches the event loop. A timeout of 0 specifies an indefinite timeout, // and will cause dispatch to block until the next event is available. A timeout // of -1 will return immediately if no events are available.
// // Portable use of the timeout argument supports only millisecond granularity of // up to 24 days (INT_MAX milliseconds). Negative values other than -1 will // cause the program to abort. // // Returns false if there are no events to process. 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 (len(loop.events) == 0) { return false; }; // TODO: Deal with signals const maxev = len(loop.events); assert(maxev <= types::INT_MAX: size, "ev::dispatch: too many events"); const events = rt::epoll_pwait(
loop.fd, &loop.events[0], maxev: int, millis, null)!; return true;
loop.fd, &loop.events[0], maxev: int, millis, null)!; for (let i = 0z; i < maxev; i += 1) { const ev = &loop.events[i]; const file = ev.data.ptr: *file; if (ev.events == 0) { continue; }; switch (file.op) { case op::NONE => abort("Invalid pending operation"); case op::READV => readv_finish(file, ev); case op::WRITEV => writev_finish(file, ev); }; }; return events != 0;
};