perf(io): remove expired native channel waiters in constant time

[sebastian-griego]

A timed channel operation can delay unrelated IO when it expires behind many blocked operations. chan_cut scans the channel queue to find the previous waiter. Even when timeouts fire in queue order, removing T timed waiters behind B blocked waiters requires O(B × T) queue traversal on the IO loop.

Use the existing IoWork.prev field to remove each expired channel waiter in O(1). Maintain both links when a waiter enters the queue, leaves on timeout, or wakes in FIFO order. The waiter and its timer use separate IoWork objects, so this adds no allocation or struct field.

Add chan_waiter_links.bend to check sender and receiver queues. It covers head, middle, tail, and singleton removal; enqueue after timeout; wake before timeout; close; and activation reuse. The test checks results without a performance threshold.

This targets native IO performance. It is separate from #1250 and #1442, which concern the global poll and timer queues. This change only updates the channel waiter ring.

Measurement

The Base-only program below parks 32,768 blocking senders and then starts 4,096 sends with 100 ms timeouts. A separate fiber sleeps for 100 ms. Every run returned all 4,096 expected timeout results.

Platform	Original 100 ms timer	Changed 100 ms timer	Method
Apple M5 Max, macOS 26.6.2, Apple Clang 21	376 ms	110 ms	Median of three alternating pairs after warmup
Debian 13 ARM64, Clang 19, Linux container	418 ms	106 ms	One before/after pair

Both versions use -O3 and --threads 1. The baseline is e5527d5be0c441d00f1f7d2fb61eb464fa478e5c. The changed binaries use C emitted from this branch. These results measure this workload; global timer insertion costs remain.

Validation

• Repository gate: 54/54.
• macOS: 22 IO tests pass in the interpreter, native runtime with 1 and 4 threads, and Bun.
• Linux: 15 channel and deadlock tests pass on both versions with 1 and 4 threads (60 runs).
• The new regression passes under ASan and UBSan. Removing any of the four required link updates makes it fail.
• A scratch ASan/UBSan test passes one million randomized queue operations. It checks FIFO order, both links, tail identity, and singleton removal.
• The full test gate requires the maintainer cluster and was not run.

Timing reproduction

Save this as mixed_timeouts.bend in the repository root. Run it on the baseline and this branch:

bun bend2/main.ts mixed_timeouts.bend -o mixed_timeouts
./mixed_timeouts --threads 1

import Base

def plain(blocked: Chan(U32), ready: Chan(U32)) -> IO(Unit):
  do IO<Unit>:
    ack : Result<&1, &1, U32, Unit> <- Chan.send(U32, ready, 1)
    result : Result<&1, &1, U32, Unit> <- Chan.send(U32, blocked, 7)
    IO.pure(Unit, Unit{})

def score(result: Poll<&1, &1, Result<&1, &1, U32, Unit>, U32>) -> U32:
  match result:
    case Ready{result}: 0
    case Wait{rest}: Bool.pick(U32, U32.is_eq(rest,9),1,0)

def timed(blocked: Chan(U32), done: Chan(U32)) -> IO(Unit):
  do IO<Unit>:
    result : Poll<&1, &1, Result<&1, &1, U32, Unit>, U32> <- Chan.try_send(U32, blocked, 9, 100)
    sent : Result<&1, &1, U32, Unit> <- Chan.send(U32, done, score(result))
    IO.pure(Unit, Unit{})

def start_plain(n: Nat, +blocked: Chan(U32), +ready: Chan(U32)) -> IO(Unit):
  match n:
    case 0n: IO.pure(Unit, Unit{})
    case 1n+p:
      do IO<Unit>:
        Unit <- IO.spawn(Unit, plain(blocked,ready))
        start_plain(p, blocked,ready)

def start_timed(n: Nat, +blocked: Chan(U32), +done: Chan(U32)) -> IO(Unit):
  match n:
    case 0n: IO.pure(Unit, Unit{})
    case 1n+p:
      do IO<Unit>:
        Unit <- IO.spawn(Unit, timed(blocked, done))
        start_timed(p, blocked, done)

def get(got: Maybe<&1, U32>) -> U32:
  match got:
    case None{}: 0
    case Some{x}: x

def wait_all(n: Nat, +done: Chan(U32), acc: U32) -> IO(U32):
  match n:
    case 0n: IO.pure(U32, acc)
    case 1n+p:
      do IO<U32>:
        got : Maybe<&1, U32> <- Chan.recv(U32, done)
        wait_all(p, done, U32.add(acc,get(got)))

def report(started: Nat, label: String) -> IO(Unit):
  do IO<Unit>:
    now : Nat <- IO.now()
    IO.print(label ++ Nat.show(Nat.sub(now, started)))

def canary() -> IO(Unit):
  do IO<Unit>:
    started : Nat <- IO.now()
    Unit <- IO.sleep(100)
    report(started,"canary_ms=")

def run(+blockers: Nat) -> IO(Unit):
  do IO<Unit>:
    +blocked : Chan(U32) <- Chan.new(U32,0)
    +ready : Chan(U32) <- Chan.new(U32,U32.from_nat(blockers))
    +done : Chan(U32) <- Chan.new(U32,4096)
    Unit <- start_plain(blockers,blocked,ready)
    queued : U32 <- wait_all(blockers,ready,0)
    Unit <- IO.print("queued=" ++ U32.show(queued))
    started : Nat <- IO.now()
    Unit <- IO.spawn(Unit, canary())
    Unit <- start_timed(4096n,blocked,done)
    count : U32 <- wait_all(4096n,done,0)
    Unit <- report(started,"complete_ms=")
    Unit <- IO.print("count=" ++ U32.show(count))
    Chan.close(U32,blocked)

def main() -> IO(Unit):
  run(32768n)