reactor_io_tests.h

Go to the documentation of this file.

/* -*- C -*-
 * Serene programming language
 * Copyright (C) 2019-2026 Sameer Rahmani <[email protected]>
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program.  If not, see <https://www.gnu.org/licenses/>.
 */
 
#pragma once
 
// The epoll backend is Linux-only, so are these tests. They drive real pollable
// descriptors (socketpairs, loopback TCP/UDP, eventfd) end to end, submit an
// op, let the reactor thread arm it and perform the transfer on readiness, then
// reap the completion. Regular files are deliberately absent -- epoll cannot
// poll them.
#ifdef __linux__
 
#  include <arpa/inet.h>
#  include <errno.h>
#  include <fcntl.h>
#  include <netinet/in.h>
#  include <stdint.h>
#  include <stdlib.h>
#  include <string.h>
#  include <sys/epoll.h>
#  include <sys/eventfd.h>
#  include <sys/socket.h>
#  include <sys/stat.h>
#  include <unistd.h>
 
#  include "acutest.h"
#  include "base.h"
// reactor_tests.h brings in rt/reactor/internal.h and the white-box
// reactor_test_submit / reactor_test_await helpers; backend.h completes the
// vtable type so the nudge (`backend->wake`) can be called.
#  include "reactor_tests.h"
#  include "rt/reactor/backend.h"
#  include "serene/rt/fiber/io.h"
 
#  define REACTOR_IO_TESTS(X)                                                                     \
    X("io::read", test_io_read), X("io::write", test_io_write), X("io::poll", test_io_poll),      \
      X("io::accept", test_io_accept), X("io::connect_success", test_io_connect),                 \
      X("io::connect_refused", test_io_refused), X("io::recvfrom", test_io_recvfrom),             \
      X("io::sendto", test_io_sendto), X("io::recvmsg", test_io_recvmsg),                         \
      X("io::sendmsg", test_io_sendmsg), X("io::shared_fd", test_io_shared_fd),                   \
      X("io::accept_fanout", test_io_fanout), X("io::tombstone_reuse", test_io_reuse),            \
      X("io::bad_fd", test_io_bad_fd), X("io::peer_close_eof", test_io_eof),                      \
      X("io::write_epipe", test_io_epipe), X("io::fiber_read", test_io_fiber_read),               \
      X("io::fiber_write", test_io_fiber_write), X("io::fiber_sleep", test_io_fiber_sleep),       \
      X("io::fiber_pipe", test_io_fiber_pipe), X("io::fiber_drain", test_io_fiber_drain),         \
      X("io::fiber_file", test_io_fiber_file),                                                    \
      X("io::fiber_accept_connect", test_io_fiber_accept_connect),                                \
      X("io::fiber_dgram", test_io_fiber_dgram), X("io::inline_helpers", test_io_inline_helpers), \
      X("io::channel_admission", test_io_channel_admission),                                      \
      X("io::busy_retry", test_io_busy_retry), X("io::open_nonblock", test_io_open_nonblock),     \
      X("io::fiber_fifo", test_io_fiber_fifo)
 
// ---------------------------------------------------------------------------
// helpers
// ---------------------------------------------------------------------------
 
static void io_nonblock(int fd) {
  int fl = fcntl(fd, F_GETFL, 0);
  TEST_ASSERT(fl >= 0);
  TEST_ASSERT(fcntl(fd, F_SETFL, fl | O_NONBLOCK) == 0);
}
 
// Push a request and nudge the reactor, so it drains the SQ at once instead of
// waiting out the reactor loop's idle timeout.
static void io_submit(srn_reactor_t *r, size_t channel, const srn_reactor_io_request_t *req) {
  reactor_test_submit(r, channel, req);
  r->backend->wake(r);
}
 
static srn_reactor_io_completion_t io_await(srn_reactor_t *r, size_t channel) {
  return reactor_test_await(r, channel);
}
 
// A connected stream pair, both ends non-blocking.
static void io_socketpair(int sv[2]) {
  TEST_ASSERT(socketpair(AF_UNIX, SOCK_STREAM, 0, sv) == 0);
  io_nonblock(sv[0]);
  io_nonblock(sv[1]);
}
 
// A non-blocking loopback TCP listener; its bound port (network order) is
// returned through `port`.
static int io_tcp_listener(uint16_t *port) {
  int s = socket(AF_INET, SOCK_STREAM, 0);
  TEST_ASSERT(s >= 0);
  io_nonblock(s);
  struct sockaddr_in a = {.sin_family = AF_INET};
  a.sin_addr.s_addr    = htonl(INADDR_LOOPBACK);
  TEST_ASSERT(bind(s, (struct sockaddr *)&a, sizeof(a)) == 0);
  TEST_ASSERT(listen(s, 16) == 0);
  socklen_t al = sizeof(a);
  TEST_ASSERT(getsockname(s, (struct sockaddr *)&a, &al) == 0);
  *port = a.sin_port;
  return s;
}
 
// A blocking connect to a loopback port; returns the client fd. Loopback
// completes the handshake at once, so this does not stall.
static int io_tcp_connect(uint16_t port) {
  int c = socket(AF_INET, SOCK_STREAM, 0);
  TEST_ASSERT(c >= 0);
  struct sockaddr_in a = {.sin_family = AF_INET, .sin_port = port};
  a.sin_addr.s_addr    = htonl(INADDR_LOOPBACK);
  TEST_ASSERT(connect(c, (struct sockaddr *)&a, sizeof(a)) == 0);
  return c;
}
 
// A non-blocking UDP socket bound to a loopback ephemeral port (returned via
// `port`).
static int io_udp_bound(uint16_t *port) {
  int s = socket(AF_INET, SOCK_DGRAM, 0);
  TEST_ASSERT(s >= 0);
  io_nonblock(s);
  struct sockaddr_in a = {.sin_family = AF_INET};
  a.sin_addr.s_addr    = htonl(INADDR_LOOPBACK);
  TEST_ASSERT(bind(s, (struct sockaddr *)&a, sizeof(a)) == 0);
  socklen_t al = sizeof(a);
  TEST_ASSERT(getsockname(s, (struct sockaddr *)&a, &al) == 0);
  *port = a.sin_port;
  return s;
}
 
// ---------------------------------------------------------------------------
// per-op happy paths
// ---------------------------------------------------------------------------
 
// READ pulls bytes already waiting on the descriptor.
static void test_io_read() {
  MAKE_ENGINE(mm, engine);
  srn_reactor_t *r = engine->reactor;
  reactor_test_start(r, 1);
 
  int sv[2];
  io_socketpair(sv);
  TEST_ASSERT(write(sv[1], "hello", 5) == 5);
 
  char buf[16]                 = {0};
  int marker                   = 0;
  srn_reactor_io_request_t req = {
    .op         = SRN_REACTOR_IO_READ,
    .fd         = sv[0],
    .buf        = buf,
    .len        = sizeof(buf),
    .offset     = -1,
    .fiber_data = &marker
  };
  io_submit(r, 0, &req);
 
  srn_reactor_io_completion_t c = io_await(r, 0);
  TEST_CHECK(c.fiber_data == &marker);
  TEST_CHECK(c.result == 5);
  TEST_CHECK(memcmp(buf, "hello", 5) == 0);
 
  close(sv[0]);
  close(sv[1]);
  SHUTDOWN_ENGINE(mm, engine);
}
 
// WRITE pushes bytes the peer can then read.
static void test_io_write() {
  MAKE_ENGINE(mm, engine);
  srn_reactor_t *r = engine->reactor;
  reactor_test_start(r, 1);
 
  int sv[2];
  io_socketpair(sv);
 
  char out[]                   = "world";
  int marker                   = 0;
  srn_reactor_io_request_t req = {
    .op         = SRN_REACTOR_IO_WRITE,
    .fd         = sv[0],
    .buf        = out,
    .len        = 5,
    .offset     = -1,
    .fiber_data = &marker
  };
  io_submit(r, 0, &req);
 
  srn_reactor_io_completion_t c = io_await(r, 0);
  TEST_CHECK(c.result == 5);
 
  char buf[16] = {0};
  TEST_CHECK(read(sv[1], buf, sizeof(buf)) == 5);
  TEST_CHECK(memcmp(buf, "world", 5) == 0);
 
  close(sv[0]);
  close(sv[1]);
  SHUTDOWN_ENGINE(mm, engine);
}
 
// POLL reports readiness without transferring anything; `result` is the mask.
static void test_io_poll() {
  MAKE_ENGINE(mm, engine);
  srn_reactor_t *r = engine->reactor;
  reactor_test_start(r, 1);
 
  int efd = eventfd(0, EFD_NONBLOCK);
  TEST_ASSERT(efd >= 0);
  uint64_t one = 1;
  TEST_ASSERT(write(efd, &one, sizeof(one)) == (ssize_t)sizeof(one));
 
  int marker                   = 0;
  srn_reactor_io_request_t req = {
    .op = SRN_REACTOR_IO_POLL, .fd = efd, .flags = EPOLLIN, .fiber_data = &marker
  };
  io_submit(r, 0, &req);
 
  srn_reactor_io_completion_t c = io_await(r, 0);
  TEST_CHECK((c.result & EPOLLIN) != 0);
 
  close(efd);
  SHUTDOWN_ENGINE(mm, engine);
}
 
// ACCEPT yields a new descriptor and fills the peer address (family-agnostic;
// here IPv4), reporting its length on the completion.
static void test_io_accept() {
  MAKE_ENGINE(mm, engine);
  srn_reactor_t *r = engine->reactor;
  reactor_test_start(r, 1);
 
  uint16_t port;
  int lst = io_tcp_listener(&port);
 
  struct sockaddr_storage ss;
  memset(&ss, 0, sizeof(ss));
  int marker                   = 0;
  srn_reactor_io_request_t req = {
    .op         = SRN_REACTOR_IO_ACCEPT,
    .fd         = lst,
    .addr       = &ss,
    .addrlen    = sizeof(ss),
    .fiber_data = &marker
  };
  io_submit(r, 0, &req);
 
  int cli = io_tcp_connect(port);
 
  srn_reactor_io_completion_t c = io_await(r, 0);
  TEST_CHECK(!HAS_ERROR(c)); // the accepted fd is in c.result
  TEST_CHECK(c.addrlen >= sizeof(struct sockaddr_in));
  TEST_CHECK(((struct sockaddr_in *)&ss)->sin_family == AF_INET);
 
  if (!HAS_ERROR(c)) {
    close((int)c.result);
  }
  close(cli);
  close(lst);
  SHUTDOWN_ENGINE(mm, engine);
}
 
// CONNECT to a live listener succeeds with a zero result.
static void test_io_connect() {
  MAKE_ENGINE(mm, engine);
  srn_reactor_t *r = engine->reactor;
  reactor_test_start(r, 1);
 
  uint16_t port;
  int lst = io_tcp_listener(&port);
 
  int s = socket(AF_INET, SOCK_STREAM, 0);
  TEST_ASSERT(s >= 0);
  io_nonblock(s); // non-blocking so connect returns EINPROGRESS, never blocks
                  // the reactor thread
  struct sockaddr_in a = {.sin_family = AF_INET, .sin_port = port};
  a.sin_addr.s_addr    = htonl(INADDR_LOOPBACK);
 
  int marker                   = 0;
  srn_reactor_io_request_t req = {
    .op = SRN_REACTOR_IO_CONNECT, .fd = s, .addr = &a, .addrlen = sizeof(a), .fiber_data = &marker
  };
  io_submit(r, 0, &req);
 
  srn_reactor_io_completion_t c = io_await(r, 0);
  TEST_CHECK(c.result == 0);
 
  close(s);
  close(lst);
  SHUTDOWN_ENGINE(mm, engine);
}
 
// CONNECT to a dead port fails with -ECONNREFUSED (whether the kernel refuses
// at submit time or via SO_ERROR on the EPOLLOUT path).
static void test_io_refused() {
  MAKE_ENGINE(mm, engine);
  srn_reactor_t *r = engine->reactor;
  reactor_test_start(r, 1);
 
  uint16_t port;
  int lst = io_tcp_listener(&port);
  close(lst); // free the port so nothing is listening
 
  int s = socket(AF_INET, SOCK_STREAM, 0);
  TEST_ASSERT(s >= 0);
  io_nonblock(s);
  struct sockaddr_in a = {.sin_family = AF_INET, .sin_port = port};
  a.sin_addr.s_addr    = htonl(INADDR_LOOPBACK);
 
  int marker                   = 0;
  srn_reactor_io_request_t req = {
    .op = SRN_REACTOR_IO_CONNECT, .fd = s, .addr = &a, .addrlen = sizeof(a), .fiber_data = &marker
  };
  io_submit(r, 0, &req);
 
  srn_reactor_io_completion_t c = io_await(r, 0);
  TEST_CHECK(HAS_ERROR(c) && c.maybe_error->tag == CONNECTION_REFUSED);
 
  close(s);
  SHUTDOWN_ENGINE(mm, engine);
}
 
// RECVFROM receives a datagram and records the sender's address and its length.
static void test_io_recvfrom() {
  MAKE_ENGINE(mm, engine);
  srn_reactor_t *r = engine->reactor;
  reactor_test_start(r, 1);
 
  uint16_t port;
  int rcv = io_udp_bound(&port);
  int snd = socket(AF_INET, SOCK_DGRAM, 0);
  TEST_ASSERT(snd >= 0);
 
  struct sockaddr_storage ss;
  memset(&ss, 0, sizeof(ss));
  char buf[32]                 = {0};
  int marker                   = 0;
  srn_reactor_io_request_t req = {
    .op         = SRN_REACTOR_IO_RECVFROM,
    .fd         = rcv,
    .buf        = buf,
    .len        = sizeof(buf),
    .addr       = &ss,
    .addrlen    = sizeof(ss),
    .fiber_data = &marker
  };
  io_submit(r, 0, &req);
 
  struct sockaddr_in to = {.sin_family = AF_INET, .sin_port = port};
  to.sin_addr.s_addr    = htonl(INADDR_LOOPBACK);
  char msg[]            = "ping";
  TEST_ASSERT(sendto(snd, msg, 4, 0, (struct sockaddr *)&to, sizeof(to)) == 4);
 
  srn_reactor_io_completion_t c = io_await(r, 0);
  TEST_CHECK(c.result == 4);
  TEST_CHECK(memcmp(buf, "ping", 4) == 0);
  TEST_CHECK(c.addrlen >= sizeof(struct sockaddr_in));
  TEST_CHECK(((struct sockaddr_in *)&ss)->sin_family == AF_INET);
 
  close(snd);
  close(rcv);
  SHUTDOWN_ENGINE(mm, engine);
}
 
// SENDTO delivers a datagram to a named peer.
static void test_io_sendto() {
  MAKE_ENGINE(mm, engine);
  srn_reactor_t *r = engine->reactor;
  reactor_test_start(r, 1);
 
  uint16_t port;
  int rcv = io_udp_bound(&port);
  int snd = socket(AF_INET, SOCK_DGRAM, 0);
  TEST_ASSERT(snd >= 0);
  io_nonblock(snd);
 
  struct sockaddr_in to = {.sin_family = AF_INET, .sin_port = port};
  to.sin_addr.s_addr    = htonl(INADDR_LOOPBACK);
  char msg[]            = "pong";
  int marker            = 0;
 
  srn_reactor_io_request_t req = {
    .op         = SRN_REACTOR_IO_SENDTO,
    .fd         = snd,
    .buf        = msg,
    .len        = 4,
    .addr       = &to,
    .addrlen    = sizeof(to),
    .fiber_data = &marker
  };
  io_submit(r, 0, &req);
 
  srn_reactor_io_completion_t c = io_await(r, 0);
  TEST_CHECK(c.result == 4);
 
  char buf[32] = {0};
  TEST_CHECK(recvfrom(rcv, buf, sizeof(buf), 0, nullptr, nullptr) == 4);
  TEST_CHECK(memcmp(buf, "pong", 4) == 0);
 
  close(snd);
  close(rcv);
  SHUTDOWN_ENGINE(mm, engine);
}
 
// RECVMSG scatters a single read across multiple iovec segments.
static void test_io_recvmsg() {
  MAKE_ENGINE(mm, engine);
  srn_reactor_t *r = engine->reactor;
  reactor_test_start(r, 1);
 
  int sv[2];
  io_socketpair(sv);
  TEST_ASSERT(write(sv[1], "foobar", 6) == 6);
 
  char p1[3]          = {0};
  char p2[3]          = {0};
  struct iovec iov[2] = {{.iov_base = p1, .iov_len = 3}, {.iov_base = p2, .iov_len = 3}};
  struct msghdr mh;
  memset(&mh, 0, sizeof(mh));
  mh.msg_iov    = iov;
  mh.msg_iovlen = 2;
 
  int marker                   = 0;
  srn_reactor_io_request_t req = {
    .op = SRN_REACTOR_IO_RECVMSG, .fd = sv[0], .buf = &mh, .fiber_data = &marker
  };
  io_submit(r, 0, &req);
 
  srn_reactor_io_completion_t c = io_await(r, 0);
  TEST_CHECK(c.result == 6);
  TEST_CHECK(memcmp(p1, "foo", 3) == 0);
  TEST_CHECK(memcmp(p2, "bar", 3) == 0);
 
  close(sv[0]);
  close(sv[1]);
  SHUTDOWN_ENGINE(mm, engine);
}
 
// SENDMSG gathers multiple iovec segments into a single write.
static void test_io_sendmsg() {
  MAKE_ENGINE(mm, engine);
  srn_reactor_t *r = engine->reactor;
  reactor_test_start(r, 1);
 
  int sv[2];
  io_socketpair(sv);
 
  char a[]            = "foo";
  char b[]            = "bar";
  struct iovec iov[2] = {{.iov_base = a, .iov_len = 3}, {.iov_base = b, .iov_len = 3}};
  struct msghdr mh;
  memset(&mh, 0, sizeof(mh));
  mh.msg_iov    = iov;
  mh.msg_iovlen = 2;
 
  int marker                   = 0;
  srn_reactor_io_request_t req = {
    .op = SRN_REACTOR_IO_SENDMSG, .fd = sv[0], .buf = &mh, .fiber_data = &marker
  };
  io_submit(r, 0, &req);
 
  srn_reactor_io_completion_t c = io_await(r, 0);
  TEST_CHECK(c.result == 6);
 
  char buf[16] = {0};
  TEST_CHECK(read(sv[1], buf, sizeof(buf)) == 6);
  TEST_CHECK(memcmp(buf, "foobar", 6) == 0);
 
  close(sv[0]);
  close(sv[1]);
  SHUTDOWN_ENGINE(mm, engine);
}
 
// ---------------------------------------------------------------------------
// shared descriptors -- the reason the backend groups ops per fd
// ---------------------------------------------------------------------------
 
// A READ and a WRITE on one descriptor at once, the fd is armed for the union
// EPOLLIN|EPOLLOUT and both complete. Both requests are queued before the nudge
// so the reactor arms them together.
static void test_io_shared_fd() {
  MAKE_ENGINE(mm, engine);
  srn_reactor_t *r = engine->reactor;
  reactor_test_start(r, 1);
 
  int sv[2];
  io_socketpair(sv);
  TEST_ASSERT(write(sv[1], "in", 2) == 2); // make the read side ready
 
  char rbuf[8]                = {0};
  char wbuf[]                 = "out";
  int m_read                  = 0;
  int m_write                 = 0;
  srn_reactor_io_request_t rd = {
    .op         = SRN_REACTOR_IO_READ,
    .fd         = sv[0],
    .buf        = rbuf,
    .len        = sizeof(rbuf),
    .offset     = -1,
    .fiber_data = &m_read
  };
  srn_reactor_io_request_t wr = {
    .op         = SRN_REACTOR_IO_WRITE,
    .fd         = sv[0],
    .buf        = wbuf,
    .len        = 3,
    .offset     = -1,
    .fiber_data = &m_write
  };
  reactor_test_submit(r, 0, &rd);
  reactor_test_submit(r, 0, &wr);
  r->backend->wake(r);
 
  bool got_read  = false;
  bool got_write = false;
  for (int i = 0; i < 2; i++) {
    srn_reactor_io_completion_t c = io_await(r, 0);
    if (c.fiber_data == &m_read) {
      got_read = true;
      TEST_CHECK(c.result == 2);
      TEST_CHECK(memcmp(rbuf, "in", 2) == 0);
    } else if (c.fiber_data == &m_write) {
      got_write = true;
      TEST_CHECK(c.result == 3);
    } else {
      TEST_CHECK(false); // unexpected completion
    }
  }
  TEST_CHECK(got_read && got_write);
 
  char buf[8] = {0};
  TEST_CHECK(read(sv[1], buf, sizeof(buf)) == 3);
  TEST_CHECK(memcmp(buf, "out", 3) == 0);
 
  close(sv[0]);
  close(sv[1]);
  SHUTDOWN_ENGINE(mm, engine);
}
 
// Many ACCEPTs queued on one listener all complete -- fan-out on a single fd.
static void test_io_fanout() {
  MAKE_ENGINE(mm, engine);
  srn_reactor_t *r = engine->reactor;
  reactor_test_start(r, 1);
 
  uint16_t port;
  int lst = io_tcp_listener(&port);
 
  enum { K = 3 };
  int markers[K];
  struct sockaddr_storage ss[K];
  for (int i = 0; i < K; i++) {
    memset(&ss[i], 0, sizeof(ss[i]));
    srn_reactor_io_request_t req = {
      .op         = SRN_REACTOR_IO_ACCEPT,
      .fd         = lst,
      .addr       = &ss[i],
      .addrlen    = sizeof(ss[i]),
      .fiber_data = &markers[i]
    };
    reactor_test_submit(r, 0, &req);
  }
  r->backend->wake(r);
 
  int clients[K];
  for (int i = 0; i < K; i++) {
    clients[i] = io_tcp_connect(port);
  }
 
  int accepted = 0;
  for (int i = 0; i < K; i++) {
    srn_reactor_io_completion_t c = io_await(r, 0);
    if (!HAS_ERROR(c)) {
      accepted++;
      close((int)c.result);
    }
  }
  TEST_CHECK(accepted == K);
 
  for (int i = 0; i < K; i++) {
    close(clients[i]);
  }
  close(lst);
  SHUTDOWN_ENGINE(mm, engine);
}
 
// ---------------------------------------------------------------------------
// lifecycle and errors
// ---------------------------------------------------------------------------
 
// After an fd's op completes its list goes dormant; a later op on the same fd
// integer revives it (the tombstone reuse path).
static void test_io_reuse() {
  MAKE_ENGINE(mm, engine);
  srn_reactor_t *r = engine->reactor;
  reactor_test_start(r, 1);
 
  int sv[2];
  io_socketpair(sv);
 
  TEST_ASSERT(write(sv[1], "a", 1) == 1);
  char b1[4]                   = {0};
  int m1                       = 0;
  srn_reactor_io_request_t rq1 = {
    .op         = SRN_REACTOR_IO_READ,
    .fd         = sv[0],
    .buf        = b1,
    .len        = sizeof(b1),
    .offset     = -1,
    .fiber_data = &m1
  };
  io_submit(r, 0, &rq1);
  srn_reactor_io_completion_t c1 = io_await(r, 0);
  TEST_CHECK(c1.result == 1 && b1[0] == 'a');
 
  // Same fd again, its list was tombstoned, so this must re-arm it.
  TEST_ASSERT(write(sv[1], "b", 1) == 1);
  char b2[4]                   = {0};
  int m2                       = 0;
  srn_reactor_io_request_t rq2 = {
    .op         = SRN_REACTOR_IO_READ,
    .fd         = sv[0],
    .buf        = b2,
    .len        = sizeof(b2),
    .offset     = -1,
    .fiber_data = &m2
  };
  io_submit(r, 0, &rq2);
  srn_reactor_io_completion_t c2 = io_await(r, 0);
  TEST_CHECK(c2.result == 1 && b2[0] == 'b');
 
  close(sv[0]);
  close(sv[1]);
  SHUTDOWN_ENGINE(mm, engine);
}
 
// Arming a closed descriptor fails the op back with -EBADF rather than hanging.
static void test_io_bad_fd() {
  MAKE_ENGINE(mm, engine);
  srn_reactor_t *r = engine->reactor;
  reactor_test_start(r, 1);
 
  int fd = dup(0);
  TEST_ASSERT(fd >= 0);
  close(fd); // fd is now an invalid descriptor number
 
  char buf[4]                  = {0};
  int marker                   = 0;
  srn_reactor_io_request_t req = {
    .op         = SRN_REACTOR_IO_READ,
    .fd         = fd,
    .buf        = buf,
    .len        = sizeof(buf),
    .offset     = -1,
    .fiber_data = &marker
  };
  io_submit(r, 0, &req);
 
  srn_reactor_io_completion_t c = io_await(r, 0);
  TEST_CHECK(HAS_ERROR(c) && c.maybe_error->tag == BAD_DESCRIPTOR);
 
  SHUTDOWN_ENGINE(mm, engine);
}
 
// A closed peer makes the descriptor readable at EOF: READ completes with 0.
static void test_io_eof() {
  MAKE_ENGINE(mm, engine);
  srn_reactor_t *r = engine->reactor;
  reactor_test_start(r, 1);
 
  int sv[2];
  io_socketpair(sv);
  close(sv[1]); // peer gone
 
  char buf[8]                  = {0};
  int marker                   = 0;
  srn_reactor_io_request_t req = {
    .op         = SRN_REACTOR_IO_READ,
    .fd         = sv[0],
    .buf        = buf,
    .len        = sizeof(buf),
    .offset     = -1,
    .fiber_data = &marker
  };
  io_submit(r, 0, &req);
 
  srn_reactor_io_completion_t c = io_await(r, 0);
  TEST_CHECK(c.result == 0);
 
  close(sv[0]);
  SHUTDOWN_ENGINE(mm, engine);
}
 
// Writing to a closed peer completes with -EPIPE. SIGPIPE is suppressed by the
// runtime (srn_engine_make), so the write fails rather than killing the
// process.
static void test_io_epipe() {
  MAKE_ENGINE(mm, engine);
  srn_reactor_t *r = engine->reactor;
  reactor_test_start(r, 1);
 
  int sv[2];
  io_socketpair(sv);
  close(sv[1]);
 
  char out[]                   = "x";
  int marker                   = 0;
  srn_reactor_io_request_t req = {
    .op         = SRN_REACTOR_IO_WRITE,
    .fd         = sv[0],
    .buf        = out,
    .len        = 1,
    .offset     = -1,
    .fiber_data = &marker
  };
  io_submit(r, 0, &req);
 
  srn_reactor_io_completion_t c = io_await(r, 0);
  TEST_CHECK(HAS_ERROR(c) && c.maybe_error->tag == BROKEN_PIPE);
 
  close(sv[0]);
  SHUTDOWN_ENGINE(mm, engine);
}
 
// ---------------------------------------------------------------------------
// through a real fiber, the full path, srn_fiber_read/write suspends the
// fiber, the reactor performs the transfer, the worker resumes the fiber with
// the result. Driven by srn_sched_run (which activates the reactor itself), not
// the white-box helpers above.
// ---------------------------------------------------------------------------
 
typedef struct fiber_io_args_t {
  int fd;
  void *buf;
  size_t len;
  srn_io_size_result_t res;
} fiber_io_args_t;
 
// A fiber entry must return a non-null result (the launcher panics on null), so
// the IO fibers return this once they have stored their result in the args.
static int fiber_io_done;
 
static srn_fiber_result_t fiber_read_entry(srn_context_t *ctx, void *arg) {
  UNUSED(ctx);
  fiber_io_args_t *a = (fiber_io_args_t *)arg;
  SRN_WITH_RETRY(a->res.maybe_error, { a->res = srn_fiber_read(a->fd, a->buf, a->len, -1); });
  return &fiber_io_done;
}
 
static srn_fiber_result_t fiber_write_entry(srn_context_t *ctx, void *arg) {
  UNUSED(ctx);
  fiber_io_args_t *a = (fiber_io_args_t *)arg;
  a->res             = srn_fiber_write(a->fd, a->buf, a->len, -1);
  return &fiber_io_done;
}
 
// A fiber reads a socketpair via srn_fiber_read, it suspends, the reactor
// reads, the worker resumes it with the byte count. The run only returns once
// the reactor is idle too, so reaching here means the whole loop closed.
static void test_io_fiber_read() {
  MAKE_ENGINE(mm, engine);
  MAKE_CONTEXT(engine, ctx);
  srn_scheduler_t *sched = engine->scheduler;
  ASSERT_NOT_NULL(sched);
 
  int sv[2];
  io_socketpair(sv);
  TEST_ASSERT(write(sv[1], "hello", 5) == 5);
 
  char buf[16]      = {0};
  fiber_io_args_t a = {.fd = sv[0], .buf = buf, .len = sizeof(buf)};
  ASSERT_NOT_NULL(srn_fiber_spawn(ctx, fiber_read_entry, &a));
 
  srn_sched_run(sched, 1);
 
  TEST_CHECK(!HAS_ERROR(a.res) && a.res.count == 5);
  TEST_CHECK(memcmp(buf, "hello", 5) == 0);
  TEST_CHECK(srn_reactor_idle(engine->reactor));
 
  close(sv[0]);
  close(sv[1]);
  RELEASE_CONTEXT(ctx);
  SHUTDOWN_ENGINE(mm, engine);
}
 
// A fiber writes a socketpair via srn_fiber_write; the peer end then has the
// bytes.
static void test_io_fiber_write() {
  MAKE_ENGINE(mm, engine);
  MAKE_CONTEXT(engine, ctx);
  srn_scheduler_t *sched = engine->scheduler;
  ASSERT_NOT_NULL(sched);
 
  int sv[2];
  io_socketpair(sv);
 
  char out[]        = "world";
  fiber_io_args_t a = {.fd = sv[0], .buf = out, .len = 5};
  ASSERT_NOT_NULL(srn_fiber_spawn(ctx, fiber_write_entry, &a));
 
  srn_sched_run(sched, 1);
 
  TEST_CHECK(!HAS_ERROR(a.res) && a.res.count == 5);
  char buf[16] = {0};
  TEST_CHECK(read(sv[1], buf, sizeof(buf)) == 5);
  TEST_CHECK(memcmp(buf, "world", 5) == 0);
 
  close(sv[0]);
  close(sv[1]);
  RELEASE_CONTEXT(ctx);
  SHUTDOWN_ENGINE(mm, engine);
}
 
typedef struct fiber_sleep_args_t {
  uint64_t duration_ns;
  srn_error_t *err;
} fiber_sleep_args_t;
 
static srn_fiber_result_t fiber_sleep_entry(srn_context_t *ctx, void *arg) {
  UNUSED(ctx);
  fiber_sleep_args_t *a = (fiber_sleep_args_t *)arg;
  a->err                = srn_fiber_sleep_ns(a->duration_ns);
  return &fiber_io_done;
}
 
// A fiber sleeps via srn_fiber_sleep_ns, it suspends, the reactor's timer keeps
// the worker alive past the point where no fibers are runnable, and the worker
// resumes the fiber once the deadline passes. The run only returns after the
// reactor goes idle, so the wall clock has advanced at least the requested
// span.
static void test_io_fiber_sleep() {
  MAKE_ENGINE(mm, engine);
  MAKE_CONTEXT(engine, ctx);
  srn_scheduler_t *sched = engine->scheduler;
  ASSERT_NOT_NULL(sched);
 
  uint64_t delay       = 50ULL * SRN_NS_PER_MS; // 50ms
  fiber_sleep_args_t a = {.duration_ns = delay, .err = nullptr};
  ASSERT_NOT_NULL(srn_fiber_spawn(ctx, fiber_sleep_entry, &a));
 
  uint64_t before = srn_now_ns();
  srn_sched_run(sched, 1);
  uint64_t elapsed = srn_now_ns() - before;
 
  TEST_CHECK(a.err == nullptr);
  TEST_CHECK(elapsed >= delay); // resumed no earlier than the deadline
  TEST_CHECK(srn_reactor_idle(engine->reactor));
 
  RELEASE_CONTEXT(ctx);
  SHUTDOWN_ENGINE(mm, engine);
}
 
// A channel admits at most one ring's worth of in-flight operations. With
// more sleeping fibers than the ring capacity on a single worker, the first
// capacity submissions are admitted and every one beyond that fails with
// CHANNEL_BUSY, instead of overflowing the rings and stalling the reactor.
static atomic_int io_adm_ok;
static atomic_int io_adm_busy;
static atomic_int io_adm_other;
 
static srn_fiber_result_t io_adm_entry(srn_context_t *ctx, void *arg) {
  UNUSED(ctx);
  UNUSED(arg);
  srn_error_t *e = srn_fiber_sleep_ms(200ULL);
  if (e == nullptr) {
    atomic_fetch_add(&io_adm_ok, 1);
  } else if (e->tag == CHANNEL_BUSY) {
    atomic_fetch_add(&io_adm_busy, 1);
  } else {
    atomic_fetch_add(&io_adm_other, 1);
  }
  return &fiber_io_done;
}
 
static void test_io_channel_admission() {
  MAKE_ENGINE(mm, engine);
  MAKE_CONTEXT(engine, ctx);
  srn_scheduler_t *sched = engine->scheduler;
  ASSERT_NOT_NULL(sched);
 
  size_t cap   = (size_t)1 << engine->config.reactor.ring_magnitude;
  size_t extra = 32;
 
  atomic_store(&io_adm_ok, 0);
  atomic_store(&io_adm_busy, 0);
  atomic_store(&io_adm_other, 0);
 
  // All fibers submit their sleeps well within the 200ms window, so the
  // channel really holds cap in-flight timers when the extra ones arrive.
  for (size_t i = 0; i < cap + extra; i++) {
    ASSERT_NOT_NULL(srn_fiber_spawn(ctx, io_adm_entry, nullptr));
  }
 
  srn_sched_run(sched, 1);
 
  TEST_MSG(
    "ok=%d busy=%d other=%d cap=%zu", atomic_load(&io_adm_ok), atomic_load(&io_adm_busy),
    atomic_load(&io_adm_other), cap
  );
  TEST_CHECK(atomic_load(&io_adm_ok) == (int)cap);
  TEST_CHECK(atomic_load(&io_adm_busy) == (int)extra);
  TEST_CHECK(atomic_load(&io_adm_other) == 0);
  TEST_CHECK(srn_reactor_idle(engine->reactor));
 
  RELEASE_CONTEXT(ctx);
  SHUTDOWN_ENGINE(mm, engine);
}
 
// The retry idiom over srn_fiber_io_retry, with more sleeping fibers than the
// channel admits, the overflow fibers retry (yielding between attempts) until
// completions reopen the admission window, and every fiber eventually
// succeeds. No CHANNEL_BUSY leaks through the loop.
static atomic_int io_retry_ok;
static atomic_int io_retry_failed;
 
static srn_fiber_result_t io_retry_entry(srn_context_t *ctx, void *arg) {
  UNUSED(ctx);
  UNUSED(arg);
 
  srn_error_t *e;
  SRN_WITH_RETRY(e, { e = srn_fiber_sleep_ms(20ULL); });
 
  if (e == nullptr) {
    atomic_fetch_add(&io_retry_ok, 1);
  } else {
    atomic_fetch_add(&io_retry_failed, 1);
  }
  return &fiber_io_done;
}
 
static void test_io_busy_retry() {
  MAKE_ENGINE(mm, engine);
  MAKE_CONTEXT(engine, ctx);
  srn_scheduler_t *sched = engine->scheduler;
  ASSERT_NOT_NULL(sched);
 
  size_t cap = (size_t)1 << engine->config.reactor.ring_magnitude;
  size_t n   = cap + 64;
 
  atomic_store(&io_retry_ok, 0);
  atomic_store(&io_retry_failed, 0);
 
  for (size_t i = 0; i < n; i++) {
    ASSERT_NOT_NULL(srn_fiber_spawn(ctx, io_retry_entry, nullptr));
  }
 
  srn_sched_run(sched, 1);
 
  TEST_MSG("ok=%d failed=%d n=%zu", atomic_load(&io_retry_ok), atomic_load(&io_retry_failed), n);
  TEST_CHECK(atomic_load(&io_retry_ok) == (int)n);
  TEST_CHECK(atomic_load(&io_retry_failed) == 0);
  TEST_CHECK(srn_reactor_idle(engine->reactor));
 
  RELEASE_CONTEXT(ctx);
  SHUTDOWN_ENGINE(mm, engine);
}
 
// Two fibers over one pipe, a writer pushes bytes with srn_fiber_write and a
// reader pulls them with srn_fiber_read, both suspending through the reactor on
// a single worker. Whichever runs first suspends and yields the worker to the
// other, so the transfer closes the loop -- the reader sees exactly what the
// writer sent.
static void test_io_fiber_pipe() {
  MAKE_ENGINE(mm, engine);
  MAKE_CONTEXT(engine, ctx);
  srn_scheduler_t *sched = engine->scheduler;
  ASSERT_NOT_NULL(sched);
 
  int fds[2];
  TEST_ASSERT(pipe(fds) == 0);
  io_nonblock(fds[0]);
  io_nonblock(fds[1]);
 
  char out[]         = "ping-pong";
  size_t n           = sizeof(out) - 1;
  char in[16]        = {0};
  fiber_io_args_t w  = {.fd = fds[1], .buf = out, .len = n};
  fiber_io_args_t rd = {.fd = fds[0], .buf = in, .len = sizeof(in)};
 
  ASSERT_NOT_NULL(srn_fiber_spawn(ctx, fiber_write_entry, &w));
  ASSERT_NOT_NULL(srn_fiber_spawn(ctx, fiber_read_entry, &rd));
 
  srn_sched_run(sched, 1);
 
  TEST_CHECK(!HAS_ERROR(w.res) && w.res.count == n);
  TEST_CHECK(!HAS_ERROR(rd.res) && rd.res.count == n);
  TEST_CHECK(memcmp(in, out, n) == 0);
  TEST_CHECK(srn_reactor_idle(engine->reactor));
 
  close(fds[0]);
  close(fds[1]);
  RELEASE_CONTEXT(ctx);
  SHUTDOWN_ENGINE(mm, engine);
}
 
typedef struct fiber_drain_args_t {
  srn_scheduler_t *sched;
  uint64_t duration_ns;
  srn_error_t *err;
} fiber_drain_args_t;
 
// Submits a real sleep and parks on it. The op is in flight when the drain
// begins, so it must finish normally (no error), not be abandoned.
static srn_fiber_result_t fiber_drain_sleeper(srn_context_t *ctx, void *arg) {
  UNUSED(ctx);
  fiber_drain_args_t *a = (fiber_drain_args_t *)arg;
  a->err                = srn_fiber_sleep_ns(a->duration_ns);
  return &fiber_io_done;
}
 
// Begins the drain, then attempts another sleep. That submission is fenced, so
// it comes back as the CANCELED error without ever parking and the fiber
// unwinds.
static srn_fiber_result_t fiber_drain_starter(srn_context_t *ctx, void *arg) {
  UNUSED(ctx);
  fiber_drain_args_t *a = (fiber_drain_args_t *)arg;
  srn_sched_drain(a->sched);
  a->err = srn_fiber_sleep_ns(a->duration_ns);
  return &fiber_io_done;
}
 
// srn_sched_drain winds the pool down gracefully, an op already in flight when
// the drain starts still completes, while a submission attempted after the
// drain is fenced into -ECANCELED. On one worker the sleeper runs first and
// parks, then the starter drains while that op is in flight, so the order is
// deterministic. The run returns once both fibers have unwound and the reactor
// is idle -- the drain converges to the same quiescence as a natural finish.
static void test_io_fiber_drain() {
  MAKE_ENGINE(mm, engine);
  MAKE_CONTEXT(engine, ctx);
  srn_scheduler_t *sched = engine->scheduler;
  ASSERT_NOT_NULL(sched);
 
  fiber_drain_args_t sleeper = {
    .sched = sched, .duration_ns = 30ULL * SRN_NS_PER_MS, .err = nullptr
  };
  fiber_drain_args_t starter = {
    .sched = sched, .duration_ns = 30ULL * SRN_NS_PER_MS, .err = nullptr
  };
  ASSERT_NOT_NULL(srn_fiber_spawn(ctx, fiber_drain_sleeper, &sleeper));
  ASSERT_NOT_NULL(srn_fiber_spawn(ctx, fiber_drain_starter, &starter));
 
  srn_sched_run(sched, 1);
 
  TEST_CHECK(sleeper.err == nullptr);                                 // in-flight op finished
  TEST_CHECK(starter.err != nullptr && starter.err->tag == CANCELED); // post-drain submit fenced
  TEST_CHECK(srn_reactor_idle(engine->reactor));                      // converged to quiescence
 
  RELEASE_CONTEXT(ctx);
  SHUTDOWN_ENGINE(mm, engine);
}
 
// A regular file is not pollable, so srn_fiber_read on it cannot go through the
// epoll reactor; it is served by a synchronous read on the worker instead. The
// fiber still gets its bytes, and the reactor is never touched (stays idle).
static void test_io_fiber_file() {
  MAKE_ENGINE(mm, engine);
  MAKE_CONTEXT(engine, ctx);
  srn_scheduler_t *sched = engine->scheduler;
  ASSERT_NOT_NULL(sched);
 
  char path[] = "/tmp/serene_fiber_file_XXXXXX";
  int fd      = mkstemp(path);
  TEST_ASSERT(fd >= 0);
  unlink(path); // anonymous; the open fd keeps it alive
  TEST_ASSERT(write(fd, "diskdata", 8) == 8);
  TEST_ASSERT(lseek(fd, 0, SEEK_SET) == 0);
 
  char buf[16]      = {0};
  fiber_io_args_t a = {.fd = fd, .buf = buf, .len = sizeof(buf)};
  ASSERT_NOT_NULL(srn_fiber_spawn(ctx, fiber_read_entry, &a));
 
  srn_sched_run(sched, 1);
 
  TEST_CHECK(!HAS_ERROR(a.res) && a.res.count == 8);
  TEST_CHECK(memcmp(buf, "diskdata", 8) == 0);
  TEST_CHECK(srn_reactor_idle(engine->reactor)); // never reached the reactor
 
  close(fd);
  RELEASE_CONTEXT(ctx);
  SHUTDOWN_ENGINE(mm, engine);
}
 
// ---------------------------------------------------------------------------
// through a real fiber -- the C10 socket surface (srn_fiber_accept / connect /
// recvfrom / sendto / poll). Each builds a request and parks like read/write;
// accept and recvfrom additionally carry the peer address length back.
// ---------------------------------------------------------------------------
 
typedef struct fiber_accept_args_t {
  int listener;
  struct sockaddr_storage peer;
  srn_io_accept_result_t res;
} fiber_accept_args_t;
 
static srn_fiber_result_t fiber_accept_entry(srn_context_t *ctx, void *arg) {
  UNUSED(ctx);
  fiber_accept_args_t *a = (fiber_accept_args_t *)arg;
  a->res = srn_fiber_accept(a->listener, (struct sockaddr *)&a->peer, sizeof(a->peer));
  return &fiber_io_done;
}
 
typedef struct fiber_connect_args_t {
  int fd;
  struct sockaddr_in to;
  srn_error_t *err;
} fiber_connect_args_t;
 
static srn_fiber_result_t fiber_connect_entry(srn_context_t *ctx, void *arg) {
  UNUSED(ctx);
  fiber_connect_args_t *a = (fiber_connect_args_t *)arg;
  a->err                  = srn_fiber_connect(a->fd, (struct sockaddr *)&a->to, sizeof(a->to));
  return &fiber_io_done;
}
 
// One fiber accepts on a loopback listener while another connects to it, both
// through the suspending wrappers on a single worker. The acceptor gets the
// connected fd with the peer address (and its length) filled in; the connector
// gets a zero result.
static void test_io_fiber_accept_connect() {
  MAKE_ENGINE(mm, engine);
  MAKE_CONTEXT(engine, ctx);
  srn_scheduler_t *sched = engine->scheduler;
  ASSERT_NOT_NULL(sched);
 
  uint16_t port;
  int lst = io_tcp_listener(&port);
 
  int cli = socket(AF_INET, SOCK_STREAM, 0);
  TEST_ASSERT(cli >= 0);
  io_nonblock(cli); // non-blocking so connect parks instead of blocking
 
  fiber_accept_args_t acc  = {.listener = lst};
  fiber_connect_args_t con = {.fd = cli, .to = {.sin_family = AF_INET, .sin_port = port}};
  con.to.sin_addr.s_addr   = htonl(INADDR_LOOPBACK);
 
  ASSERT_NOT_NULL(srn_fiber_spawn(ctx, fiber_accept_entry, &acc));
  ASSERT_NOT_NULL(srn_fiber_spawn(ctx, fiber_connect_entry, &con));
 
  srn_sched_run(sched, 1);
 
  TEST_CHECK(!HAS_ERROR(acc.res)); // accepted fd is in acc.res.fd
  TEST_CHECK(con.err == nullptr);  // connected
  TEST_CHECK(acc.res.addrlen >= sizeof(struct sockaddr_in));
  TEST_CHECK(((struct sockaddr_in *)&acc.peer)->sin_family == AF_INET);
 
  if (!HAS_ERROR(acc.res)) {
    close(acc.res.fd);
  }
  close(cli);
  close(lst);
  RELEASE_CONTEXT(ctx);
  SHUTDOWN_ENGINE(mm, engine);
}
 
typedef struct fiber_recvfrom_args_t {
  int fd;
  char buf[32];
  struct sockaddr_storage from;
  srn_io_recvfrom_result_t res;
} fiber_recvfrom_args_t;
 
static srn_fiber_result_t fiber_recvfrom_entry(srn_context_t *ctx, void *arg) {
  UNUSED(ctx);
  fiber_recvfrom_args_t *a = (fiber_recvfrom_args_t *)arg;
  a->res                   = srn_fiber_recvfrom(
    a->fd, a->buf, sizeof(a->buf), 0, (struct sockaddr *)&a->from, sizeof(a->from)
  );
  return &fiber_io_done;
}
 
typedef struct fiber_sendto_args_t {
  int fd;
  struct sockaddr_in to;
  srn_io_size_result_t res;
} fiber_sendto_args_t;
 
static srn_fiber_result_t fiber_sendto_entry(srn_context_t *ctx, void *arg) {
  UNUSED(ctx);
  fiber_sendto_args_t *a = (fiber_sendto_args_t *)arg;
  a->res = srn_fiber_sendto(a->fd, "ping", 4, 0, (struct sockaddr *)&a->to, sizeof(a->to));
  return &fiber_io_done;
}
 
// One fiber receives a datagram while another sends it, over loopback UDP. The
// receiver gets the bytes and the sender's address length; the sender gets the
// byte count.
static void test_io_fiber_dgram() {
  MAKE_ENGINE(mm, engine);
  MAKE_CONTEXT(engine, ctx);
  srn_scheduler_t *sched = engine->scheduler;
  ASSERT_NOT_NULL(sched);
 
  uint16_t port;
  int rcv = io_udp_bound(&port);
  int snd = socket(AF_INET, SOCK_DGRAM, 0);
  TEST_ASSERT(snd >= 0);
  io_nonblock(snd);
 
  fiber_recvfrom_args_t r = {.fd = rcv};
  fiber_sendto_args_t s   = {.fd = snd, .to = {.sin_family = AF_INET, .sin_port = port}};
  s.to.sin_addr.s_addr    = htonl(INADDR_LOOPBACK);
 
  // Receiver first so it is armed before the datagram is sent.
  ASSERT_NOT_NULL(srn_fiber_spawn(ctx, fiber_recvfrom_entry, &r));
  ASSERT_NOT_NULL(srn_fiber_spawn(ctx, fiber_sendto_entry, &s));
 
  srn_sched_run(sched, 1);
 
  TEST_CHECK(!HAS_ERROR(r.res) && r.res.count == 4);
  TEST_CHECK(memcmp(r.buf, "ping", 4) == 0);
  TEST_CHECK(!HAS_ERROR(s.res) && s.res.count == 4);
  TEST_CHECK(r.res.addrlen >= sizeof(struct sockaddr_in));
 
  close(rcv);
  close(snd);
  RELEASE_CONTEXT(ctx);
  SHUTDOWN_ENGINE(mm, engine);
}
 
// The non-suspending helpers are plain syscall wrappers -- no reactor, no
// scheduler. Build a loopback listener with them, exercise an error path, and
// open a temp file, checking the null-or-error convention and the tag mapping.
static void test_io_inline_helpers() {
  srn_io_fd_result_t s = srn_fiber_socket(AF_INET, SOCK_STREAM, 0);
  TEST_CHECK(!HAS_ERROR(s));
  int lst = s.fd;
 
  struct sockaddr_in a = {.sin_family = AF_INET};
  a.sin_addr.s_addr    = htonl(INADDR_LOOPBACK);
  TEST_CHECK(srn_fiber_bind(lst, (struct sockaddr *)&a, sizeof(a)) == nullptr);
  TEST_CHECK(srn_fiber_listen(lst, 4) == nullptr);
 
  int one = 1;
  TEST_CHECK(srn_fiber_setsockopt(lst, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one)) == nullptr);
  int got      = 0;
  socklen_t gl = sizeof(got);
  TEST_CHECK(srn_fiber_getsockopt(lst, SOL_SOCKET, SO_REUSEADDR, &got, &gl) == nullptr);
 
  TEST_CHECK(srn_fiber_close(lst) == nullptr);
  // Closing the now-stale descriptor fails with a bad-descriptor error.
  srn_error_t *e = srn_fiber_close(lst);
  TEST_CHECK(e != nullptr && e->tag == BAD_DESCRIPTOR);
 
  // open() a real temp file through the helper, then a missing path.
  char path[] = "/tmp/serene_io_helper_XXXXXX";
  int tmp     = mkstemp(path);
  TEST_ASSERT(tmp >= 0);
  close(tmp);
 
  srn_io_fd_result_t o = srn_fiber_open(path, O_RDONLY, 0);
  TEST_CHECK(!HAS_ERROR(o));
  if (!HAS_ERROR(o)) {
    TEST_CHECK(srn_fiber_close(o.fd) == nullptr);
  }
  unlink(path);
 
  srn_io_fd_result_t miss = srn_fiber_open("/tmp/serene_no_such_file_zzq", O_RDONLY, 0);
  TEST_CHECK(HAS_ERROR(miss) && miss.maybe_error->tag == NOT_FOUND);
}
 
// srn_fiber_open forces O_NONBLOCK and O_CLOEXEC, so a descriptor it returns
// is always safe to hand to the suspending calls, the reactor thread must
// never end up waiting on a blocking descriptor.
static void test_io_open_nonblock() {
  srn_io_fd_result_t r = srn_fiber_open("/dev/null", O_RDONLY, 0);
  TEST_ASSERT(!HAS_ERROR(r));
 
  int fl = fcntl(r.fd, F_GETFL, 0);
  TEST_CHECK(fl >= 0);
  TEST_CHECK((fl & O_NONBLOCK) != 0);
 
  int fdfl = fcntl(r.fd, F_GETFD, 0);
  TEST_CHECK(fdfl >= 0);
  TEST_CHECK((fdfl & FD_CLOEXEC) != 0);
 
  TEST_CHECK(srn_fiber_close(r.fd) == nullptr);
}
 
// A FIFO opened through srn_fiber_open is non-blocking, so both ends are
// pollable and a write/read pair crosses it through the reactor like any
// socket or pipe. The read end opens first, with O_NONBLOCK an O_RDONLY open
// returns at once instead of blocking the opener until a writer appears.
static void test_io_fiber_fifo() {
  MAKE_ENGINE(mm, engine);
  MAKE_CONTEXT(engine, ctx);
  srn_scheduler_t *sched = engine->scheduler;
  ASSERT_NOT_NULL(sched);
 
  char dir[] = "/tmp/serene_fifo_XXXXXX";
  TEST_ASSERT(mkdtemp(dir) != nullptr);
  char path[128];
  TEST_ASSERT(snprintf(path, sizeof(path), "%s/fifo", dir) > 0);
  TEST_ASSERT(mkfifo(path, 0600) == 0);
 
  srn_io_fd_result_t rend = srn_fiber_open(path, O_RDONLY, 0);
  TEST_ASSERT(!HAS_ERROR(rend));
  srn_io_fd_result_t wend = srn_fiber_open(path, O_WRONLY, 0);
  TEST_ASSERT(!HAS_ERROR(wend));
 
  char out[]         = "fifo-ping";
  size_t n           = sizeof(out) - 1;
  char in[16]        = {0};
  fiber_io_args_t w  = {.fd = wend.fd, .buf = out, .len = n};
  fiber_io_args_t rd = {.fd = rend.fd, .buf = in, .len = sizeof(in)};
 
  ASSERT_NOT_NULL(srn_fiber_spawn(ctx, fiber_write_entry, &w));
  ASSERT_NOT_NULL(srn_fiber_spawn(ctx, fiber_read_entry, &rd));
 
  srn_sched_run(sched, 1);
 
  TEST_CHECK(!HAS_ERROR(w.res) && w.res.count == n);
  TEST_CHECK(!HAS_ERROR(rd.res) && rd.res.count == n);
  TEST_CHECK(memcmp(in, out, n) == 0);
  TEST_CHECK(srn_reactor_idle(engine->reactor));
 
  TEST_CHECK(srn_fiber_close(rend.fd) == nullptr);
  TEST_CHECK(srn_fiber_close(wend.fd) == nullptr);
  TEST_ASSERT(unlink(path) == 0);
  TEST_ASSERT(rmdir(dir) == 0);
  RELEASE_CONTEXT(ctx);
  SHUTDOWN_ENGINE(mm, engine);
}
 
#endif // __linux__