Reactor vs Proactor - How libuv, Nginx, Netty, Tokio & the Go Netpoller Work

# A tiny reactor (the libuv/Nginx/Netty-NIO shape): one loop, readiness events,
# timers, and callbacks. The handler does the read itself once told "ready".
import heapq, itertools, selectors, socket, time

class Reactor:
    def __init__(self):
        self.sel = selectors.DefaultSelector()
        self.timers, self.seq = [], itertools.count()
        self.log = []
    def on_readable(self, sock, cb):
        sock.setblocking(False)
        self.sel.register(sock, selectors.EVENT_READ, cb)
    def call_later(self, delay, cb):
        heapq.heappush(self.timers, (time.monotonic() + delay, next(self.seq), cb))
    def run(self, until):
        while not until():
            # Phase 1: run due timers. Phase 2: wait for I/O no longer than next timer.
            now = time.monotonic()
            while self.timers and self.timers[0][0] <= now:
                heapq.heappop(self.timers)[2]()
            timeout = max(0, self.timers[0][0] - now) if self.timers else 0.1
            for key, _ in self.sel.select(timeout):
                key.data(key.fileobj)        # dispatch: handler performs the read

r = Reactor()
pairs = [socket.socketpair() for _ in range(3)]
replies = []
def handler(sock):
    data = sock.recv(64)                     # ready => non-blocking read succeeds
    r.log.append(f"read {data.decode()}")
    sock.send(data[::-1])
    r.sel.unregister(sock)
for i, (_, srv) in enumerate(pairs):
    r.on_readable(srv, handler)
r.call_later(0.02, lambda: r.log.append("timer fired (20ms)"))
for i, (cli, _) in enumerate(pairs):
    cli.send(f"req{i}".encode())
r.run(until=lambda: len(r.log) == 4)
print("\n".join(r.log))
print("replies:", [cli.recv(64).decode() for cli, _ in pairs])

read req0
read req1
read req2
timer fired (20ms)
replies: ['0qer', '1qer', '2qer']

// Proactor shape (IOCP, io_uring, Boost.Asio on Windows): you START an operation
// with a buffer, and the completion handler receives the finished result.
// Simulated device; compare with the reactor where the handler does the read.
type Completion = { op: string; bytes: number; data?: string };

class Proactor {
  private queue: Array<() => Completion> = [];
  private handlers: Array<(c: Completion) => void> = [];
  // Initiate: the "kernel" owns the buffer until completion.
  asyncRead(fd: number, size: number, onDone: (c: Completion) => void) {
    this.queue.push(() => ({ op: `read fd=${fd}`, bytes: size, data: `payload-${fd}` }));
    this.handlers.push(onDone);
  }
  asyncWrite(fd: number, data: string, onDone: (c: Completion) => void) {
    this.queue.push(() => ({ op: `write fd=${fd}`, bytes: data.length }));
    this.handlers.push(onDone);
  }
  // Event loop: dequeue completions (like GetQueuedCompletionStatus / reaping a CQ).
  run() {
    while (this.queue.length) {
      const work = this.queue.shift()!; const h = this.handlers.shift()!;
      h(work());
    }
  }
}

const p = new Proactor();
for (const fd of [7, 8]) {
  p.asyncRead(fd, 4096, (c) => {
    console.log(`completed ${c.op}: ${c.bytes}B data=${c.data}`);  // data already in buffer
    p.asyncWrite(fd, c.data!.toUpperCase(), (w) => console.log(`completed ${w.op}: ${w.bytes}B`));
  });
}
p.run();
console.log("reactor: 'fd is readable, you read it'; proactor: 'your read finished, here are the bytes'");

completed read fd=7: 4096B data=payload-7
completed read fd=8: 4096B data=payload-8
completed write fd=7: 9B
completed write fd=8: 9B
reactor: 'fd is readable, you read it'; proactor: 'your read finished, here are the bytes'