I/O from an actor: awaiting sockets, files, timers and HTTP
On this page
Note: aio, actorhttp, the HTTP and TLS packages, and iocore and iobuf beneath them run on Windows and on Linux. Batching of datagrams (many in one call) is not built yet.
A program that talks to the world spends most of its time waiting: for a connection, for bytes from a peer, for a disk, for a clock. A thread that blocks while it waits is a core doing nothing. In Cone an actor's behaviour awaits the operation instead, exactly as it awaits another actor's answer: the actor goes on with its other messages, and the rest of the behaviour runs when the answer comes. This page covers the packages that do it and each kind of operation: sockets, datagrams, files, timers, names, file metadata, child processes and HTTP. Afterwards you can write a server, a client or a pipeline that keeps every core busy while it waits. Here three timers wait on one actor at once:
mod kitchen;
import stdio use *;
import time use Duration;
import sync use *;
import actors;
import aio;
actor Timer {
// Waits without holding a thread, then says so
pub async do ring(self, what &'static imm str, ms i64, done Sender[i64]) {
let r = await aio.sleep(Duration.fromMillis(ms));
match r {
case let e Error { print <- what, ": the timer failed\n"; }
else { print <- what, " is ready after ", ms, " ms\n"; }
}
var d = done;
d <- ms; // tell main this one is done
}
}
fn main() {
var tx Sender[i64];
var rx Receiver[i64];
tx, rx = Channel[i64].make();
// Three timers on one actor: none of them holds a worker while it waits
let t = new Timer();
t.ring("The tea", 300i64, tx.clone());
t.ring("The toast", 200i64, tx.clone());
t.ring("The egg", 400i64, tx.clone());
// A behaviour paused on I/O does not keep the program running,
// so main waits until all three have rung
each i in 1 .. 3 {
rx.recv();
}
}
It prints, in the order the timers come due rather than the order they were sent:
The toast is ready after 200 ms The tea is ready after 300 ms The egg is ready after 400 ms
One actor handled all three messages, and each ring paused at its await so the next could start. Note the last lines of main. A program ends when no actor is queued or running, and an actor paused on I/O is neither: without the wait, the program would end at once and the timers would never ring (The I/O thread, and the program's end). Here main waits on a channel, a queue between threads which each timer sends to as it finishes; a long-running server waits the same way for word that it is done.
The I/O packages#
What is built is in library packages, which the standard library chapter lists. From the plainest to the most capable:
- Blocking, synchronous I/O: fs (files and folders), path and env (beneath them, the posix package is the C library's own POSIX functions on Linux and maps POSIX onto the Windows C runtime on Windows), and process, which runs a program and waits for it, written over Windows' own kernel32 or over posix_spawn on Linux. Fine for a tool that does one thing at a time; a thread that calls them waits.
- Asynchronous I/O, completion-based: iocore, the loop the rest are built over: a thread hands it a job and later gets exactly one answer for it, carrying the buffer back. With it come iobuf (owned buffers), timewheel (its timers), time and calendar and diag (where an operation's time went). The I/O loop underneath opens it up.
- HTTP and TLS over it (HTTP, TLS and URLs): http1, HTTP/1.1's connection core, sans-IO; httpconn, which joins it to real sockets on a loop, client and server; tls, TLS 1.2 and 1.3 over OpenSSL, also sans-IO, with a server's certificate checked by the operating system; url, URLs as data; and httpclient, which fetches an http:// or https:// URL in one call, with connection pooling, retries and redirects.
- I/O an actor awaits: aio. An actor's behaviour may await a socket's operation (listen, accept, connect, read, write), a file's (open, read at an offset, write at an offset), a timer (aio.sleep(d)), a name looked up (aio.resolve(host)), a file's metadata (aio.stat(path)) or a child process (aio.run(args)), and pauses there until the operation's answer comes. TCP sockets, UDP sockets and files.
- HTTP an actor awaits: actorhttp. An actor's behaviour may await client.get(url), for http:// or https://, and gets the whole response, or the response as it arrives, in chunks or as server-sent events; and an actor may serve: await each request of a listening server and answer it, whole or streamed (below).
Most programs need only the last two, and the rest of this page is about them. The iocore loop underneath, for a program that drives it itself, is the next page's.
Awaiting I/O#
The aio package makes each operation; awaited, the behaviour pauses, the actor goes on with its other messages, and when the operation has finished, the rest of the behaviour runs with its answer. A server and a client, each an actor, the client sending five bytes to the server over a TCP connection on this machine:
// The Socket an answer holds, swapped out of it: Socket.none() if it failed
fn socketOf(r Result[Socket, IoError]) Socket {
var res = r;
var s = Socket.none();
match &mut res {
case var ok &mut Ok {s <=> *ok;}
case let e &mut Error {say("failed", u64.from((*e).code));}
}
s;
}
actor Client {
pool Pool;
pub async do send(self, port u16) {
let conn = socketOf(await aio.connect(SocketAddr.localhost(port)));
var buf = pool.mustTake(64usize);
buf.append("hello");
let sent = await conn.write(buf); // pauses until all of it is sent
match sent {
case let e Error {say("write failed", u64.from(e.code));}
else {}
}
} // conn closes here
}
actor Server {
pool Pool;
client Client;
pub async do serve(self) u64 {
let listener = socketOf(await aio.listen(SocketAddr.localhost(0u16)));
let addr = await listener.localAddr(); // the port the system chose
match addr {
case let a Ok {client.send(a.port());}
case let e Error {say("no address", u64.from(e.code));}
}
let conn = socketOf(await listener.accept());
let none = await conn.read(IoBuf.empty()); // a buffer with no room
match none {
case let e Error {say("no room: error", u64.from(e.code));}
else {}
}
var total = 0u64;
var more = true;
while more {
let got = await conn.read(pool.mustTake(4096usize));
match got {
case let b Ok {
if b.len() == 0usize {
say("end of stream", total); // the peer closed
more = false;
} else {
say("received", u64.from(b.len()));
total = total + u64.from(b.len());
}
}
case let e Error {
say("read failed", u64.from(e.code));
more = false;
}
}
}
total;
}
}
(say prints a word and a number on a line; Pool and IoBuf are iobuf's, SocketAddr and IoError iocore's, and Socket aio's.) Sent serve(), the server prints:
no room: error 122 received 5 end of stream 5
Each await there pauses one behaviour, and only it. While the server waits for a connection, the client waits for its connect; the worker threads run neither until an answer comes.
What an operation answers#
Each operation declares its own answer, and the await's value is exactly that: await adds nothing to it. An operation that can fail answers as a synchronous call that can fail does, with a Result: the success, or the operating system's failure, an IoError. Its code is Windows' error number for the failure, on every system: on Linux the errno is turned into the Windows number that names the same condition (ERROR_CONNECTION_REFUSED for ECONNREFUSED), and one with no such number is its errno above 0x10000000. Its message() is the system's words for it.
- aio.listen(addr), aio.connect(addr), listener.accept(): a Socket. Cone does not yet move a value out of a matched variant's field, so it comes out by a swap, as socketOf does.
- conn.read(buf): the buffer, given back with what arrived appended to it. The buffer is moved into the read, which is how the system may write into it while the behaviour waits, and the read's answer is how it comes back. At the end of the stream, when the peer has closed, it comes back as it went, its length unchanged. A buffer with no room fails with 122, as above. A failed read's buffer goes home to its pool.
- conn.write(buf): the count written, which is all of the buffer; the buffer goes home.
- s.localAddr(): the address the socket is bound to.
- aio.openFile(path, access): a File, out of the Result by a swap as a Socket is. f.readAt(offset, buf) and f.writeAt(offset, buf) answer as read and write do (Files).
An operation's answer should be used. An await on one that is a statement of its own drops the failure unread, so it is warned. The program still compiles, and the answer is dropped as any unused value is:
pub async do careless(self, conn Socket) {
await conn.write(pool.mustTake(64usize)); // warned: the answer is thrown away
}
Warning 3005: This 'await' waits for an operation, which answers with a Result, and the answer is thrown away. An operation's answer says whether it worked, a failure among what it may say, as a synchronous call that can fail answers, so a failure here goes unread: bind it, 'let r = await ...', and match on it.
An operation is made by the call (conn.read(buf)) and started by the await; one made and never awaited does nothing, and is dropped. That is why one behaviour cannot start two operations and await them both: three timers at once took three ring messages above. An operation that did not finish, cancelled or past its deadline, answers like one that failed (next).
Cancelling an operation, and limiting it#
A read from a peer that never sends would wait for ever. So every operation can be limited, and is answered exactly once, with one of four outcomes: done (the Ok answer), failed (an Error whose code is the system's), cancelled, or timed out. The last two are Errors too. An IoError has a kind(), an IoErrorKind: Os for a failure of the system (its code and message() are the system's), Cancelled or TimedOut (its code is 0), or Truncated (a datagram cut to the room that took it, below; its code is the system's, 10040). The kind is what e.isCancelled() and e.isTimedOut() read, and aio.isCancelled(e) and aio.isTimedOut(e) are the same questions, so no system error, whatever its number, is taken for either. The buffer an operation was given goes home to its pool.
An operation is limited by the In form of its call, which takes an IoLimits: conn.readIn(buf, lim), conn.writeIn(buf, lim), listener.acceptIn(lim), aio.connectIn(addr, lim), a file's f.readAtIn(offset, buf, lim) and f.writeAtIn(offset, buf, lim), and a UDP socket's u.recvFromIn(buf, lim) and u.sendToIn(addr, buf, lim). An IoLimits holds a cancel group (iocore's CancelGroup, made with CancelGroup.make(); Cancel groups has the other ways) and an absolute deadline, an Instant: IoLimits.within(group), IoLimits.until(deadline), or both with new IoLimits(Some[CancelGroup](group), deadline). The operation ends when the group is cancelled, or at the deadline, the tighter of it and the group's (a group made with CancelGroup.until(d), or a childUntil(d) of one, carries a deadline of its own, and a child's is never later than its parent's). A group is a value: it goes in a message to another actor, or to a thread, and its one cancel() ends every operation in it, in whichever actors they are.
actor Reader {
pool Pool;
pub async do fetch(self, conn Socket, group CancelGroup) u64 {
let until = Instant.now() + Duration.fromSeconds(5i64);
let got = await conn.readIn(pool.mustTake(4096usize), new IoLimits(Some[CancelGroup](group), until));
match got {
case let b Ok {u64.from(b.len());}
case let e Error {
if aio.isCancelled(e) {say("cancelled", 0u64);}
elif aio.isTimedOut(e) {say("too slow", 0u64);}
else {say("failed", u64.from(e.code));}
0u64;
}
}
}
}
Whoever holds group, in this actor (a message to itself, which runs while fetch is paused), in another, or on a plain thread, may call group.cancel(), and the read ends. A cancel and the operation's completion that race are settled by iocore: the answer is whichever came first, never both. Cancelled is a state: an operation started in a group already cancelled, or with a deadline already past, is answered at once. Closing a socket, by dropping it, cancels the operations in flight on it.
A cancel does not undo what a write already sent, or a read already took: an operation that ended cancelled or timed out may have done part of its work, and its answer does not say how much.
Sockets#
A Socket moves: it may be sent in a message to another actor. Dropping it closes the socket, after whatever operations were started on it before. A socket in a paused behaviour waits with the behaviour, as any value does (What crosses an await), so it stays open until the behaviour ends.
UDP sockets#
aio.bindUdp(addr) binds a UDP socket on the I/O thread (port 0 has the system choose one, which u.localAddr() tells) and answers a UdpSocket, out of the Result by a swap as a Socket is. A UdpSocket is a kind of its own beside Socket and File: read, write and accept are not on it, and recvFrom and sendTo are not on the others, so a call of the wrong kind does not compile. It is not connected: each datagram names where it goes.
- u.recvFrom(buf): one datagram, appended to the buffer, which is moved in. The answer is a Datagram, whose buf is the buffer given back, its length grown by the datagram's, and whose from is the sender's SocketAddr. A datagram of no bytes is a datagram, not an end: the length does not grow. A buffer with no room fails with 122, as a read's does.
- A datagram longer than the buffer's room is cut: the answer is an Error whose IoError is of the kind Truncated (aio.isTruncated(e) or e.isTruncated(); its code is 10040, WSAEMSGSIZE, on both systems). The rest of the datagram is lost, the next one is whole, and the buffer goes home to its pool as a failed read's does, the bytes that fitted with it.
- u.sendTo(addr, buf): all of the buffer as one datagram to addr, the count its length, the buffer going home. UDP promises no delivery: the answer says the system took the datagram. One too large for the network fails with 10040.
Both take In forms, u.recvFromIn(buf, lim) and u.sendToIn(addr, buf, lim), and answer exactly once. A UdpSocket moves and closes when dropped, as a Socket does. One receive at a time per socket is the supported use; datagrams that arrive while none is outstanding wait in the socket's buffer, which the operating system bounds.
actor Echo {
pool Pool;
pub async do once(self, u UdpSocket) { // u: bound earlier, by bindUdp
let r = await u.recvFrom(pool.mustTake(1500usize));
match r {
case let d Ok {
var reply = pool.mustTake(64usize);
reply.append("got it");
let sent = await u.sendTo(d.from, reply); // back to the sender
}
case let e Error {
if aio.isTruncated(e) {say("too long", 0u64);} // longer than 1500 bytes
}
}
}
}
Files#
aio.openFile(path, access) opens a file on the I/O thread, with iocore's Access: Read an existing file; Create a new one, or an existing one emptied, to read and write; Update an existing one or a new one, its bytes kept, to read and write. It answers a File, or the system's failure (a name that is not there is ERROR_FILE_NOT_FOUND, 2, on both systems). The path is copied. f.readAt(offset, buf) and f.writeAt(offset, buf) work at a byte offset, with no position kept in the file, so any number may be in flight on one file at once. A read appends what is there to the buffer, up to its room, and from the end of the file or past it comes back with the buffer as it went; a buffer with no room fails with 122, as a socket's does. A write answers the count written, all of its buffer. Both take the In forms above, and answer exactly once, done, failed, cancelled or timed out. openFile has no In form: opening is one call to the system, which cannot be cancelled once begun.
actor Patcher {
pool Pool;
pub async do patch(self, name Path) u64 {
let f = fileOf(await aio.openFile(&name, Access.Update)); // fileOf: as socketOf
var b = pool.mustTake(5usize);
b.append("hello");
let w = await f.writeAt(100u64, b); // five bytes at offset 100
match w {
case let n Ok {u64.from(n);} // 5
case let e Error {u64.from(e.code);}
}
} // f closes here
}
A File, like a Socket, moves: it may be sent in a message to another actor, and dropping it closes the file, after the operations started on it before, cancelling any still in flight. On Windows the operations are the completion port's; on Linux they run on a thread of the blocking pool (pread and pwrite), and answer the same. fileio, the file service with classes and a queue-depth governor, is a package of its own, not built yet.
Sleeping#
aio.sleep(d) and aio.sleepUntil(at) are timers an actor awaits, as the kitchen above does: the behaviour pauses, its actor and its worker go on with other messages, and the rest of the behaviour runs when the time comes. d is a Duration, counted from the await; at is an absolute Instant (one already past is answered at once, as is a zero Duration). Both answer Result[void, IoError]: Ok when the time has come, an Error of kind Cancelled or TimedOut when the limits ended it first. Their In forms, aio.sleepIn(d, lim) and aio.sleepUntilIn(at, lim), take the IoLimits of the other operations. A group's cancel ends the sleep as Cancelled, at once if the group is cancelled already. A sleep due after the deadline (the limits' own, or the group's, the tighter) is answered TimedOut at the deadline instead; one due at or before it is Ok at its time, as iocore's timer in a cancel group is (the time chapter shows the same on a loop).
pub async do pause(self, group CancelGroup) {
// Half a second, unless the group is cancelled first
let r = await aio.sleepIn(Duration.fromMillis(500i64), IoLimits.within(group));
match r {
case let e Error {say("cancelled", 0u64);}
else {say("half a second has passed", 0u64);}
}
}
Names, file metadata and child processes#
Three things have no completion form of their own, and run on a thread of iocore's blocking pool: the behaviour pauses, a pool thread does the work, and the answer comes back as the other operations' do.
- aio.resolve(host, port) looks a name up (or an address written as text): a List[SocketAddr], every address found, each with the port (default 0). A name with no address is the system's failure (11001, WSAHOST_NOT_FOUND, for a name that is not there).
- aio.stat(path) asks a file's metadata: iocore's FileStat, with its size, attributes, modified time and isDir(). A path that is not there is ERROR_FILE_NOT_FOUND, 2.
- aio.run(args, opts, dir) and aio.runLine(line, opts, dir) run a child process and wait for it, as the process package's synchronous run and runLine do, with the same Options and the same Output: its exit code (an answer, not an error), and what it wrote to its standard output and standard error, apart. The Error is a program that could not be started, an IoError whose code is the system's number. The synchronous process package stays beside it.
Each has an In form taking IoLimits (resolveIn, statIn, runIn, runLineIn). A job still queued for a thread, which a cancel group's cancel or a deadline takes out of its queue, is answered Cancelled or TimedOut without having run. A job a thread has taken cannot be stopped: a cancel waits for it to end, and a deadline answers TimedOut at the deadline while the work goes on, its answer dropped. So a child process is not stopped by its limits: give its Options a timeout, which stops it and everything it started. At most four children run at once (the general lane's threads); more wait their turn.
actor Builder {
pub async do build(self) u64 {
var args = List[String].empty();
args.push(String.from("conec"));
args.push(String.from("app.cone"));
// The compile runs on a pool thread; this actor handles its other messages meanwhile
let r = await aio.run(args.view(), new process.Options(timeout: Duration.fromSeconds(60i64)));
match r {
case let ok Ok {u64.from(ok.exitCode);}
case let e Error {u64.from(e.code);} // 2: no program of that name
}
}
}
What is left out: a child's standard input and output as streams while it runs, a spawn apart from the wait (the process package offers neither), and a file's metadata by its open File.
HTTP an actor awaits#
The actorhttp package (a working name, as netmsg is; the final name is yet to be chosen) makes HTTP and HTTPS fetches awaitable the same way: an actor writes await client.get(url) and the behaviour pauses until the whole response is in. The fetch is httpclient's (its pools of connections per server, its retries and redirects, TLS under https://), run on a thread of actorhttp's own, and the answer comes back as a message, as an aio operation's does:
actor Weather {
client Client;
pub async do show(self, url String, done OneshotSender[u64]) {
// The whole response, awaited: this actor handles its other messages
// meanwhile. It is given ten seconds, and the answer says how it ended
let r = await client.getIn(url.view(), IoLimits.until(Instant.now() + Duration.fromSeconds(10i64)));
match r {
case let ok Ok {
say("status", u64.from(ok.status));
printStr(ok.body.view());
printStr("\n");
}
case let e Error {
if e.isTimedOut() {printStr("too slow\n");}
elif e.isCancelled() {printStr("cancelled\n");}
else {printStr(e.message().view()); printStr("\n");}
}
}
var d = done;
d.send(1u64);
}
}
Fetching from a small server on this machine that answers with the weather, it prints status 200 and the body, sunny, 21 C. The done it is handed is the sending end of a one-shot channel, on which main waits, as the kitchen's waited on its channel.
A Client (Client.standard(), or Client.make(config, maxBody) with an httpclient.Config and the most body bytes a response may have) is a handle that moves: it may be sent in a message, and clone() makes another handle to the same client, so any number of actors fetch through one set of connection pools at once. The last handle's going ends the client, and every fetch of it in flight is answered cancelled. get(url) and send(request) (an httpclient.Request, with its method, header fields, body, intent and redirect limit) answer a Result: the success is httpclient's Response, with the status, the header fields, the whole body as bytes, the URL that answered after any redirects, and its timing. An HTTP error status such as 404 is an answer, not an Error. The Error is a FetchError: the network or the client refusing (a name not found, a refused connection, a certificate refused, a body longer than the limit, a URL that does not parse), with httpclient's Failure in its failure saying at which stage and why, or the fetch cancelled or past its deadline, which e.isCancelled() and e.isTimedOut() read; never a panic. getIn(url, lim) and sendIn(request, lim) take the same IoLimits as aio's operations (a cancel group and an absolute deadline), and a request's own deadline is kept beside it, the tighter winning.
A program that wants one connection itself has actorhttp.connect(host, port) and actorhttp.connectSecure(host, port, tlsConfig), which answer a Conn that moves like a Socket: conn.request(head, body) writes a request head (RequestHead.make("GET", "/", "example.com")) and a body and answers the whole response; conn.requestIn(head, body, lim) adds limits. The connection carries one exchange at a time; a request cancelled or timed out on the wire resets it (HTTP/1.1's only way to stop one), and later requests on it fail as closed. A connectSecure connection is made when the TCP connection is: the TLS handshake runs under the first request, and a certificate the client's configuration refuses ends that request failed at the TLS stage (or at the read or closed stage, if the connection broke before the request was asked for). Dropping a Conn closes the connection.
A response can instead be read as it arrives. client.stream(request) is answered when the response's head is in, with a BodyStream (the status and header fields are fields of it) whose read() awaits the next piece of the body: Some(piece), then None when the body is whole, or an Error as above. client.events(request) is the same for server-sent events (the sse package's, HTTP, TLS and URLs): an EventStream whose next() awaits the next sse.Event. A reader that awaits slowly slows the sender (the connection is told to send more only as the actor takes what is there), and dropping the stream cancels the rest of the fetch.
The other way round, actorhttp.listen(addr, config) answers a Server (a handle that moves and clone()s; its last handle's going closes it), and await server.next() answers the next request, whole, as an Exchange: the Request (method, target, header fields, body, the client's address) and the way to answer it. A server usually has one actor looping on next() and giving each Exchange, which moves, to an actor of its own. The actor answers once: ex.respond(head, body), or ex.start(head, length) then await ex.write(bytes) as often as it likes and await ex.finish() (a body of unknown length goes chunked), or ex.startEvents() then await ex.sendEvent(&event) for a stream of server-sent events. A write is answered when the connection has taken the bytes, so an actor that awaits each is slowed to its peer's speed; a peer that has left answers the write Closed. ServerConfig.secure(tlsConfig) makes the connections TLS; its other fields bound the request body a server takes (past it, 413) and the requests that may wait for an actor (past it, 503). An Exchange dropped without an answer is answered 500. The package's header lists each choice; tests/server.cone, events.cone and tlsserver.cone are these, run.
A cancel and a completion that race, and a fetch dropped with its client, are settled as an aio operation's are: exactly one answer. At the program's end, a behaviour still paused on a fetch is abandoned, as one paused on a socket is (The I/O thread, and the program's end).
Not built yet#
- A socket's half-close, awaited; fileio, the file service.
- A request's body streamed from an actor or to one (a client sends it whole; a server delivers it whole, up to a limit), a stream that reconnects, Upgrade and WebSocket, and httpclient and httpconn running on aio's own sockets (actorhttp keeps a thread and a loop of its own).
You can now serve, fetch, read files and sleep from an actor without holding a thread, and limit or cancel any of it. Every one of those answers came from one loop on a thread of its own; the next page opens that loop up, for a program that drives it itself.