Runloom benchmark report

"""Std name: runloom_epoll_py_sync  (this file ALSO backs runloom_iouring_py_sync,
launched byte-for-byte with env RUNLOOM_IOURING_LOOP=1).

Server tier 1 (epoll) / tier 3 (io_uring): runloom default backend, ZERO
optimized -- the naive, object-heavy path.

Spec: wrapped python calls, no direct C calls, python objects.
    listener = runloom.sync.tcp_listen(...)
    while True:
        conn, _ = listener.accept()
        runloom.go(handle, conn)      # real name: runloom.fiber

The handler uses recv() (allocates a bytes per read) + sendall(bytes) on the
high-level runloom.sync.Socket facade -- deliberately the slow tier.
Tier 3 is byte-for-byte this file; the orchestrator just exports
RUNLOOM_IOURING_LOOP=1 (spec: "same code as 1 but io_uring loop").
"""
import argparse
import os

import runloom
import runloom.sync as rs


def handle(conn):
    try:
        while True:
            data = conn.recv(65536)          # python bytes alloc per read
            if not data:
                break
            conn.sendall(data)
    except OSError:
        pass
    finally:
        try:
            conn.close()
        except Exception:
            pass


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--host", default="10.99.0.1")
    ap.add_argument("--port", type=int, default=9000)
    ap.add_argument("--hubs", type=int, default=int((os.cpu_count() or 1) * 0.7))
    ap.add_argument("--token", default="")   # for targeted pkill by the orchestrator
    args = ap.parse_args()

    def root():
        ln = rs.tcp_listen(args.host, args.port, backlog=4096)
        port = ln.getsockname()[1]
        print("LISTENING %d" % port, flush=True)
        while True:
            conn, _ = ln.accept()
            runloom.fiber(handle, conn)

    runloom.run(args.hubs, root)


if __name__ == "__main__":
    main()

"""Server tier 2: runloom_c.serve C scaffold (listen/accept/spawn in C) with a
regular PYTHON handler that uses the C-level TCPConn methods recv_into/send_all.

Spec: uses C calls so faster than the sync wrappers, but the handler is still a
plain python function (no Cython). The C scaffold runs N SO_REUSEPORT acceptors.
"""
import argparse
import os

import runloom
import runloom_c

CHUNK = 65536


def handle(conn):
    # conn is a runloom_c.TCPConn. recv_into/send_all are single C calls (no
    # bytes alloc), but this is still a python frame dispatched per round trip.
    buf = bytearray(CHUNK)
    mv = memoryview(buf)
    try:
        while True:
            n = conn.recv_into(buf)
            if not n:
                break
            conn.send_all(mv[:n])
    except OSError:
        pass


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--host", default="10.99.0.1")
    ap.add_argument("--port", type=int, default=9000)
    ap.add_argument("--hubs", type=int, default=int((os.cpu_count() or 1) * 0.7))
    ap.add_argument("--token", default="")
    args = ap.parse_args()

    def root():
        port, listeners = runloom_c.serve(
            args.host, args.port, handle,
            acceptors=args.hubs, backlog=4096)
        print("LISTENING %d" % port, flush=True)
        runloom.sleep(float("inf"))

    runloom.run(args.hubs, main_fn=root)


if __name__ == "__main__":
    main()

"""Std name: runloom_iouring_cython_tcpcon  (this file ALSO backs the epoll tier
runloom_epoll_cython_tcpcon and the ..._opt tier; the orchestrator selects the
backend + optimize per spec).

Server tiers 4 & 5: runloom_c.serve + the zero-PyObject CYTHON handler. Backend
is io_uring when the orchestrator sets RUNLOOM_IOURING_LOOP=1, else epoll.

  tier 4: no optimize()
  tier 5: runloom.optimize("throughput") first  (--optimize throughput)

The handler (handler_cy.handler) calls runloom's cooperative recv/send as plain
C functions via the runloom_c.__tcp_capi__ capsule, so the per-request hot loop
allocates no Python objects (proven by disasm_check.sh).
"""
import argparse
import os
import sys

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))  # find handler_cy*.so

import runloom
import runloom_c


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--host", default="10.99.0.1")
    ap.add_argument("--port", type=int, default=9000)
    ap.add_argument("--hubs", type=int, default=int((os.cpu_count() or 1) * 0.7))
    ap.add_argument("--optimize", default="none", choices=["none", "throughput"])
    ap.add_argument("--token", default="")
    args = ap.parse_args()

    # optimize() must run BEFORE runloom.run() (decision: tier 5 = tier 4 + this).
    if args.optimize == "throughput":
        eff = runloom.optimize("throughput")
        print("OPTIMIZE %s" % eff, flush=True)

    import handler_cy  # imported after optimize so the capsule/runtime are set

    def root():
        port, listeners = runloom_c.serve(
            args.host, args.port, handler_cy.handler,
            acceptors=args.hubs, backlog=4096)
        print("LISTENING %d" % port, flush=True)
        runloom.sleep(float("inf"))

    runloom.run(args.hubs, main_fn=root)


if __name__ == "__main__":
    main()

"""Std name: runloom_iouring_cdef_tcpcon  (this file ALSO backs
runloom_epoll_cdef_tcpcon -- same handler, no io_uring env, so it runs on the
epoll readiness backend; the orchestrator selects the backend per spec).

Server tier: runloom_c.serve + a tstate-free Cython CDEF handler (the c_entry
fast path); io_uring proactor when the orchestrator sets RUNLOOM_IOURING_LOOP=1,
else the epoll readiness backend.

The difference from runloom_iouring_cython_tcpcon.py: handler_cdef.handler is a
runloom_c.c_handler PyCapsule (a cdef C function), so serve() spawns it via
runloom_mn_fiber_c -> the g->c_entry path -> NO Python frame, NO per-park tstate
save/restore. Zero PyObjects in the hot loop AND zero tstate juggle per round
trip -- the all-C echo's advantage with a custom handler.
"""
import argparse
import os
import sys

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))  # find handler_cdef*.so

import runloom
import runloom_c


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--host", default="10.99.0.1")
    ap.add_argument("--port", type=int, default=9000)
    ap.add_argument("--hubs", type=int, default=int((os.cpu_count() or 1) * 0.7))
    ap.add_argument("--optimize", default="none", choices=["none", "throughput"])
    ap.add_argument("--token", default="")
    args = ap.parse_args()

    if args.optimize == "throughput":
        print("OPTIMIZE %s" % runloom.optimize("throughput"), flush=True)

    import handler_cdef  # builds the capsule at import

    def root():
        port, listeners = runloom_c.serve(
            args.host, args.port, handler_cdef.handler,
            acceptors=args.hubs, backlog=4096)
        print("LISTENING %d" % port, flush=True)
        runloom.sleep(float("inf"))

    runloom.run(args.hubs, main_fn=root)


if __name__ == "__main__":
    main()

# cython: language_level=3, boundscheck=False, wraparound=False, cdivision=True, freethreading_compatible=True
"""Zero-PyObject Cython echo handler for runloom_c.serve (benchmark tiers 4 & 5).

The whole point of this tier: the per-request hot loop must create NO Python
objects, so the cost of an echo round-trip is `recv()` + `send()` + a netpoll
park and nothing else -- no method dispatch, no bytes box, no Py_buffer fill.

We get there by calling the C functions runloom_tcpconn_c_recv_into /
runloom_tcpconn_c_send_all *directly* (via the runloom_c.__tcp_capi__ capsule's
function-pointer table), passing a raw stack buffer.  `disasm_check.sh` objdumps
the compiled `handler` symbol and asserts there is NO call to any Py_/_Py_/
PyObject_ symbol between the recv and send calls.

Built by build_cy.py against -I<repo>/src/runloom_c.  `freethreading_compatible`
is REQUIRED: without it, importing this module on 3.13t silently re-enables the
GIL and destroys the M:N parallelism we are trying to measure.
"""
from cpython.object cimport PyObject
from cpython.pycapsule cimport PyCapsule_GetPointer

cdef extern from "runloom_tcp_capi.h":
    ctypedef struct RunloomTCPCAPI:
        Py_ssize_t (*recv_into)(PyObject *conn, void *buf, Py_ssize_t n)
        Py_ssize_t (*send_all)(PyObject *conn, const void *buf, Py_ssize_t n)
    const char *RUNLOOM_TCP_CAPI_CAPSULE_NAME

# Resolve the pointer table once at import (no RTLD_GLOBAL games, no undefined
# symbols -- the capsule is the supported cross-module C-API hand-off).
cdef const RunloomTCPCAPI *_capi = NULL

cdef int _load_capi() except -1:
    global _capi
    import runloom_c
    cap = runloom_c.__tcp_capi__
    _capi = <const RunloomTCPCAPI *>PyCapsule_GetPointer(
        cap, RUNLOOM_TCP_CAPI_CAPSULE_NAME)
    if _capi is NULL:
        raise ImportError("runloom_c.__tcp_capi__ capsule pointer is NULL")
    return 0

_load_capi()

# Chunk size for the streaming echo.  64 KiB lives comfortably on the fiber's
# 512 KiB stack and is large enough that the 1.5 MB bandwidth test loops only
# ~24x per message while the small-payload req/s test fits in one read.
DEF CHUNK = 65536

# Work knob for the work curve: the SAME FNV-1a byte hash, run INLINE in this
# zero-PyObject Cython handler. So the FULL request path is native -- capi recv,
# native FNV, fold, capi send -- no interpreted recv_into/send_all/fold wrapper
# and no per-call boxing. This is the state-of-the-art optimized runloom handler
# (the line that competes with Go), not a Python def calling a compiled function.
# _work is set once via set_work() before serve() spawns any fiber. work=0 = echo.
cdef int _work = 0


def set_work(int w):
    global _work
    _work = w


cdef unsigned int _fnv(const unsigned char *buf, Py_ssize_t n, int passes) noexcept nogil:
    cdef unsigned int h = 2166136261u   # FNV-1a offset basis
    cdef Py_ssize_t i
    cdef int p
    for p in range(passes):
        for i in range(n):
            h = (h ^ buf[i]) * 16777619u   # FNV-1a prime, native wraparound
    return h


def handler(conn):
    """serve() hands us a runloom_c.TCPConn.  recv -> (optional inline FNV) ->
    send, until EOF. Zero PyObjects in the loop body either way."""
    cdef PyObject *c = <PyObject *>conn
    cdef char buf[CHUNK]
    cdef Py_ssize_t n
    cdef unsigned int h
    while True:
        n = _capi.recv_into(c, buf, CHUNK)
        if n <= 0:
            break
        if _work > 0:
            h = _fnv(<const unsigned char *>buf, n, _work)
            buf[0] = <char>(<unsigned char>buf[0] ^ <unsigned char>(h & 0xffu))   # fold -> no elision
        if _capi.send_all(c, buf, n) < 0:
            break
    # Close explicitly so the fd's netpoll registration is released NOW, not at
    # TCPConn dealloc.  Under connection churn that dealloc lags, leaving the
    # closed peer's fd in epoll, level-triggered-readable (EPOLLIN/RDHUP) with no
    # parker -> it re-fires every epoll_wait forever (the pump dispatch misses,
    # restashes, re-fires), pinning every hub in a futex storm so no fiber can
    # resume = a congestion-collapse livelock.  conn_churn investigation.
    conn.close()

# cython: language_level=3, boundscheck=False, wraparound=False, freethreading_compatible=True
"""Tstate-free Cython `cdef` handler for runloom_c.serve(handler=<capsule>).

This is the *c_entry* fast path with custom logic. Unlike handler_cy (a Python
`def`, so serve() spawns it as a full Python fiber that carries a PyThreadState
and pays tstate_save/restore on every park), this handler is exposed to serve()
as a `runloom_c.c_handler` PyCapsule wrapping a `cdef` C function. serve() spawns
it via runloom_mn_fiber_c -> the `g->c_entry` path in runloom_g_entry, which skips
ALL Python-frame / tstate setup. So the request loop has:
  * zero PyObjects (like handler_cy), AND
  * zero tstate save/restore per park (unlike handler_cy) -- the all-C echo's
    advantage, now available to a custom handler.

It runs entirely `nogil` (there is no tstate to hold), calling runloom's raw-fd
cooperative recv/send via the runloom_c.__tcp_capi__ capsule (the proactor when
RUNLOOM_IOURING_LOOP is on, else readiness + wait_fd).
"""
from libc.stdint cimport intptr_t
from cpython.pycapsule cimport PyCapsule_GetPointer, PyCapsule_New

# Function-pointer types for the raw-fd capi (noexcept nogil: callable from the
# tstate-free, GIL-less c_entry handler).
ctypedef Py_ssize_t (*fd_recv_t)(int fd, void *buf, Py_ssize_t n) noexcept nogil
ctypedef Py_ssize_t (*fd_send_t)(int fd, const void *buf, Py_ssize_t n) noexcept nogil
ctypedef void (*fd_close_t)(int fd) noexcept nogil

cdef extern from "runloom_tcp_capi.h":
    ctypedef struct RunloomTCPCAPI:
        fd_recv_t fd_recv
        fd_send_t fd_send_all
        fd_close_t fd_close
    const char *RUNLOOM_TCP_CAPI_CAPSULE_NAME
    const char *RUNLOOM_C_HANDLER_CAPSULE_NAME

cdef fd_recv_t _fd_recv = NULL
cdef fd_send_t _fd_send_all = NULL
cdef fd_close_t _fd_close = NULL

# Work knob for the cross-runtime work curve: the SAME FNV-1a byte hash as
# work_cy / py_fnv / goFnv, but here it runs INSIDE the tstate-free nogil cdef
# handler -- so unlike srv_runloom_work.py's Python `def` handler (which wraps the
# compiled work in interpreted recv_into/send_all/fold), the ENTIRE request path
# is native C. This is the "fully-native runloom handler" line for the
# Cython-vs-cdef-vs-Go comparison. _work is set once via set_work() before serve()
# spawns any fiber, so the nogil read needs no lock. work=0 -> plain echo.
cdef int _work = 0


def set_work(int w):
    global _work
    _work = w


cdef unsigned int _fnv(const unsigned char *buf, Py_ssize_t n, int passes) noexcept nogil:
    cdef unsigned int h = 2166136261u   # FNV-1a offset basis
    cdef Py_ssize_t i
    cdef int p
    for p in range(passes):
        for i in range(n):
            h = (h ^ buf[i]) * 16777619u   # FNV-1a prime, native wraparound
    return h


cdef int _load() except -1:
    global _fd_recv, _fd_send_all, _fd_close
    import runloom_c
    cdef RunloomTCPCAPI *capi = <RunloomTCPCAPI *>PyCapsule_GetPointer(
        runloom_c.__tcp_capi__, RUNLOOM_TCP_CAPI_CAPSULE_NAME)
    if capi is NULL:
        raise ImportError("runloom_c.__tcp_capi__ capsule pointer is NULL")
    _fd_recv = capi.fd_recv
    _fd_send_all = capi.fd_send_all
    _fd_close = capi.fd_close
    return 0


_load()


cdef void echo_handler(void *arg) noexcept nogil:
    """serve() hands us the accepted fd (via intptr_t). recv -> (optional FNV
    work) -> send, until EOF. No tstate, no Python objects, no GIL -- the c_entry
    fast path. With _work>0 the ENTIRE request path is native C (the fully-native
    runloom handler line vs Go); _work==0 is the plain echo."""
    cdef int fd = <int><intptr_t>arg
    cdef char buf[16384]
    cdef Py_ssize_t n
    cdef unsigned int h
    while True:
        n = _fd_recv(fd, buf, 16384)
        if n <= 0:
            break
        if _work > 0:
            h = _fnv(<const unsigned char *>buf, n, _work)
            buf[0] = <char>(<unsigned char>buf[0] ^ <unsigned char>(h & 0xffu))   # fold in -> no elision
        if _fd_send_all(fd, buf, n) < 0:
            break
    _fd_close(fd)


# The capsule serve() detects (RUNLOOM_C_HANDLER_CAPSULE_NAME = "runloom_c.c_handler").
handler = PyCapsule_New(<void *>echo_handler, RUNLOOM_C_HANDLER_CAPSULE_NAME, NULL)

"""Std name: asyncio_epoll_py_proto  (this file ALSO backs uvloop_libuv_py_proto,
launched with --loop uvloop).

Baseline server: canonical asyncio Protocol echo (and uvloop via --loop).

Single-threaded (decision #4: run on the GIL build, its best case). uvloop uses
the identical protocol, only the event-loop policy changes.

--work N applies the SAME FNV-1a byte hash as the runloom work curve (py_fnv,
identical constants) N times over each chunk before echoing, folded into byte 0
so it can't be elided. work=0 is the plain echo. This is the interpreted-Python
reference for the cross-runtime handler work curve (1 core, like the echo bench).
"""
import argparse
import asyncio
import socket

FNV_OFF = 2166136261        # 0x811c9dc5
FNV_PRIME = 16777619        # 0x01000193


def py_fnv(buf, n, passes):
    """Identical to srv_runloom_work.py:py_fnv -- pure inline arithmetic."""
    h = FNV_OFF
    for _ in range(passes):
        for i in range(n):
            h = ((h ^ buf[i]) * FNV_PRIME) & 0xffffffff
    return h


class EchoProtocol(asyncio.Protocol):
    def __init__(self, work):
        self.work = work

    def connection_made(self, transport):
        self.transport = transport
        sock = transport.get_extra_info("socket")
        if sock is not None:
            try:
                sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
            except OSError:
                pass

    def data_received(self, data):
        if self.work:
            b = bytearray(data)
            h = py_fnv(b, len(b), self.work)
            b[0] = (b[0] ^ (h & 0xff)) & 0xff   # fold in -> no elision
            self.transport.write(b)
        else:
            self.transport.write(data)   # echo (work=0)


async def amain(host, port, work):
    loop = asyncio.get_running_loop()
    server = await loop.create_server(lambda: EchoProtocol(work), host, port,
                                      backlog=4096, reuse_address=True)
    print("LISTENING %d" % server.sockets[0].getsockname()[1], flush=True)
    async with server:
        await server.serve_forever()


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--host", default="10.99.0.1")
    ap.add_argument("--port", type=int, default=9000)
    ap.add_argument("--loop", default="asyncio", choices=["asyncio", "uvloop"])
    ap.add_argument("--work", type=int, default=0, help="FNV passes per chunk (0 = echo)")
    ap.add_argument("--token", default="")
    args = ap.parse_args()
    if args.loop == "uvloop":
        import uvloop
        uvloop.install()
    asyncio.run(amain(args.host, args.port, args.work))


if __name__ == "__main__":
    main()

"""Baseline server: gevent StreamServer echo (greenlet + libev).

Single-threaded (decision #4: GIL build, best case). One greenlet per connection.

--work N applies the SAME FNV-1a byte hash as the runloom work curve (py_fnv,
identical constants) N times over each chunk before echoing, folded into byte 0.
work=0 is the plain echo. Interpreted-Python reference (1 core) for the
cross-runtime handler work curve.
"""
import argparse
import socket

from gevent.server import StreamServer

FNV_OFF = 2166136261        # 0x811c9dc5
FNV_PRIME = 16777619        # 0x01000193
WORK = 0                    # set from --work in main()


def py_fnv(buf, n, passes):
    """Identical to srv_runloom_work.py:py_fnv -- pure inline arithmetic."""
    h = FNV_OFF
    for _ in range(passes):
        for i in range(n):
            h = ((h ^ buf[i]) * FNV_PRIME) & 0xffffffff
    return h


def handle(sock, addr):
    try:
        sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
    except OSError:
        pass
    while True:
        data = sock.recv(65536)
        if not data:
            break
        if WORK:
            b = bytearray(data)
            h = py_fnv(b, len(b), WORK)
            b[0] = (b[0] ^ (h & 0xff)) & 0xff   # fold in -> no elision
            sock.sendall(b)
        else:
            sock.sendall(data)   # echo (work=0)


def main():
    global WORK
    ap = argparse.ArgumentParser()
    ap.add_argument("--host", default="10.99.0.1")
    ap.add_argument("--port", type=int, default=9000)
    ap.add_argument("--work", type=int, default=0, help="FNV passes per chunk (0 = echo)")
    ap.add_argument("--token", default="")
    args = ap.parse_args()
    WORK = args.work
    server = StreamServer((args.host, args.port), handle, backlog=4096)
    server.init_socket()
    server.start()
    print("LISTENING %d" % server.address[1], flush=True)
    server.serve_forever()


if __name__ == "__main__":
    main()

// Baseline server: Go net echo, one goroutine per connection.
// Spec: cap Go's cores to int(cpu*0.7) via GOMAXPROCS.
//
// Acceptors: by default the server opens N SO_REUSEPORT listeners (N =
// GOMAXPROCS), one accept loop each, so the kernel load-balances incoming SYNs
// across N accept queues -- the same architecture as runloom's per-hub reuseport
// acceptors. This matters ONLY for connection-churn (accept in the hot loop); the
// persistent req/s / bandwidth paths accept once at establishment and are
// unaffected. Pass -acceptors 1 to reproduce the old single-Accept() baseline.
//
// -work N applies the SAME FNV-1a byte hash as the runloom work curve (identical
// constants) N times over each chunk before echoing, folded into byte 0 so it
// can't be elided. work=0 is the plain echo. This is the compiled/native
// reference for the cross-runtime handler work curve (multi-core via GOMAXPROCS).
package main

import (
	"context"
	"flag"
	"fmt"
	"net"
	"os"
	"runtime"
	"syscall"
)

const (
	fnvOff   uint32 = 2166136261 // 0x811c9dc5
	fnvPrime uint32 = 16777619   // 0x01000193
	// SO_REUSEPORT (Linux); literal so we need no x/sys/unix dependency.
	soReusePort = 0xf
)

// goFnv: native uint32 FNV-1a, wraparound is automatic (no mask needed, unlike
// the Python twin) -- the same work, expressed the natural way in each language.
func goFnv(buf []byte, passes int) uint32 {
	h := fnvOff
	for p := 0; p < passes; p++ {
		for _, b := range buf {
			h = (h ^ uint32(b)) * fnvPrime
		}
	}
	return h
}

func handle(c *net.TCPConn, work int) {
	defer c.Close()
	c.SetNoDelay(true)
	buf := make([]byte, 65536)
	for {
		n, err := c.Read(buf)
		if err != nil {
			return
		}
		if work > 0 {
			h := goFnv(buf[:n], work)
			buf[0] ^= byte(h & 0xff) // fold in -> no elision
		}
		if _, err := c.Write(buf[:n]); err != nil {
			return
		}
	}
}

// listenReuseport binds addr with SO_REUSEPORT set, so multiple listeners can
// share the same host:port and the kernel distributes accepts across them.
func listenReuseport(addr string) (net.Listener, error) {
	lc := net.ListenConfig{
		Control: func(network, address string, c syscall.RawConn) error {
			var serr error
			if cerr := c.Control(func(fd uintptr) {
				serr = syscall.SetsockoptInt(int(fd), syscall.SOL_SOCKET, soReusePort, 1)
			}); cerr != nil {
				return cerr
			}
			return serr
		},
	}
	return lc.Listen(context.Background(), "tcp", addr)
}

func acceptLoop(ln net.Listener, work int) {
	for {
		c, err := ln.Accept()
		if err != nil {
			return
		}
		go handle(c.(*net.TCPConn), work)
	}
}

func main() {
	host := flag.String("host", "10.99.0.1", "")
	port := flag.Int("port", 9000, "")
	gomax := flag.Int("gomaxprocs", int(float64(runtime.NumCPU())*0.7), "")
	work := flag.Int("work", 0, "FNV passes per chunk (0 = echo)")
	acceptors := flag.Int("acceptors", 0, "SO_REUSEPORT accept loops (0 = gomaxprocs; 1 = old single-Accept baseline)")
	flag.String("token", "", "")
	flag.Parse()
	runtime.GOMAXPROCS(*gomax)

	nacc := *acceptors
	if nacc <= 0 {
		nacc = *gomax
	}
	if nacc < 1 {
		nacc = 1
	}

	addr := fmt.Sprintf("%s:%d", *host, *port)
	// First listener reports the real port (covers port==0); the rest bind that
	// resolved address with SO_REUSEPORT.
	first, err := listenReuseport(addr)
	if err != nil {
		fmt.Fprintln(os.Stderr, "listen:", err)
		os.Exit(1)
	}
	la := first.Addr().(*net.TCPAddr)
	realAddr := fmt.Sprintf("%s:%d", *host, la.Port)
	fmt.Printf("LISTENING %d\n", la.Port)
	os.Stdout.Sync()

	go acceptLoop(first, *work)
	for i := 1; i < nacc; i++ {
		ln, err := listenReuseport(realAddr)
		if err != nil {
			fmt.Fprintln(os.Stderr, "listen acceptor", i, ":", err)
			os.Exit(1)
		}
		go acceptLoop(ln, *work)
	}
	select {} // accept loops run in goroutines; block main forever
}

"""Work-curve server: ONE program, one knob (--work N = FNV-1a passes over the
payload), three handler implementations spanning interpreted -> state-of-the-art:

  --handler py      interpreted Python: a `def` doing conn.recv_into / py_fnv /
                    fold / conn.send_all -- every step in the interpreter.
  --handler cython  the FASTEST / properly-optimized path: the zero-PyObject
                    Cython handler (handler_cy) with the FNV INLINE -- capi recv,
                    native FNV, fold, capi send. No Python def wrapper, no
                    per-call boxing; the whole request path is native. This is
                    runloom's state of the art (the line that competes with Go).
  --handler cdef    handler_cdef: the same native work but on the tstate-free
                    c_entry path (no Python frame at all) -- the extreme.

--work 0 is a pure echo in every mode. The work is pure inline arithmetic (no
stdlib, no blockpool offload), so it runs on the fiber's hub -- a valid per-hub
measurement. The compiled handlers (cython/cdef) set their work knob via
set_work() before serve() spawns any fiber.
"""
import argparse
import os
import sys

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))  # find handler_cy/cdef *.so

import runloom
import runloom_c

CHUNK = 65536
FNV_OFF = 2166136261        # 0x811c9dc5
FNV_PRIME = 16777619        # 0x01000193


def py_fnv(buf, n, passes):
    """Pure-Python FNV-1a -- the interpreted twin of the inline FNV in the
    compiled handlers (handler_cy / handler_cdef)."""
    h = FNV_OFF
    for _ in range(passes):
        for i in range(n):
            h = ((h ^ buf[i]) * FNV_PRIME) & 0xffffffff
    return h


def make_handle(work_fn, passes):
    def handle(conn):
        buf = bytearray(CHUNK)
        mv = memoryview(buf)
        try:
            if passes == 0:
                while True:                       # --work 0 == echo
                    n = conn.recv_into(buf)
                    if not n:
                        break
                    conn.send_all(mv[:n])
            else:
                while True:
                    n = conn.recv_into(buf)
                    if not n:
                        break
                    h = work_fn(buf, n, passes)   # the only variable: py vs compiled
                    buf[0] = (buf[0] ^ (h & 0xff)) & 0xff   # fold in -> no dead-code elision
                    conn.send_all(mv[:n])
        except OSError:
            pass
    return handle


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--host", default="10.99.0.1")
    ap.add_argument("--port", type=int, default=9000)
    ap.add_argument("--hubs", type=int, default=int((os.cpu_count() or 1) * 0.7))
    ap.add_argument("--handler", default="py", choices=["py", "cython", "cdef"])
    ap.add_argument("--work", type=int, default=0, help="FNV passes over the payload (0 = echo)")
    ap.add_argument("--token", default="")
    args = ap.parse_args()

    # cython / cdef: the work runs INLINE inside the compiled, zero-PyObject
    # handler -- the whole request path is native (the Go-comparable line). py:
    # the interpreted Python def. Compiled handlers take the work via set_work().
    if args.handler == "cython":
        import handler_cy            # zero-PyObject Cython def, work inline
        handler_cy.set_work(args.work)
        srv_handler = handler_cy.handler
    elif args.handler == "cdef":
        import handler_cdef          # tstate-free c_entry, work inline
        handler_cdef.set_work(args.work)
        srv_handler = handler_cdef.handler
    else:
        srv_handler = make_handle(py_fnv, args.work)   # interpreted baseline

    def root():
        port, listeners = runloom_c.serve(args.host, args.port, srv_handler,
                                          acceptors=args.hubs, backlog=4096)
        print("LISTENING %d" % port, flush=True)
        runloom.sleep(float("inf"))

    runloom.run(args.hubs, main_fn=root)


if __name__ == "__main__":
    main()

// loadgen -- the Go closed-loop echo load generator (spec: client language = go).
//
// Method (spec): open `conns` persistent keepalive connections to the server,
// disable Nagle, then on each connection repeatedly send the payload and read it
// back (closed loop).  Count round-trips over a fixed measurement window after a
// ramp/warmup phase; req/s = round-trips / window.  A lock-free log-bucket
// histogram records per-round-trip latency for p50/p99/p99.9 (no sampling bias,
// bounded memory).  Emits one JSON object on stdout.
//
// Decisions baked in: TCP_NODELAY is set once per connection at setup, never in
// the per-request loop (decision #6).  GOMAXPROCS is the client core budget
// (spec: int(cpu*0.25)).  Per-connection buffers are sized to the payload, so
// the small-payload req/s run and the few-connection 1.5 MB bandwidth run both
// stay within memory.
package main

import (
	"encoding/json"
	"flag"
	"fmt"
	"io"
	"math"
	"net"
	"os"
	"runtime"
	"sync"
	"sync/atomic"
	"time"
)

// ---- lock-free latency histogram (log-spaced buckets, ~1.04x each) ----
const (
	nBuckets = 512
	logBase  = 1.04
)

var lnBase = math.Log(logBase)

type hist struct {
	b [nBuckets]int64
}

func bucketOf(ns int64) int {
	if ns < 1 {
		ns = 1
	}
	us := float64(ns) / 1000.0
	if us < 1 {
		return 0
	}
	i := int(math.Log(us) / lnBase)
	if i < 0 {
		i = 0
	}
	if i >= nBuckets {
		i = nBuckets - 1
	}
	return i
}

func bucketLowUS(i int) float64 { return math.Pow(logBase, float64(i)) }

func (h *hist) record(ns int64) { atomic.AddInt64(&h.b[bucketOf(ns)], 1) }

func (h *hist) merge(o *hist) {
	for i := range h.b {
		h.b[i] += o.b[i]
	}
}

func (h *hist) total() int64 {
	var t int64
	for _, c := range h.b {
		t += c
	}
	return t
}

// percentile returns the bucket-low latency in microseconds at quantile q.
func (h *hist) pct(q float64) float64 {
	total := h.total()
	if total == 0 {
		return 0
	}
	target := int64(q * float64(total))
	var cum int64
	for i, c := range h.b {
		cum += c
		if cum >= target {
			return bucketLowUS(i)
		}
	}
	return bucketLowUS(nBuckets - 1)
}

func main() {
	addr := flag.String("addr", "10.99.0.1:9000", "server address")
	conns := flag.Int("conns", 64, "concurrent persistent connections")
	payload := flag.Int("payload", 1024, "bytes sent+echoed per round trip")
	ramp := flag.Float64("ramp", 2.0, "ramp/warmup seconds (not counted)")
	measure := flag.Float64("measure", 5.0, "measurement window seconds")
	gomax := flag.Int("gomaxprocs", runtime.NumCPU()/4, "GOMAXPROCS (client cores)")
	flag.Parse()
	runtime.GOMAXPROCS(*gomax)

	var measuring atomic.Bool
	var totalReqs atomic.Int64
	var connErrs atomic.Int64
	var establishErrs atomic.Int64
	var stop atomic.Bool

	sendTemplate := make([]byte, *payload)
	for i := range sendTemplate {
		sendTemplate[i] = byte(i)
	}

	connsList := make([]net.Conn, 0, *conns)
	var listMu sync.Mutex
	perHist := make([]hist, *conns)
	var wg sync.WaitGroup

	worker := func(idx int, c *net.TCPConn) {
		defer wg.Done()
		send := make([]byte, *payload)
		copy(send, sendTemplate)
		recvbuf := make([]byte, *payload)
		h := &perHist[idx]
		for !stop.Load() {
			t0 := time.Now()
			if _, err := c.Write(send); err != nil {
				if !stop.Load() {
					connErrs.Add(1)
				}
				return
			}
			if _, err := io.ReadFull(c, recvbuf); err != nil {
				if !stop.Load() {
					connErrs.Add(1)
				}
				return
			}
			if measuring.Load() {
				totalReqs.Add(1)
				h.record(time.Since(t0).Nanoseconds())
			}
		}
	}

	// Establish all connections first (spec: connect num conns, then measure).
	// Dial in PARALLEL (capped) so high connection counts don't serialise into
	// a multi-second ramp -- sequential dials made the 8k+ rungs time out.
	d := net.Dialer{Timeout: 5 * time.Second}
	var estWg sync.WaitGroup
	sem := make(chan struct{}, 512)
	for i := 0; i < *conns; i++ {
		estWg.Add(1)
		sem <- struct{}{}
		go func(idx int) {
			defer estWg.Done()
			defer func() { <-sem }()
			c, err := d.Dial("tcp", *addr)
			if err != nil {
				establishErrs.Add(1)
				return
			}
			tcp := c.(*net.TCPConn)
			tcp.SetNoDelay(true)
			listMu.Lock()
			connsList = append(connsList, c)
			listMu.Unlock()
			wg.Add(1)
			go worker(idx, tcp)
		}(i)
	}
	estWg.Wait()

	live := len(connsList)
	if live == 0 {
		out := map[string]any{"error": "no connections established",
			"establish_errors": establishErrs.Load(), "requested_conns": *conns}
		b, _ := json.Marshal(out)
		fmt.Println(string(b))
		os.Exit(1)
	}

	time.Sleep(time.Duration(*ramp * float64(time.Second)))
	measuring.Store(true)
	t0 := time.Now()
	time.Sleep(time.Duration(*measure * float64(time.Second)))
	measuring.Store(false)
	elapsed := time.Since(t0).Seconds()

	stop.Store(true)
	listMu.Lock()
	for _, c := range connsList {
		c.SetDeadline(time.Now())
	}
	listMu.Unlock()
	doneCh := make(chan struct{})
	go func() { wg.Wait(); close(doneCh) }()
	select {
	case <-doneCh:
	case <-time.After(2 * time.Second):
	}

	var agg hist
	for i := range perHist {
		agg.merge(&perHist[i])
	}
	reqs := totalReqs.Load()
	rps := float64(reqs) / elapsed
	out := map[string]any{
		"requested_conns":  *conns,
		"live_conns":       live,
		"payload":          *payload,
		"gomaxprocs":       *gomax,
		"measure_s":        elapsed,
		"reqs":             reqs,
		"rps":              rps,
		"bytes_per_s":      rps * float64(*payload) * 2, // sent + echoed
		"p50_us":           agg.pct(0.50),
		"p99_us":           agg.pct(0.99),
		"p999_us":          agg.pct(0.999),
		"max_us":           agg.pct(1.0),
		"conn_errors":      connErrs.Load(),
		"establish_errors": establishErrs.Load(),
	}
	b, _ := json.Marshal(out)
	fmt.Println(string(b))
}

// churn_loadgen -- connection-CHURN load generator: the metric the persistent
// loadgen deliberately avoids.  Each worker loops: dial -> send payload -> read
// it back -> CLOSE -> repeat, as hard as it can.  So the server pays
// accept + spawn-a-handler + serve + teardown for EVERY counted unit, in the hot
// loop -- this is conn/s (a.k.a. "new connection per request"), where the
// per-connection spawn cost actually lands.  Counterpart to loadgen.go (req/s on
// persistent keepalive connections).  Emits one JSON object on stdout.
//
// 1 request per connection, so conn/s == req/s for this workload, but every
// request is a fresh connection -> a fresh server-side handler spawn.
package main

import (
	"encoding/json"
	"flag"
	"fmt"
	"io"
	"math"
	"net"
	"runtime"
	"strings"
	"sync"
	"sync/atomic"
	"time"
)

const (
	nBuckets = 512
	logBase  = 1.04
)

var lnBase = math.Log(logBase)

type hist struct{ b [nBuckets]int64 }

func bucketOf(ns int64) int {
	if ns < 1 {
		ns = 1
	}
	us := float64(ns) / 1000.0
	if us < 1 {
		return 0
	}
	i := int(math.Log(us) / lnBase)
	if i < 0 {
		i = 0
	}
	if i >= nBuckets {
		i = nBuckets - 1
	}
	return i
}
func bucketLowUS(i int) float64 { return math.Pow(logBase, float64(i)) }
func (h *hist) record(ns int64)  { atomic.AddInt64(&h.b[bucketOf(ns)], 1) }
func (h *hist) merge(o *hist) {
	for i := range h.b {
		h.b[i] += o.b[i]
	}
}
func (h *hist) total() int64 {
	var t int64
	for _, c := range h.b {
		t += c
	}
	return t
}
func (h *hist) pct(q float64) float64 {
	total := h.total()
	if total == 0 {
		return 0
	}
	target := int64(q * float64(total))
	var cum int64
	for i, c := range h.b {
		cum += c
		if cum >= target {
			return bucketLowUS(i)
		}
	}
	return bucketLowUS(nBuckets - 1)
}

func main() {
	addr := flag.String("addr", "10.99.0.1:9000", "server address")
	workers := flag.Int("conns", 64, "concurrent dialers (in-flight connections)")
	payload := flag.Int("payload", 64, "bytes sent+echoed per connection")
	ramp := flag.Float64("ramp", 2.0, "ramp/warmup seconds (not counted)")
	measure := flag.Float64("measure", 5.0, "measurement window seconds")
	gomax := flag.Int("gomaxprocs", runtime.NumCPU()/4, "GOMAXPROCS (client cores)")
	// Source-IP fan-out (the netns analog of big_100's 127/8 fragment trick): the
	// churn client ACTIVELY closes every connection, so it accumulates TIME_WAIT
	// and burns ephemeral ports on its own 4-tuples.  A single source IP has only
	// ~64k ephemeral ports, and at a high conn/s rate the TIME_WAIT backlog
	// (rate x fin_timeout) blows past that and dials start failing -- which caps
	// the measured conn/s, not the server.  Binding each connect to a rotating
	// source IP from a block on the client veth gives each IP its own independent
	// ephemeral/TIME_WAIT pool, multiplying capacity by the number of IPs.  Empty
	// -> the old single unbound dialer (so the persistent loadgen is unchanged).
	srcips := flag.String("srcips", "", "comma-separated client source IPs to "+
		"rotate connects across (TIME_WAIT/ephemeral fan-out); empty = unbound")
	flag.Parse()
	runtime.GOMAXPROCS(*gomax)

	// One dialer per source IP (LocalAddr port 0 -> kernel picks an ephemeral
	// port on that IP).  Built once, shared read-only across workers.
	var dialers []*net.Dialer
	for _, s := range strings.Split(*srcips, ",") {
		s = strings.TrimSpace(s)
		if s == "" {
			continue
		}
		ip := net.ParseIP(s)
		if ip == nil {
			fmt.Printf("{\"error\":\"bad -srcips entry %q\"}\n", s)
			return
		}
		dialers = append(dialers, &net.Dialer{
			Timeout:   5 * time.Second,
			LocalAddr: &net.TCPAddr{IP: ip},
		})
	}
	if len(dialers) == 0 {
		dialers = []*net.Dialer{{Timeout: 5 * time.Second}}
	}
	nDialers := len(dialers)

	var measuring atomic.Bool
	var totalConns atomic.Int64
	var dialErrs, ioErrs atomic.Int64
	var stop atomic.Bool

	send := make([]byte, *payload)
	for i := range send {
		send[i] = byte(i)
	}
	perHist := make([]hist, *workers)
	var wg sync.WaitGroup

	worker := func(idx int) {
		defer wg.Done()
		recvbuf := make([]byte, *payload)
		mine := make([]byte, *payload)
		copy(mine, send)
		h := &perHist[idx]
		di := idx % nDialers // offset per worker, then rotate per connection
		for !stop.Load() {
			d := dialers[di]
			di++
			if di >= nDialers {
				di = 0
			}
			t0 := time.Now()
			c, err := d.Dial("tcp", *addr)        // a NEW connection -> server spawns a handler
			if err != nil {
				if !stop.Load() {
					dialErrs.Add(1)
				}
				continue
			}
			tcp := c.(*net.TCPConn)
			tcp.SetNoDelay(true)
			ok := true
			if _, err := tcp.Write(mine); err != nil {
				ok = false
			}
			if ok {
				if _, err := io.ReadFull(tcp, recvbuf); err != nil {
					ok = false
				}
			}
			tcp.Close()
			if !ok {
				if !stop.Load() {
					ioErrs.Add(1)
				}
				continue
			}
			if measuring.Load() {
				totalConns.Add(1)
				h.record(time.Since(t0).Nanoseconds())
			}
		}
	}

	wg.Add(*workers)
	for i := 0; i < *workers; i++ {
		go worker(i)
	}

	time.Sleep(time.Duration(*ramp * float64(time.Second)))
	measuring.Store(true)
	t0 := time.Now()
	time.Sleep(time.Duration(*measure * float64(time.Second)))
	measuring.Store(false)
	elapsed := time.Since(t0).Seconds()
	stop.Store(true)

	doneCh := make(chan struct{})
	go func() { wg.Wait(); close(doneCh) }()
	select {
	case <-doneCh:
	case <-time.After(8 * time.Second):
	}

	var agg hist
	for i := range perHist {
		agg.merge(&perHist[i])
	}
	conns := totalConns.Load()
	cps := float64(conns) / elapsed
	out := map[string]any{
		"workers":     *workers,
		"payload":     *payload,
		"gomaxprocs":  *gomax,
		"measure_s":   elapsed,
		"conns":       conns,
		"conns_per_s": cps,
		"p50_us":      agg.pct(0.50),
		"p99_us":      agg.pct(0.99),
		"p999_us":     agg.pct(0.999),
		"dial_errors": dialErrs.Load(),
		"io_errors":   ioErrs.Load(),
		// Aliases so this is a drop-in for measure.ladder (same saturation
		// machinery as the req/s benchmark). One request per fresh connection,
		// so conn/s IS the req/s for this workload; live_conns is the count of
		// concurrently in-flight dialers (the driven concurrency level); the
		// error fields map churn's dial/io failures onto the ladder's tally.
		"rps":              cps,
		"live_conns":       int64(*workers),
		"establish_errors": dialErrs.Load(),
		"conn_errors":      ioErrs.Load(),
	}
	b, _ := json.Marshal(out)
	fmt.Println(string(b))
}

#!/usr/bin/env python3
"""Run the whole benchmark suite (perf + speed + memory), full mode, in order,
and capture an environment snapshot.  Each sub-orchestrator writes its own JSON
into benchmark/results/; gen_report.py turns them into the consolidated HTML.

Usage: python3 run_all.py [--quick]
"""
import argparse
import json
import os
import subprocess
import sys
import time

sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "harness"))
import config
import env

HERE = os.path.dirname(os.path.abspath(__file__))
PY = config.FT_PYTHON


def phase(name, argv):
    print("\n" + "=" * 70, flush=True)
    print("== PHASE: %s ==" % name, flush=True)
    print("=" * 70, flush=True)
    t0 = time.time()
    rc = subprocess.run([PY, os.path.join(HERE, argv[0])] + argv[1:]).returncode
    print("== %s done in %.0fs (rc=%d) ==" % (name, time.time() - t0, rc), flush=True)
    return rc


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--quick", action="store_true")
    args = ap.parse_args()
    q = ["--quick"] if args.quick else []

    os.makedirs(config.RESULTS_DIR, exist_ok=True)
    # environment snapshot up front
    info = env.capture()
    with open(os.path.join(config.RESULTS_DIR, "env.json"), "w") as f:
        json.dump(info, f, indent=2)
    print("\n".join(env.header_lines(info)), flush=True)

    phase("performance (req/s + bandwidth)", ["run_perf.py"] + q)
    phase("speed (spawn/ctxswitch/rtt/http)", ["run_speed.py"] + q)
    phase("memory (RSS per fiber + 1M)", ["run_mem.py"] + q)
    print("\nALL PHASES COMPLETE. results in", config.RESULTS_DIR, flush=True)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Performance benchmark orchestrator: req/s and bandwidth for the 5 runloom
tiers + asyncio + uvloop + gevent + go.

For each server config and each metric (small-payload req/s, 1.5 MB bandwidth) it
brings the server up in the server netns (pinned, fd-raised, debug-off), then has
the Go loadgen in the client netns walk the connection ladder until req/s
plateaus, recording the full curve, latency percentiles, and which side
(server/client) was CPU-bound at the peak.

Usage:
    python3 run_perf.py [--quick] [--only NAME[,NAME...]] [--metric reqps|bandwidth|both]
"""
import argparse
import json
import os
import sys
import time

sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "harness"))
import config
import topo
import measure

LOADGEN = os.path.join(config.CLIENTS_DIR, "loadgen")
SD = config.SERVERS_DIR
P = config.FT_PYTHON
G = config.GIL_PYTHON

_port_counter = [config.BASE_PORT]


def next_port():
    _port_counter[0] += 1
    return _port_counter[0]


def py(*rest):
    return [P, *rest]


# Each spec: name, label, interpreter tag, server "cores" (shown raw alongside throughput),
# cpu pin spec, gil flag, env, and make_argv(port, token).
def build_specs():
    many = config.SERVER_CPU_SPEC
    one = str(config.SERVER_CPUS[0])
    HUBS = config.HUBS
    GO = config.GO_SERVER_CORES

    def rl(script, extra=()):
        def mk(port, token):
            return py(os.path.join(SD, script), "--host", config.SRV_IP,
                      "--port", str(port), "--hubs", str(HUBS),
                      "--token", token, *extra)
        return mk

    def base(script, interp, extra=()):
        def mk(port, token):
            return [G if interp == "gil" else P, os.path.join(SD, script),
                    "--host", config.SRV_IP, "--port", str(port),
                    "--token", token, *extra]
        return mk

    def gomk(port, token):
        return [os.path.join(SD, "go_netpoll_native_net"), "-host", config.SRV_IP,
                "-port", str(port), "-gomaxprocs", str(GO), "-token", token]

    return [
        dict(name="runloom_sync", label="Runloom sync wrappers (epoll, py handler)",
             interp="3.13t FT", cores=HUBS, cpus=many, gil_off=True, env={},
             make=rl("runloom_epoll_py_sync.py")),
        dict(name="runloom_c", label="Runloom C scaffold (py handler, C TCPConn)",
             interp="3.13t FT", cores=HUBS, cpus=many, gil_off=True, env={},
             make=rl("runloom_epoll_py_tcpcon.py")),
        dict(name="runloom_c_cython", label="Runloom C scaffold + Cython C handler (epoll)",
             interp="3.13t FT", cores=HUBS, cpus=many, gil_off=True, env={},
             make=rl("runloom_iouring_cython_tcpcon.py", ("--optimize", "none"))),
        dict(name="runloom_iouring", label="Runloom io_uring loop (py handler)",
             interp="3.13t FT", cores=HUBS, cpus=many, gil_off=True,
             env={"RUNLOOM_IOURING_LOOP": "1"}, make=rl("runloom_epoll_py_sync.py")),
        dict(name="runloom_cython", label="Runloom io_uring + Cython C handler",
             interp="3.13t FT", cores=HUBS, cpus=many, gil_off=True,
             env={"RUNLOOM_IOURING_LOOP": "1"},
             make=rl("runloom_iouring_cython_tcpcon.py", ("--optimize", "none"))),
        dict(name="runloom_cython_opt", label="Runloom io_uring + Cython + optimize(throughput)",
             interp="3.13t FT", cores=HUBS, cpus=many, gil_off=True,
             env={"RUNLOOM_IOURING_LOOP": "1"},
             make=rl("runloom_iouring_cython_tcpcon.py", ("--optimize", "throughput"))),
        dict(name="runloom_cdef", label="Runloom io_uring + Cython cdef handler (tstate-free c_entry)",
             interp="3.13t FT", cores=HUBS, cpus=many, gil_off=True,
             env={"RUNLOOM_IOURING_LOOP": "1"},
             make=rl("runloom_iouring_cdef_tcpcon.py")),
        dict(name="runloom_cdef_epoll", label="Runloom epoll + Cython cdef handler (tstate-free c_entry)",
             interp="3.13t FT", cores=HUBS, cpus=many, gil_off=True, env={},
             make=rl("runloom_iouring_cdef_tcpcon.py")),
        dict(name="asyncio", label="asyncio Protocol (GIL, 1 core)",
             interp="3.13 GIL", cores=1, cpus=one, gil_off=False, env={},
             make=base("asyncio_epoll_py_proto.py", "gil", ("--loop", "asyncio"))),
        dict(name="uvloop", label="uvloop (GIL, 1 core)",
             interp="3.13 GIL", cores=1, cpus=one, gil_off=False, env={},
             make=base("asyncio_epoll_py_proto.py", "gil", ("--loop", "uvloop"))),
        dict(name="gevent", label="gevent StreamServer (GIL, 1 core)",
             interp="3.13 GIL", cores=1, cpus=one, gil_off=False, env={},
             make=base("gevent_libev_py_stream.py", "gil")),
        dict(name="go", label="Go net (GOMAXPROCS=%d)" % GO,
             interp="go", cores=GO, cpus=many, gil_off=True, env={}, make=gomk),
    ]


def server_factory(spec, port, token):
    def factory():
        argv = spec["make"](port, token)
        cmd = topo.ns_cmd(config.SRV_NS, argv, cpus=spec["cpus"],
                          extra_env=spec["env"], gil_off=spec["gil_off"],
                          raise_fd=True)
        srv = measure.Server(cmd, token, spec["name"])
        srv.start(timeout=40)
        time.sleep(0.5)  # let acceptors arm
        return srv
    return factory


METRICS = {
    "reqps": dict(payload=config.PAYLOAD_SMALL, ladder=config.CONN_LADDER),
    "bandwidth": dict(payload=config.PAYLOAD_LARGE,
                      ladder=[1, 2, 4, 8, 16, 32, 64, 128]),
}


def run(quick, only, which_metrics):
    os.makedirs(config.RESULTS_DIR, exist_ok=True)
    specs = build_specs()
    if only:
        specs = [s for s in specs if s["name"] in only]
    reps = 1 if quick else config.REPS
    ramp = 1.0 if quick else config.RAMP_S
    measure_s = 2.0 if quick else config.MEASURE_S
    gomax = config.CLIENT_CORES

    print("== topology setup ==", flush=True)
    topo.setup()
    results = {"meta": config.summary(), "quick": quick, "servers": {}}
    try:
        for spec in specs:
            results["servers"][spec["name"]] = {
                "label": spec["label"], "interp": spec["interp"],
                "cores": spec["cores"], "metrics": {}}
            for metric in which_metrics:
                mc = METRICS[metric]
                lad = mc["ladder"]
                if quick:
                    lad = lad[::3] or lad[:1]
                port = next_port()
                token = "RLBENCH_%s_%s_%d" % (spec["name"], metric, port)
                print("\n== %s / %s (payload=%dB, cores=%d) =="
                      % (spec["name"], metric, mc["payload"], spec["cores"]), flush=True)
                try:
                    srv_cpus = [int(c) for c in spec["cpus"].split(",")]
                    out = measure.ladder(
                        server_factory(spec, port, token), LOADGEN,
                        "%s:%d" % (config.SRV_IP, port), mc["payload"], lad,
                        reps, ramp, measure_s, gomax, config.PLATEAU_PATIENCE,
                        server_cpus=srv_cpus)
                    out["payload"] = mc["payload"]
                    results["servers"][spec["name"]]["metrics"][metric] = out
                    pk = out["peak"]
                    print("  -> PEAK rps=%.0f @ conns=%s  bottleneck=%s"
                          % (pk["rps_median"], pk.get("conns"), out["bottleneck_at_peak"]),
                          flush=True)
                except Exception as e:
                    print("  !! FAILED: %r" % e, flush=True)
                    results["servers"][spec["name"]]["metrics"][metric] = {"error": repr(e)}
                # make sure nothing lingers
                import subprocess
                subprocess.run(["sudo", "-n", "pkill", "-9", "-f", token],
                               capture_output=True)
                time.sleep(0.5)
    finally:
        topo.teardown()

    out_path = os.path.join(config.RESULTS_DIR,
                            "perf_quick.json" if quick else "perf.json")
    # Merge into any existing results with the same quick flag, so a targeted
    # `--only <name>` re-run ADDS that server to the matrix instead of clobbering
    # the others (used to splice in extra configs without re-running everything).
    if os.path.exists(out_path):
        try:
            with open(out_path) as f:
                prev = json.load(f)
            if bool(prev.get("quick")) == bool(quick):
                merged = prev.get("servers", {})
                merged.update(results["servers"])
                results["servers"] = merged
        except Exception:
            pass
    with open(out_path, "w") as f:
        json.dump(results, f, indent=2)
    print("\nwrote", out_path, flush=True)
    return out_path


if __name__ == "__main__":
    ap = argparse.ArgumentParser()
    ap.add_argument("--quick", action="store_true")
    ap.add_argument("--only", default="")
    ap.add_argument("--metric", default="both", choices=["reqps", "bandwidth", "both"])
    args = ap.parse_args()
    only = set(filter(None, args.only.split(",")))
    metrics = ["reqps", "bandwidth"] if args.metric == "both" else [args.metric]
    run(args.quick, only, metrics)

#!/usr/bin/env python3
"""Speed benchmark orchestrator: spawn / context-switch / HTTP req-s / TCP RTT
for [runloom, go, asyncio, greenlet, uvloop].

spawn + ctxswitch are pure-scheduler (no network); the orchestrator runs an n=0
startup baseline and subtracts it. rtt + http launch a Go target in the server
netns and run each runtime's client in the client netns.

Usage: python3 run_speed.py [--quick] [--only runtime,...] [--metric ...]
"""
import argparse
import json
import os
import subprocess
import sys
import time

sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "harness"))
import config
import topo
import measure

SP = os.path.join(config.SUITE_DIR, "speed")
SD = config.SERVERS_DIR
P = config.FT_PYTHON
G = config.GIL_PYTHON
SPEED_GO = os.path.join(SP, "speed_go")
SRV_GO = os.path.join(SD, "go_netpoll_native_net")           # echo target for rtt
MANY = config.SERVER_CPU_SPEC
ONE_S = str(config.SERVER_CPUS[0])
ONE_C = str(config.CLIENT_CPUS[0])
HUBS = config.HUBS
N_FULL = 1_000_000
N_SWITCH_FULL = 4_000_000
_port = [config.BASE_PORT + 400]


def nextport():
    _port[0] += 1
    return _port[0]


def parse_json(out):
    for line in reversed(out.strip().splitlines()):
        line = line.strip()
        if line.startswith("{"):
            return json.loads(line)
    raise RuntimeError("no JSON in: %s" % out[-500:])


def run_local(argv, cpus, gil_off, extra_env=None, raise_fd=True, timeout=300):
    cmd = topo.pinned_cmd(argv, cpus=cpus, extra_env=extra_env,
                          gil_off=gil_off, raise_fd=raise_fd)
    r = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
    if r.returncode != 0 and not r.stdout.strip():
        raise RuntimeError("cmd failed rc=%d: %s\n%s" % (r.returncode, " ".join(argv), r.stderr[-800:]))
    return parse_json(r.stdout)


# ---- runtime client commands (argv builders) ----
def argv_for(rt, metric, n, host=None, port=None, conns=64, ramp=1.0, measure_s=3.0,
             gomax=None):
    if rt == "go":
        if metric in ("spawn", "ctxswitch"):
            return [SPEED_GO, "-metric", metric, "-n", str(n), "-gomaxprocs", str(gomax or HUBS)]
        if metric == "rtt":
            return [SPEED_GO, "-metric", "rtt", "-addr", "%s:%d" % (host, port), "-n", str(n)]
        return [SPEED_GO, "-metric", "httpclient", "-addr", "%s:%d" % (host, port),
                "-conns", str(conns), "-gomaxprocs", str(gomax or config.CLIENT_CORES),
                "-ramp", str(ramp), "-measure", str(measure_s)]
    if rt == "runloom":
        base = [P, os.path.join(SP, "runloom_epoll_py_fiber.py"), "--metric", metric]
        if metric in ("spawn", "ctxswitch"):
            return base + ["--n", str(n), "--hubs", str(gomax or HUBS)]
        if metric == "rtt":
            return base + ["--host", host, "--port", str(port), "--n", str(n)]
        return base + ["--host", host, "--port", str(port), "--conns", str(conns),
                       "--hubs", str(gomax or config.CLIENT_CORES),
                       "--ramp", str(ramp), "--measure", str(measure_s)]
    if rt == "runloom_c":
        base = [P, os.path.join(SP, "run_centry.py"), "--metric", metric]
        if metric in ("spawn", "ctxswitch"):
            return base + ["--n", str(n), "--hubs", str(gomax or HUBS)]
        # centry doesn't support network metrics yet
        raise ValueError("runloom_c only supports spawn/ctxswitch metrics")
    if rt in ("asyncio", "uvloop"):
        base = [G, os.path.join(SP, "speed_asyncio.py"), "--metric", metric, "--loop", rt]
        if metric in ("spawn", "ctxswitch"):
            return base + ["--n", str(n)]
        if metric == "rtt":
            return base + ["--host", host, "--port", str(port), "--n", str(n)]
        return base + ["--host", host, "--port", str(port), "--conns", str(conns),
                       "--ramp", str(ramp), "--measure", str(measure_s)]
    if rt == "greenlet":
        base = [G, os.path.join(SP, "greenlet_native_py_coro.py"), "--metric", metric]
        if metric in ("spawn", "ctxswitch"):
            return base + ["--n", str(n)]
        if metric == "rtt":
            return base + ["--host", host, "--port", str(port), "--n", str(n)]
        return base + ["--host", host, "--port", str(port), "--conns", str(conns),
                       "--ramp", str(ramp), "--measure", str(measure_s)]
    raise ValueError(rt)


RUNTIMES = ["runloom", "runloom_c", "go", "asyncio", "greenlet", "uvloop"]


def cpus_for(rt, metric):
    # single-threaded runtimes -> 1 core; multi-core -> server set (spawn/ctx) or
    # client set (network clients).
    if rt in ("asyncio", "uvloop", "greenlet"):
        return ONE_S if metric in ("spawn", "ctxswitch") else ONE_C
    return MANY if metric in ("spawn", "ctxswitch") else config.CLIENT_CPU_SPEC


def gil_off_for(rt):
    return rt in ("runloom", "runloom_c", "go")  # go ignores it; GIL runtimes need GIL on


def cores_of(res, rt, metric):
    return res.get("cores", 1)


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--quick", action="store_true")
    ap.add_argument("--only", default="")
    ap.add_argument("--metric", default="all")
    args = ap.parse_args()
    only = set(filter(None, args.only.split(","))) or set(RUNTIMES)
    metrics = (["spawn", "ctxswitch", "rtt", "http"] if args.metric == "all"
               else [args.metric])
    nspawn = 100_000 if args.quick else N_FULL
    nswitch = 1_000_000 if args.quick else N_SWITCH_FULL
    nrtt = 20_000 if args.quick else 100_000
    ramp, meas = (1.0, 2.0) if args.quick else (config.RAMP_S, config.MEASURE_S)
    conns = 256

    os.makedirs(config.RESULTS_DIR, exist_ok=True)
    results = {"meta": config.summary(), "quick": args.quick, "metrics": {}}

    # ---- spawn ----
    if "spawn" in metrics:
        results["metrics"]["spawn"] = {}
        for rt in [r for r in RUNTIMES if r in only]:
            try:
                base = run_local(argv_for(rt, "spawn", 0), cpus_for(rt, "spawn"),
                                 gil_off_for(rt), raise_fd=True)
                full = run_local(argv_for(rt, "spawn", nspawn), cpus_for(rt, "spawn"),
                                 gil_off_for(rt), raise_fd=True, timeout=600)
                sec = max(full["seconds"] - base["seconds"], 1e-9)
                rate = nspawn / sec
                results["metrics"]["spawn"][rt] = {
                    "n": nspawn, "seconds": sec, "rate_per_s": rate,
                    "cores": full.get("cores", 1)}
                print("spawn   %-9s %.3fs  %.0f/s  (%.2f us/task, %d cores)"
                      % (rt, sec, rate, sec * 1e6 / nspawn, full.get("cores", 1)), flush=True)
            except Exception as e:
                print("spawn   %-9s FAILED %r" % (rt, e), flush=True)
                results["metrics"]["spawn"][rt] = {"error": repr(e)}

    # ---- ctxswitch ----
    if "ctxswitch" in metrics:
        results["metrics"]["ctxswitch"] = {}
        for rt in [r for r in RUNTIMES if r in only]:
            # runloom: this is a pure-CPU yield loop, which spuriously trips the
            # ATTACHED/CPU-preempt watchdog (a feature for I/O workloads that park
            # back to hub_main, never this microbenchmark). Measure the
            # representative cooperative-yield cost with it OFF -- consistent with
            # the c_entry capstone, which is also preempt-off. Other runtimes
            # ignore these vars.
            cx_env = {"RUNLOOM_PREEMPT": "0", "RUNLOOM_SYSMON": "0"} if rt == "runloom" else None
            try:
                base = run_local(argv_for(rt, "ctxswitch", 0), cpus_for(rt, "ctxswitch"),
                                 gil_off_for(rt), extra_env=cx_env, raise_fd=True)
                full = run_local(argv_for(rt, "ctxswitch", nswitch), cpus_for(rt, "ctxswitch"),
                                 gil_off_for(rt), extra_env=cx_env, raise_fd=True, timeout=600)
                sw = full["switches"]
                sec = max(full["seconds"] - base["seconds"], 1e-9)
                ns = sec * 1e9 / sw
                results["metrics"]["ctxswitch"][rt] = {
                    "switches": sw, "seconds": sec, "ns_per_switch": ns,
                    "cores": full.get("cores", 1)}
                print("ctxsw   %-9s %.0f ns/switch  (%d switches, %d cores)"
                      % (rt, ns, sw, full.get("cores", 1)), flush=True)
            except Exception as e:
                print("ctxsw   %-9s FAILED %r" % (rt, e), flush=True)
                results["metrics"]["ctxswitch"][rt] = {"error": repr(e)}

    # ---- network metrics: need the topology + a Go target ----
    if "rtt" in metrics or "http" in metrics:
        topo.setup()
        try:
            if "rtt" in metrics:
                results["metrics"]["rtt"] = {}
                port = nextport()
                token = "RLSPEED_echo_%d" % port
                echo = measure.Server(topo.ns_cmd(config.SRV_NS,
                    [SRV_GO, "-host", config.SRV_IP, "-port", str(port),
                     "-gomaxprocs", str(HUBS), "-token", token],
                    cpus=MANY, raise_fd=True, gil_off=True), token, "go-echo")
                echo.start(timeout=20)
                time.sleep(0.5)
                try:
                    for rt in [r for r in RUNTIMES if r in only]:
                        try:
                            res = _netns_client(rt, "rtt", nrtt, config.SRV_IP, port, ONE_C)
                            results["metrics"]["rtt"][rt] = res
                            print("rtt     %-9s %.0f ns/rtt" % (rt, res["ns_per_rtt"]), flush=True)
                        except Exception as e:
                            print("rtt     %-9s FAILED %r" % (rt, e), flush=True)
                            results["metrics"]["rtt"][rt] = {"error": repr(e)}
                finally:
                    echo.stop()

            if "http" in metrics:
                results["metrics"]["http"] = {}
                port = nextport()
                token = "RLSPEED_httpd_%d" % port
                httpd = measure.Server(topo.ns_cmd(config.SRV_NS,
                    [SPEED_GO, "-metric", "httpd", "-host", config.SRV_IP,
                     "-port", str(port), "-gomaxprocs", str(HUBS), "-token", token],
                    cpus=MANY, raise_fd=True, gil_off=True), token, "go-httpd")
                httpd.start(timeout=20)
                time.sleep(0.5)
                try:
                    for rt in [r for r in RUNTIMES if r in only]:
                        try:
                            res = _netns_client(rt, "http", 0, config.SRV_IP, port,
                                                config.CLIENT_CPU_SPEC if rt in ("runloom", "go")
                                                else ONE_C, conns=conns, ramp=ramp, measure_s=meas)
                            results["metrics"]["http"][rt] = res
                            print("http    %-9s %.0f rps  (%d cores)" % (rt, res["rps"], res.get("cores", 1)), flush=True)
                        except Exception as e:
                            print("http    %-9s FAILED %r" % (rt, e), flush=True)
                            results["metrics"]["http"][rt] = {"error": repr(e)}
                finally:
                    httpd.stop()
        finally:
            topo.teardown()

    out = os.path.join(config.RESULTS_DIR, "speed_quick.json" if args.quick else "speed.json")
    # Merge into any existing file so a SUBSET run (e.g. --metric ctxswitch) does
    # NOT wipe the other metrics, and a runtime that errored this pass keeps its
    # prior good value rather than overwriting it with the error.
    if os.path.exists(out):
        try:
            with open(out) as f:
                prev = json.load(f)
            pm = dict(prev.get("metrics", {}))
            for metric, rtmap in results["metrics"].items():
                dest = dict(pm.get(metric, {}))
                for rt, val in rtmap.items():
                    if isinstance(val, dict) and "error" in val and rt in dest:
                        continue  # keep the prior good value
                    dest[rt] = val
                pm[metric] = dest
            prev["meta"] = results["meta"]
            prev["quick"] = results["quick"]
            prev["metrics"] = pm
            results = prev
        except Exception as e:
            print("merge into existing %s failed (%r); overwriting" % (out, e), flush=True)
    with open(out, "w") as f:
        json.dump(results, f, indent=2)
    print("\nwrote", out, flush=True)


def _netns_client(rt, metric, n, host, port, cpus, conns=64, ramp=1.0, measure_s=3.0):
    """Run a runtime's network client inside the CLIENT netns."""
    argv = argv_for(rt, metric, n, host=host, port=port, conns=conns,
                    ramp=ramp, measure_s=measure_s)
    cmd = topo.ns_cmd(config.CLI_NS, argv, cpus=cpus, gil_off=gil_off_for(rt),
                      raise_fd=True)
    r = subprocess.run(cmd, capture_output=True, text=True, timeout=ramp + measure_s + 120)
    return parse_json(r.stdout)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Memory benchmark orchestrator.

Matrix (spec): columns = [go, runloom py handler, runloom py+optimize(memory),
runloom c handler]; rows = [empty bytes/fiber, with-socket bytes/fiber,
1M fibers total RSS].

Per-fiber rows use a baseline (n=0) RSS subtracted from an n=K RSS, so the number
is the *incremental* used memory per fiber, not fixed interpreter/runtime
overhead. The 1M row is the absolute resident set for a million live fibers.
All numbers are USED memory (VmRSS / PSS), never virtual size.

Usage: python3 run_mem.py [--quick]
"""
import argparse
import json
import os
import subprocess
import sys

sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "harness"))
import config
import topo

MEM = os.path.join(config.SUITE_DIR, "memory")
MEM_GO = os.path.join(MEM, "mem_go")
MEM_RL = os.path.join(MEM, "mem_runloom.py")
P = config.FT_PYTHON
MANY = config.SERVER_CPU_SPEC


def parse_json(out):
    for line in reversed(out.strip().splitlines()):
        line = line.strip()
        if line.startswith("{"):
            return json.loads(line)
    raise RuntimeError("no JSON in output: %s" % out[-400:])


def run(argv, gil_off, timeout=600):
    cmd = topo.pinned_cmd(argv, cpus=MANY, gil_off=gil_off, raise_fd=True)
    r = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
    try:
        return parse_json(r.stdout)
    except Exception as e:
        raise RuntimeError("%r (rc=%d) stderr=%s" % (e, r.returncode, r.stderr[-400:]))


# config -> (builder(state, n) -> argv, gil_off)
def go_argv(state, n):
    return [MEM_GO, "-state", state, "-n", str(n)]


def rl_argv(handler, optimize):
    def b(state, n):
        a = [P, MEM_RL, "--state", state, "--n", str(n), "--handler", handler,
             "--hubs", str(config.HUBS)]
        if optimize != "none":
            a += ["--optimize", optimize]
        return a
    return b


CONFIGS = [
    ("go", go_argv, False),
    ("runloom_py", rl_argv("py", "none"), True),
    ("runloom_py_optmem", rl_argv("py", "memory"), True),
    ("runloom_c", rl_argv("c", "none"), True),
]


def per_unit(builder, gil_off, state, k):
    base = run(builder(state, 0), gil_off)
    full = run(builder(state, k), gil_off)
    rss_b, rss_f = base.get("rss_bytes") or 0, full.get("rss_bytes") or 0
    pss_b, pss_f = base.get("pss_bytes"), full.get("pss_bytes")
    out = {"n": k, "rss_total": rss_f, "rss_baseline": rss_b,
           "bytes_per_fiber_rss": (rss_f - rss_b) / k if k else None}
    if pss_b is not None and pss_f is not None:
        out["bytes_per_fiber_pss"] = (pss_f - pss_b) / k
    return out


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--quick", action="store_true")
    args = ap.parse_args()
    n_empty = 50_000 if args.quick else 200_000
    n_socket = 5_000 if args.quick else 20_000
    n_million = 200_000 if args.quick else 1_000_000

    topo.ensure_kernel_ceilings()   # vm.max_map_count / fs.nr_open for 1M fibers
    os.makedirs(config.RESULTS_DIR, exist_ok=True)
    results = {"meta": config.summary(), "quick": args.quick, "configs": {}}

    for name, builder, gil_off in CONFIGS:
        results["configs"][name] = {}
        # Warm any first-touch lazy init BEFORE the differential runs, so it is
        # charged to NEITHER the n=0 baseline NOR the n=K full.  The --handler c
        # column compiles its parker on the fly (Cython); a cold-cache compile
        # would inflate only the first of the two processes per_unit spawns and
        # bias bytes/fiber.  This throwaway n=0 run populates the on-disk compile
        # cache so the baseline and full processes both just import it.  Cheap
        # no-op for the go / py columns.
        try:
            run(builder("empty", 0), gil_off, timeout=300)
        except Exception:
            pass
        # empty bytes/fiber
        try:
            e = per_unit(builder, gil_off, "empty", n_empty)
            results["configs"][name]["empty"] = e
            print("empty   %-20s %6.0f B/fiber RSS  (n=%d, total=%.1f MiB)"
                  % (name, e["bytes_per_fiber_rss"], e["n"], e["rss_total"] / 2**20), flush=True)
        except Exception as ex:
            print("empty   %-20s FAILED %r" % (name, ex), flush=True)
            results["configs"][name]["empty"] = {"error": repr(ex)}
        # with-socket bytes/fiber
        try:
            s = per_unit(builder, gil_off, "socket", n_socket)
            results["configs"][name]["socket"] = s
            print("socket  %-20s %6.0f B/fiber RSS  (n=%d, total=%.1f MiB)"
                  % (name, s["bytes_per_fiber_rss"], s["n"], s["rss_total"] / 2**20), flush=True)
        except Exception as ex:
            print("socket  %-20s FAILED %r" % (name, ex), flush=True)
            results["configs"][name]["socket"] = {"error": repr(ex)}
        # 1M total (empty)
        try:
            m = run(builder("empty", n_million), gil_off, timeout=900)
            results["configs"][name]["million"] = {
                "n": n_million, "rss_total": m.get("rss_bytes"),
                "pss_total": m.get("pss_bytes"),
                "rss_per_fiber": (m.get("rss_bytes") or 0) / n_million}
            print("1M      %-20s %.2f GiB total RSS  (%.0f B/fiber, n=%d)"
                  % (name, (m.get("rss_bytes") or 0) / 2**30,
                     (m.get("rss_bytes") or 0) / n_million, n_million), flush=True)
        except Exception as ex:
            print("1M      %-20s FAILED %r" % (name, ex), flush=True)
            results["configs"][name]["million"] = {"error": repr(ex)}

    out = os.path.join(config.RESULTS_DIR, "mem_quick.json" if args.quick else "mem.json")
    with open(out, "w") as f:
        json.dump(results, f, indent=2)
    print("\nwrote", out, flush=True)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Connection-CHURN benchmark -- conn/s, the metric the req/s benchmark avoids,
measured against the SAME servers as the performance benchmark and driven to the
SAME saturation.

The persistent req/s benchmark (run_perf.py) establishes connections ONCE and
loops requests on them, so the server never spawns a handler under load.  This
one does the opposite: the client opens a NEW connection, sends one request,
reads the echo, and CLOSES -- repeated as hard as it can.  So the server pays
accept + spawn-a-handler + serve + teardown for EVERY counted connection, in the
hot loop.  This is where per-connection fiber/goroutine/coroutine spawn actually
lands -- the "spawn a handler per request" case every reader assumes.

Same servers, same load machinery as the performance benchmark, on purpose:

  * Servers: run_perf.build_specs() -- the identical 11 specs (7 runloom tiers +
    asyncio + uvloop + gevent + go), launched the identical way via
    run_perf.server_factory() (pinned cores, fd-raised, debug off, same HUBS).

  * Load: measure.ladder() -- the identical climb-until-it-plateaus saturation
    logic, with the per-core server- vs client-bound CPU check.  Here the ladder
    rung is the number of concurrent DIALERS (in-flight connection churn); the
    churn loadgen reports conn/s under the `rps` key so ladder drives it exactly
    as it drives the persistent loadgen.

Writes results/conn_churn.json in the same {meta, servers:{name:{peak,curve,...}}}
shape as perf.json.
"""
import argparse
import json
import os
import subprocess
import sys
import time

sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "harness"))
import config
import topo
import measure
import run_perf                       # reuse build_specs() + server_factory()

CHURN = os.path.join(config.CLIENTS_DIR, "churn_loadgen")
PAYLOAD = config.PAYLOAD_SMALL
# Ladder rung = number of concurrent dialers (in-flight connection churn).  Same
# ladder the req/s benchmark uses, so "same load"; the plateau rule stops the
# climb once conn/s stops growing, so heavier per-unit churn just plateaus sooner.
LADDER = config.CONN_LADDER


def _churn_sysctls():
    """Connection churn floods TIME_WAIT and burns ephemeral ports -- widen the
    range and allow TW reuse in BOTH netns so dials don't start failing."""
    for ns in (config.CLI_NS, config.SRV_NS):
        subprocess.run(["sudo", "-n", "ip", "netns", "exec", ns, "sysctl", "-w",
                        "net.ipv4.tcp_tw_reuse=1",
                        "net.ipv4.ip_local_port_range=1024 65535",
                        "net.ipv4.tcp_fin_timeout=5"],
                       capture_output=True)


def run(quick, only):
    os.makedirs(config.RESULTS_DIR, exist_ok=True)
    specs = run_perf.build_specs()
    if only:
        specs = [s for s in specs if s["name"] in only]
    reps = 1 if quick else config.REPS
    ramp = 1.0 if quick else config.RAMP_S
    measure_s = 2.0 if quick else config.MEASURE_S
    gomax = config.CLIENT_CORES
    lad = (LADDER[::3] or LADDER[:1]) if quick else LADDER

    print("== topology setup ==", flush=True)
    topo.setup()
    _churn_sysctls()
    results = {"meta": config.summary(), "quick": quick, "metric": "conn/s",
               "servers": {}}
    port = config.BASE_PORT + 700
    try:
        for spec in specs:
            port += 1
            token = "RLCHURN_%s_%d" % (spec["name"], port)
            entry = {"label": spec["label"], "interp": spec["interp"],
                     "cores": spec["cores"]}
            print("\n== churn %s (payload=%dB, cores=%d) =="
                  % (spec["name"], PAYLOAD, spec["cores"]), flush=True)
            try:
                srv_cpus = [int(c) for c in spec["cpus"].split(",")]
                out = measure.ladder(
                    run_perf.server_factory(spec, port, token), CHURN,
                    "%s:%d" % (config.SRV_IP, port), PAYLOAD, lad,
                    reps, ramp, measure_s, gomax, config.PLATEAU_PATIENCE,
                    server_cpus=srv_cpus, src_ips=config.CLI_SRC_IPS)
                out["payload"] = PAYLOAD
                entry.update(out)
                pk = out["peak"]
                print("  -> PEAK %.0f conn/s @ dialers=%s  bottleneck=%s"
                      % (pk.get("rps_median", 0), pk.get("conns"),
                         out["bottleneck_at_peak"]), flush=True)
            except Exception as e:
                entry["error"] = repr(e)
                print("  !! FAILED: %r" % e, flush=True)
            results["servers"][spec["name"]] = entry
            subprocess.run(["sudo", "-n", "pkill", "-9", "-f", token],
                           capture_output=True)
            time.sleep(0.5)
    finally:
        topo.teardown()

    out_path = os.path.join(config.RESULTS_DIR, "conn_churn.json")
    with open(out_path, "w") as f:
        json.dump(results, f, indent=2)
    print("\nwrote", out_path, flush=True)
    return out_path


if __name__ == "__main__":
    ap = argparse.ArgumentParser()
    ap.add_argument("--quick", action="store_true")
    ap.add_argument("--only", default="")
    args = ap.parse_args()
    only = set(filter(None, args.only.split(",")))
    run(args.quick, only)

#!/usr/bin/env python3
"""Spawn rate vs N -- spawn/s for every runtime as N climbs 1k -> 1M.

No bounding, no tricks: spawn N tasks, drain, measure N/seconds.  All runtimes
front-load the same way (Go's bench does too).  Each runtime runs at its native
config: runloom + go on HUBS cores (GIL off), asyncio/uvloop/greenlet on 1 core
(GIL on) -- same as the cross-runtime report.  Writes results/spawn_curve.json
for the report curve.  RUNLOOM_SYSMON/PREEMPT off (microbenchmark watchdog noise).

WARM steady-state, boot excluded for ALL runtimes.  The runloom programs are
invoked with --warm WARM: they run WARM extra in-process passes and report the
best timed pass, so the one-time runloom.run() scheduler boot is NOT in the timed
window.  Go and the GIL loops are already warm at main() (their runtime/loop boots
before they start their own timer), so no flag is needed for them -- this is the
SAME basis, not a runloom-specific trick.  (Without --warm, runloom would carry a
~39 ms scheduler boot inside the timed window that Go never pays in its window --
a startup race, not a per-spawn comparison.)

This is NAKED spawn -- the WORST case: create+run+destroy with no I/O to amortize
over.  Warm on this box at 1M, c_entry and Go are AT PARITY (~2.2M each, 8-run
medians 2.23M vs 2.24M, ranges overlapping, ranking flips run-to-run); fiber_fast
~1.9M (~0.85x Go).  The rate still climbs with N for every runtime -- a per-run
fixed cost (front-load loop + drain) amortizing; runloom's residual (~19 ms) is
larger than Go's (~5 ms), so its small-N rates sag more.  The per-spawn SLOPE is
what matters: warm, runloom's marginal cost per fiber (~440 ns) is within noise of
Go's (~410 ns).

Two runloom spawn entries are reported:
  runloom_py -- runloom.fiber_fast: the fair apples-to-apples vs Go's `go f()`
                (a thin Python spawn, no per-spawn work).
  runloom_c  -- the pure-C c_entry path (no Python frame): the scheduler ceiling.
NOTE: the DEFAULT runloom.fiber adds the grow-down auto-sizer (small right-sized
stacks, an RSS feature Go lacks).  Its learned size spawns down the DEFERRED
stack-alloc path, so the default is ~1.4M/s warm (small-stacks AND fast, ~1.7M
under optimize("throughput")), not the old ~7x-slower eager-alloc number; the
optimize("throughput")/("memory") swap
runloom.fiber between fiber_fast and grow-down.
"""
import json
import os
import subprocess
import sys

sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "harness"))
import config

HERE = os.path.dirname(os.path.abspath(__file__))
FT = config.FT_PYTHON
GIL = config.GIL_PYTHON
GO = os.path.join(HERE, "speed_go")
HUBS = 8
MANY = "16-%d" % (16 + HUBS - 1)   # the 8 server cores
ONE = "16"                          # single-core runtimes
NS = [1000, 3000, 10000, 30000, 100000, 300000, 1000000]
REPS = 2
WARM = 4   # extra in-process passes for the runloom programs (scheduler boot excluded)

# label + how to invoke spawn for each runtime
RTS = [
    ("runloom_py", "Runloom (M:N) — Python fiber"),
    ("runloom_c",  "Runloom (M:N) — pure C fiber (c_entry)"),
    ("go",         "Go (GOMAXPROCS=%d)" % HUBS),
    ("asyncio",    "asyncio (1 core)"),
    ("uvloop",     "uvloop (1 core)"),
    ("greenlet",   "greenlet (1 core)"),
]


def spec(rt, n):
    """-> (argv, cpus, gil_off)"""
    if rt == "go":
        return [GO, "-metric", "spawn", "-n", str(n), "-gomaxprocs", str(HUBS)], MANY, True
    if rt == "runloom_py":
        return [FT, os.path.join(HERE, "runloom_epoll_py_fiber.py"), "--metric", "spawn",
                "--n", str(n), "--hubs", str(HUBS), "--warm", str(WARM)], MANY, True
    if rt == "runloom_c":
        return [FT, os.path.join(HERE, "run_centry.py"), "--metric", "spawn",
                "--n", str(n), "--hubs", str(HUBS), "--warm", str(WARM)], MANY, True
    if rt in ("asyncio", "uvloop"):
        return [GIL, os.path.join(HERE, "speed_asyncio.py"), "--metric", "spawn",
                "--loop", rt, "--n", str(n)], ONE, False
    if rt == "greenlet":
        return [GIL, os.path.join(HERE, "greenlet_native_py_coro.py"), "--metric", "spawn",
                "--n", str(n)], ONE, False
    raise ValueError(rt)


def _run(argv, cpus, gil_off):
    env = dict(os.environ, PYTHONPATH=os.path.join(config.REPO, "src"),
               PYTHON_GIL="0" if gil_off else "1",
               RUNLOOM_SYSMON="0", RUNLOOM_PREEMPT="0")
    out = subprocess.run(["taskset", "-c", cpus] + argv,
                         capture_output=True, text=True, env=env, timeout=300).stdout
    for line in reversed(out.strip().splitlines()):
        if line.startswith("{"):
            return json.loads(line)
    raise RuntimeError("no json")


def rate(rt, n):
    argv, cpus, gil = spec(rt, n)
    best = 0.0
    for _ in range(REPS):
        best = max(best, n / _run(argv, cpus, gil)["seconds"])
    return best


def main():
    res = {rt: {} for rt, _ in RTS}
    labels = {rt: lab for rt, lab in RTS}
    print("hubs=%d  reps=%d (best)  raw spawn/s = N / whole-run-seconds\n" % (HUBS, REPS))
    print("%-10s" % "N" + "".join("%14s" % rt for rt, _ in RTS))
    for n in NS:
        row = "%-10d" % n
        for rt, _ in RTS:
            try:
                r = rate(rt, n)
                res[rt][n] = r
                row += "%14.0f" % r
            except Exception:
                res[rt][n] = None
                row += "%14s" % "ERR"
        print(row, flush=True)
    out = os.path.join(config.RESULTS_DIR, "spawn_curve.json")
    json.dump({"hubs": HUBS, "NS": NS, "reps": REPS, "labels": labels, "rates": res},
              open(out, "w"), indent=2)
    print("\nwrote", out)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""The handler work-curve: interpreted handler vs the fully-optimized one.

ONE server (srv_runloom_work.py), ONE knob (--work N = FNV-1a passes over the
payload), the same runtime: --handler py (an interpreted Python def doing
recv_into/py_fnv/fold/send_all) vs --handler cython (the zero-PyObject Cython
handler with the FNV INLINE -- native recv, FNV, fold, send, no Python wrapper).
The cython line is runloom's state of the art (it tracks Go in work_xrt_sweep).

`--work 0` IS the echo (the handler skips the work call), so the leftmost point
of the curve consolidates the echo load and should reproduce the echo numbers
(~600k saturated) -- a built-in cross-check. As N grows the interpreted curve
bends down while the compiled curve holds; that gap is the thing echo could
never show (every handler optimization ties on echo -- no handler CPU).

The work is PURE inline arithmetic (FNV xor/mul loop) -- nothing runloom routes
to the blockpool, so it runs on the fiber's hub and the per-core CPU accounting
stays valid. A hashlib/json/struct call would offload or converge to native and
erase the signal; stated in the report as the honest framing.

Short saturating ladder per point: as work rises the server gets more
CPU-bound and saturates at LOWER conn counts, so [1024,2048,4096] (2048 was the
proven echo saturation knee) captures every point. Writes results/work_curve.json.
"""
import json
import os
import subprocess
import sys
import time

sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "harness"))
import config
import topo
import measure

SD = config.SERVERS_DIR
P = config.FT_PYTHON
LOADGEN = os.path.join(config.CLIENTS_DIR, "loadgen")
MANY = config.SERVER_CPU_SPEC
HUBS = config.HUBS
SRV_CPUS = [int(c) for c in MANY.split(",")]

PAYLOAD = 1024                       # 1 KiB: enough bytes for the work to register
WORKS = [0, 1, 2, 4, 8, 16, 32, 64]  # FNV passes; 0 == echo (the lowest curve point)
HANDLERS = ["py", "cython"]
LADDER = [1024, 2048, 4096]          # short saturating ladder (work>0 saturates earlier)
REPS = 2


def factory(handler, work, port, token):
    def f():
        argv = [P, os.path.join(SD, "srv_runloom_work.py"), "--host", config.SRV_IP,
                "--port", str(port), "--hubs", str(HUBS), "--token", token,
                "--handler", handler, "--work", str(work)]
        cmd = topo.ns_cmd(config.SRV_NS, argv, cpus=MANY, extra_env={},
                          gil_off=True, raise_fd=True)
        srv = measure.Server(cmd, token, "work_%s_%d" % (handler, work))
        srv.start(timeout=40)
        time.sleep(0.5)
        return srv
    return f


def run_point(handler, work, port):
    token = "RLWORK_%s_%d_%d" % (handler, work, port)
    label = "%s work=%d" % (handler, work)
    print("\n== %s (payload=%dB) ==" % (label, PAYLOAD), flush=True)
    out = measure.ladder(
        factory(handler, work, port, token), LOADGEN,
        "%s:%d" % (config.SRV_IP, port), PAYLOAD, LADDER,
        REPS, config.RAMP_S, config.MEASURE_S, config.CLIENT_CORES,
        len(LADDER), server_cpus=SRV_CPUS)        # patience = full ladder: take the max
    pk = out["peak"]
    print("  -> %-16s peak=%.0f rps @ conns=%s  server_cpu=%.0f%%  bottleneck=%s"
          % (label, pk["rps_median"], pk.get("conns"),
             (pk.get("server_cpu_util") or 0) * 100, out["bottleneck_at_peak"]),
          flush=True)
    return out


def main():
    topo.setup()
    results = {h: {} for h in HANDLERS}
    port = 9400
    try:
        for handler in HANDLERS:
            for work in WORKS:
                port += 1
                try:
                    results[handler][str(work)] = run_point(handler, work, port)
                except Exception as e:
                    print("  !! %s work=%d FAILED %r" % (handler, work, e), flush=True)
                    results[handler][str(work)] = {"error": repr(e)}
                subprocess.run(["sudo", "-n", "pkill", "-9", "-f",
                                "RLWORK_%s_%d_%d" % (handler, work, port)],
                               capture_output=True)
                time.sleep(0.4)
    finally:
        topo.teardown()

    meta = {"payload": PAYLOAD, "works": WORKS, "ladder": LADDER, "reps": REPS,
            "hubs": HUBS, "server_cpus": MANY}
    out = os.path.join(config.RESULTS_DIR, "work_curve.json")
    with open(out, "w") as f:
        json.dump({"meta": meta, "results": results}, f, indent=2)

    # headline: the curve + the py-vs-cython speedup at each work level
    print("\n=== handler work curve (peak rps, 1 KiB payload) ===")
    print("  %-6s %12s %12s %9s   %s" % ("work", "py", "cython", "cy/py", "bottleneck(py|cy)"))

    def pk(h, w):
        r = results[h].get(str(w), {})
        p = r.get("peak", {})
        return p.get("rps_median", 0.0), r.get("bottleneck_at_peak", "?")

    for w in WORKS:
        py, bpy = pk("py", w)
        cy, bcy = pk("cython", w)
        spd = (cy / py) if py else 0.0
        tag = "  (echo)" if w == 0 else ""
        print("  %-6s %12.0f %12.0f %8.2fx   %s|%s%s"
              % (w, py, cy, spd, bpy[:6], bcy[:6], tag))
    print("\nwrote", out)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Cross-runtime handler work curve: the same FNV --work knob across EVERY
runtime, reported as RAW peak throughput (the core count each ran on is recorded
alongside, not divided out).

The runloom-only curve (work_sweep.py -> work_curve.json) isolates "what does
compiling the handler buy" within one runtime. This puts that in context: the
identical FNV-1a byte hash runs in each runtime's natural handler language --

  interpreted Python : runloom_py (M:N), asyncio, uvloop, gevent  (1 core each
                       for the event loops, like the echo benchmark)
  compiled / native  : runloom_cython (M:N), go (GOMAXPROCS)

The prediction is two bands: the interpreted runtimes cluster together and the
compiled ones cluster together -- i.e. for CPU-bound handler work the dominant
variable is the handler LANGUAGE, not the runtime. runloom's own advantage is
that it gets the compiled band (Cython) while keeping M:N parallelism across all
cores; a single asyncio process serialises the same work onto one core. Compare
within a matched core count (runloom-cython vs Go run on the same cores). Nothing
cherry-picked: same algorithm, every runtime, raw throughput with cores shown.

Separate artifact (results/work_xrt.json) so the committed 8-point isolation
curve is untouched. Short ladder; work>0 is CPU-bound and saturates at low conns.
"""
import json
import os
import subprocess
import sys
import time

sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "harness"))
import config
import topo
import measure

SD = config.SERVERS_DIR
FT = config.FT_PYTHON
GIL = config.GIL_PYTHON
LOADGEN = os.path.join(config.CLIENTS_DIR, "loadgen")
MANY = config.SERVER_CPU_SPEC
ONE = str(config.SERVER_CPUS[0])
HUBS = config.HUBS
GO = config.GO_SERVER_CORES

PAYLOAD = 1024
WORKS = [0, 1, 4, 16, 64]          # echo -> heavy; 5 points span the dynamic range
LADDER = [1024, 4096]              # work>0 saturates low; 4096 covers echo on 44 cores
REPS = 2


def s(script):
    return os.path.join(SD, script)


def build_runtimes():
    def rl(handler):
        def mk(port, token, w):
            return [FT, s("srv_runloom_work.py"), "--host", config.SRV_IP,
                    "--port", str(port), "--hubs", str(HUBS), "--token", token,
                    "--handler", handler, "--work", str(w)]
        return mk

    def aio(loop):
        def mk(port, token, w):
            return [GIL, s("asyncio_epoll_py_proto.py"), "--host", config.SRV_IP,
                    "--port", str(port), "--loop", loop, "--work", str(w),
                    "--token", token]
        return mk

    def gev(port, token, w):
        return [GIL, s("gevent_libev_py_stream.py"), "--host", config.SRV_IP,
                "--port", str(port), "--work", str(w), "--token", token]

    def gomk(port, token, w):
        return [s("go_netpoll_native_net"), "-host", config.SRV_IP, "-port", str(port),
                "-gomaxprocs", str(GO), "-work", str(w), "-token", token]

    return [
        # The cdef c_entry (tstate-free) handler matched this Cython handler to
        # within noise, so the cross-runtime comparison shows just ONE compiled
        # runloom line -- the relatable "compile your hot handler in Cython" path.
        # (The cdef-vs-cython tstate-bypass detail lives in its own report
        # section, suite/servers/handler_cdef.pyx, not here.)
        dict(name="runloom_cython", label="Runloom (M:N) — Cython handler (compiled)",
             kind="compiled", cores=HUBS, cpus=MANY, gil_off=True, env={}, make=rl("cython")),
        dict(name="go", label="Go net (GOMAXPROCS=%d)" % GO,
             kind="compiled", cores=GO, cpus=MANY, gil_off=True, env={}, make=gomk),
        dict(name="runloom_py", label="Runloom (M:N) — Python handler",
             kind="interpreted", cores=HUBS, cpus=MANY, gil_off=True, env={}, make=rl("py")),
        dict(name="asyncio", label="asyncio Protocol (1 core)",
             kind="interpreted", cores=1, cpus=ONE, gil_off=False, env={}, make=aio("asyncio")),
        dict(name="uvloop", label="uvloop (1 core)",
             kind="interpreted", cores=1, cpus=ONE, gil_off=False, env={}, make=aio("uvloop")),
        dict(name="gevent", label="gevent StreamServer (1 core)",
             kind="interpreted", cores=1, cpus=ONE, gil_off=False, env={}, make=gev),
    ]


def factory(rt, port, token, w):
    def f():
        argv = rt["make"](port, token, w)
        cmd = topo.ns_cmd(config.SRV_NS, argv, cpus=rt["cpus"], extra_env=rt["env"],
                          gil_off=rt["gil_off"], raise_fd=True)
        srv = measure.Server(cmd, token, "%s_w%d" % (rt["name"], w))
        srv.start(timeout=40)
        time.sleep(0.5)
        return srv
    return f


def run_point(rt, w, port):
    token = "RLXRT_%s_%d_%d" % (rt["name"], w, port)
    srv_cpus = [int(c) for c in rt["cpus"].split(",")]
    print("\n== %s work=%d (cores=%d) ==" % (rt["name"], w, rt["cores"]), flush=True)
    out = measure.ladder(
        factory(rt, port, token, w), LOADGEN, "%s:%d" % (config.SRV_IP, port),
        PAYLOAD, LADDER, REPS, config.RAMP_S, config.MEASURE_S,
        config.CLIENT_CORES, len(LADDER), server_cpus=srv_cpus)
    pk = out["peak"]
    rps = pk["rps_median"]
    print("  -> %-16s peak=%.0f rps  cores=%d  cpu=%.0f%%  bottleneck=%s"
          % (rt["name"], rps, rt["cores"], (pk.get("server_cpu_util") or 0) * 100,
             out["bottleneck_at_peak"]), flush=True)
    out["_cores"] = rt["cores"]
    return out


def main():
    import argparse
    ap = argparse.ArgumentParser()
    ap.add_argument("--only", default="", help="comma list of runtime names to (re)run; merges into work_xrt.json")
    a = ap.parse_args()
    only = set(filter(None, a.only.split(",")))
    topo.setup()
    runtimes = [rt for rt in build_runtimes() if not only or rt["name"] in only]
    results = {rt["name"]: {} for rt in runtimes}
    port = 9600
    try:
        for rt in runtimes:
            for w in WORKS:
                port += 1
                try:
                    results[rt["name"]][str(w)] = run_point(rt, w, port)
                except Exception as e:
                    print("  !! %s work=%d FAILED %r" % (rt["name"], w, e), flush=True)
                    results[rt["name"]][str(w)] = {"error": repr(e)}
                subprocess.run(["sudo", "-n", "pkill", "-9", "-f",
                                "RLXRT_%s_%d_%d" % (rt["name"], w, port)],
                               capture_output=True)
                time.sleep(0.4)
    finally:
        topo.teardown()

    rt_meta = {rt["name"]: {"cores": rt["cores"], "label": rt["label"], "kind": rt["kind"]}
               for rt in runtimes}
    out = os.path.join(config.RESULTS_DIR, "work_xrt.json")
    # merge-on-write: if --only re-ran a subset, keep the other runtimes' data
    merged_results, merged_rt = dict(results), dict(rt_meta)
    if os.path.exists(out):
        old = json.load(open(out))
        old_res = dict(old.get("results", {})); old_res.update(results); merged_results = old_res
        old_rt = dict(old.get("meta", {}).get("runtimes", {})); old_rt.update(rt_meta); merged_rt = old_rt
    meta = {"payload": PAYLOAD, "works": WORKS, "ladder": LADDER, "reps": REPS,
            "runtimes": merged_rt}
    with open(out, "w") as f:
        json.dump({"meta": meta, "results": merged_results}, f, indent=2)

    print("\n=== cross-runtime work curve (PER-CORE rps, 1 KiB payload) ===")
    head = "  %-8s" % "work" + "".join("%16s" % rt["name"] for rt in runtimes)
    print(head)
    for w in WORKS:
        row = "  %-8d" % w
        for rt in runtimes:
            r = results[rt["name"]].get(str(w), {})
            rps = r.get("peak", {}).get("rps_median")
            row += "%16s" % (("%.0f" % (rps / rt["cores"])) if rps else "-")
        print(row)
    print("\nwrote", out)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Focused io_uring-vs-epoll comparison, to settle whether the loop backend's
"+20% over epoll" reproduces and whether it transfers to a real handler.

Two tests, each epoll vs RUNLOOM_IOURING_LOOP=1, same server otherwise:

  Test 1  all-C 8-byte echo (serve handler=None -> runloom_io_c_echo, a
          tstate-free c_entry fiber). This is the ORIGINAL +20% condition:
          tiny payload (syscall-count-bound, where batching wins) + no Python
          state (cheap always-park). Payload = 8 bytes.

  Test 2  Cython C handler at 1 KiB (runloom_iouring_cython_tcpcon.py). A REAL handler on
          a Python-tstate fiber. The capi now routes through the Stage-2 proactor
          (loop_recv) under the loop backend, so this asks: does the batching win
          survive the per-park tstate cost at a realistic payload? Payload = 1 KiB.

Writes results/iouring_test.json.

Results (2026-06-19, Xeon E5-2696 v3, free-threaded 3.13t; full record in
../IOURING_TSTATE_FINDINGS.md):
  Test 1  8-byte all-C echo : epoll 654k vs io_uring 659k (both client-bound);
          server ceiling 1.14M -> 1.21M = +6%. Modest: the all-C epoll path is
          already near-optimal for a tiny tstate-free echo.
  Test 2  1 KiB Cython      : epoll 455k (server-bound) vs io_uring 639k
          (client-bound); server ceiling 533k -> 1.16M = +2.17x, server CPU
          85% -> 55%. The proactor batching beats epoll decisively for a real
          handler -- and runloom_cython on io_uring becomes the FASTEST runloom
          config in the suite. "io_uring loses on loopback" was an artifact of
          driving it through the readiness path instead of loop_recv.
"""
import json
import os
import sys
import time

sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "harness"))
import config
import topo
import measure

SD = config.SERVERS_DIR
P = config.FT_PYTHON
LOADGEN = os.path.join(config.CLIENTS_DIR, "loadgen")
MANY = config.SERVER_CPU_SPEC
HUBS = config.HUBS
SRV_CPUS = [int(c) for c in MANY.split(",")]


def factory(script, extra, env, port, token):
    def f():
        argv = [P, os.path.join(SD, script), "--host", config.SRV_IP,
                "--port", str(port), "--hubs", str(HUBS), "--token", token] + extra
        cmd = topo.ns_cmd(config.SRV_NS, argv, cpus=MANY, extra_env=env,
                          gil_off=True, raise_fd=True)
        srv = measure.Server(cmd, token, script)
        srv.start(timeout=40)
        time.sleep(0.5)
        return srv
    return f


def run_one(name, script, extra, env, payload, port):
    token = "RLIOU_%s_%d" % (name, port)
    print("\n== %s (payload=%dB, env=%s) ==" % (name, payload, env or "epoll"), flush=True)
    out = measure.ladder(
        factory(script, extra, env, port, token), LOADGEN,
        "%s:%d" % (config.SRV_IP, port), payload, config.CONN_LADDER,
        3, config.RAMP_S, config.MEASURE_S, config.CLIENT_CORES,
        config.PLATEAU_PATIENCE, server_cpus=SRV_CPUS)
    pk = out["peak"]
    out["_payload"] = payload
    print("  -> %-24s peak=%.0f rps @ conns=%s  bottleneck=%s  server-ceiling=%s"
          % (name, pk["rps_median"], pk.get("conns"), out["bottleneck_at_peak"],
             ("%.0f" % out["server_ceiling_est"]) if out.get("server_ceiling_est") else "-"),
          flush=True)
    return out


def main():
    topo.setup()
    results = {}
    port = 9300
    try:
        IOU = {"RUNLOOM_IOURING_LOOP": "1"}
        cases = [
            # io_uring vs epoll (8-byte all-C echo + 1 KiB Cython handler)
            ("cecho_epoll", "srv_runloom_cecho.py", [], {}, 8),
            ("cecho_iouring", "srv_runloom_cecho.py", [], IOU, 8),
            ("cython_epoll", "runloom_iouring_cython_tcpcon.py", ["--optimize", "none"], {}, 1024),
            ("cython_iouring_proactor", "runloom_iouring_cython_tcpcon.py", ["--optimize", "none"], IOU, 1024),
            # tstate bypass: a Python-fiber Cython handler vs a tstate-free cdef
            # c_entry handler, at 8 bytes (op-bound -- where per-park tstate cost
            # should matter) and 1 KiB (I/O-bound -- where it should wash out).
            ("cython_iouring_8b", "runloom_iouring_cython_tcpcon.py", ["--optimize", "none"], IOU, 8),
            ("cdef_iouring_8b", "runloom_iouring_cdef_tcpcon.py", [], IOU, 8),
            ("cdef_iouring_1k", "runloom_iouring_cdef_tcpcon.py", [], IOU, 1024),
        ]
        for name, script, extra, env, payload in cases:
            port += 1
            try:
                results[name] = run_one(name, script, extra, env, payload, port)
            except Exception as e:
                print("  !! %s FAILED %r" % (name, e), flush=True)
                results[name] = {"error": repr(e)}
            import subprocess
            subprocess.run(["sudo", "-n", "pkill", "-9", "-f", "RLIOU_%s_%d" % (name, port)],
                           capture_output=True)
            time.sleep(0.5)
    finally:
        topo.teardown()
    out = os.path.join(config.RESULTS_DIR, "iouring_test.json")
    with open(out, "w") as f:
        json.dump(results, f, indent=2)
    # headline
    print("\n=== io_uring vs epoll summary ===")
    def peak(n):
        r = results.get(n, {})
        return r.get("peak", {}).get("rps_median", 0), r.get("server_ceiling_est") or 0
    for a, b, label in [("cecho_epoll", "cecho_iouring", "all-C 8B echo"),
                        ("cython_epoll", "cython_iouring_proactor", "Cython 1KB handler")]:
        pa, ca = peak(a); pb, cb = peak(b)
        if pa and pb:
            print("  %-20s epoll peak=%.0f (ceil %.0f) | io_uring peak=%.0f (ceil %.0f) | "
                  "io_uring/epoll = %.2fx peak, %.2fx ceiling"
                  % (label, pa, ca, pb, cb, pb / pa, (cb / ca) if ca else 0))
    print("\nwrote", out)


if __name__ == "__main__":
    main()

"""Active (batch) spawn measurement -- the fiber_n fleet-launch path, measured
IN-SUITE on this box so the report cites a committed number instead of prose.

For each N it times the end-to-end create+run+destroy of N no-op fibers two ways:
  - naked : for _ in range(N): runloom.fiber(noop)   (one at a time)
  - batch : runloom_c.fiber_n(noop, N)               (one bulk C call)
under default config and under runloom.optimize("throughput") (warm-stack arena +
bulk + FRESH + parallel create). An n=0 empty-run baseline is subtracted to remove
scheduler startup/teardown. rate = N / (wall - baseline), median of R reps.

HONEST FRAMING: there is NO Go batch-spawn equivalent (Go has no bulk-spawn API),
so this is a runloom *capability* measurement, NOT a Go comparison. The only
like-for-like spawn comparison vs Go is naked single-spawn (spawn_curve.json),
which Go wins. Run pinned to a single NUMA node (the cross-NUMA g-arena traffic
otherwise dominates and is a pinning artifact, not a spawn cost).
"""
import argparse
import json
import os
import statistics
import time

import runloom
import runloom_c


def noop():
    pass


def _empty():
    pass


def _baseline(hubs, reps=3):
    ts = []
    for _ in range(reps):
        t0 = time.perf_counter()
        runloom.run(hubs, _empty)
        ts.append(time.perf_counter() - t0)
    return min(ts)   # min = least startup noise


def measure(mode, n, hubs, base, reps):
    rates, walls = [], []
    for _ in range(reps):
        if mode == "naked":
            def root():
                for _ in range(n):
                    runloom.fiber(noop)
        else:
            def root():
                runloom_c.fiber_n(noop, n)
        t0 = time.perf_counter()
        runloom.run(hubs, root)
        wall = time.perf_counter() - t0
        walls.append(wall)
        rates.append(n / max(wall - base, 1e-9))
    return {"rate_per_s": statistics.median(rates),
            "wall_median_s": statistics.median(walls), "reps": reps}


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--hubs", type=int, default=8)
    ap.add_argument("--reps", type=int, default=3)
    ap.add_argument("--ns", default="100000,300000,1000000")
    ap.add_argument("--optimize", default="none", choices=["none", "throughput"])
    ap.add_argument("--out", default="")
    args = ap.parse_args()
    ns = [int(x) for x in args.ns.split(",")]

    eff = runloom.optimize("throughput") if args.optimize == "throughput" else None
    base = _baseline(args.hubs)

    rates = {}
    for n in ns:
        rates[n] = {"naked": measure("naked", n, args.hubs, base, args.reps),
                    "batch": measure("batch", n, args.hubs, base, args.reps)}
        nk = rates[n]["naked"]["rate_per_s"]
        bt = rates[n]["batch"]["rate_per_s"]
        print("N=%-8d naked=%10.0f/s  batch=%10.0f/s  (batch/naked %.2fx)"
              % (n, nk, bt, bt / max(nk, 1)))

    out = {"meta": {"hubs": args.hubs, "reps": args.reps, "baseline_s": base,
                    "optimize": args.optimize, "optimize_eff": eff,
                    "cpu_affinity": sorted(os.sched_getaffinity(0)),
                    "ncores_pinned": len(os.sched_getaffinity(0)),
                    "note": "rate = N/(wall - empty-run baseline), median of reps; "
                            "NO Go batch equivalent exists -- capability, not comparison"},
           "rates": rates}
    if args.out:
        with open(args.out, "w") as f:
            json.dump(out, f, indent=1)
        print("wrote", args.out)


if __name__ == "__main__":
    main()

"""The [runloom] column of the speed benchmark. All metrics run under the REAL
M:N scheduler runloom.run(hubs) -- never run(1), which is the different M:1
cooperative scheduler (decision #5).

  --metric spawn      : spawn N no-op fibers, drain  -> seconds (orchestrator
                        subtracts an n=0 startup baseline)
  --metric ctxswitch  : 2-fiber unbuffered-Chan ping-pong, N round-trips -> seconds
  --metric rtt        : 1 fiber, N sequential round-trips to a Go echo server
  --metric http       : H fibers, keepalive HTTP/1.1 GET vs a Go httpd, windowed req/s
"""
import argparse
import json
import os
import socket
import time

import runloom
import runloom_c
import runloom.sync as rs

HUBS_DEFAULT = int((os.cpu_count() or 1) * 0.7)


def noop():
    pass


def m_spawn(n, hubs, stack_size=0, warm=0):
    # Naked single-spawn rate.  Default (stack_size=0) uses runloom.fiber_fast --
    # a thin Python spawn with no per-spawn work, the apples-to-apples vs Go's
    # `go f()`.  The DEFAULT runloom.fiber adds the grow-down auto-sizer (small
    # right-sized stacks, an RSS feature Go lacks); its learned size now spawns
    # down the DEFERRED stack-alloc path, so the default is ~1.9M/s warm
    # (small-stacks AND fast) -- not the old ~7x-slower eager-alloc number.
    # optimize("throughput")/("memory") swaps runloom.fiber between fiber_fast and
    # grow-down.  stack_size>0 pins each fiber's C stack (decomposition variant).
    # warm>0: run `warm` extra in-process passes first, report the BEST timed pass,
    # so the one-time runloom.run() scheduler boot is excluded -- the same basis Go
    # is measured on (its runtime is already up at main()).
    if stack_size > 0:
        def root():
            for _ in range(n):
                runloom.fiber(noop, stack_size=stack_size)
    else:
        def root():
            f = runloom.fiber_fast
            for _ in range(n):
                f(noop)
    best = None
    for _ in range(warm + 1):
        t0 = time.perf_counter()
        runloom.run(hubs, root)
        dt = time.perf_counter() - t0
        best = dt if best is None else min(best, dt)
    return {"seconds": best, "n": n, "cores": hubs,
            "stack_size": stack_size, "warm": warm}


def _make_distinct_worker(K, yobj):
    # The real contention is SHARED CLOSURE CELLS, not the code object (proven in
    # SCHEDULER_SCALING_FINDINGS.md "CORRECTION" + suite/speed/hot_diag.py: one
    # SHARED code object scales fine; one SHARED closure's cells do not).  This
    # worker reads `sy`/`K` as GLOBALS in its own dict, so it has NO shared cells
    # -- which is why it scales.  (The `shared` mode uses a nested closure, so all
    # fibers share its cells == the wall.)  User-facing fix: @runloom.hot.
    g = {"sy": yobj, "K": K, "__builtins__": __builtins__}
    exec(compile("def w():\n for _ in range(K):\n  sy()", "<w>", "exec"), g)
    return g["w"]


def m_ctxswitch(n, hubs, distinct=False):
    # "Context switch under load": G concurrent fibers each yield K times, so the
    # hubs stay full of ready work and switches are same-hub re-dispatch -- the
    # realistic cost a loaded server pays, NOT a 2-fiber ping-pong (which forces
    # a cross-hub wake of a freshly-parked idle hub every op: ~30us, pathological
    # and unrepresentative). G*K == n total switches.
    #
    # The yield object is runloom_c.sched_yield, an IMMORTAL process-lifetime
    # singleton (module_init.c.inc), so the per-yield refcount-contention layer
    # is already gone for both modes.  --distinct ALSO de-shares the code object
    # (above), so the only residual cost is the per-yield Python frame itself,
    # which is per-hub-parallel and therefore scales.  shared (default) == the
    # naive "one handler fn for every fiber" server; distinct == the fixed path.
    G = max(2, hubs * 16)
    K = max(1, n // G)
    sched_yield = runloom_c.sched_yield

    if distinct:
        workers = [_make_distinct_worker(K, sched_yield) for _ in range(G)]

        def root():
            for w in workers:
                runloom.fiber(w)
    else:
        def worker():
            for _ in range(K):
                sched_yield()

        def root():
            for _ in range(G):
                runloom.fiber(worker)
    t0 = time.perf_counter()
    runloom.run(hubs, root)
    return {"seconds": time.perf_counter() - t0, "n": n, "cores": hubs,
            "switches": G * K, "fibers": G, "yields_each": K,
            "mode": "distinct" if distinct else "shared"}


def _recvn(sock, n):
    got = 0
    while got < n:
        b = sock.recv(n - got)
        if not b:
            return False
        got += len(b)
    return True


def m_rtt(host, port, n, payload):
    out = {}

    def root():
        s = rs.tcp_connect(host, port)
        try:
            s.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
        except Exception:
            pass
        msg = b"\xab" * payload
        for _ in range(1000):           # warmup
            s.sendall(msg)
            _recvn(s, payload)
        t0 = time.perf_counter()
        for _ in range(n):
            s.sendall(msg)
            _recvn(s, payload)
        out["seconds"] = time.perf_counter() - t0
        s.close()
    runloom.run(2, root)
    return {"ns_per_rtt": out["seconds"] * 1e9 / n, "n": n, "payload": payload,
            "cores": 1}


HTTP_REQ = b"GET / HTTP/1.1\r\nHost: b\r\nConnection: keep-alive\r\n\r\n"


def _http_once(sock):
    sock.sendall(HTTP_REQ)
    data = b""
    while b"\r\n\r\n" not in data:
        chunk = sock.recv(4096)
        if not chunk:
            return False
        data += chunk
    header, _, rest = data.partition(b"\r\n\r\n")
    cl = 0
    for line in header.split(b"\r\n"):
        if line[:15].lower() == b"content-length:":
            cl = int(line.split(b":", 1)[1])
    body = rest
    while len(body) < cl:
        chunk = sock.recv(4096)
        if not chunk:
            return False
        body += chunk
    return True


def m_http(host, port, hubs, conns, ramp, measure):
    counters = bytearray(8 * conns)  # sharded counters (race-free, 1 writer each)
    import struct
    state = {"measuring": False, "stop": False, "live": 0}

    def worker(idx):
        try:
            s = rs.tcp_connect(host, port)
        except Exception:
            return
        try:
            s.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
        except Exception:
            pass
        c = 0
        while not state["stop"]:
            if not _http_once(s):
                break
            if state["measuring"]:
                c += 1
        struct.pack_into("<q", counters, idx * 8, c)
        try:
            s.close()
        except Exception:
            pass

    def timer():
        runloom.sleep(ramp)
        state["measuring"] = True
        t0 = time.perf_counter()
        runloom.sleep(measure)
        state["measuring"] = False
        state["stop"] = True
        state["elapsed"] = time.perf_counter() - t0

    def root():
        for i in range(conns):
            runloom.fiber(worker, i)
        runloom.fiber(timer)

    runloom.run(hubs, root)
    total = sum(struct.unpack_from("<q", counters, i * 8)[0] for i in range(conns))
    elapsed = state.get("elapsed", measure)
    return {"rps": total / elapsed, "reqs": total, "measure_s": elapsed,
            "conns": conns, "cores": hubs}


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--metric", required=True,
                    choices=["spawn", "ctxswitch", "rtt", "http"])
    ap.add_argument("--n", type=int, default=1_000_000)
    ap.add_argument("--hubs", type=int, default=HUBS_DEFAULT)
    ap.add_argument("--host", default="10.99.0.1")
    ap.add_argument("--port", type=int, default=9100)
    ap.add_argument("--payload", type=int, default=64)
    ap.add_argument("--conns", type=int, default=64)
    ap.add_argument("--ramp", type=float, default=1.0)
    ap.add_argument("--measure", type=float, default=3.0)
    ap.add_argument("--distinct", action="store_true",
                    help="ctxswitch: give each fiber its own code object "
                         "(de-shares co_code_adaptive; the fixed Python path)")
    ap.add_argument("--stack-size", type=int, default=0,
                    help="spawn: pin each fiber's C stack size in bytes (0 = default)")
    ap.add_argument("--warm", type=int, default=0,
                    help="spawn: extra in-process passes before timing (boot-excluded warm rate)")
    args = ap.parse_args()

    if args.metric == "spawn":
        res = m_spawn(args.n, args.hubs, stack_size=args.stack_size, warm=args.warm)
    elif args.metric == "ctxswitch":
        res = m_ctxswitch(args.n, args.hubs, distinct=args.distinct)
    elif args.metric == "rtt":
        res = m_rtt(args.host, args.port, args.n, args.payload)
    else:
        res = m_http(args.host, args.port, args.hubs, args.conns, args.ramp, args.measure)
    res.update({"runtime": "runloom", "metric": args.metric})
    print(json.dumps(res))


if __name__ == "__main__":
    main()

"""The [asyncio] and [uvloop] columns of the speed benchmark (single-threaded,
GIL build -- decision #4). --loop selects the event loop policy.

  spawn      : gather N no-op coroutines  -> seconds (orchestrator baselines n=0)
  ctxswitch  : G tasks each `await sleep(0)` K times (loaded-yield) -> seconds
  rtt        : 1 stream, N sequential round-trips to a Go echo server
  http       : H tasks, keepalive HTTP/1.1 GET vs a Go httpd, windowed req/s
"""
import argparse
import asyncio
import json
import socket
import time

HTTP_REQ = b"GET / HTTP/1.1\r\nHost: b\r\nConnection: keep-alive\r\n\r\n"


async def m_spawn(n):
    async def noop():
        return
    t0 = time.perf_counter()
    if n:
        await asyncio.gather(*[noop() for _ in range(n)])
    return {"seconds": time.perf_counter() - t0, "n": n, "cores": 1}


async def m_ctxswitch(n):
    G = 64
    K = max(1, n // G)

    async def worker():
        for _ in range(K):
            await asyncio.sleep(0)
    t0 = time.perf_counter()
    await asyncio.gather(*[worker() for _ in range(G)])
    return {"seconds": time.perf_counter() - t0, "switches": G * K,
            "fibers": G, "yields_each": K, "cores": 1}


async def m_rtt(host, port, n, payload):
    reader, writer = await asyncio.open_connection(host, port)
    sock = writer.get_extra_info("socket")
    if sock:
        sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
    msg = b"\xab" * payload
    for _ in range(1000):
        writer.write(msg)
        await writer.drain()
        await reader.readexactly(payload)
    t0 = time.perf_counter()
    for _ in range(n):
        writer.write(msg)
        await writer.drain()
        await reader.readexactly(payload)
    dt = time.perf_counter() - t0
    writer.close()
    try:
        await writer.wait_closed()
    except Exception:
        pass
    return {"ns_per_rtt": dt * 1e9 / n, "n": n, "payload": payload, "cores": 1}


def _content_length(header):
    for line in header.split(b"\r\n"):
        if line[:15].lower() == b"content-length:":
            return int(line.split(b":", 1)[1])
    return 0


async def m_http(host, port, conns, ramp, measure):
    state = {"measuring": False, "stop": False}
    counters = [0] * conns

    async def worker(idx):
        try:
            reader, writer = await asyncio.open_connection(host, port)
        except Exception:
            return
        sock = writer.get_extra_info("socket")
        if sock:
            sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
        try:
            while not state["stop"]:
                writer.write(HTTP_REQ)
                await writer.drain()
                header = await reader.readuntil(b"\r\n\r\n")
                cl = _content_length(header)
                if cl:
                    await reader.readexactly(cl)
                if state["measuring"]:
                    counters[idx] += 1
        except Exception:
            pass
        finally:
            writer.close()

    tasks = [asyncio.create_task(worker(i)) for i in range(conns)]
    await asyncio.sleep(ramp)
    state["measuring"] = True
    t0 = time.perf_counter()
    await asyncio.sleep(measure)
    state["measuring"] = False
    state["stop"] = True
    elapsed = time.perf_counter() - t0
    for t in tasks:
        t.cancel()
    await asyncio.gather(*tasks, return_exceptions=True)
    total = sum(counters)
    return {"rps": total / elapsed, "reqs": total, "measure_s": elapsed,
            "conns": conns, "cores": 1}


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--metric", required=True,
                    choices=["spawn", "ctxswitch", "rtt", "http"])
    ap.add_argument("--loop", default="asyncio", choices=["asyncio", "uvloop"])
    ap.add_argument("--n", type=int, default=1_000_000)
    ap.add_argument("--host", default="10.99.0.1")
    ap.add_argument("--port", type=int, default=9100)
    ap.add_argument("--payload", type=int, default=64)
    ap.add_argument("--conns", type=int, default=64)
    ap.add_argument("--ramp", type=float, default=1.0)
    ap.add_argument("--measure", type=float, default=3.0)
    args = ap.parse_args()
    if args.loop == "uvloop":
        import uvloop
        uvloop.install()
    rt = args.loop

    if args.metric == "spawn":
        res = asyncio.run(m_spawn(args.n))
    elif args.metric == "ctxswitch":
        res = asyncio.run(m_ctxswitch(args.n))
    elif args.metric == "rtt":
        res = asyncio.run(m_rtt(args.host, args.port, args.n, args.payload))
    else:
        res = asyncio.run(m_http(args.host, args.port, args.conns, args.ramp, args.measure))
    res.update({"runtime": rt, "metric": args.metric})
    print(json.dumps(res))


if __name__ == "__main__":
    main()

"""The [greenlet] column of the speed benchmark (GIL build, single-threaded).

Raw `greenlet` has no scheduler or I/O loop, so:
  * spawn / ctxswitch use raw greenlet (the purest cooperative switch);
  * rtt / http use gevent (greenlet + libev), the greenlet ecosystem's I/O layer.
The report labels these "greenlet (gevent for I/O)".

  spawn      : create N greenlets, switch into each  -> seconds
  ctxswitch  : round-robin hub over G greenlets, K rounds (loaded-yield) -> seconds
  rtt        : gevent socket, N sequential round-trips to a Go echo server
  http       : H greenlets, keepalive HTTP/1.1 GET vs a Go httpd, windowed req/s
"""
import argparse
import json
import time

import greenlet

HTTP_REQ = b"GET / HTTP/1.1\r\nHost: b\r\nConnection: keep-alive\r\n\r\n"


def m_spawn(n):
    def noop():
        pass
    t0 = time.perf_counter()
    gs = [greenlet.greenlet(noop) for _ in range(n)]
    for g in gs:
        g.switch()
    return {"seconds": time.perf_counter() - t0, "n": n, "cores": 1}


def m_ctxswitch(n):
    G = 64
    K = max(1, n // (2 * G))
    hub = greenlet.getcurrent()

    def worker():
        for _ in range(K):
            hub.switch()
    workers = [greenlet.greenlet(worker) for _ in range(G)]
    t0 = time.perf_counter()
    for _ in range(K):
        for w in workers:
            w.switch()
    dt = time.perf_counter() - t0
    return {"seconds": dt, "switches": 2 * G * K, "fibers": G, "rounds": K,
            "cores": 1}


def m_rtt(host, port, n, payload):
    import socket
    from gevent import socket as gsocket
    s = gsocket.create_connection((host, port))
    s.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
    msg = b"\xab" * payload

    def recvn(k):
        got = 0
        while got < k:
            b = s.recv(k - got)
            if not b:
                return False
            got += len(b)
        return True

    for _ in range(1000):
        s.sendall(msg)
        recvn(payload)
    t0 = time.perf_counter()
    for _ in range(n):
        s.sendall(msg)
        recvn(payload)
    dt = time.perf_counter() - t0
    s.close()
    return {"ns_per_rtt": dt * 1e9 / n, "n": n, "payload": payload, "cores": 1}


def _content_length(header):
    for line in header.split(b"\r\n"):
        if line[:15].lower() == b"content-length:":
            return int(line.split(b":", 1)[1])
    return 0


def m_http(host, port, conns, ramp, measure):
    import socket
    import gevent
    from gevent import socket as gsocket
    state = {"measuring": False, "stop": False}
    counters = [0] * conns

    def worker(idx):
        try:
            s = gsocket.create_connection((host, port))
            s.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
        except Exception:
            return
        buf = b""
        while not state["stop"]:
            s.sendall(HTTP_REQ)
            buf = b""
            while b"\r\n\r\n" not in buf:
                d = s.recv(4096)
                if not d:
                    return
                buf += d
            header, _, rest = buf.partition(b"\r\n\r\n")
            cl = _content_length(header)
            body = rest
            while len(body) < cl:
                d = s.recv(4096)
                if not d:
                    return
                body += d
            if state["measuring"]:
                counters[idx] += 1
        s.close()

    greenlets = [gevent.spawn(worker, i) for i in range(conns)]
    gevent.sleep(ramp)
    state["measuring"] = True
    t0 = time.perf_counter()
    gevent.sleep(measure)
    state["measuring"] = False
    state["stop"] = True
    elapsed = time.perf_counter() - t0
    gevent.killall(greenlets, block=True, timeout=2)
    total = sum(counters)
    return {"rps": total / elapsed, "reqs": total, "measure_s": elapsed,
            "conns": conns, "cores": 1}


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--metric", required=True,
                    choices=["spawn", "ctxswitch", "rtt", "http"])
    ap.add_argument("--n", type=int, default=1_000_000)
    ap.add_argument("--host", default="10.99.0.1")
    ap.add_argument("--port", type=int, default=9100)
    ap.add_argument("--payload", type=int, default=64)
    ap.add_argument("--conns", type=int, default=64)
    ap.add_argument("--ramp", type=float, default=1.0)
    ap.add_argument("--measure", type=float, default=3.0)
    args = ap.parse_args()

    if args.metric == "spawn":
        res = m_spawn(args.n)
    elif args.metric == "ctxswitch":
        res = m_ctxswitch(args.n)
    elif args.metric == "rtt":
        res = m_rtt(args.host, args.port, args.n, args.payload)
    else:
        res = m_http(args.host, args.port, args.conns, args.ramp, args.measure)
    res.update({"runtime": "greenlet", "metric": args.metric})
    print(json.dumps(res))


if __name__ == "__main__":
    main()

// speed_go -- the Go side of the speed benchmark (the [go] column) plus the two
// fixed Go targets the other runtimes are measured against.
//
// Subcommands (-metric):
//   spawn       : spawn N goroutines (each wg.Done), drain -> seconds
//   ctxswitch   : 2-goroutine unbuffered-channel ping-pong, N round-trips -> ns/switch
//   rtt         : TCP client; N sequential round-trips to -addr echo server -> ns/RTT
//   httpd       : HTTP server target for the http-req/s metric (prints LISTENING)
//   httpclient  : concurrent HTTP GET load vs -addr for -measure s -> req/s
package main

import (
	"encoding/json"
	"flag"
	"fmt"
	"io"
	"net"
	"net/http"
	"os"
	"runtime"
	"sync"
	"sync/atomic"
	"time"
)

func emit(m map[string]any) {
	b, _ := json.Marshal(m)
	fmt.Println(string(b))
}

func spawn(n, gomax int) {
	runtime.GOMAXPROCS(gomax)
	var wg sync.WaitGroup
	wg.Add(n)
	t0 := time.Now()
	for i := 0; i < n; i++ {
		go func() { wg.Done() }()
	}
	wg.Wait()
	dt := time.Since(t0).Seconds()
	emit(map[string]any{"runtime": "go", "metric": "spawn", "n": n,
		"cores": gomax, "seconds": dt, "rate_per_s": float64(n) / dt})
}

func ctxswitch(n, gomax int) {
	// Loaded-yield (matches the other runtimes): G goroutines each Gosched K
	// times so the run queues stay full and switches are real re-dispatch, not a
	// 2-party ping-pong that idles all but two threads.
	runtime.GOMAXPROCS(gomax)
	G := gomax * 16
	if G < 2 {
		G = 2
	}
	K := n / G
	if K < 1 {
		K = 1
	}
	var wg sync.WaitGroup
	wg.Add(G)
	t0 := time.Now()
	for i := 0; i < G; i++ {
		go func() {
			for j := 0; j < K; j++ {
				runtime.Gosched()
			}
			wg.Done()
		}()
	}
	wg.Wait()
	dt := time.Since(t0)
	switches := G * K
	emit(map[string]any{"runtime": "go", "metric": "ctxswitch", "n": n,
		"cores": gomax, "switches": switches, "fibers": G,
		"seconds": dt.Seconds(),
		"ns_per_switch": float64(dt.Nanoseconds()) / float64(switches)})
}

func rtt(addr string, n, payload int) {
	c, err := net.Dial("tcp", addr)
	if err != nil {
		emit(map[string]any{"runtime": "go", "metric": "rtt", "error": err.Error()})
		os.Exit(1)
	}
	c.(*net.TCPConn).SetNoDelay(true)
	send := make([]byte, payload)
	recv := make([]byte, payload)
	// warmup
	for i := 0; i < 1000; i++ {
		c.Write(send)
		io.ReadFull(c, recv)
	}
	t0 := time.Now()
	for i := 0; i < n; i++ {
		if _, err := c.Write(send); err != nil {
			break
		}
		if _, err := io.ReadFull(c, recv); err != nil {
			break
		}
	}
	dt := time.Since(t0)
	emit(map[string]any{"runtime": "go", "metric": "rtt", "n": n, "payload": payload,
		"ns_per_rtt": float64(dt.Nanoseconds()) / float64(n)})
}

func httpd(host string, port, gomax int) {
	runtime.GOMAXPROCS(gomax)
	body := []byte("OK\n")
	mux := http.NewServeMux()
	mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "text/plain")
		w.Write(body)
	})
	ln, err := net.Listen("tcp", fmt.Sprintf("%s:%d", host, port))
	if err != nil {
		fmt.Fprintln(os.Stderr, err)
		os.Exit(1)
	}
	fmt.Printf("LISTENING %d\n", ln.Addr().(*net.TCPAddr).Port)
	os.Stdout.Sync()
	srv := &http.Server{Handler: mux}
	srv.Serve(ln)
}

func httpclient(addr string, conns, gomax int, ramp, measure float64) {
	runtime.GOMAXPROCS(gomax)
	url := "http://" + addr + "/"
	tr := &http.Transport{MaxIdleConns: conns * 2, MaxIdleConnsPerHost: conns * 2,
		MaxConnsPerHost: conns * 2, DisableCompression: true}
	client := &http.Client{Transport: tr}
	var measuring atomic.Bool
	var reqs atomic.Int64
	var errs atomic.Int64
	var stop atomic.Bool
	var wg sync.WaitGroup
	for i := 0; i < conns; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for !stop.Load() {
				resp, err := client.Get(url)
				if err != nil {
					errs.Add(1)
					continue
				}
				io.Copy(io.Discard, resp.Body)
				resp.Body.Close()
				if measuring.Load() {
					reqs.Add(1)
				}
			}
		}()
	}
	time.Sleep(time.Duration(ramp * float64(time.Second)))
	measuring.Store(true)
	t0 := time.Now()
	time.Sleep(time.Duration(measure * float64(time.Second)))
	measuring.Store(false)
	elapsed := time.Since(t0).Seconds()
	stop.Store(true)
	go func() { wg.Wait() }()
	time.Sleep(200 * time.Millisecond)
	emit(map[string]any{"runtime": "go", "metric": "http", "conns": conns,
		"cores": gomax, "measure_s": elapsed, "reqs": reqs.Load(),
		"rps": float64(reqs.Load()) / elapsed, "errors": errs.Load()})
}

func main() {
	metric := flag.String("metric", "spawn", "spawn|ctxswitch|rtt|httpd|httpclient")
	n := flag.Int("n", 1000000, "")
	gomax := flag.Int("gomaxprocs", runtime.NumCPU(), "")
	addr := flag.String("addr", "127.0.0.1:9100", "")
	host := flag.String("host", "127.0.0.1", "")
	port := flag.Int("port", 9100, "")
	payload := flag.Int("payload", 64, "")
	conns := flag.Int("conns", 64, "")
	ramp := flag.Float64("ramp", 1.0, "")
	measure := flag.Float64("measure", 3.0, "")
	flag.String("token", "", "")
	flag.Parse()
	switch *metric {
	case "spawn":
		spawn(*n, *gomax)
	case "ctxswitch":
		ctxswitch(*n, *gomax)
	case "rtt":
		rtt(*addr, *n, *payload)
	case "httpd":
		httpd(*host, *port, *gomax)
	case "httpclient":
		httpclient(*addr, *conns, *gomax, *ramp, *measure)
	default:
		fmt.Fprintln(os.Stderr, "unknown metric", *metric)
		os.Exit(2)
	}
}

"""Memory probe for the runloom columns: hold N parked fibers in a given state
and report USED memory (RSS / PSS -- NOT virtual size).

Configs (via flags):
  --handler py            interpreted handler: the fiber's entry runs in the
                          CPython evaluator.  For 'socket' it holds a 64 KiB
                          bytearray (matches runloom_epoll_py_tcpcon) -> heap-faulted.
  --handler c             COMPILED handler: the fiber's entry is native (Cython)
                          code, so its frozen C stack carries NO
                          _PyEval_EvalFrameDefault interpreter activation.  For
                          'socket' it is the stack-buffer Cython handler; for
                          'empty' the bare parker itself is compiled (see below).
  --optimize memory       call runloom.optimize("memory") first

States (--state):
  empty   : N fibers parked on a shared Chan recv (bare fiber: g + stack).
            With --handler c the parker is COMPILED to native code, so a parked
            fiber's resident C stack is ~1 page (no eval-loop activation) vs ~2
            pages interpreted -- the real, ~2x-cheaper cost of a compiled fiber.
            (Before this, the 'c' column silently re-ran the Python parker and
            read identical to --handler py.)
  socket  : N fibers each holding a socketpair end, parked on recv (+ the
            handler's per-conn buffer)

The measurer fiber waits for everyone to settle, snapshots /proc/self RSS+PSS,
prints JSON, and hard-exits (the worker fibers are parked forever).
"""
import argparse
import json
import os
import socket as sk

import runloom
import runloom_c


def rss_bytes():
    with open("/proc/self/status") as f:
        for line in f:
            if line.startswith("VmRSS:"):
                return int(line.split()[1]) * 1024
    return None


def pss_bytes():
    try:
        with open("/proc/self/smaps_rollup") as f:
            for line in f:
                if line.startswith("Pss:"):
                    return int(line.split()[1]) * 1024
    except Exception:
        return None


def compile_empty_parker(shared):
    """Compile the bare 'empty' parker to native (Cython) code.

    The point of the --handler c / empty column: a parked fiber whose frozen C
    stack carries the compiled cyfunction frame instead of a
    _PyEval_EvalFrameDefault interpreter activation.  That single eval frame
    (~448 B on 3.13t) is what tips the live park chain across a second 4 KiB
    page, so dropping it nearly halves a parked fiber's resident stack (~2 pages
    -> ~1 page).  Reuses benchmark/bench/cycompile (the on-the-fly handler
    compiler), so this measures the SAME source as the py parker, just compiled.

    Spawned with no args, so runloom.fiber uses the callable directly (no
    arg-binding lambda wrapper) -- the only frame between runloom_g_entry and
    Chan.recv is the native parker, the cleanest apples-to-apples vs --handler py.
    Raises (never silently falls back to the Python parker) if compilation is
    unavailable -- a silent fallback is the exact bug this column had.
    """
    import sys
    here = os.path.dirname(os.path.abspath(__file__))
    sys.path.insert(0, os.path.join(here, "..", "..", "bench"))   # cycompile
    import cycompile

    def parker():
        shared.recv()                       # park forever; compiled to native C

    # The preamble declares `shared` as a module global so Cython compiles the
    # bare name as a global load (not an undeclared-name error); bind the real
    # channel into the compiled module right after import.
    (cy,) = cycompile.compile_funcs([parker], preamble="shared = None")
    cy.__globals__["shared"] = shared
    return cy


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--state", required=True, choices=["empty", "socket"])
    ap.add_argument("--n", type=int, required=True)
    ap.add_argument("--handler", default="py", choices=["py", "c"])
    ap.add_argument("--optimize", default="none", choices=["none", "memory"])
    ap.add_argument("--hubs", type=int, default=int((os.cpu_count() or 1) * 0.7))
    ap.add_argument("--settle", type=float, default=0.0)
    args = ap.parse_args()

    if args.optimize == "memory":
        runloom.optimize("memory")
    handler_fn = None
    if args.handler == "c" and args.state == "socket":
        import sys
        sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)),
                                        "..", "servers"))
        import handler_cy
        handler_fn = handler_cy.handler

    n = args.n
    settle = args.settle or max(3.0, n / 150000.0)
    shared = runloom_c.Chan()
    peers = []   # keep socketpair peer ends alive so recv never returns

    def worker_empty():
        shared.recv()                       # park forever (no sender)

    def worker_socket():
        a, b = sk.socketpair()
        peers.append(b)                     # keep peer open -> recv blocks
        afd = a.detach()
        conn = runloom_c.TCPConn(afd)
        if args.handler == "py":
            buf = bytearray(65536)          # the py-handler's per-conn heap buffer
            conn.recv_into(buf)             # parks forever
        else:
            handler_fn(conn)                # Cython handler: stack buffer, parks

    if args.state == "socket":
        worker = worker_socket
    elif args.handler == "c":
        worker = compile_empty_parker(shared)   # COMPILED bare fiber (~1 page)
    else:
        worker = worker_empty                   # interpreted bare fiber (~2 pages)

    def measurer():
        runloom.sleep(settle)
        out = {"runtime": "runloom", "handler": args.handler,
               "optimize": args.optimize, "state": args.state, "n": n,
               "hubs": args.hubs, "rss_bytes": rss_bytes(), "pss_bytes": pss_bytes()}
        print(json.dumps(out), flush=True)
        os._exit(0)

    def root():
        for _ in range(n):
            runloom.fiber(worker)
        runloom.fiber(measurer)

    runloom.run(args.hubs, root)


if __name__ == "__main__":
    main()

// Memory probe for the [go] column: hold N blocked goroutines in a given state
// and report USED memory (VmRSS from /proc/self/status -- not virtual size).
//
//	-state empty   : N goroutines blocked on a channel receive (bare goroutine)
//	-state socket  : N goroutines each holding a socketpair end, blocked on Read
package main

import (
	"bufio"
	"encoding/json"
	"flag"
	"fmt"
	"net"
	"os"
	"strconv"
	"strings"
	"sync"
	"syscall"
	"time"
)

func vmRSSBytes() int64 {
	f, err := os.Open("/proc/self/status")
	if err != nil {
		return -1
	}
	defer f.Close()
	sc := bufio.NewScanner(f)
	for sc.Scan() {
		line := sc.Text()
		if strings.HasPrefix(line, "VmRSS:") {
			fields := strings.Fields(line)
			kb, _ := strconv.ParseInt(fields[1], 10, 64)
			return kb * 1024
		}
	}
	return -1
}

func main() {
	state := flag.String("state", "empty", "empty|socket")
	n := flag.Int("n", 100000, "")
	settle := flag.Float64("settle", 0, "")
	flag.Parse()

	s := *settle
	if s == 0 {
		s = 3.0
		if v := float64(*n) / 150000.0; v > s {
			s = v
		}
	}

	done := make(chan struct{}) // never closed -> receivers block forever
	var peers []*os.File
	var conns []net.Conn
	var mu sync.Mutex

	for i := 0; i < *n; i++ {
		if *state == "socket" {
			fds, err := syscall.Socketpair(syscall.AF_UNIX, syscall.SOCK_STREAM, 0)
			if err != nil {
				continue
			}
			// Hand the read end to the runtime netpoller (net.Conn) instead of a
			// blocking os.File.Read -- otherwise each blocked Read pins an OS
			// thread and 10k+ of them hit Go's thread-exhaustion limit (and
			// inflate RSS with thread stacks). This matches how a real Go server
			// holds idle connections: epoll-managed, one goroutine, no thread.
			syscall.SetNonblock(fds[0], true)
			f := os.NewFile(uintptr(fds[0]), "a")
			c, err := net.FileConn(f)
			f.Close()
			if err != nil {
				syscall.Close(fds[1])
				continue
			}
			peer := os.NewFile(uintptr(fds[1]), "b")
			mu.Lock()
			conns = append(conns, c)
			peers = append(peers, peer)
			mu.Unlock()
			go func(c net.Conn) {
				// match the runloom handler buffer (mem_runloom.py bytearray(65536))
				// and the perf-path Go server (srv_go.go make([]byte,65536)) so the
				// w/socket comparison holds an equal-size handler buffer per task.
				buf := make([]byte, 65536)
				// Fault the buffer in (make its pages resident) -- the ACTUAL
				// under-load situation: a live connection's Read FILLS this buffer,
				// and CPython's bytearray(65536) is eagerly resident on allocation.
				// Without this touch Go's make([]byte) stays lazily unfaulted while
				// the probe parks on Read, which UNDERSTATES Go's real
				// per-active-connection RSS and makes the w/socket column
				// apples-to-oranges vs CPython.  One write per 4 KiB page = resident.
				for i := 0; i < len(buf); i += 4096 {
					buf[i] = 1
				}
				c.Read(buf) // parks via the netpoller: peer never writes
			}(c)
		} else {
			go func() { <-done }()
		}
	}

	time.Sleep(time.Duration(s * float64(time.Second)))
	out := map[string]any{"runtime": "go", "state": *state, "n": *n,
		"rss_bytes": vmRSSBytes(), "pss_bytes": nil}
	b, _ := json.Marshal(out)
	fmt.Println(string(b))
	_ = peers
	_ = conns
}

/* Part of runloom_tcp.c -- #included into that translation unit, not compiled
 * standalone.  Implements the zero-PyObject C entry points declared in
 * runloom_tcp_capi.h.  Mirrors RunloomTCPConn_recv_into / _send_all exactly
 * (same FSM classification, same io_uring branch, same netpoll park), but
 * without Py_buffer / PyArg / PyLong / PyErr -- so a Cython handler that calls
 * these has an allocation-free, PyObject-free hot loop. */

#include "runloom_tcp_capi.h"

/* recv up to n bytes into buf.  Returns bytes read (0 = EOF), -1 on error. */
Py_ssize_t runloom_tcpconn_c_recv_into(PyObject *conn, void *buf, Py_ssize_t n)
{
    RunloomTCPConn *self = (RunloomTCPConn *)conn;
    int fd;

    if (self->closed || self->fd < 0) { errno = EBADF; return -1; }
    if (n <= 0) return 0;
    fd = self->fd;

#if defined(__linux__)
    /* Stage-2 single-shot io_uring PROACTOR: when the loop backend is on, submit
     * the recv as an SQE and inline-drain -- on a data-ready socket the CQE is
     * already posted (FAST_POLL did the recv) so the fiber skips the park
     * entirely.  This is the +20%-on-loopback path the all-C echo uses; without
     * it a Cython handler on RUNLOOM_IOURING_LOOP would do a readiness recv() +
     * an io_uring-bridged park (io_uring's cost, none of its win). */
    if (runloom_iouring_loop_enabled()) {
        runloom_iouring_ring_t *ring = runloom_mn_current_iouring_ring();
        if (ring != NULL) {
            runloom_iouring_ssize_t r =
                runloom_iouring_loop_recv(ring, fd, buf, (size_t)n, 0);
            return (r < 0) ? -1 : (Py_ssize_t)r;   /* errno set by loop_recv */
        }
    }
    if (runloom_tcpconn_use_iouring(self)) {
        runloom_iouring_ssize_t r;
        if (runloom_iouring_pbuf_available()) {
            if (self->ms == NULL) self->ms = runloom_iouring_ms_open(fd);
            if (self->ms != NULL) {
                r = runloom_iouring_ms_recv(self->ms, buf, (size_t)n);
                return (r < 0) ? -1 : (Py_ssize_t)r;
            }
        }
        r = runloom_iouring_recv(fd, buf, (size_t)n, 0);
        return (r < 0) ? -1 : (Py_ssize_t)r;
    }
#endif

    for (;;) {
#if defined(RUNLOOM_OS_WINDOWS)
        long long r = (long long)recv((SOCKET)fd, (char *)buf, (int)n, 0);
#else
        long long r = (long long)recv(fd, buf, (size_t)n, 0);
#endif
        RUNLOOM_IO_SWITCH(runloom_io_classify(RUNLOOM_IO_RECV, r, runloom_sock_errno())) {
        case RUNLOOM_IO_READY:
        case RUNLOOM_IO_EOF:
            return (Py_ssize_t)r;        /* >0 data, or 0 = orderly EOF */
        case RUNLOOM_IO_WOULDBLOCK_R:
        case RUNLOOM_IO_INTR:
            break;                        /* not ready -> park on READ */
        case RUNLOOM_IO_ERROR:
            return -1;                    /* errno already set by recv() */
        case RUNLOOM_IO_PROGRESS:
        case RUNLOOM_IO_WOULDBLOCK_W:
        case RUNLOOM_IO_RETRY:
        case RUNLOOM_IO_EVENT_COUNT:
            runloom_fsm_violation("runloom_tcp_capi.recv_into",
                                  RUNLOOM_IO_RECV, -1, __FILE__, __LINE__);
        } RUNLOOM_IO_SWITCH_END
        if (runloom_netpoll_wait_fd_coop(fd, RUNLOOM_NETPOLL_READ, -1LL) < 0) {
            if (!errno) errno = ECANCELED;
            return -1;
        }
    }
}

/* send exactly n bytes from buf.  Returns n on success, -1 on error. */
Py_ssize_t runloom_tcpconn_c_send_all(PyObject *conn, const void *buf, Py_ssize_t n)
{
    RunloomTCPConn *self = (RunloomTCPConn *)conn;
    int fd;
    Py_ssize_t sent = 0;

    if (self->closed || self->fd < 0) { errno = EBADF; return -1; }
    if (n <= 0) return 0;
    fd = self->fd;

#if defined(__linux__)
    /* Stage-2 single-shot io_uring proactor send (see recv_into above). */
    if (runloom_iouring_loop_enabled()) {
        runloom_iouring_ring_t *ring = runloom_mn_current_iouring_ring();
        if (ring != NULL) {
            while (sent < n) {
                runloom_iouring_ssize_t r =
                    runloom_iouring_loop_send(ring, fd, (const char *)buf + sent,
                                              (size_t)(n - sent), 0);
                if (r < 0) return -1;
                sent += (Py_ssize_t)r;
                if (r == 0) break;
            }
            return sent;
        }
    }
    if (runloom_tcpconn_use_iouring(self)) {
        while (sent < n) {
            runloom_iouring_ssize_t r =
                runloom_iouring_send(fd, (const char *)buf + sent,
                                     (size_t)(n - sent), 0);
            if (r < 0) return -1;
            sent += (Py_ssize_t)r;
            if (r == 0) break;            /* defensive */
        }
        return sent;
    }
#endif

    while (sent < n) {
#if defined(RUNLOOM_OS_WINDOWS)
        long long r = (long long)send((SOCKET)fd, (const char *)buf + sent,
                                      (int)(n - sent), 0);
#else
        long long r = (long long)send(fd, (const char *)buf + sent,
                                      (size_t)(n - sent), 0);
#endif
        RUNLOOM_IO_SWITCH(runloom_io_classify(RUNLOOM_IO_SEND, r, runloom_sock_errno())) {
        case RUNLOOM_IO_PROGRESS:
            sent += (Py_ssize_t)r;
            continue;                     /* re-check sent < n */
        case RUNLOOM_IO_WOULDBLOCK_W:
        case RUNLOOM_IO_INTR:
            break;                        /* not writable -> park on WRITE */
        case RUNLOOM_IO_ERROR:
            return -1;
        case RUNLOOM_IO_READY:
        case RUNLOOM_IO_EOF:
        case RUNLOOM_IO_WOULDBLOCK_R:
        case RUNLOOM_IO_RETRY:
        case RUNLOOM_IO_EVENT_COUNT:
            runloom_fsm_violation("runloom_tcp_capi.send_all",
                                  RUNLOOM_IO_SEND, -1, __FILE__, __LINE__);
        } RUNLOOM_IO_SWITCH_END
        if (runloom_netpoll_wait_fd_coop(fd, RUNLOOM_NETPOLL_WRITE, -1LL) < 0) {
            if (!errno) errno = ECANCELED;
            return -1;
        }
    }
    return sent;
}

/* ============================================================================
 * Raw-fd, tstate-free I/O for c_entry handlers (no PyObject, no TCPConn, no
 * tstate).  Mirrors the all-C echo (module_io.c.inc runloom_io_c_echo): Stage-2
 * io_uring proactor when the loop backend + a hub ring are live, else readiness
 * recv/send + runloom_netpoll_wait_fd.  A Cython cdef handler cimports these (via
 * the __tcp_capi__ capsule) and runs the request loop with zero Python state.
 * ============================================================================ */
Py_ssize_t runloom_tcp_c_fd_recv(int fd, void *buf, Py_ssize_t n)
{
    if (n <= 0) return 0;
#if defined(__linux__)
    if (runloom_iouring_loop_enabled()) {
        runloom_iouring_ring_t *ring = runloom_mn_current_iouring_ring();
        if (ring != NULL) {
            runloom_iouring_ssize_t r =
                runloom_iouring_loop_recv(ring, fd, buf, (size_t)n, 0);
            return (r < 0) ? -1 : (Py_ssize_t)r;
        }
    }
#endif
    for (;;) {
#if defined(RUNLOOM_OS_WINDOWS)
        long long r = (long long)recv((SOCKET)fd, (char *)buf, (int)n, 0);
#else
        long long r = (long long)recv(fd, buf, (size_t)n, 0);
#endif
        RUNLOOM_IO_SWITCH(runloom_io_classify(RUNLOOM_IO_RECV, r, runloom_sock_errno())) {
        case RUNLOOM_IO_READY:
        case RUNLOOM_IO_EOF:
            return (Py_ssize_t)r;
        case RUNLOOM_IO_WOULDBLOCK_R:
        case RUNLOOM_IO_INTR:
            break;
        case RUNLOOM_IO_ERROR:
            return -1;
        case RUNLOOM_IO_PROGRESS:
        case RUNLOOM_IO_WOULDBLOCK_W:
        case RUNLOOM_IO_RETRY:
        case RUNLOOM_IO_EVENT_COUNT:
            runloom_fsm_violation("runloom_tcp_capi.fd_recv",
                                  RUNLOOM_IO_RECV, -1, __FILE__, __LINE__);
        } RUNLOOM_IO_SWITCH_END
        /* _coop maps the POSITIVE RUNLOOM_NETPOLL_CANCELLED sentinel to -1 so a
         * cancelled park UNWINDS instead of slipping the "< 0" test and busy-
         * spinning on a cancelled-but-open fd under connection churn (matches the
         * TCPConn methods + the cooperative socket fast paths; audit finding B3). */
        if (runloom_netpoll_wait_fd_coop(fd, RUNLOOM_NETPOLL_READ, -1LL) < 0) {
            if (!errno) errno = ECANCELED;
            return -1;
        }
    }
}

Py_ssize_t runloom_tcp_c_fd_send_all(int fd, const void *buf, Py_ssize_t n)
{
    Py_ssize_t sent = 0;
    if (n <= 0) return 0;
#if defined(__linux__)
    if (runloom_iouring_loop_enabled()) {
        runloom_iouring_ring_t *ring = runloom_mn_current_iouring_ring();
        if (ring != NULL) {
            while (sent < n) {
                runloom_iouring_ssize_t r = runloom_iouring_loop_send(
                    ring, fd, (const char *)buf + sent, (size_t)(n - sent), 0);
                if (r < 0) return -1;
                sent += (Py_ssize_t)r;
                if (r == 0) break;
            }
            return sent;
        }
    }
#endif
    while (sent < n) {
#if defined(RUNLOOM_OS_WINDOWS)
        long long r = (long long)send((SOCKET)fd, (const char *)buf + sent,
                                      (int)(n - sent), 0);
#else
        long long r = (long long)send(fd, (const char *)buf + sent,
                                      (size_t)(n - sent), 0);
#endif
        RUNLOOM_IO_SWITCH(runloom_io_classify(RUNLOOM_IO_SEND, r, runloom_sock_errno())) {
        case RUNLOOM_IO_PROGRESS:
            sent += (Py_ssize_t)r;
            continue;
        case RUNLOOM_IO_WOULDBLOCK_W:
        case RUNLOOM_IO_INTR:
            break;
        case RUNLOOM_IO_ERROR:
            return -1;
        case RUNLOOM_IO_READY:
        case RUNLOOM_IO_EOF:
        case RUNLOOM_IO_WOULDBLOCK_R:
        case RUNLOOM_IO_RETRY:
        case RUNLOOM_IO_EVENT_COUNT:
            runloom_fsm_violation("runloom_tcp_capi.fd_send_all",
                                  RUNLOOM_IO_SEND, -1, __FILE__, __LINE__);
        } RUNLOOM_IO_SWITCH_END
        /* _coop: unwind a cancelled park instead of busy-spinning (see fd_recv). */
        if (runloom_netpoll_wait_fd_coop(fd, RUNLOOM_NETPOLL_WRITE, -1LL) < 0) {
            if (!errno) errno = ECANCELED;
            return -1;
        }
    }
    return sent;
}

void runloom_tcp_c_fd_close(int fd)
{
    runloom_netpoll_release_if_idle(fd);
#if defined(RUNLOOM_OS_WINDOWS)
    closesocket((SOCKET)fd);
#else
    close(fd);
#endif
}

/* The exported pointer table for the PyCapsule fallback. */
static const RunloomTCPCAPI runloom_tcp_capi_table = {
    runloom_tcpconn_c_recv_into,
    runloom_tcpconn_c_send_all,
    runloom_tcp_c_fd_recv,
    runloom_tcp_c_fd_send_all,
    runloom_tcp_c_fd_close,
};

/* Called once from module init (module_init.c.inc) to attach the capsule. */
int runloom_tcpconn_capi_register(PyObject *module)
{
    PyObject *cap = PyCapsule_New((void *)&runloom_tcp_capi_table,
                                  RUNLOOM_TCP_CAPI_CAPSULE_NAME, NULL);
    if (cap == NULL) return -1;
    if (PyModule_AddObject(module, "__tcp_capi__", cap) < 0) {
        Py_DECREF(cap);
        return -1;
    }
    return 0;
}

"""Shared configuration for the Runloom benchmark suite.

Everything that the spec pins to os.cpu_count() is derived here so every program
agrees on the same numbers, and the report can print them as the assumed
constraints.  Core *placement* implements scoping decision #3 (client and server
on disjoint cores so they don't steal each other's CPU) + #5 (per-core =
saturated multi-core / hub_count).
"""
import os

# ---------------------------------------------------------------------------
# Machine-derived sizing (the spec's int(cpu*k) knobs)
# ---------------------------------------------------------------------------
CPU_COUNT = os.cpu_count() or 1

HUBS = int(CPU_COUNT * 0.7)          # runloom M:N hubs  (spec: int(cpu*0.7))
GO_SERVER_CORES = int(CPU_COUNT * 0.7)  # go GOMAXPROCS    (spec: int(cpu*0.7))
CLIENT_CORES = int(CPU_COUNT * 0.25)    # go loadgen cores (spec: int(cpu*0.25))

# ---------------------------------------------------------------------------
# Disjoint CPU placement (decision #3).
#   client  -> the first CLIENT_CORES cpus (NUMA node 0 on the 2-node box)
#   server  -> the next HUBS cpus, starting after the client set
# They never overlap, so the loadgen cannot steal a server hub's core.  On the
# 64-vCPU / 2-NUMA test box this puts the client wholly in node0 (0-15) and the
# server in node0's tail + node1 (16-59); the gap is documented in the report.
# ---------------------------------------------------------------------------
_all = list(range(CPU_COUNT))
CLIENT_CPUS = _all[:CLIENT_CORES]
SERVER_CPUS = _all[CLIENT_CORES:CLIENT_CORES + max(HUBS, GO_SERVER_CORES)]
# taskset -c strings
CLIENT_CPU_SPEC = ",".join(map(str, CLIENT_CPUS))
SERVER_CPU_SPEC = ",".join(map(str, SERVER_CPUS))

# ---------------------------------------------------------------------------
# Network topology (decision #3: veth pair across two netns)
# ---------------------------------------------------------------------------
SRV_NS = "rl_srv"
CLI_NS = "rl_cli"
VETH_SRV = "rl_vsrv"
VETH_CLI = "rl_vcli"
SRV_IP = "10.99.0.1"
CLI_IP = "10.99.0.2"
PREFIX = 24
BASE_PORT = 9000

# Client source-IP fan-out for the CONNECTION-CHURN benchmark.  The churn client
# actively closes every connection, so it accumulates TIME_WAIT and burns
# ephemeral ports on its own 4-tuples; a single source IP caps out at ~64k ports,
# and at a high conn/s rate the TIME_WAIT backlog (rate x fin_timeout) exceeds
# that and dials begin to fail -- capping the MEASURED conn/s, not the server.
# This is the netns/veth analog of big_100's 127/8 fragment trick: spread the
# connects across a block of source IPs on the client veth so each gets its own
# independent ephemeral/TIME_WAIT pool.  All live in the client /24 (on-link, so
# the server's connected route returns replies with no extra routing).  The
# primary CLI_IP is first; BENCH_CLI_SRC_IPS overrides the count.
CLI_SRC_IP_COUNT = int(os.environ.get("BENCH_CLI_SRC_IPS", "32"))
CLI_SRC_IPS = ["10.99.0.%d" % (2 + i) for i in range(max(1, CLI_SRC_IP_COUNT))]

# Spec sysctls, applied INSIDE the server netns (they are namespaced).
NS_SYSCTLS = {
    "net.ipv4.tcp_wmem": "4096 16384 2097152",   # 2 MB max
    "net.ipv4.tcp_rmem": "4096 87380 2097152",   # 2 MB max
    "net.core.somaxconn": "65535",               # accept backlog ceiling
    "net.ipv4.tcp_tw_reuse": "1",
}

# fd ceiling for millions-of-connections runs (decision: raise per-block).
FD_LIMIT = 8_388_608

# ---------------------------------------------------------------------------
# Payloads (decision #1)
#   req/s headline  -> small payload, measures scheduling/syscall overhead
#   bandwidth (GB/s)-> the spec's 1.5 MB buffer, measures copy/IO throughput
# ---------------------------------------------------------------------------
PAYLOAD_SMALL = 1024                 # 1 KiB request for the req/s metric
PAYLOAD_LARGE = 1536 * 1024          # 1.5 MiB for the bandwidth metric
CLIENT_BUF = 1536 * 1024             # fixed 1.5 MB client buffer (spec)

# ---------------------------------------------------------------------------
# Measurement (decision #8: rigorous stop rule + ladder)
# ---------------------------------------------------------------------------
RAMP_S = 2.0                         # establish + warm connections before timing
MEASURE_S = 5.0                      # timed window
REPS = 3                             # independent reps per ladder rung
# Geometric connection ladder; stop when 2 consecutive rungs fail to beat the
# best rung's bootstrap CI.  Capped at 32768 so the PERSISTENT req/s benchmark --
# which establishes this many CONCURRENT connections from the single client IP --
# stays under the ~64k ephemeral-port ceiling.  (The CHURN benchmark fans its
# connects across CLI_SRC_IPS, so it is no longer bound by that ceiling at these
# rungs; the cap is the persistent benchmark's constraint.)
CONN_LADDER = [16, 32, 64, 128, 256, 512, 1024, 2048, 4096, 8192, 16384, 32768]
PLATEAU_PATIENCE = 2

# ---------------------------------------------------------------------------
# Interpreters (decision #4 + #7)
# ---------------------------------------------------------------------------
PYENV = os.path.expanduser("~/.pyenv/versions")
FT_PYTHON = os.path.join(PYENV, "3.13.13t", "bin", "python3")   # runloom (GIL off)
GIL_PYTHON = os.path.join(PYENV, "3.13.13", "bin", "python3")   # asyncio/uvloop/gevent best-case

# Repo paths
HARNESS_DIR = os.path.dirname(os.path.abspath(__file__))
SUITE_DIR = os.path.dirname(HARNESS_DIR)
BENCH_DIR = os.path.dirname(SUITE_DIR)
REPO = os.path.dirname(BENCH_DIR)
SRC = os.path.join(REPO, "src")
RESULTS_DIR = os.path.join(BENCH_DIR, "results")
SERVERS_DIR = os.path.join(SUITE_DIR, "servers")
CLIENTS_DIR = os.path.join(SUITE_DIR, "clients")


def git_commit():
    """Short HEAD sha of the repo the suite is running from (provenance: which
    commit produced these numbers).  '?' if git is unavailable."""
    import subprocess
    try:
        return subprocess.run(["git", "-C", REPO, "rev-parse", "--short", "HEAD"],
                              capture_output=True, text=True, timeout=10).stdout.strip() or "?"
    except Exception:
        return "?"


def base_env(gil_off=True):
    """A clean child env: PYTHONPATH=src, GIL toggled, RUNLOOM_DEBUG cleared
    (decision #7: as-shipped release, debug OFF)."""
    e = dict(os.environ)
    e["PYTHONPATH"] = SRC + (os.pathsep + e["PYTHONPATH"] if e.get("PYTHONPATH") else "")
    e["PYTHON_GIL"] = "0" if gil_off else "1"
    e.pop("RUNLOOM_DEBUG", None)
    return e


def summary():
    return {
        "git_commit": git_commit(),
        "cpu_count": CPU_COUNT,
        "hubs": HUBS,
        "go_server_cores": GO_SERVER_CORES,
        "client_cores": CLIENT_CORES,
        "client_cpus": CLIENT_CPU_SPEC,
        "server_cpus": SERVER_CPU_SPEC,
        "payload_small_bytes": PAYLOAD_SMALL,
        "payload_large_bytes": PAYLOAD_LARGE,
        "ramp_s": RAMP_S,
        "measure_s": MEASURE_S,
        "reps": REPS,
        "conn_ladder": CONN_LADDER,
        "sysctls": NS_SYSCTLS,
        "fd_limit": FD_LIMIT,
        "ft_python": FT_PYTHON,
        "gil_python": GIL_PYTHON,
    }


if __name__ == "__main__":
    import json
    print(json.dumps(summary(), indent=2))

"""System setup for the network benchmarks: the veth-pair / two-netns topology,
disjoint CPU pinning, fd-limit raising, and namespaced sysctls (decisions #3,
#7, #8 and the spec's system-wide requirements).

Why a veth pair across two netns instead of loopback:
  * On this Docker host every loopback packet traverses the host nft ruleset
    (~14% throughput tax) -- a fresh netns has an empty ruleset.
  * Client and server in *separate* netns over a veth pair cross a real device
    queue, so io_uring is not hidden behind the loopback fast path, and the
    loadgen and server cannot contend on the same `lo`.

Everything here shells out through `sudo -n` (passwordless sudo confirmed on the
box).  Commands are returned as argv lists for subprocess; env vars are passed
explicitly via an `env` prefix because `sudo` strips the environment.
"""
import os
import shutil
import subprocess
import sys

from config import (SRV_NS, CLI_NS, VETH_SRV, VETH_CLI, SRV_IP, CLI_IP, PREFIX,
                    NS_SYSCTLS, FD_LIMIT, SRC, FT_PYTHON, CLI_SRC_IPS)


def _sudo(*argv, check=True, quiet=False):
    cmd = ["sudo", "-n", *argv]
    r = subprocess.run(cmd, capture_output=True, text=True)
    if r.returncode != 0 and check and not quiet:
        sys.stderr.write("CMD FAILED: %s\n%s\n" % (" ".join(cmd), r.stderr))
    return r


def ensure_kernel_ceilings():
    """Raise the global kernel ceilings that bite at millions of fds/fibers.
    These persist until reboot; safe to re-apply."""
    for key, val in {
        "fs.nr_open": str(FD_LIMIT),
        # ~2 VMAs per fiber stack (the stack mapping + its guard-page mprotect).
        # 1M plain fibers need ~2M VMAs, so the old 2,000,000 ceiling sat right at
        # the wall: mmap/mprotect starts failing at the limit and the spawn path
        # degrades into a syscall-retry storm -> the 1M-fiber RSS bench timed out
        # at 900s. 4M gives headroom (with it raised, the same run completes in
        # ~86s). optimize("memory")'s warm-stack arena uses far fewer VMAs and was
        # never affected.
        "vm.max_map_count": "4000000",
        "net.core.somaxconn": "65535",
    }.items():
        _sudo("sysctl", "-w", "%s=%s" % (key, val), quiet=True)


def teardown():
    """Delete both netns (removes the veth pair with them).  Idempotent."""
    for ns in (SRV_NS, CLI_NS):
        _sudo("ip", "netns", "del", ns, check=False, quiet=True)
    # A stray veth (e.g. a half-built setup) lingers in the root ns; clean it.
    _sudo("ip", "link", "del", VETH_SRV, check=False, quiet=True)


def setup():
    """(Re)create the two-netns veth topology + namespaced sysctls.  Idempotent:
    tears down any stale instance first."""
    teardown()
    ensure_kernel_ceilings()
    # netns
    _sudo("ip", "netns", "add", SRV_NS)
    _sudo("ip", "netns", "add", CLI_NS)
    # veth pair, one end into each ns
    _sudo("ip", "link", "add", VETH_SRV, "type", "veth", "peer", "name", VETH_CLI)
    _sudo("ip", "link", "set", VETH_SRV, "netns", SRV_NS)
    _sudo("ip", "link", "set", VETH_CLI, "netns", CLI_NS)
    # addresses + up
    _sudo("ip", "netns", "exec", SRV_NS, "ip", "addr", "add",
          "%s/%d" % (SRV_IP, PREFIX), "dev", VETH_SRV)
    _sudo("ip", "netns", "exec", CLI_NS, "ip", "addr", "add",
          "%s/%d" % (CLI_IP, PREFIX), "dev", VETH_CLI)
    # Client source-IP fan-out block (the netns analog of big_100's 127/8 trick):
    # extra addresses on the client veth so the churn loadgen can spread its
    # connects across many source IPs and avoid TIME_WAIT/ephemeral exhaustion.
    # All are in the client /24 (CLI_IP's subnet), so they need no extra route;
    # CLI_IP itself is already added above.  Idempotent ("add" of an existing
    # addr just warns, which _sudo swallows under quiet).
    for ip in CLI_SRC_IPS:
        if ip == CLI_IP:
            continue
        _sudo("ip", "netns", "exec", CLI_NS, "ip", "addr", "add",
              "%s/%d" % (ip, PREFIX), "dev", VETH_CLI, check=False, quiet=True)
    for ns, dev in ((SRV_NS, VETH_SRV), (CLI_NS, VETH_CLI)):
        _sudo("ip", "netns", "exec", ns, "ip", "link", "set", dev, "up")
        _sudo("ip", "netns", "exec", ns, "ip", "link", "set", "lo", "up")
    # spec sysctls -- namespaced, so set them INSIDE the server ns
    for key, val in NS_SYSCTLS.items():
        _sudo("ip", "netns", "exec", SRV_NS, "sysctl", "-w", "%s=%s" % (key, val), quiet=True)
    # sanity: client can reach server
    r = _sudo("ip", "netns", "exec", CLI_NS, "ping", "-c", "1", "-W", "1", SRV_IP, check=False)
    if r.returncode != 0:
        raise RuntimeError("veth topology setup failed: client cannot ping server\n" + r.stderr)
    return {"srv_ns": SRV_NS, "cli_ns": CLI_NS, "srv_ip": SRV_IP, "cli_ip": CLI_IP}


def _env_prefix(extra_env=None, gil_off=True):
    """Explicit env passed through sudo's environment scrub."""
    env = {
        "PYTHONPATH": SRC,
        "PYTHON_GIL": "0" if gil_off else "1",
        "PATH": os.environ.get("PATH", "/usr/bin:/bin"),
        "HOME": os.environ.get("HOME", "/root"),
    }
    if extra_env:
        env.update(extra_env)
    return ["env"] + ["%s=%s" % (k, v) for k, v in env.items()]


def ns_cmd(ns, argv, cpus=None, extra_env=None, gil_off=True, raise_fd=True):
    """Build an argv list that runs `argv` inside netns `ns`, pinned to `cpus`
    (a 'c0,c1,...' taskset spec), with RLIMIT_NOFILE raised and env injected."""
    cmd = ["sudo", "-n", "ip", "netns", "exec", ns]
    if raise_fd:
        cmd += ["prlimit", "--nofile=%d:%d" % (FD_LIMIT, FD_LIMIT)]
    if cpus:
        cmd += ["taskset", "-c", cpus]
    cmd += _env_prefix(extra_env, gil_off)
    cmd += list(argv)
    return cmd


def pinned_cmd(argv, cpus=None, extra_env=None, gil_off=True, raise_fd=False):
    """For benchmarks that don't need the network (spawn/ctxswitch/memory):
    just pin + env (+ optional fd raise via sudo prlimit, root)."""
    cmd = []
    if raise_fd:
        cmd += ["sudo", "-n", "prlimit", "--nofile=%d:%d" % (FD_LIMIT, FD_LIMIT)]
    if cpus:
        cmd += ["taskset", "-c", cpus]
    if raise_fd:
        # sudo scrubs env -> inject explicitly
        cmd += _env_prefix(extra_env, gil_off)
    else:
        # no sudo: caller passes env= to Popen; still set the basics inline
        cmd += _env_prefix(extra_env, gil_off)
    cmd += list(argv)
    return cmd


if __name__ == "__main__":
    import json
    action = sys.argv[1] if len(sys.argv) > 1 else "setup"
    if action == "setup":
        print(json.dumps(setup(), indent=2))
    elif action == "teardown":
        teardown()
        print("torn down")
    elif action == "demo-cmd":
        print(" ".join(ns_cmd(SRV_NS, [FT_PYTHON, "-c", "print(1)"], cpus="32,33")))

"""Measurement primitives shared by the network benchmarks: launch a server and
wait for its LISTENING line, sample per-core CPU to decide server- vs client-
bound, run the Go loadgen, and walk the connection ladder with a rigorous stop
rule + bootstrap CI (decision #8).
"""
import json
import os
import queue
import statistics
import subprocess
import threading
import time

import config


# --------------------------------------------------------------------------
# per-core CPU utilisation from /proc/stat  (server-bound check, decision #8)
# --------------------------------------------------------------------------
def _proc_stat():
    out = {}
    with open("/proc/stat") as f:
        for line in f:
            if line.startswith("cpu") and line[3].isdigit():
                parts = line.split()
                cpu = int(parts[0][3:])
                vals = list(map(int, parts[1:]))
                idle = vals[3] + (vals[4] if len(vals) > 4 else 0)  # idle + iowait
                total = sum(vals)
                out[cpu] = (total - idle, total)  # (busy, total)
    return out


def cpu_group_util(snap0, snap1, cpus):
    """Mean busy fraction across `cpus` between two /proc/stat snapshots."""
    fr = []
    for c in cpus:
        if c in snap0 and c in snap1:
            db = snap1[c][0] - snap0[c][0]
            dt = snap1[c][1] - snap0[c][1]
            if dt > 0:
                fr.append(db / dt)
    return (sum(fr) / len(fr)) if fr else 0.0


# --------------------------------------------------------------------------
# server process lifecycle
# --------------------------------------------------------------------------
class Server:
    def __init__(self, cmd, token, name):
        self.cmd = cmd
        self.token = token
        self.name = name
        self.proc = None
        self.port = None
        self.lines = []
        self._q = queue.Queue()

    def _drain(self, stream, tag):
        for line in iter(stream.readline, ""):
            self.lines.append("[%s] %s" % (tag, line.rstrip()))
            self._q.put(line.rstrip())
        stream.close()

    def start(self, timeout=30.0):
        self.proc = subprocess.Popen(
            self.cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
            text=True, start_new_session=True)
        threading.Thread(target=self._drain, args=(self.proc.stdout, "out"), daemon=True).start()
        threading.Thread(target=self._drain, args=(self.proc.stderr, "err"), daemon=True).start()
        deadline = time.time() + timeout
        while time.time() < deadline:
            try:
                line = self._q.get(timeout=0.5)
            except queue.Empty:
                if self.proc.poll() is not None:
                    raise RuntimeError("server %s exited before LISTENING:\n%s"
                                       % (self.name, "\n".join(self.lines)))
                continue
            if line.startswith("LISTENING"):
                self.port = int(line.split()[1])
                return self.port
        raise RuntimeError("server %s never printed LISTENING:\n%s"
                           % (self.name, "\n".join(self.lines)))

    def stop(self):
        # Servers run as root via `sudo ip netns exec ...`; kill by unique token.
        subprocess.run(["sudo", "-n", "pkill", "-9", "-f", self.token],
                       capture_output=True)
        if self.proc:
            try:
                self.proc.terminate()
                self.proc.wait(timeout=3)
            except Exception:
                try:
                    self.proc.kill()
                except Exception:
                    pass
        time.sleep(0.3)


# --------------------------------------------------------------------------
# loadgen invocation
# --------------------------------------------------------------------------
def run_loadgen(loadgen_bin, addr, conns, payload, ramp, measure, gomax,
                server_cpus=None, src_ips=None):
    """Run the Go loadgen in the client netns, pinned to client cpus.  Returns
    (result_dict, server_util, client_util).  server_cpus is the server's ACTUAL
    pinned core set (so a 1-core server's util isn't diluted across 44 cores).
    src_ips (churn only) is a list of client source IPs the loadgen rotates its
    connects across to dodge TIME_WAIT/ephemeral-port exhaustion."""
    import topo
    server_cpus = server_cpus if server_cpus is not None else config.SERVER_CPUS
    argv = [loadgen_bin, "-addr", addr, "-conns", str(conns),
            "-payload", str(payload), "-ramp", str(ramp),
            "-measure", str(measure), "-gomaxprocs", str(gomax)]
    if src_ips:
        argv += ["-srcips", ",".join(src_ips)]
    cmd = topo.ns_cmd(config.CLI_NS, argv, cpus=config.CLIENT_CPU_SPEC,
                      raise_fd=True, gil_off=True)
    s0 = _proc_stat()
    t0 = time.time()
    r = subprocess.run(cmd, capture_output=True, text=True,
                       timeout=ramp + measure + 45)
    dt = time.time() - t0
    s1 = _proc_stat()
    srv_u = cpu_group_util(s0, s1, server_cpus)
    cli_u = cpu_group_util(s0, s1, config.CLIENT_CPUS)
    out = r.stdout.strip().splitlines()
    res = None
    for line in reversed(out):
        line = line.strip()
        if line.startswith("{"):
            res = json.loads(line)
            break
    if res is None:
        raise RuntimeError("loadgen produced no JSON (conns=%d):\nSTDOUT:%s\nSTDERR:%s"
                           % (conns, r.stdout, r.stderr))
    return res, srv_u, cli_u


def bootstrap_ci(xs, iters=2000, q=0.95):
    """Nonparametric bootstrap CI of the median (decision #8)."""
    if len(xs) < 2:
        return (xs[0], xs[0]) if xs else (0.0, 0.0)
    import random
    rnd = random.Random(12345)
    meds = []
    n = len(xs)
    for _ in range(iters):
        sample = [xs[rnd.randrange(n)] for _ in range(n)]
        meds.append(statistics.median(sample))
    meds.sort()
    lo = meds[int((1 - q) / 2 * iters)]
    hi = meds[int((1 + q) / 2 * iters)]
    return (lo, hi)


def ladder(server_factory, loadgen_bin, addr, payload, ladder_conns,
           reps, ramp, measure, gomax, patience, server_cpus=None, src_ips=None):
    """Bring up ONE server, sweep the connection ladder, find peak rps with a
    rigorous stop rule.  Returns the full curve + peak summary.  server_cpus is
    the server's actual pinned core set for the server-bound check.  src_ips
    (churn only) is forwarded to the loadgen for source-IP fan-out.

    server_factory() -> a started Server (already LISTENING).  We own stopping it.
    """
    srv = server_factory()
    curve = []
    best = {"rps_median": -1.0}
    misses = 0
    try:
        for conns in ladder_conns:
            reps_rps, reps_lat, srv_us, cli_us, errs = [], [], [], [], 0
            for _ in range(reps):
                res, su, cu = run_loadgen(loadgen_bin, addr, conns, payload,
                                          ramp, measure, gomax, server_cpus=server_cpus,
                                          src_ips=src_ips)
                reps_rps.append(res["rps"])
                reps_lat.append(res)
                srv_us.append(su)
                cli_us.append(cu)
                errs += res.get("conn_errors", 0) + res.get("establish_errors", 0)
            med = statistics.median(reps_rps)
            lo, hi = bootstrap_ci(reps_rps)
            rung = {
                "conns": conns,
                "rps_median": med,
                "rps_ci": [lo, hi],
                "rps_reps": reps_rps,
                "server_cpu_util": statistics.median(srv_us),
                "client_cpu_util": statistics.median(cli_us),
                "p50_us": statistics.median([r["p50_us"] for r in reps_lat]),
                "p99_us": statistics.median([r["p99_us"] for r in reps_lat]),
                "p999_us": statistics.median([r["p999_us"] for r in reps_lat]),
                "live_conns": statistics.median([r["live_conns"] for r in reps_lat]),
                "errors": errs,
            }
            curve.append(rung)
            # plateau detection: a rung "improves" only if its median beats the
            # incumbent peak's CI *upper* bound -- a real gain, not noise.  A run
            # that fails to is a miss; `patience` consecutive misses stop the
            # sweep (and we keep climbing through marginal new maxima meanwhile).
            peak_ci_hi = best["rps_ci"][1] if best["rps_median"] >= 0 else -1.0
            if med > peak_ci_hi:
                best = rung
                misses = 0
            else:
                misses += 1
                if med > best["rps_median"]:
                    best = rung   # marginal new max -- keep for reporting
            print("  conns=%-6d rps=%-12.0f ci=[%.0f,%.0f] srvCPU=%.0f%% cliCPU=%.0f%% p99=%.0fus err=%d %s"
                  % (conns, med, lo, hi, rung["server_cpu_util"] * 100,
                     rung["client_cpu_util"] * 100, rung["p99_us"], errs,
                     "(miss %d)" % misses if misses else ""), flush=True)
            if misses >= patience:
                break
    finally:
        srv.stop()
    # bottleneck attribution at the peak
    su = best.get("server_cpu_util", 0.0) or 0.0
    cu = best.get("client_cpu_util", 0.0) or 0.0
    bottleneck = "server" if su >= 0.85 else (
        "client" if cu >= 0.85 else "neither_saturated")
    # When NOT server-bound, estimate the server's true ceiling by extrapolating
    # its measured CPU utilisation to 100% (decision: addresses the 16-core
    # client not saturating fast servers). Clearly an estimate, not a measurement.
    server_ceiling_est = (best["rps_median"] / su) if su > 0.05 else None
    return {"curve": curve, "peak": best, "bottleneck_at_peak": bottleneck,
            "server_ceiling_est": server_ceiling_est,
            "server_ceiling_note": (
                "extrapolated = peak_rps / server_cpu_util; the %d-core client was "
                "the bottleneck at peak" % config.CLIENT_CORES)
            if bottleneck != "server" else "server was the bottleneck (measured)"}

# Spawn + conn/s performance campaign — consolidated summary

One-page index of the spawn-tuning + connection-churn work. Detailed docs:
[`spawn_experiments.md`](spawn_experiments.md) (the A–F campaign + lock-in),
[`spawn_above_1m.md`](spawn_above_1m.md) (the >1M push),
[`conn_cpu.md`](conn_cpu.md) (the passive/conn-CPU decomposition + comparison),
[`spawn_cost.md`](spawn_cost.md) (current spawn-cost note).

## Headline numbers (this box: 64c / 2 NUMA, FT CPython 3.13t)

| metric | what it is | runloom | vs Go |
|---|---|--:|---|
| **active spawn** — fleet launch (`fiber_n`) | create+run+destroy N at once | **~2.41M/s** warm | runloom's spawn ceiling; **Go has no batch API** — a capability, not a like-for-like comparison |
| naked spawn — 1 issuer | one at a time, nothing batched | warm (8-run medians): `c_entry` **~2.23M**; `fiber_fast` **~1.91M**; default `fiber` **~1.4M** | **`c_entry` ≈ Go (~2.2M each, at parity within noise** — ranges overlap, ranking flips run-to-run). `fiber_fast` ~0.85× Go, default `fiber` ~0.6× / 1.5× behind `c_entry`. Boot excluded for all (`--warm`); the default's grow-down learned size spawns down the DEFERRED stack-alloc path → fast AND small-stacked. `optimize("throughput")`→`fiber_fast`, `optimize("memory")`→grow-down |
| single-`fiber()` loop + warm arena | server-style per-event spawn | 242k → **363k** (secure scrub) | — |
| passive — conn/s (churn) | one handler/conn, full TCP lifecycle | **~75–78k/s** (client-bound, 0 dial errors) | **parity** — Go on matched N `SO_REUSEPORT` acceptors ≈ runloom (same acceptor architecture both sides) |
| req/s — persistent / keep-alive | the browser case, spawn ~0% | competitive | ≈ Go **within loadgen noise (client-bound)**; reported raw with cores shown — single-core uvloop/asyncio are server-bound on their one core |
| ctxswitch | yield/resume under load | — | competitive (refcount tax = 1.65× lever) |

## What landed (keepers, gate-green on the branch)

1. **Resident-memset stack scrub = default** (`RUNLOOM_STACK_SCRUB_RESIDENT`, opt-out
   `=0`). Secure wipe of only the touched pages — no per-fiber TLB-shootdown IPI.
   **1.5× on the single-spawn path** (242k→363k), keeps the security guarantee.
2. **`optimize("throughput")` wires the validated spawn fast-path**: warm-stack arena
   + bulk + `FRESH=1` + **`RUNLOOM_GON_PCREATE=auto`** (parallel bulk-create). Delivers
   **~1.5M active spawn/s** for `fiber_n`. `optimize("memory")` turns the RAM parts off.
3. **Parallel bulk-create** (`RUNLOOM_GON_PCREATE`) — TSan-clean, the lever that took
   active spawn 804k → 1.54M.

## The big insight (active spawn)

`total = (create × run)/(create + run)` — a harmonic mean. create was **serial at
1.32M/s** (the per-g template loop, ~72% of it cold-write bound), run parallel at
2.07M. So **every spawn lever was capped at 804k until create parallelized.** Spawning
8 builder threads fill disjoint arena slices (Pass A), then Pass B (coro-fill) parallelized
the same way → create 7.5M → **total ~2.0M @8h, 2.23M @16h** in that tuned campaign. (In-suite on
this box the reproducible figure is **~1.6M**, and Go's naked spawn is **~2.15M** — so this is a
runloom *capability*, since Go has no batch-spawn API, **not** a like-for-like Go win.) Beyond 16 hubs
the run/drain phase is `batch->live`-contention-bound (32h drops to 1.22M); the per-fiber
shard is the next lever, but 8/16h already clear Go.

## Three stories we PROVED WRONG (the campaign's real value)

1. **"The +2× spawn win = killing CPython's mimalloc QSBR purge."** WRONG. A patched
   CPython didn't reproduce it; `strace -k` showed the per-fiber madvise is runloom's
   own **security stack-scrub**. Fixed the docs; shipped the secure resident scrub.
2. **"runloom spins where Go sleeps on conn/s."** WRONG — a `--hubs 1` strace artifact.
   The real multi-hub server **sleeps** in `clock_nanosleep`/`epoll_wait(1ms)`;
   `sched_yield` is 0.5%. (Caught by the adversarial design panel.)
3. **"runloom uses 3× Go's CPU on conn/s."** WRONG — unsaturated, client-bottlenecked,
   mismatched ladder rungs. Saturated on the same cores: **runloom_c = 93% of Go.**

Common thread: a benchmark number is a claim, and the deception is in the dropped
part — which side saturated, how many cores, warm/cold, which metric. The fixes were
always *strace the syscall caller / check who saturated / measure per-core*, never
*trust the throughput delta*.

## Metric framing (now baked into the benchmark report legend)

- **Active spawn** (`fiber_n` fleet) vs **passive spawn** (one handler per connection).
  The parallel-create win is active-only; servers single-spawn, so it doesn't move
  conn/s.
- **conn/s** (churn, new conn/request) peaks ≈ **75–78k** for the fast tiers incl. Go
  (kernel TCP bound, client-bound here) vs **req/s** (keep-alive) = 100k–1M+ (spawn
  ~0%). **Browsers are keep-alive → they hit req/s, not churn.**

## Open / next

- **Active >1.5M:** the run/drain-phase scaling project (shard `batch->live`, the
  scheduler contention) + the ~27k-instr CPython per-fiber floor. `SCHEDULER_SCALING`.
- **Passive last ~7%:** micro-wins (`accept4` drops 2 fcntl/conn; pool the absurd 64KB
  handler buffer; drop the per-conn `functools.partial`; scrub SP-watermark vs
  mincore), and **rebuild `handler_cdef`** to measure the compiled-handler tier (should
  fully close to Go). `runloom_c.serve()` also has a 2-hub serving floor (Go runs on 1).

# Pushing runloom naked-spawn past 1,000,000/s — the plan

> **Note:** the "past Go" framing in this campaign compares runloom's **batch**
> `fiber_n` against Go's **naked** spawn, which is not like-for-like — Go has no
> batch-spawn API. Warm steady-state on the current box, batch `fiber_n` reaches
> **~2.41M/s** — runloom's spawn ceiling, a *capability* with no Go equivalent.
>
> Naked single-spawn (warm steady-state, boot excluded for all — the rate a server
> sustains): the pure-C `c_entry` path and Go are **at parity, ~2.2M/s each**
> (8-run medians 2.23M vs 2.24M, ranges overlapping, ranking flips run-to-run).
> `fiber_fast` **~1.91M (~0.85× Go)**, the **default** `runloom.fiber` **~1.4M
> (~0.6× Go)**, 1.5× behind `c_entry` (its grow-down learned size spawns down the
> deferred alloc path — small-stacked AND fast). See `spawn_cost.md`. The figures
> below are the tuned **batch**-campaign record, not the in-suite report numbers
> and not the naked-spawn result.

Companion to [`spawn_experiments.md`](spawn_experiments.md). That campaign locked in
**804k spawn/s** (the bulk `fiber_n` + FRESH path) and proved memory is solved. This
doc is the vetted plan for the next magnitude, produced by a 7-lever design panel,
each design **adversarially verified against the actual code** before ranking.

## The one truth that orders everything

The bulk path's total is a **harmonic mean** of two phases (verified in
`mn_sched_init_fini.c.inc`, `bulk.py`'s `t0/t1/t2`):

```
total = (create × run) / (create + run)
      = (1.32M × 2.07M) / (1.32M + 2.07M) = 804k
```

- **create = 1.32M/s, SERIAL** — three single-thread passes on the spawner:
  template-memcpy (`mn_sched_init_fini.c.inc:716`), the `runloom_coro_bulk_init`
  coro-field loop, and the `Py_INCREF(callable)` loop (`:764`).
- **run = 2.07M/s, parallel** across the 8 hubs — already **1.6× above** create.

**Therefore every spawn lever except parallelizing `create` is arithmetically
capped at 804k.** This collapses the ranking: the run-phase and scheduler levers
cannot move spawn/s while create is the binding constraint. To cross 1M, create must
rise to ~2.0–2.85M → total 1.02M–1.20M.

## Ranked plan (adjusted predictions are the skeptic's, not the optimist's)

| # | lever | adj. spawn/s | conf | verdict |
|---|---|--:|---|---|
| **1** | **parallelize create** | **1.05M** | med | the ONLY lever that crosses 1M; it's the binding constraint |
| 2 | pure-C-entry fast lane | 950k | med | doesn't move the Python-callable number; best *instrument* for Wall #3 + a real C fast-lane |
| 3 | kill `sched_yield` idle-spin | 820k | low | the 38% is **unattributed** (likely steal-CAS spin, not nanosleep) — measure first |
| 4 | drain fast-path | 804k | low | run is already 1.6× above create → moves nothing; its real win (`RUNLOOM_NO_CTX_COPY`) already ships |
| 5 | full NUMA pipeline | 880k | low | mostly #1 re-skinned; run *degrades* cross-node (refcount wall) — fold node-affine first-touch into #1 |
| 6 | pool CPython per-fiber state | 815k | low | datastack already TLS-pooled, tstate shared-per-hub — already done; falsify with two zero-build probes |
| 7 | biased refcount (create-side) | 805k | low | **REJECT** — `RUNLOOM_GON_NOREFCNT` already exists; INCREF is ~0.18% of create |

**Combined realistic ceiling if the top levers stack: ~1.2M** (create→~2.85M on one
NUMA node's bandwidth × the unchanged 2.07M run).

## What the adversaries already falsified (don't re-do these)
- `RUNLOOM_NO_CTX_COPY` (drain) and `RUNLOOM_GON_NOREFCNT` (create-side refcount)
  **already exist** — A/B them to log the null result, don't re-implement.
- Hubs are **not idle-dispatchable** at `fiber_n` time — they run `runloom_hub_main`
  continuously (rest in `cond_timedwait`/`nanosleep`); waking them puts cond-wait
  latency **on the create critical path**. Parallel-create must use **dedicated
  builders or a pre-armed hub-side build descriptor**, not "dispatch to idle hubs."
- Datastack is already TLS-pooled (CAP 4096); per-g tstate is opt-in slow-mode.

## Levers the panel flagged as MISSED (candidates after #1)
- **Single-spawn path** goid atomic + slab alloc — the 363–456k path *real servers*
  use (vs the batch `fiber_n` the bench uses). #1 is benchmark-shaped unless servers
  call `fiber_n`. → also a **product decision**: reshape the public spawn API toward
  batch-launch?
- **Steal-backoff / bounded steal scan** — the smaller, lower-risk version of #3 if
  attribution shows the 38% is steal-CAS spin.
- **Patched-CPython attack on the 27k-instr floor** itself (Wall #3) — the only path
  *past* ~1.8M.

## Execution order (this is what gets run)
1. **Measure-first (zero risk):** decompose the three create passes at N=300k — which
   dominates? (`RUNLOOM_GON_NOREFCNT` isolates the INCREF pass for free.)
2. **Prototype parallel create:** a true barrier-gated parallel region (dedicated
   builder threads or pre-armed hub build descriptors that DON'T pay `cond_wait`),
   block-partition contiguous arena slices `[h·n/H, (h+1)·n/H)`, pure-C stores on
   builders, INCREF stays on the issuer. Measure create/s over `--hubs {1,2,4,8}` to
   find the single-NUMA-controller bandwidth saturation point; `bulk.py`'s
   `done-delta==n` is the correctness gate.
3. **Re-measure total** — does it cross 1M? Fold in node-affine first-touch (#5) as a
   variant if create saturates one node's bandwidth.
4. Then, only if create clears ~2M and drain becomes the wall: A/B `RUNLOOM_NO_CTX_COPY`.

_Results below as each step runs._

## Results — lever #1 (parallel create): **GOAL MET, 804k → 1.54M spawn/s**

**Step 1 — decompose create (N=300k, `RUNLOOM_GON_TIMING=1`):** the three serial
passes are NOT equal — Pass A dominates:

| pass | time | share | what it is |
|---|--:|--:|---|
| A — g-template loop | ~185 ms | **~72%** | memcpy g-template + fields + per-hub list link, into the **cold** g-arena |
| B — `coro_bulk_init` | ~45 ms | ~25% | coro structs + stack-arena fields |
| C — `Py_INCREF` loop | **0.1 ms** | ~0.04% | confirms the refcount lever (#7) is **dead** |

Pass A is **cold-write / cache-latency bound** (writing ~N×sizeof(g) of fresh
arena), so it parallelizes across cores' load/store + memory controllers.

**Step 2 — parallel Pass A (`RUNLOOM_GON_PCREATE=P`):** P builder threads each fill a
contiguous slice `[b·n/P,(b+1)·n/P)` into its own per-hub partial lists; the issuer
merges (list concatenation) after join. Pure-C disjoint-slice writes, INCREF stays on
the issuer. Pass A: **185 ms → 30 ms (6.4×)**.

**Step 3 — scaling (N=300k, 8 hubs, cores 16–23 = node0):**

| `PCREATE` | create/s | run/s | **total/s** |
|---|--:|--:|--:|
| 0 (serial) | 1.21M | 2.14M | 763k |
| 2 | 1.91M | 2.17M | 995k |
| 4 | 3.11M | 2.40M | 1.33M |
| **8** | **3.76M** | 2.62M | **1.51M** |
| 16 | 3.67M | 2.51M | 1.48M (over-subscribes 8 cores) |
| **8 + `NO_CTX_COPY=1`** | 3.72M | **2.64M** | **1.54M** |

**Outcome:** **1.54M spawn/s** — 1.9× the locked-in 804k, comfortably past 1M, and it
matches the harmonic-mean model (`create=3.72M, run=2.64M → 1.55M`). P=8 is optimal
(= the 8 node0 cores). At N=1M it's 1.21M (larger arenas cost more/fiber). Correctness:
`test_spawn_bulk_lifecycle` green 3/3 with PCREATE=8 (disjoint-slice writes + local
per-hub lists merged after `pthread_join` = no data race; serial fallback on
spawn/alloc failure).

**The constraint has flipped:** create (3.7M) now exceeds run (2.6M), so **run is the
new binding constraint**. `NO_CTX_COPY` (the pre-existing run lever the plan predicted)
took run 2.34M→2.64M. Parallelizing Pass B would raise create further but can't move
total while run-bound — the next magnitude is a **run/drain-phase** lever (the per-fiber
resume+complete cost = the CPython Wall #3, now exposed).

**Keep?** `RUNLOOM_GON_PCREATE=8` is a real, gated, correctness-checked **~1.9× win**
for the bulk fleet-launch path. Productionize: a **persistent builder pool** (the
prototype spawns fresh pthreads per call — negligible at large N, but a pool removes
even that for mid-N), then wire into `optimize("throughput")`.

### Then I chased the new wall (the run/drain phase) — it's the fundamental frontier
With create parallelized, **run is the binding constraint, and it does NOT scale** —
the opposite of create:

| hubs (= builders), N=1M, node0 | create/s | run/s | total/s |
|---|--:|--:|--:|
| 8 | 2.80M | 2.15M | 1.20M |
| 16 | 3.13M | **2.32M (peak)** | **1.33M** |
| 24 | 3.43M | 1.85M ↓ | 1.20M |
| 32 | 3.41M | 1.54M ↓↓ | 1.06M |

Create keeps scaling with builders (2.8→3.4M); **run *degrades* past 16 hubs** — the
signature of **cache-line contention**, not idle: more cores make it *worse*. I ruled
out the obvious suspect — immortalizing the shared callable (the per-fiber decref
bounce) was **noise** (+8%/−8%/−3% at 8/16/32 hubs). So the run wall is the
**scheduler's run-phase scaling**: the per-fiber `batch->live` atomic (one cache line
hammered by all hubs on every completion), the `__sched_yield` idle-spin during an
uneven drain (27% of run in the profile), and the per-fiber CPython lifecycle floor.

**Conclusion: ~1.5M is the achievable ceiling with the current run phase** (8 hubs,
parallel create). Going meaningfully past it is the **run-phase scaling project** —
shard `batch->live`, steal-backoff / kill the drain idle-spin, and ultimately the 27k-
instr CPython floor (Wall #3). That's `SCHEDULER_SCALING_FINDINGS.md` territory, a
larger effort than the single create lever — and the next real frontier.

## Results — parallelize_passB: **2.05M @8h / 2.23M @16h, PAST Go**

Parallelized Pass B (coro-fill, `RUNLOOM_GON_PCREATE_B`) like Pass A: Pass B 32ms→5.7ms,
create 4.3M→7.5M, total 1.54M→1.80M @8h(N=300k); **2.05M @8h / 2.23M @16h @N=1M — past
Go's 1.8M.** TSan-clean, gate-green, in `optimize("throughput")`. 32h degrades (1.22M) =
the run-phase `batch->live` contention (shard deferred; 8/16h already beat Go).

## Status
- **Goal met: 804k → 1.54M spawn/s** (1.9×, past 1M) via lever #1, parallel create.
- Run phase characterized as the next (fundamental, non-scaling) wall.
- Next: validate `PCREATE` concurrency under TSan; productionize (builder pool +
  `optimize("throughput")`); then the run-phase scaling project for the next magnitude.

# Closing the conn/s CPU gap to Go — measured, corrected, ranked

> **Note:** the "cdef ~110% — beats Go" headline below is from an older 2-core side
> run whose cdef figure used a since-superseded build, so treat the per-core
> decomposition here as indicative, not a current ranking. The acceptor architecture
> is now matched — both runloom and the Go baseline run N `SO_REUSEPORT` acceptors —
> so connection churn is a like-for-like comparison; the suite churn section reports
> **parity** (~75–78k conn/s, client-bound).

The passive (server) story: at ~30k conn/s runloom and Go match on *throughput*, but
the `--quick` churn run showed runloom at ~51% server CPU vs Go ~16% — the "~3× CPU"
gap. This doc is the code-verified decomposition + a vetted plan to close it.

> **Two corrections up front (the adversarial design panel caught both — the same
> discipline that caught the QSBR mis-attribution):**
> 1. **There is NO `sched_yield` busy-spin.** My first profile (a `--hubs 1` strace)
>    showed `sched_yield` dominant and I framed it as "runloom spins where Go sleeps."
>    Wrong. On the real **multi-hub** server, idle hubs **block in `clock_nanosleep` /
>    `epoll_wait(1ms)`** (70% of syscall *time* is sleeping); `sched_yield` is 0.51%.
>    The hubs sleep. Grep confirms no `sched_yield` in the M:N hub idle loop.
> 2. **The "3× CPU" is not a clean apples-to-apples number.** The `--quick` run measured
>    runloom at 128 dialers vs Go at 16 dialers (different ladder rungs) — and runloom
>    was client-bottlenecked (server 50% idle). A clean same-operating-point comparison
>    is still owed (see "first experiment").

## Measured per-connection cost (multi-hub strace + 2-core saturated perf)

**Syscalls per connection: runloom ~10 vs Go ~6.** The 4 extra are addressable:

| syscall | runloom/conn | Go/conn | note |
|---|--:|--:|---|
| accept | 1 | 1 | Go uses `accept4` (flags baked in) |
| **fcntl (set_nonblock)** | **2** | **0** | a blocking accepted fd → GETFL+SETFL; `accept4(SOCK_NONBLOCK)` removes both |
| setsockopt (TCP_NODELAY) | 1 | 1 | shared |
| recvfrom (data + EOF) | 2 | ~2 | shared (echo reads to EOF) |
| sendto | 1 | 1 | shared |
| **mincore** | **1** | **0** | the resident stack-scrub (now default) does a syscall per teardown |
| close (FIN teardown) | 1 | 1 | **shared kernel** — ~31% of busy CPU; runloom issues NO extra `epoll_ctl(DEL)` (relies on kernel fd-auto-remove, like Go) |
| sched_yield / epoll / nanosleep | ~few | ~few | idle rest + park; **sleeping, not spinning** |

Busy-CPU hotspots (2-core saturated perf): `close` FIN path ~31% (shared kernel,
loopback-softirq-amplified), **`memset` 4.65%** (the handler's `bytearray(65536)`
zeroed *per connection*), **`mincore` 2.7%** (the scrub), `nft_do_chain` 3.5% (netns
artifact, gone on a real NIC), syscall entry/exit tax ~6%.

## Addressable vs irreducible (the honest floor)

**Most of the 35-point gap (51→16) is IRREDUCIBLE at this workload:**
- `close()` FIN teardown (~31% busy) — both runtimes pay it; runloom adds nothing.
- loopback softirq + `nft_do_chain` (~3.5%) — netns artifacts, **discount on real NIC**.
- the syscall entry/exit tax on the syscalls **Go also issues**.
- **the Python-handler tax** — a CPython frame dispatched per round-trip
  (`PyObject_CallFunctionObjArgs` + a per-conn `functools.partial` + free-threaded
  refcount churn). Go runs compiled goroutine code. **This is several points and is
  structural: it does not go away while the handler is Python.**

**Addressable (runloom-specific), ~10–16 points total, all small + diffuse:**

| # | lever | adj. pts | what |
|---|---|--:|---|
| 1 | acceptor-local spawn + idle-pump 1ms→5ms | ~9 (low conf) | pin the handler to the acceptor's hub (kill ~30k/s cross-hub wake kicks); widen the idle pump cadence. **NOT** a true `-1` block — the `:767` pump branch has no `sub_head` re-check, so the timeout IS the lost-wake backstop; keep it TIMED. |
| 2 | shrink the 64KB per-conn buffer → 4KB | ~1 (compounds) | `runloom_epoll_py_tcpcon.py` allocs+zeros 64KB per conn to echo 1KB. 1-line fix. |
| 3 | `accept4(SOCK_NONBLOCK\|SOCK_CLOEXEC)` | ~1 | drop the 2 redundant `fcntl`/conn; mirrors the all-C acceptor that already does this |
| 4 | drop the per-conn `functools.partial` | ~1 | pass `conn` as the fiber arg directly; remove a per-conn heap alloc + dispatch |
| 5 | scrub: SP-watermark instead of `mincore` | ~1 | keep the secure wipe but track touched depth, no per-teardown syscall (or `RUNLOOM_STACK_SCRUB=0` for servers that hold no secrets) |

**Realistic floor: ~35% server CPU** (vs Go 16%) for the **Python-handler** tier.

## The structural lever (the one that actually approaches Go)

The single biggest "toward Go" move is **not** in the scheduler — it's the **handler
tier**. The Python frame per round-trip is the irreducible runloom-vs-Go cost; the repo
already ships compiled tiers (`runloom_cdef` = tstate-free c_entry Cython handler,
`runloom_cython`) that remove it. **To match Go's CPU/conn you serve with a compiled
handler, not a Python one** — that is what the `cdef`/`cython` server tiers are for, and
the right comparison is *go vs runloom_cdef*, not *go vs runloom_c (py handler)*.

## Plan
1. **First (mandatory): a clean same-operating-point measurement** — strace -c + perf
   the EXACT default server under `conn_churn.py` at the dialer rung that lands ~30k
   conn/s, ≥3 reps, real-NIC or netns (discount `nft`), and measure runloom_c (py
   handler) AND runloom_cdef (compiled) vs go at the *same* dialers. This replaces the
   muddy `--quick` 51%-vs-16% with an honest number, and tells us how much the compiled
   tier alone closes.
2. Land the cheap, zero-risk addressable wins (#2 buffer, #3 accept4, #4 drop partial) —
   each ~1 pt, trivial, correctness-neutral; they compound once #1 lands.
3. Acceptor-local spawn + idle-pump widen (#1) — the largest single lever, gated,
   strace-proven, TIMED backstop (never `-1`).
4. Report runloom_cdef vs go as the real "passive CPU" headline.

## Results — the clean saturated comparison (this overturns the "3× CPU")

The `--quick` "51% vs 16%" was never a real number: different ladder rungs (runloom
@128 dialers vs go @16) and the runloom servers were **client-bottlenecked** (server
~50% idle). So I forced the server to be the bottleneck — pinned every server to the
**same 2 cores**, hit it with a 16-core client, ran each **separately**, and confirmed
saturation (CPU → ~200%).
[`conn_compare.py`](../../tests/experiments/spawn_perf/conn_compare.py), loopback, 1KB payload:

| server (2 cores, saturated, ~198% CPU) | conn/s per core | vs go |
|---|--:|--:|
| **runloom_cdef (compiled handler)** | **8,538** | **110% — BEATS go** |
| go (GOMAXPROCS=2) | 7,783–7,886 | 100% |
| runloom_c (Python handler) | 7,077–7,288 | **~91% of go** |

(conn/s on 2 saturated cores; per-server runs, ~±5% run-to-run. The 3rd server in a
sequence hits ephemeral-port/TIME_WAIT exhaustion from the prior churn bursts — run
each first/fresh to avoid the "no listen" artifact.)

**So runloom's conn/s CPU efficiency is ~93% of Go's — within ~7%, not 3× worse.**
The "3×" was a measurement artifact (unsaturated, client-bound, mismatched rungs).
With the *Python* handler, on the same saturated cores, runloom does **93% of Go's
connections per core**. (The cdef/compiled tier should close most of the remaining 7%;
its prebuilt `.so` is stale vs the recent ext rebuilds — rebuild `handler_cdef` to
measure it.)

**Architectural note found along the way:** `runloom_c.serve()` requires **≥2 hubs**
(the M:N model needs a hub to host the accept loops), so runloom has a 2-core serving
floor where Go runs on 1. Not a CPU-efficiency issue, but worth knowing.

**Bottom line:** the passive (conn/s) story is far better than the muddy first run
suggested — runloom is ~Go-competitive per core. The addressable micro-wins (accept4,
4KB buffer, drop the partial, scrub mincore) would chip at the last ~7%, and the
compiled handler tier is the way to fully close it; none of it is the "fix the
scheduler, it's 3× off" story the bad measurement implied.

_status: clean saturated comparison done — runloom_c ≈ 91% of Go, cdef 110%._

### Micro-wins — ran them, mostly dead ends (honest)
- **accept4** (drop 2 fcntl/conn, +CLOEXEC): **throughput-neutral when saturated** —
  13.5k vs 13.9k conn/s on/off (noise). The fcntls weren't the bottleneck; freed
  microseconds get re-absorbed on a saturated core. **Kept** (default-on) for the
  CLOEXEC correctness + fewer syscalls, but it is **not** a throughput win.
- **64KB handler buffer**: **symmetric with Go** — `go_netpoll_native_net.go` also does
  `make([]byte, 65536)` per connection. So it's apples-to-apples, *not* a
  runloom-specific cost; shrinking runloom's alone would be cheating the comparison.
  No change.
- **per-conn `functools.partial`**: the **only** genuinely runloom-specific passive
  overhead (Go does `go handleConn(c)` directly; runloom builds a partial per conn in
  `module_io.c.inc`). Small (a heap alloc), likely also saturated-throughput-neutral.

**Conclusion: there is little runloom-specific passive overhead left to remove.**
runloom is already Go-competitive on conn/s (91% Python tier, 110% compiled tier); the
remaining ~9% on the Python tier *is* the interpreter handler tax, and the **cdef
compiled handler removes it and beats Go**. That's the real passive answer, not the
micro-wins.

# io_uring & thread-state: what this benchmark actually found

This is the consolidated record of a deep investigation that ran alongside the
benchmark suite. It corrects an earlier, wrong conclusion ("io_uring loses on
loopback") and lands on a concrete, measured win plus a buildable next step.
Numbers are on the box in `report.html`'s header (Xeon E5-2696 v3, 64 vCPU,
free-threaded 3.13t, veth across two netns).

## TL;DR

- **io_uring's loop backend WINS on this loopback-class path — when driven
  through the Stage-2 proactor.** Routing the Cython handler's recv/send through
  `runloom_iouring_loop_recv/send` took it from **439k req/s (server-bound)** to
  **639k (client-bound), 1.16M server-ceiling** at 1 KiB — the fastest runloom
  config measured, **+40% peak / +2.17× server ceiling** vs the same handler on
  epoll, at ~half the server CPU (85% → 55%).
- The earlier "io_uring loses" was an **artifact of driving it wrong**: the
  io_uring tiers used the *readiness* path (`recv()` + `wait_fd_coop` + the
  epoll→ring bridge) — io_uring's overhead with none of its win.
- The famous **"+20% over epoll"** reference is real but **conditional** (8-byte
  payload + a tstate-free `c_entry` fiber) and its writeup mis-attributed the
  mechanism. Magnitude is setup-dependent: **+6%** ceiling for an 8-byte all-C
  echo (epoll already near-optimal there), **+117%** ceiling for a 1 KiB Cython
  handler (epoll baseline has more to reclaim).

## Five execution models in a trenchcoat

The reason this was hard: runloom stacks several models whose assumptions
disagree, and the surprising behavior lives at the *seams*, not in any one model:

1. **reactor (epoll)** "tell me when ready, I'll syscall" vs **proactor
   (io_uring)** "do the syscall, tell me when done."
2. **Go-style M:N stackful coroutines** grafted onto **CPython's one-tstate-per-
   thread** model (hence the per-fiber tstate snapshot, or per-g tstate).
3. **free-threaded CPython** underneath: atomic refcounts + cross-thread merges +
   STW GC — a per-core tax a single-threaded event loop never pays.
4. **handler dispatch**: a Python callable (full Python fiber + tstate) vs a
   `c_entry` C function pointer (no Python state at all).
5. even inside io_uring: **single-shot inline-drain** (`ring_do`) vs **deferred
   batch, always-park** (`loop_io`).

Every pairwise model is documented; nobody documents the cross product. "io_uring
always-parks" is cheap under `c_entry` and expensive under a tstate; "batching
wins" is true at 8 B and irrelevant at 1.5 MB; "free-threading scales" is true in
absolute and false per-core.

## The io_uring loop backend: how it actually works

`RUNLOOM_IOURING_LOOP=1` replaces the hub's *block primitive* — the hub blocks
directly in `io_uring_submit_and_wait_timeout` instead of `epoll_wait`
(`io_uring_l_loop.c.inc`). The recv/send themselves are submitted as SQEs.

- `runloom_iouring_loop_recv/send` → **`runloom_iouring_loop_io`** (l.450):
  `enqueue_op` (deferred, **no per-op syscall**) then **always** park
  (`runloom_sched_park_current` + `runloom_coro_yield`). It does NOT inline-drain.
- The **win is batching** (l.415 comment): the hub's single
  `submit_and_wait_timeout` submits *all* parked connections' SQEs and reaps
  *all* ready CQEs at once. N connections → ~1 syscall instead of N.
- The +20% writeup credited an **inline-drain that skips the park** — that is a
  *different* function, `runloom_iouring_ring_do` (`io_uring_l_ring.c.inc:291`),
  not the path `loop_recv` takes. So the mechanism in that writeup is wrong even
  though the number was real.

### Why my original io_uring tiers lost

The capi (`runloom_tcpconn_c_recv_into`) under `RUNLOOM_IOURING_LOOP=1` but
without the per-conn `RUNLOOM_TCPCONN_IOURING` fell through to plain `recv()` +
`runloom_netpoll_wait_fd_coop`. That's a *readiness* recv plus an io_uring-
**bridged** park (eventfd → epoll → pump → drain), i.e. io_uring's bookkeeping on
top of an epoll-equivalent recv. Strictly worse than plain epoll.

### The fix

`runloom_tcp_capi.c.inc` now checks `runloom_iouring_loop_enabled()` first and, if
a hub ring exists, drives recv/send through `runloom_iouring_loop_recv/send` (the
proactor) — exactly mirroring the all-C echo. This is what flipped tiers 4/5 from
loss to win. (Side effect: tier 5's `optimize` multishot — which the all-C echo
itself flags as "slower on loopback" — is now superseded by the proactor.)

### Measured (iouring_compare.py, 3 reps, bootstrap-CI ladder)

| test | epoll | io_uring | peak | server ceiling |
|---|--:|--:|:--:|:--:|
| 8-byte all-C echo (`handler=None`, `c_entry`) | 654k (client) | 659k (client) | +1% | 1.14M → 1.21M (+6%) |
| 1 KiB Cython handler | 455k (server) | 639k (client) | **+40%** | 533k → 1.16M (**+2.17×**) |

## thread-state: the "omit it when there isn't one" optimization

`runloom_g_entry` (`runloom_sched_core.c.inc:302`) has a **C-only entry**: a fiber
spawned with a C function pointer (`g->c_entry != NULL`, via
`runloom_mn_fiber_c`) **skips all Python-frame and tstate setup** — no datastack,
no frames, no snapshot. The all-C echo uses this; that's the *other half* of its
speed (alongside the 8-byte payload). A Cython handler does **not** — `serve()`
spawns a Python callable as a full Python fiber, so it carries a tstate and pays
`runloom_tstate_save/restore` on every `loop_io` park.

Default tstate mode is **per-hub snapshot** (`runloom_resolve_migratable_mode()`
returns 0). The per-g full `PyThreadState` mode (`RUNLOOM_PER_G_TSTATE`) is
**gated off** (a per-g tstate's mimalloc heap migrating across hubs SEGVs under
churn). Measured per-fiber empty RSS:

| mode | B/fiber | 1M fibers |
|---|--:|--:|
| default (per-hub snapshot) | ~8,800 | 8.8 GB |
| per-g full PyThreadState (gated) | ~26,700 | ~26.7 GB |
| Go goroutine (reference) | ~2,700 | 2.5 GB |

So a full per-fiber `PyThreadState` is **~18 KB/fiber**. "Skipping the tstate"
(`c_entry`) saves that whole 18 KB in per-g mode; in the default snapshot mode it
saves the datastack-chunk + frame-faulting portion of the 8.8 KB.

## Why no handler optimization could win here: echo has no handler CPU

The deepest finding, and the one that explains every "tie" below: **a TCP echo
does essentially no CPU work in the handler** — it shuttles bytes from `recv` to
`send`, with no parse, compute, or serialize step. `perf` confirms it: both the
Python and Cython servers are **kernel-TCP-send-bound** (~37% in `__send`,
handler code <1% self). So the handler *implementation* is irrelevant here —
Python `def`, Cython `def`, and tstate-free `cdef` all do ~nothing, and the CPU
is in the kernel I/O path regardless.

That is why **every handler-side optimization ties on echo**: the zero-PyObject
capi, the tstate bypass, the `c_entry` path — there is no handler compute for
them to reduce. The only optimization that moved the needle (the io_uring
proactor) helps the **I/O path**, not the handler. To demonstrate a handler
optimization you need a **CPU-bound handler** (HTTP/JSON parse, hashing, a real
protocol), where the handler actually burns cycles; echo is precisely the
workload where they are guaranteed to be equal. Read the tiers below with that
lens: they measure the I/O path, not the handler.

**This prediction is now demonstrated.** `WORK_CURVE_EXPERIMENT.md` gives the
handler real CPU via a `--work N` knob (an FNV byte-hash, pure inline arithmetic).
At `work=0` the handler skips the work and an interpreted vs a fully-native
handler are dead-even (0.99×, reproducing echo) &mdash; the built-in cross-check.
As the knob grows the interpreted handler collapses while the fully-native Cython
handler (work inlined, zero-PyObject) holds nearly flat: gap **164× at work=64**.
And cross-runtime (per core), that fully-native handler **matches-to-beats Go**
&mdash; ahead through ~work 4 (runloom's faster I/O), within ~8% at heaviest
compute &mdash; with `cython ≈ cdef` re-confirming the tstate bypass buys ~nothing
on throughput. So the runtime (I/O + scheduler) is the solved, Go-beating part;
the only residual cost is the interpreter, which a compiled handler removes.
(Honest caveat: delegate the work to a C lib and every runtime re-converges.)

## Built: the cdef / c_entry handler tier (an honest negative result)

The `serve()` C-handler path is **built**: it now accepts a `runloom_c.c_handler`
PyCapsule wrapping a `cdef` C function and spawns it via `runloom_mn_fiber_c` →
the `g->c_entry` path (no Python frame, no per-park tstate). The plumbing:
`runloom_tcp_capi.{h,c.inc}` exposes raw-fd tstate-free `fd_recv`/`fd_send_all`/
`fd_close` (proactor or readiness) via the `__tcp_capi__` capsule;
`module_io.c.inc` adds the capsule dispatch + a parameterized C acceptor;
`servers/handler_cdef.pyx` is a `nogil` cdef handler; tier `runloom_iouring_cdef_tcpcon.py`.

**The tstate-bypass buys ~nothing on throughput** (measured, `iouring_compare.py`):

| payload | `cython` (tstate) ceiling | `cdef` (tstate-free) ceiling | delta |
|---|--:|--:|--:|
| 8 B (op-bound) | 1,166,138 | 1,169,111 | +0.25% |
| 1 KiB (I/O-bound) | 1,131,765 | 1,158,341 | +2.3% |

Within noise at **both** payloads — even at 8 bytes, where the per-park tstate
cost should bite hardest. The reason is a *good* one for runloom: the default
per-hub **snapshot** tstate mode is already cheap (`runloom_tstate_save/restore`
snaps a few recursion-depth ints, not a `PyThreadState`), so there's almost
nothing to bypass; the expensive per-g `PyThreadState` (~18 KB/fiber) is gated
off. **Takeaway: use the Cython `def` handler (`handler_cy`) on the proactor** —
as fast as the tstate-free `cdef` path with far less ceremony. The `cdef`/
`c_entry` path is a working, correct feature whose remaining value is per-fiber
*memory* (no datastack/frames), not speed.

## The "anomaly" — investigated to ground, RESOLVED: there is none

Arc (kept because the dead ends are instructive): the matrix showed
`runloom_c_cython` (Cython on epoll) **server-bound at ~425k (86% CPU)** while
`runloom_c` (Python on epoll) was **client-bound at ~620k (63% CPU)** — an
apparent ~2× server-CPU-per-request gap for a *no-work* echo, where the Cython
handler is the *leaner* one. A first loopback A/B looked equal → I called it an
artifact; a fresh netns re-measure reproduced the 425k → I **retracted** that and
reopened it. Two hypotheses were on the table: (a) a hidden shared `PyObject`
refcounted/locked per request, contending across hubs only at saturation; (b) the
64 KB stack buffer's cache footprint.

Settled by two clean tests:

1. **Saturation (2048 conns, netns, the load the ladder never reached):**
   cython **604,946** rps ≈ py **598,527** rps — **equal.** The cython server was
   never walled at 425k; "server-bound 425k" was the **ladder's plateau/bottleneck
   label misfiring at sub-saturation** (512–1024 conns), reproduced because the
   re-measure hit the same conn-count, not a real wall.
2. **Instruction level (`objdump`):** `runloom_tcpconn_c_send_all` (cython capi) =
   **115 insns, ZERO `Py_` calls** (just `send`/`wait_fd`/io_uring/`errno`);
   `RunloomTCPConn_send_all` (py method) = 134 insns with `PyArg_ParseTuple` +
   `PyBuffer_Release`×4 + `PyLong`×2 + `PyErr`×3. **No hidden Python object** in
   the Cython path — it is strictly leaner.

Saturation profiles are identical kernel-TCP-send proportions; the only lock/atomic
frames are kernel TCP spin locks (`_raw_spin_lock_irqsave` 0.87% in *both*) plus a
tiny futex delta (cython 3.0% vs py 2.1% = cross-hub park/wake). **Both hypotheses
ruled out; the handlers are equal.** The 2× was a measurement artifact.

**Lesson banked into the suite:** a "server-bound" peak isn't real until you push
connections past the loadgen's knee — confirm by saturating, not by trusting the
ladder's bottleneck label or a client-bound server's extrapolated ceiling. Full
trace: `results/anomaly_notes.md`, `results/sat_top_{cython,py}.txt`.

# Runloom Benchmark Suite — Original Specification (archived verbatim)

> Archived from the user's original prompt for reproducibility. This is the
> source-of-truth spec the benchmark suite implements. Decisions and deviations
> agreed during scoping are recorded at the bottom under "Scoping decisions".

---

You are building a benchmark suite for Runloom. The project will be saved in a directory in the Runloom repo called benchmark. There will be different benchmark tests for aspects of the run time.

Requirements:
	- pin all programs to CPU cores
	- ensure debug mode is not used on any programs (as it interferes with benches)
	- use fresh netns to bypass firewall throttling for tests
	- system wide changes before bench ensure:
sysctl -w net.ipv4/tcp_wmem="4096 16384 2097152"  # 2 MB max
sysctl -w net.ipv4/tcp_rmem="4096 87380 2097152"
	- ensure fd limit can handle millions of connections
	- programs launched from vs code shell inherit limits -- ensure you take this into consideration

Performance benchmark: testing the requests / second of [runloom in various configurations, go]

Client:
	- Language = go
	- process cores = int(os.cpu_count() * 0.25)
	- fixed 1.5 mb buffer initalized at start
	- disable naggle for client socks
	- method:
		- num_no cons are connected to server
		- the buffer is sent to the server and read back
	- methodology:
		- start with low con no
		- measure req/s
		- continue to raise until req/s doesn't increase
		- return the max req/s as the result for the server

Servers:
	1. runloom default epoll/kqueue/wsa (zero optimized / bad):
		- uses wrapped python calls, no direct C calls, and python objects
		- hubs = int(os.cpu_count() * 0.7)
		- method:
			listener = runloom.sync.tcp_listen("127.0.0.1", 9000)
			while True:
				conn, _ = listener.accept()
				runloom.go(handle, conn)

	2. Runloom default epoll/kqueue/wsa (C optimized):
		- pros: uses C calls so should be faster than the sync / runloom api wrappers
		- cons: still a regular python func handler (no CPython.)
		- hubs = int(os.cpu_count() * 0.7)
		- method:
			def main():
				# C server scaffold: listen, accept, spawn all in C
				# Returns (bound_port, [listener_objects])
				port, listeners = runloom_c.serve(
					host="127.0.0.1",
					port=9000,
					handler=handle,
					acceptors=hubs,      # hubs accept-loop fibers (one per core, with SO_REUSEPORT)
					backlog=128
				)

				print(f"Server listening on {port}")
				# Server runs indefinitely; close listeners to stop
				try:
					runloom.sleep(float('inf'))
				except KeyboardInterrupt:
					for ln in listeners:
						ln.close()

			runloom.run(hubs, main)

	3. Runloom using io_uring (very optimized)
		- less security than default backends
		- python handlers still used here
		- hubs = int(os.cpu_count() * 0.7)
		- same code as 1. "runloom default zero optimized" but RUNLOOM_IOURING_LOOP=1

	4. Runloom using io_uring and Cython handlers
		- insecure but fast, handlers completely portable to C
		- hubs = int(os.cpu_count() * 0.7)
		- same code as 2 but handlers use Cython and RUNLOOM_IOURING_LOOP=1 backend.
		- this is VERY important:
			- the actual handler function for the call needs to be written in such a way that it doesnt output any Python objects otherwise you lose all the speed improvements. this needs to be confirmed by looking at the assembly directly. i will provide sample code.

			import os
			import socket as _sk
			import cython
			import runloom
			import runloom_c

			PORT   = 8080
			hubs   = int(os.cpu_count() * 0.7)
			CHUNK  = 1 << 18
			RCVBUF = SNDBUF = 1 << 20
			stop   = bytearray(1)

			def handler(conn):
				buf = bytearray(CHUNK)
				mv  = memoryview(buf)
				n: cython.int
				stop_flag: cython.bint

				while True:
					stop_flag = stop[0]
					if stop_flag:
						break
					n = conn.recv_into(buf, CHUNK)
					if not n:
						break
					conn.send_all(mv[:n])

			def root():
				port, listeners = runloom_c.serve("0.0.0.0", PORT, handler,
												  acceptors=hubs, backlog=4096)
				while True:
					runloom.sleep(3600)

			runloom.run(hubs, root)

	5. Runloom using io_uring and Cython handlers
		do the same as 4 but run optimize(throughput)

	asyncio
		- single threaded
		- use basic canonical python 3 asyncio protocol handlers

	gevent
		- you decide
	uvloop
		- same asyncio
	go
		- cap go's cores to int(os.cpu_count() * 0.7)

Speed benchmark
	- heading row: [runloom, go, asyncio, greenlet, uvloop]
		- time to spawn 1 mil empty fibers / coroutines
		- context switching time
		- http reqs / second (against a go server with int(os.cpu_count() * 0.7) cores
		- network TCP overhead (round trip to loopback) to a go server.

memory benchmark
	- heading: state: [go, runloom python handler, runloom python handler with optimize(memory), runloom c handler]
		- empty just spawned
		- spawned with a socket
		- 1 million

	- make it clear exactly that its used memory, not virtual memory ranges or w/e.

reporting:
	- scale everything down to 1 core e.g. asyncio stays the same but any multi-core programs get divided by core_no
	- sort by best performing to least
	- record hardware / os details for the machine that ran the bench at the top

save all results and code to be reused in the benchmark folder. i want a single consolodated view of all the data in a html file and also generate the relevant sections of the current read me (use a subset of the full data -- choose most relevant.) I want the read me sections to have footnotes that link back to the detailed benchmark html page in the benchmark directory. and the html page should be able to load code portions of the bench mark programs as well as the assumed constraints to produce reliable data. I'd also like it if the html page linked back to the backend profiling work that should already be saved in the pre-existing benchmark dir for (linux, mac, windows.) Save this full text prompt into prompt for archiving.

If you have any ideas for improvements to the benchmark that might improve accuracy before proceeding please interrupt me for ideas / discussions.

---

## Scoping decisions (agreed during setup, 2026-06-19)

1. **req/s payload (Q1 → a):** the headline **req/s** metric uses a **small payload (64 B–1 KB)** so it measures scheduling/syscall overhead, not loopback memcpy bandwidth. The **1.5 MB** buffer is kept as a **separate "bandwidth (GB/s)"** metric. Both are reported, clearly labelled.
2. **Zero-PyObject Cython tier (Q2 → b):** servers 4 & 5 use a **real Cython handler** that `cdef extern`s Runloom's cooperative recv/send as **plain C functions** and calls them **directly** (compiled-in C, no `PyObject_CallMethod`), pulling the connection's C handle once at entry. This requires a small addition to `runloom_c` exposing a C-level recv/send API. The `handler=None` all-C echo is *not* used for tiers 4/5 because it does not demonstrate a Cython-handler optimization. Zero-PyObject in the hot loop is proven via `objdump` of the compiled `.so`.
3. **Topology (Q3 → a):** client and server run in **separate network namespaces joined by a veth pair**, pinned to **disjoint NUMA nodes**. This isolates client/server contention and avoids the loopback fast-path that hides Runloom's io_uring win. Spec sysctls are applied **inside** the server netns (they are namespaced).
4. **Single-threaded baselines (Q4 → a):** asyncio / uvloop / gevent run on **GIL-enabled CPython 3.13** (their best case — no free-threaded atomic-refcount tax); Runloom and Go run on **free-threaded 3.13t**. gevent is installed on the GIL build. The interpreter/build is labelled per row. (Raw `greenlet` for the speed microbenchmark is already present and runs as-is.)
5. **Per-core normalization:** per-core figures are the **saturated M:N throughput divided by hub count** (and Go divided by GOMAXPROCS); single-threaded runtimes are already 1 core. We do **not** measure `run(1)` as "Runloom per core" — that is the M:1 cooperative scheduler, a different runtime than the M:N work-stealer. Raw saturated numbers are shown next to the divided ones so scaling efficiency is visible.
6. **TCP_NODELAY:** set once on listener + client socket templates (uniformly across every backend, outside the per-request hot loop), not via per-connection `setsockopt` in handler code — so it cannot creep in as per-call overhead. (Linux has no kernel-global nodelay sysctl.)
7. **Build:** as-shipped **release** build (`-O2`, fortify on), `RUNLOOM_DEBUG` unset and verified, no sanitizers. Built and run on `~/.pyenv/versions/3.13.13t` (free-threaded, Cython 3.2.5 present).
8. **Validity guards baked in:** geometric connection ladder with a rigorous
   stop rule (a rung "improves" only if its median req/s beats the incumbent
   peak's bootstrap-CI **upper** bound; `PLATEAU_PATIENCE` consecutive misses end
   the sweep), `REPS` independent reps per rung with a nonparametric bootstrap CI
   of the median, full curve + p50/p99/p99.9 reported, and per-core CPU
   utilisation (from `/proc/stat`, server-core group vs client-core group)
   recorded every rung to attribute the bottleneck.

## Implementation-refined behaviours (discovered while building, 2026-06-19)

These refine the decisions above based on what the real runtime/box required:

9. **The spec's API names were idealised; the real ones are used:**
   `runloom.go` -> `runloom.fiber`; `conn.send_all` is the C `TCPConn` method
   (the `runloom.sync.Socket` facade uses `sendall`); `runloom.run(n, main_fn)`.
   `runloom_c.serve`, `runloom.optimize("throughput")`, and
   `RUNLOOM_IOURING_LOOP=1` are all real and used as written. Debug is the
   `RUNLOOM_DEBUG` env var (default build is `-O2 -DNDEBUG` release), not a build
   flag -- the suite clears it and `env.py` records the proof.

10. **Zero-PyObject Cython handler is delivered via a new C-API (decision #2b):**
    `src/runloom_c/runloom_tcp_capi.{h,c.inc}` exposes
    `runloom_tcpconn_c_recv_into` / `_send_all` (the same epoll/io_uring
    cooperative core as the `TCPConn` methods, no Py_buffer/PyArg/PyLong), handed
    to the Cython module through the `runloom_c.__tcp_capi__` PyCapsule. The
    handler is built `freethreading_compatible=True` (or importing it would
    silently re-enable the GIL and kill M:N). `disasm_check.sh` objdumps the
    handler's implementation function and asserts the per-request loop is exactly
    two indirect capi calls + a stack-canary epilogue -- zero PyObject traffic.

11. **The 16-core client (spec: int(cpu*0.25)) cannot saturate the fast servers**
    for a symmetric echo (client does the same recv/send work as the server).
    So at the plateau the *client* is often CPU-bound, not the server. Rather than
    violate the spec's client budget, every rung records both CPU groups, the
    peak is tagged `bottleneck_at_peak` (server/client/neither), and when not
    server-bound a **`server_ceiling_est = peak_rps / server_cpu_util`** is
    reported (extrapolating the server's measured utilisation to 100%). This keeps
    the fast tiers distinguishable instead of all flat-lining at the client's
    limit, and is labelled an estimate, not a measurement.

12. **Topology is concrete (decision #3a):** two real netns (`rl_srv`, `rl_cli`)
    joined by a veth pair (`10.99.0.1` <-> `10.99.0.2`), passwordless `sudo`
    confirmed on the box. The 2-NUMA split is node0=cpu0-31, node1=cpu32-63;
    client pins to cpu0-15, server to cpu16-59 (disjoint, documented NUMA span).
    Spec sysctls (`tcp_wmem`/`tcp_rmem` 2 MB max) are set inside the server netns;
    `ip_local_port_range` is widened in the client netns so >32k connections fit
    under the ephemeral-port ceiling; `fs.nr_open`/`vm.max_map_count`/`somaxconn`
    kernel ceilings are raised; `RLIMIT_NOFILE` is raised per-exec via
    `prlimit` inside the netns (the editor-shell 4096 cap does not propagate).

13. **req/s vs bandwidth use different ladders:** the small-payload req/s metric
    walks connections up to 32768 (scheduling-bound); the 1.5 MB bandwidth metric
    uses a short ladder (1..128 connections) because a handful of streams already
    saturate the path -- and a deep ladder at 1.5 MB/conn would blow out memory.
    The Go loadgen establishes all connections in **parallel** (capped) so a
    high-connection rung does not serialise into a multi-second ramp.

## io_uring & thread-state investigation (2026-06-19, full record in `../IOURING_TSTATE_FINDINGS.md`)

14. **io_uring's loop backend WINS here — once driven through the Stage-2
    proactor.** The first cut showed the io_uring tiers *losing* (runloom_cython
    439k, server-bound) because the capi fell through to the *readiness* path
    (`recv()` + `wait_fd_coop` + the epoll→ring bridge): io_uring's bookkeeping
    with none of its win. The fix routes the capi through
    `runloom_iouring_loop_recv/send` (the proactor) when `RUNLOOM_IOURING_LOOP`
    is on. Result: runloom_cython 1 KiB went **439k → 639k (client-bound), server
    ceiling 533k → 1.16M** = +40% peak / **+2.17× ceiling**, at ~half the server
    CPU. So "io_uring loses on loopback" was an artifact of mis-driving it; the
    corrected suite reports io_uring as a major win via the proactor.

15. **The "+20% over epoll" reference reconciled.** Real but conditional: it's an
    **8-byte ping-pong** on a **tstate-free `c_entry`** fiber, and its writeup
    mis-attributed the mechanism (it credits an inline-drain skip-park in
    `ring_do`; the all-C echo actually calls `loop_io`, which *always* parks —
    the real win is **batching** one `submit_and_wait_timeout` across N parked
    conns). Magnitude is setup-dependent: **+6%** ceiling for the 8-byte all-C
    echo (epoll already near-optimal), **+117%** ceiling for the 1 KiB Cython
    handler. Measured head-to-head in `suite/iouring_compare.py` (8B `handler=None`
    + 1 KiB Cython, each epoll vs `RUNLOOM_IOURING_LOOP=1`).

16. **The tstate "omit-if-absent" optimization (`g->c_entry`) and its cost.**
    `runloom_g_entry` skips ALL Python-frame/tstate setup for a fiber spawned via
    `runloom_mn_fiber_c` (a C function pointer) — the all-C echo's other speed
    half. A Cython handler is a Python callable, so it gets a full Python fiber +
    tstate and pays `tstate_save/restore` on every proactor park. Default tstate
    mode is **per-hub snapshot** (no per-fiber `PyThreadState`); the per-g mode is
    gated off (mimalloc-heap migration SEGV). Measured: a full per-g
    `PyThreadState` is **~18 KB/fiber** (26.7 KB vs 8.8 KB snapshot vs 2.7 KB
    go-goroutine). **Buildable next step**: a `cdef`/C-pointer handler path for
    `serve()` (capsule → `mn_fiber_c` → `c_entry`) would give a *custom* handler
    the tstate-free fast path — should top even the proactor Cython tier on CPU
    and be lighter per fiber. Evidence backs it; not yet built.

17. **Open anomaly (reproducible):** the Cython handler on *epoll* is server-bound
    at 455k (533k ceiling) — *slower* than the Python handler on epoll (988k
    ceiling), despite being zero-PyObject and zero-alloc. It **vanishes under the
    io_uring proactor**, implicating the capi's epoll readiness path specifically.
    Flagged for a focused `perf`/flamegraph pass; not yet explained.

## Follow-up built + investigated (2026-06-19, branch feat/cdef-handler → main)

18. **`cdef`/`c_entry` handler tier BUILT — honest negative on throughput.**
    `serve()` now accepts a `runloom_c.c_handler` PyCapsule (a `cdef` C function)
    and spawns it via `runloom_mn_fiber_c` → the tstate-free `g->c_entry` path
    (raw-fd capi `fd_recv`/`fd_send_all`/`fd_close` + `module_io.c.inc` dispatch +
    `handler_cdef.pyx` + tier `runloom_iouring_cdef_tcpcon.py`). Measured: the tstate-bypass
    buys ~nothing — `cdef` vs `cython` ceiling +0.25% at 8 B, +2.3% at 1 KiB (both
    noise). The default per-hub **snapshot** tstate is already cheap (snaps a few
    ints, not a `PyThreadState`), so there's nothing to bypass. Use `handler_cy`
    (Cython `def`) on the proactor; the `cdef` path's value is per-fiber memory.

19. **Anomaly #17 RESOLVED: there is no anomaly.** Arc: looked ~2× (cython epoll
    server-bound 425k vs py client-bound 620k) → called it an artifact → retracted
    when a netns re-measure reproduced 425k → settled definitively by saturation +
    objdump. At **2048 conns** (the load the ladder never reached): cython 604,946
    ≈ py 598,527 — **equal**; the "425k server-bound" was the ladder's plateau label
    misfiring at sub-saturation conn counts. Objdump: the cython capi `send_all` is
    **115 insns, ZERO `Py_` calls** vs the py method's 134 insns + 10 `Py_` calls —
    **no hidden Python object**, the Cython path is strictly leaner. Both hypotheses
    (hidden PyObject; shared-object lock contention) ruled out. **Lesson banked:**
    confirm a "server-bound" peak by saturating past the loadgen knee, don't trust
    the ladder's bottleneck label. Full trace: `../IOURING_TSTATE_FINDINGS.md` +
    `results/anomaly_notes.md`.

## Work-curve, scaling, bench-consolidation & report refinements (2026-06-19, branch feat/cdef-handler)

20. **The "work" experiment — handler CPU vs throughput (`../WORK_CURVE_EXPERIMENT.md`).**
    Echo ties every handler optimization because it has no handler CPU. So: ONE
    server (`servers/srv_runloom_work.py`), ONE knob (`--work N` = an FNV-1a byte
    hash over the payload, repeated N times, folded into the reply so it can't be
    elided), swept into a curve. The work is **pure inline arithmetic** — no
    stdlib/`hashlib`/`json` and nothing runloom routes to the blockpool — so it
    runs on the fiber's hub (a valid per-hub measurement); a thread-offloaded call
    would wreck the per-core accounting. **`--work 0` IS the echo** (lowest point),
    so the one program consolidates the echo load and reproduces it as a built-in
    cross-check.

21. **"cython" means the FASTEST fully-native handler, not a Python `def` wrapping
    compiled work.** First cut made `--handler cython` an interpreted `def` calling
    a compiled `work_cy.fnv_work` — which isolated "what does compiling the *work*
    buy" but carried Python per-request overhead (recv_into/send_all/fold/call-box)
    and so trailed Go ~2× at light work. Per the project owner, the cython line must
    be the **state of the art**: the FNV is now **inline inside the zero-PyObject
    Cython handler** (`handler_cy`, work knob via `set_work()`), and a third
    `--handler cdef` runs it in the tstate-free `c_entry` handler (`handler_cdef`).
    `disasm_check.sh` re-confirms the per-request loop stays PyObject-free.
    **Result:** a fully-native runloom handler **matches-to-beats Go across the
    curve** — runloom-cython/cdef are *ahead* of Go through ~work 4 (faster I/O) and
    within ~8% at heaviest compute. The 2× gap was 100% the Python `def` wrapper.
    `cython ≈ cdef` (re-confirms the tstate bypass is worth ~nothing on throughput).
    The handler **language** is the only thing that ever separates the field;
    interpreted Python (runloom-py/asyncio/uvloop/gevent) collapses ~100–200× under
    work. Takeaway: runloom's runtime (I/O + scheduler) is the solved, Go-beating
    part; the residual is interpreter cost, orthogonal to runloom (future work:
    auto-compile handlers — seed in `bench/poc_compile_handler.py`).

22. **Cross-runtime work curve, per core (`work_xrt_sweep.py` → `results/work_xrt.json`).**
    The same `--work` FNV in every runtime's natural handler language, reported
    per core, with `--only NAME,...` + merge-on-write so individual runtimes can be
    re-measured without re-running the rest. Read the **heaviest-work column** for
    the true capacity comparison (the only point all runtimes are server-bound):
    two bands by language (compiled runloom-cdef ≈ runloom-cython ≈ Go ~6.5–7k/core;
    interpreted ~30–40/core). Light-work columns are loadgen-bound for the compiled
    runtimes (so non-monotonicity there is the measurement, not the runtime); at
    `w=0` echo the single-thread event loops lead per core (no FT/M:N tax on pure
    I/O), which inverts the instant the handler does work.

23. **Scheduler micro-benchmark scaling — spawn/ctxswitch (`../SCHEDULER_SCALING_FINDINGS.md`).**
    No timer-boundary artifact: `run_speed.py` already subtracts an **n=0 lifecycle
    baseline** for both spawn and ctxswitch, so the published gaps vs Go are genuine.
    Hub-scaling curve: runloom ctxswitch is **~350 ns/switch (beats greenlet 465 &
    Go 676) and flat to ~16 hubs**, then a cliff (3,452 ns @32, 12,130 @44).
    `perf` localised the wall: it is **free-threaded CPython `_PyCriticalSection`/
    `_PyMutex` contention** (a futex → cross-NUMA IPI storm; 24% `native_write_msr`)
    from 704 identical-worker fibers sharing interpreter state — **runloom's yield is
    ~2% of the profile.** So the "25 µs" headline is a CPython object-lock artifact,
    not runloom. spawn is a separate, milder story (genuinely ~16 µs/task heavy path,
    mild anti-scaling) — own profile TODO.

24. **`bench/` consolidation (commit bf896e7).** Audited before deleting: `bench/`
    is the **live local perf gate** (`scripts/bench.sh` runs the `bench.micro`/
    `bench.mn` package against committed `bench/results/` baselines) plus profiling
    drivers (`bench/profile/`) referenced by `docs/dev/PERHUB_EPOLL.md` — NOT
    superseded by `benchmark/`. Only the 20 `bench/bench_*.py` one-shots (which the
    repo's own README flags as superseded, imported by nothing) + the generated
    `.cyc_build/` cache were removed. A full move under `benchmark/` would break
    `scripts/bench.sh` + the docs and is deliberately left for a reference-updating
    pass.

25. **Report (`report.html`) presentation.** Inline SVG line charts (no JS deps) for
    the work + cross-runtime curves, log-y by default with a **focused linear
    "compiled handlers vs Go" chart** where the log scale would hide the small
    ratios; **clickable legends** (`tglSeries`) to toggle any series (isolate one
    runloom config vs Go). Fixed HTML double-escaping (authored entities in
    `code_block` titles / table headers were run through `esc()`), and emphasised
    the winning row (trophy + accent border, marked on the capacity table).