The I/O loop underneath: iocore and the I/O thread
On this page
Note: iocore (sockets, timers, files, cancel groups and the blocking pool) and iobuf run on Windows and on Linux. Not built yet: batching of datagrams (many in one call), and more than one I/O thread.
Every socket, file and timer an actor awaited on the last page was answered by one loop, iocore's, on a thread of its own. Most programs never touch it. This page is for those that do: a plain thread that wants many operations in flight without actors, a library that builds a service of its own over the loop, or a program that must know what becomes of I/O still waiting when it ends. Afterwards you can drive the loop yourself, and say what an await on I/O does underneath. Here are the kitchen's three timers again, with no actor: one thread hands the loop three jobs and waits for their answers.
mod timers;
import stdio use *;
import collections use List;
import time use Instant, Duration;
import iocore use *;
fn main() {
var lp = Loop.mustMake();
let now = Instant.now();
// Three jobs for the loop: wake at these times.
// Each carries a token, which comes back in its answer
lp.after(now + Duration.fromMillis(300i64), Target.Here(1u64)); // the tea
lp.after(now + Duration.fromMillis(200i64), Target.Here(2u64)); // the toast
lp.after(now + Duration.fromMillis(400i64), Target.Here(3u64)); // the egg
// One thread, waiting for all three answers
var out = List[Completion].empty();
while out.len() < 3usize {
lp.poll(forever(), &mut out); // the only place the loop waits
}
each c in out {
print <- "timer ", c.token, " is done\n";
}
}
It prints, in the order the timers come due:
timer 2 is done timer 1 is done timer 3 is done
Nothing here pauses. lp.after(at, to) hands the loop a job and returns at once; lp.poll(until, out) is the one place the thread waits, until an answer arrives or the time it was given comes (forever() never comes), and it appends the answers to out. The token is the program's own number for the job: the loop hands it back and never reads it, so the program knows which job each answer is for. The kitchen's actor did the same, with aio doing these steps for it (The I/O thread, and the program's end).
What the loop does#
iocore is a completion loop: a program hands a loop a job (receive into this buffer, send these bytes, accept a connection, read a file at an offset, wake at a time) and later gets exactly one answer for it, carrying the buffer back. One thread drives a loop: it submits to it and polls it, and another thread that wants a loop has its own. On Windows the loop is an I/O completion port. On Linux it is epoll: a socket's operation is tried at once, and waits in a queue of the socket's if it would block; a file's runs on a thread of the blocking pool. An operation's answers are the same on both.
A job can carry a deadline and belong to a cancel group (below), and a blocking pool runs work that has no completion form: the names, file metadata and child processes an actor awaits run there. Beneath the loop lie iobuf, owned buffers lent to an operation by moving them; timewheel, its timers; time and calendar, time values; and diag, records of where an operation's time went. Timers, deadlines and cancelling in I/O covers the loop's timers and the answers they give.
An answer goes where its job's Target says. Target.Here(token), as above, sends it to this loop's own poll. Others hand it to another loop, to a channel (Target.Chan), to a thread waiting on a one-shot channel (Target.Once), or to another runtime, which is how the I/O thread under aio turns each answer into a message to the actor awaiting it.
Datagrams (UDP) on the loop#
The loop's datagrams are the same as the UDP sockets an actor awaits, driven by hand. lp.bindUdp(addr) makes a UDP socket bound to an address (port 0 has the system choose one, which lp.localAddr(h) tells). It is not connected: each datagram names where it goes, and each one received names where it came from. Two operations, one datagram each, are answered like the others: Op.SendTo(sock, addr, buf) sends all of the buffer as one datagram, and Op.RecvFrom(sock, buf) takes one into the buffer's room, its sender in the answer's peer, a SocketAddr. Deadlines, cancel groups and the handle's close end them as they end a TCP socket's operations, the buffer back in the answer.
import stdio use *;
import collections use List;
import iobuf use *;
import iocore use *;
fn main() {
var lp = Loop.mustMake();
let pool = Pool.make();
let server = lp.mustBindUdp(SocketAddr.localhost(0u16)); // port 0: the system chooses
let client = lp.mustBindUdp(SocketAddr.localhost(0u16));
var port = 0u16;
match lp.localAddr(server) {
case let a Ok {port = a.port();}
case let e Error {}
}
lp.submit(Op.RecvFrom(server, pool.mustTake(1500usize)), Target.Here(1u64));
var hello = pool.mustTake(64usize);
hello.append("hello");
lp.submit(Op.SendTo(client, SocketAddr.localhost(port), hello), Target.Here(2u64));
var out = List[Completion].empty();
while out.len() < 2usize {
lp.poll(forever(), &mut out); // the only place the loop waits
}
each c in out {
if c.token == 1u64 {
printUInt(u64.from(c.n));
printStr(" bytes from ");
printStr(c.peer.text().view());
printStr("\n");
}
}
}
This prints a line like the following, the client's port the one the system chose:
5 bytes from 127.0.0.1:50526
- A datagram of no bytes is a datagram: RecvFrom answers Done(0) and names its sender. It is not the end of anything, as a stream's Done(0) is.
- A datagram longer than the receive's room is cut, as UDP cuts it: the answer is Failed with error 10040 (WSAEMSGSIZE, on Linux too), the room filled with its first bytes and the sender named; the rest of that datagram is lost, and the next receive takes the next datagram whole. A SendTo of more than UDP carries fails with the same number.
- UDP promises no delivery: Done on a SendTo says the system took the datagram. Datagrams that arrive while no receive is waiting are held by the socket, up to what the system allows, and lost beyond it. Take them with one receive at a time on a socket; the order in which several outstanding receives on one socket are filled is not promised.
- On Windows a datagram sent to a port nobody holds would make the sender's next receive fail (Winsock reports the ICMP "port unreachable" on it); that is turned off, so the receive goes on waiting, as it does on Linux.
The netmsg package is a game's protocol built on these two operations: acknowledged packets, reliable messages and latest-wins values between a server and its clients.
Cancel groups#
Work in flight on the loop can be asked to end. lp.cancel(id) asks one operation to. A CancelGroup asks many at once. A group is a value: copying it makes another handle on the same group, it is handed to many operations on any loops, and it crosses threads. Its one cancel(), from any thread, asks every operation in it to end, and a loop asleep in poll wakes at once to act on it.
An operation joins a group as it is submitted, with lp.submitIn(op, to, group, deadline), or lp.afterIn(at, to, group) for a timer. A cancel is a request, so each operation still answers exactly once, and its answer says how it ended: Cancelled if the cancel caught it, or Done (or Failed) if it had finished first, its buffer back either way. Cancelled is a state, not an event: from the cancel on, group.isCancelled() is true, and an operation submitted into the group is answered Cancelled at once, its buffer back, and never started. cancel() is idempotent: it answers true the first time and false after.
import stdio use *;
import collections use List;
import time use Instant, Duration;
import iocore use *;
fn main() {
var lp = Loop.mustMake();
let group = CancelGroup.make();
let hour = Duration.fromSeconds(3600i64);
lp.afterIn(Instant.now() + hour, Target.Here(1u64), group); // two timers, an hour away,
lp.afterIn(Instant.now() + hour, Target.Here(2u64), group); // in the group
lp.after(Instant.now() + hour, Target.Here(3u64)); // and one outside it
printStr(if group.cancel() {"first cancel\n";} else {"cancel again\n";});
printStr(if group.cancel() {"first cancel\n";} else {"cancel again\n";});
var out = List[Completion].empty();
while out.len() < 2usize {
lp.poll(forever(), &mut out); // the only place the loop waits
}
var cancelled = 0u64;
each c in out {
match c.outcome() {
case is Cancelled {cancelled = cancelled + 1u64;}
else {}
}
}
printUInt(cancelled);
printStr(" cancelled, ");
printUInt(lp.inFlight());
printStr(" still waiting\n");
}
This prints:
first cancel cancel again 2 cancelled, 1 still waiting
The timer outside the group is still in flight; the loop cancels it when it dies. Groups nest and carry deadlines, which are absolute: an Instant, never a duration.
- group.child() is cancelled with group and by its own cancel, which does not reach group.
- CancelGroup.until(deadline) is a group whose operations answer TimedOut by the deadline, a timer among them: one due after the deadline answers TimedOut at it, one due before answers Done at its own time. group.childUntil(d) is a child whose deadline is the tighter of its parent's and d, so deadlines are inherited as the tighter, all the way down.
- group.shield(d) is for cleanup after a cancel: the group's cancel and deadline do not reach the shield, its own deadline d bounds it, and its own cancel still ends it.
The same groups limit the operations an actor awaits, through IoLimits. What an actor cannot do is cancel a paused behaviour itself, short of the operation it awaits (Not built yet).
The I/O thread, and the program's end#
The operations an actor awaits through aio run on one thread of their own, the I/O thread, which the first operation awaited starts: it waits in iocore's completion loop, and hands each answer to the actor awaiting it as a message, in its mailbox. The worker threads never wait on I/O. An operation in flight keeps the actor awaiting it alive, whether or not anything else holds the actor, as a request to another actor does. There is one I/O thread; more than one is not built yet.
But it does not keep the program alive. At the program's end, the final its entry runs waits until no actor is queued or running (an actor paused on I/O is neither; The runtime starts and stops by itself) and then ends the I/O: every operation still in flight is cancelled and every socket closed, and the answers are not delivered. Each behaviour still paused on one is abandoned: the rest of it never runs, and every value it held when it paused is finalized, in the order it would have died in had the behaviour ended there, once nothing else holds its actor. A server waiting for a connection that never comes ends this way; so does a read from a peer that never sends, and an HTTP fetch still in flight. A program whose work is waiting on I/O therefore keeps main waiting until it is done, as the kitchen does on its channel.
Or main can end by calling actors.waitUntilIdle(). It returns once no actor is queued or running and no awaited operation is outstanding: every timer or I/O operation an actor awaits has been answered, and whatever its answer started has run. The runtime stays running, and the final then ends the program as above.
mod idle;
import stdio use *;
import time use Duration;
import actors;
import aio;
actor Timer {
pub async do ring(self, millis i64) {
let r = await aio.sleep(Duration.fromMillis(millis));
print <- "rang\n";
}
}
fn main() {
let t = new Timer();
t.ring(200i64);
actors.waitUntilIdle(); // returns once "rang" is printed
print <- "done\n";
}
It prints rang, then done. Without the call, main returns at once, and the sleeping behaviour is abandoned as above.
- A program that never goes idle, such as a listening server, never returns from it. There is no time limit.
- Called before any actor exists, it returns at once. Called from one of the runtime's own workers, it is a panic.
- Call it only as main's last statement: a message another thread sends while it waits may arrive after it returns.
- Every await counts, so it covers aio, actorhttp and any other package that hands an actor something to await.
Interim: how a program ends with I/O in flight is to be designed again. This call stands in until then.
You can now drive the iocore loop from a plain thread, send datagrams on it, cancel its work in groups, and say what happens to I/O still waiting when a program ends. The I/O thread answers the actors; the worker threads run them, and the next page opens up the runtime that keeps those workers.