Lindenii Project Forge
Implement ev::readable, ev::writable
use errors; use io; use net; use rt; export type op = enum u64 {
NONE = 0, READV = 1 << 0, WRITEV = 1 << 1, ACCEPT = 1 << 2, CONNECT = 1 << 3,
NONE = 0, READV = 1 << 0, WRITEV = 1 << 1, READABLE = 1 << 16, WRITABLE = 2 << 16, ACCEPT = 3 << 16, CONNECT = 4 << 16,
}; export type fflags = enum uint { BLOCKING = 1 << 31, }; export type file = struct { fd: io::file, ev: *loop, flags: fflags, op: op, cb: nullable *void, cb2: nullable *void, user: nullable *void, // Operation-specific data union { struct { rvbuf: rt::iovec, rvec: []rt::iovec, wvbuf: rt::iovec, wvec: []rt::iovec, }, sockflags: net::sockflags, }, }; // 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 => if (err: int == 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)!; }; 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 *void) 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) *void = { return file.user as *void; }; // 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; }; // 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) {
if (file.op & op::READV != 0 || file.op & op::READABLE != 0) {
events |= rt::EPOLLIN | rt::EPOLLHUP; };
if (file.op & op::WRITEV != 0) {
if (file.op & op::WRITEV != 0 || file.op & op::WRITABLE != 0) {
events |= rt::EPOLLOUT | rt::EPOLLHUP; }; if (file.op & op::ACCEPT != 0) { events |= rt::EPOLLIN; }; if (file.op & op::CONNECT != 0) { events |= rt::EPOLLOUT; }; 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.rvbuf = io::mkvector(buf); // XXX: Bit of a hack to avoid allocating a slice const vec = (&file.rvbuf: *[*]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::READV == 0); if (file.flags & fflags::BLOCKING != 0) { const r = io::readv(file.fd, vec...); cb(file, r); return req { ... }; }; file.op |= op::READV; file.cb = cb; file.rvec = vec; file_epoll_ctl(file); return req { ... }; }; fn readv_ready(file: *file, ev: *rt::epoll_event) void = { assert(file.op & op::READV != 0); assert(file.cb != null); const cb = file.cb: *readcb; file.op &= ~op::READV; file_epoll_ctl(file); if (ev.events & rt::EPOLLHUP != 0) { cb(file, io::EOF); } else { const r = io::readv(file.fd, file.rvec...); cb(file, r); }; }; // 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.wvbuf = io::mkvector(buf); // XXX: Bit of a hack to avoid allocating a slice const vec = (&file.wvbuf: *[*]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::WRITEV == 0); if (file.flags & fflags::BLOCKING != 0) { const r = io::writev(file.fd, vec...); cb(file, r); return req { ... }; }; file.op |= op::WRITEV; file.cb2 = cb; file.wvec = vec; file_epoll_ctl(file); return req { ... }; }; fn writev_ready(file: *file, ev: *rt::epoll_event) void = { assert(file.op & op::WRITEV != 0); assert(file.cb != null); const r = io::writev(file.fd, file.wvec...); const cb = file.cb2: *writecb; file.op &= ~op::WRITEV; file_epoll_ctl(file); cb(file, r); };
use errors; use io; use rt; use time; use types; export type loop = struct { fd: io::file, events: []rt::epoll_event, 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. 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, // XXX: Should the number of events be customizable? events = alloc([rt::epoll_event { ... }...], 256), 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; }; // 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; }; 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 = rt::epoll_pwait( loop.fd, &loop.events[0], maxev: int, millis, null)!; for (let i = 0; i < nevent; i += 1) { const ev = &loop.events[i]; const file = ev.data.ptr: *file; if (ev.events == 0) { continue; };
if (file.op == op::NONE) { abort("Invalid pending operation"); };
const pending = file.op;
if (ev.events & (rt::EPOLLIN | rt::EPOLLHUP) != 0
&& file.op & op::READV != 0) {
&& pending & op::READV != 0) {
readv_ready(file, ev); }; if (ev.events & (rt::EPOLLOUT | rt::EPOLLHUP) != 0
&& file.op & op::WRITEV != 0) {
&& pending & op::WRITEV != 0) {
writev_ready(file, ev); };
if (file.op & op::ACCEPT != 0) {
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);
}; if (file.op & op::CONNECT != 0) {
case op::CONNECT =>
connect_ready(file, ev);
case => yield;
}; }; 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 rt; // A callback for a [[readable]] operation. export type readablecb = fn(file: *file) void; // Executes the callback when a given file is readable. Cannot be combined with // [[read]] or [[readv]]. export fn readable( file: *file, cb: *readablecb, ) req = { assert(file.op & op::READABLE == 0 && file.op & op::READV == 0); if (file.flags & fflags::BLOCKING != 0) { cb(file); return req { ... }; }; file.op |= op::READABLE; file.cb = cb; file_epoll_ctl(file); return req { ... }; }; fn readable_ready(file: *file, ev: *rt::epoll_event) void = { assert(file.op & op::READABLE != 0); assert(file.cb != null); const cb = file.cb: *readablecb; file.op &= ~op::READABLE; file_epoll_ctl(file); cb(file); }; // A callback for a [[writable]] operation. export type writablecb = fn(file: *file) void; // Executes the callback when a given file is writable. Cannot be combined with // [[write]] or [[writev]]. export fn writable( file: *file, cb: *writablecb, ) req = { assert(file.op & op::WRITABLE == 0 && file.op & op::WRITEV == 0); if (file.flags & fflags::BLOCKING != 0) { cb(file); return req { ... }; }; file.op |= op::WRITABLE; file.cb = cb; file_epoll_ctl(file); return req { ... }; }; fn writable_ready(file: *file, ev: *rt::epoll_event) void = { assert(file.op & op::WRITABLE != 0); assert(file.cb != null); const cb = file.cb: *writablecb; file.op &= ~op::WRITABLE; file_epoll_ctl(file); cb(file); };