Lindenii Project Forge
Initial UDP support
use errors; use ev; use log; use net; use net::ip; use os; type state = struct { loop: *ev::loop, src: ip::addr, port: u16, buf: [os::BUFSIZ]u8, wbuf: []u8, }; export fn main() void = { const loop = ev::newloop()!; defer ev::finish(&loop); const sock = match (ev::listen_udp(&loop, ip::LOCAL_V4, 12345)) { case let err: net::error => log::fatalf("Error: listen: {}", net::strerror(err)); case let err: errors::error => log::fatalf("Error: listen: {}", errors::strerror(err)); case let sock: *ev::file => yield sock; }; defer ev::close(sock); let state = state { loop = &loop, ... }; ev::setuser(sock, &state); ev::recvfrom(sock, &recv, state.buf); log::println("Listening on 127.0.0.1:12345"); for (ev::dispatch(&loop, -1)!) void; }; fn recv(sock: *ev::file, r: ((size, ip::addr, u16) | net::error)) void = { const state = ev::getuser(sock): *state; const (n, src, port) = match (r) { case let err: net::error => log::println("Error: recv:", net::strerror(err)); ev::stop(state.loop); return; case let packet: (size, ip::addr, u16) => yield packet; }; // TODO: Make ev send/write all data and drop these fields and the need // to manually manage a write buffer and re-call sendto in &send state.src = src; state.port = port; log::printfln("{} bytes from {}:{}", n, ip::string(src), port); state.wbuf = state.buf[..n]; ev::sendto(sock, &send, state.wbuf, src, port); }; fn send(sock: *ev::file, r: (size | net::error)) void = { const state = ev::getuser(sock): *state; const n = match (r) { case let err: net::error => log::println("Error: send:", net::strerror(err)); ev::stop(state.loop); return; case let n: size => yield n; }; static delete(state.wbuf[..n]); if (len(state.wbuf) != 0) { ev::sendto(sock, &send, state.wbuf, state.src, state.port); } else { ev::recvfrom(sock, &recv, state.buf); }; };
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 = 4 << 16, SIGNAL = 5 << 16, TIMER = 6 << 16,
READABLE = 1 << 16, WRITABLE = 2 << 16, ACCEPT = 3 << 16, CONNECT_TCP = 4 << 16, CONNECT_UDP = 5 << 16, SIGNAL = 6 << 16, TIMER = 7 << 16, RECVFROM = 8 << 16, SENDTO = 9 << 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, 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, ) (*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)!; }; if (file.op == op::SIGNAL) { signal_restore(file); }; 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 || file.op & op::READABLE != 0) { events |= rt::EPOLLIN | rt::EPOLLHUP; }; if (file.op & op::WRITEV != 0 || file.op & op::WRITABLE != 0) { events |= rt::EPOLLOUT | rt::EPOLLHUP; }; switch (file.op) { case op::ACCEPT => events |= rt::EPOLLIN;
case op::CONNECT =>
case op::CONNECT_TCP =>
events |= rt::EPOLLOUT; case op::SIGNAL => events |= rt::EPOLLIN; case op::TIMER => events &= ~rt::EPOLLONESHOT; events |= rt::EPOLLIN;
case op::RECVFROM => events |= rt::EPOLLIN; case op::SENDTO => events |= rt::EPOLLOUT;
case => yield; }; 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 errors; use io; use rt; use time; use types; use unix::signal; 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; }; const pending = file.op; if (ev.events & (rt::EPOLLIN | rt::EPOLLHUP) != 0 && pending & op::READV != 0) { readv_ready(file, ev); }; if (ev.events & (rt::EPOLLOUT | rt::EPOLLHUP) != 0 && pending & 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 => connect_ready(file, ev);
case op::CONNECT_TCP => connect_tcp_ready(file, ev);
case op::SIGNAL => signal_ready(file, ev); case op::TIMER => timer_ready(file, ev);
case op::RECVFROM => recvfrom_ready(file, ev); case op::SENDTO => sendto_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 net; use net::ip; use net::tcp;
use net::udp;
use rt; // Creates a socket which listens for incoming TCP connections on the given // IP address and port. export fn listen_tcp( loop: *loop, addr: ip::addr, port: u16, opts: tcp::listen_option... ) (*file | net::error | errors::error) = { const sock = tcp::listen(addr, port, opts...)?;
return register(loop, sock)?; }; // Creates a socket which listens for incoming UDP packets on the given IP // address and port. export fn listen_udp( loop: *loop, addr: ip::addr, port: u16, opts: udp::listen_option... ) (*file | net::error | errors::error) = { const sock = udp::listen(addr, port, opts...)?;
return register(loop, sock)?; }; export type connectcb = fn(result: (*file | net::error), user: nullable *void) void; // Creates a socket and connects to a given IP address and port over TCP. // // The variadic arguments accept [[net::sockflags]] and/or no more than one user // data pointer. If the user data pointer is provided, it will be passed to the // callback. This allows the user to pass a state object through the connection // process: // // let user: state = // ... // ev::connect_tcp(&loop, &connected, addr, port, &user); // // fn connected(result: (*ev::file | net::error), user: nullable *void) void = { // let user = user: *state; // }; // // The user data object provided will be assigned to the [[file]] which is // provided to the callback after the connection is established. // // If you don't need a user data object you can just omit it: // // ev::connect_tcp(&loop, &connected, addr, port, &user); export fn connect_tcp( loop: *loop, cb: *connectcb, addr: ip::addr, port: u16, opts: (net::sockflags | *void)... ) (void | net::error | errors::error) = { // XXX: This doesn't let us set keepalive let opt: net::sockflags = 0; let user: nullable *void = null; for (let i = 0z; i < len(opts); i += 1) { match (opts[i]) { case let o: net::sockflags => opt |= o; case let u: *void => assert(user == null); user = u; }; }; const sock = tcp::connect(addr, port, opt | net::sockflags::NONBLOCK)?; let file = register(loop, sock)?; file.user = user; file.cb = cb;
file.op = op::CONNECT;
file.op = op::CONNECT_TCP;
file_epoll_ctl(file); };
fn connect_ready(
// Creates a UDP socket on this event loop and sets the default destination to // the given address. export fn connect_udp( loop: *loop, dest: ip::addr, port: u16, opts: udp::connect_option... ) (*file | net::error | errors::error) = { const sock = udp::connect(dest, port, opts...)?; const file = register(loop, sock)?; return file; }; fn connect_tcp_ready(
sock: *file, ev: *rt::epoll_event, ) void = {
assert(sock.op == op::CONNECT);
assert(sock.op == op::CONNECT_TCP);
assert(ev.events & rt::EPOLLOUT != 0); assert(sock.cb != null); const cb = sock.cb: *connectcb; sock.op = op::NONE; file_epoll_ctl(sock); let errno = 0i, optsz = size(int): u32; rt::getsockopt(sock.fd, rt::SOL_SOCKET, rt::SO_ERROR, &errno, &optsz)!; if (errno != 0) { cb(errors::errno(errno), sock.user); close(sock); } else { // XXX: If the user puts NONBLOCK into the opts provided at // [[connect_tcp]] we could try to preserve that here const fl = rt::fcntl(sock.fd, rt::F_GETFL, void)!; rt::fcntl(sock.fd, rt::F_SETFL, fl & ~rt::O_NONBLOCK)!; cb(sock, sock.user); }; }; // A callback for an [[accept]] operation. export type acceptcb = fn(file: *file, result: (*file | net::error)) void; // Schedules an accept operation on a socket. export fn accept( sock: *file, cb: *acceptcb, flags: net::sockflags... ) req = { assert(sock.op == op::NONE); let fl: net::sockflags = 0; for (let i = 0z; i < len(flags); i += 1) { fl |= flags[i]; }; sock.op = op::ACCEPT; sock.cb = cb; sock.sockflags = fl; file_epoll_ctl(sock); return req { ... }; }; fn accept_ready( sock: *file, ev: *rt::epoll_event, ) void = { assert(sock.op == op::ACCEPT); assert(ev.events & rt::EPOLLIN != 0); assert(sock.cb != null); const cb = sock.cb: *acceptcb; sock.op = op::NONE; file_epoll_ctl(sock); const r = tcp::accept(sock.fd, sock.sockflags); match (r) { case let fd: net::socket => // TODO: Bubble up errors from here? const file = register(sock.ev, fd)!; cb(sock, file); case let err: net::error => cb(sock, err); }; };
// TODO: Support recv & send in parallel // Callback for a [[recvfrom]] operation. The second parameter is either an // error or a tuple of the number of bytes received and the IP address and port // of the sender. export type recvfromcb = fn( file: *file, r: ((size, ip::addr, u16) | net::error), ) void; // Schedules a receive operation on a socket. export fn recvfrom( sock: *file, cb: *recvfromcb, buf: []u8, ) req = { assert(sock.op == op::NONE); sock.op = op::RECVFROM; sock.cb = cb; sock.sendrecv.rbuf = buf; file_epoll_ctl(sock); return req { ... }; }; fn recvfrom_ready( sock: *file, ev: *rt::epoll_event, ) void = { assert(sock.op == op::RECVFROM); assert(sock.cb != null); const cb = sock.cb: *recvfromcb; sock.op = op::NONE; file_epoll_ctl(sock); let src: ip::addr = ip::ANY_V4, port = 0u16; match (udp::recvfrom(sock.fd, sock.sendrecv.rbuf, &src, &port)) { case let err: net::error => cb(sock, err); case let n: size => cb(sock, (n, src, port)); }; }; export type sendtocb = fn(file: *file, r: (size | net::error)) void; // Schedules a send operation on a socket. export fn sendto( sock: *file, cb: *sendtocb, buf: []u8, dest: ip::addr, port: u16, ) req = { assert(sock.op == op::NONE); sock.op = op::SENDTO; sock.cb = cb; sock.sendrecv.sbuf = buf; sock.sendrecv.dest = dest; sock.sendrecv.port = port; file_epoll_ctl(sock); return req { ... }; }; fn sendto_ready( sock: *file, ev: *rt::epoll_event, ) void = { assert(sock.op == op::SENDTO); assert(sock.cb != null); const cb = sock.cb: *sendtocb; sock.op = op::NONE; file_epoll_ctl(sock); const r = udp::sendto( sock.fd, sock.sendrecv.sbuf, sock.sendrecv.dest, sock.sendrecv.port, ); cb(sock, r); };