diff --git a/benchmarks/CMakeLists.txt b/benchmarks/CMakeLists.txt index a7ea70f..a0fccbd 100644 --- a/benchmarks/CMakeLists.txt +++ b/benchmarks/CMakeLists.txt @@ -116,6 +116,16 @@ if(ARKNET_BUILD_TESTS) endif() arknet_add_check(connection_socket_operation_trace 10 arknet_connection_socket_trace_test) endif() +if(UNIX) + # POSIX poll(2) or kqueue peer without Asio or strands; speaks the native control protocol. + add_executable(arknet_poll_connection_control poll_connections.cpp) + target_compile_features(arknet_poll_connection_control PRIVATE cxx_std_20) + target_link_libraries(arknet_poll_connection_control PRIVATE Threads::Threads) + if(ARKNET_BUILD_TESTS) + arknet_add_check(benchmark_poll_connections_help 10 arknet_poll_connection_control --help) + arknet_add_check(benchmark_poll_connections_self_test 20 arknet_poll_connection_control --self-test) + endif() +endif() if(ARKNET_BUILD_TESTS AND UNIX) add_executable(arknet_asio_control_process_test ../tests/connection_load_native_control.cpp) add_dependencies(arknet_asio_control_process_test arknet_asio_connection_control arknet_connection_benchmark) diff --git a/benchmarks/CONNECTIONS.md b/benchmarks/CONNECTIONS.md index 527f921..8538a2c 100644 --- a/benchmarks/CONNECTIONS.md +++ b/benchmarks/CONNECTIONS.md @@ -191,6 +191,25 @@ HELLO, START, DATA, FIN and acknowledgement identity/sequence rules match the original benchmark. Payload is limited to 1–4023 bytes, preserving the original conservative 4096-byte receive-cap validation. +Two diagnostic isolation switches change one mechanism each; their defaults keep +the behavior above, and the final `result` records both. `--session-strands off` +binds every session handler to the context executor instead of a per-session +strand; with a single IO worker this still serializes all handlers. +`--write-buffers joined` copies the length prefix and protocol frame into one +buffer per write instead of passing two buffers; the extra copy is not included in +the declared buffer budget. Pair both peers with the same switches when isolating a +mechanism, and compare against interleaved default runs. + +`--io-histogram on` wraps each socket in a counting stream and adds `io_histogram` +to the `result`: per direction, successful `async_read_some`/`async_write_some` +calls, bytes, short completions and log2 histograms of requested and completed +bytes. `async_read_until` issues zero-byte reads when a frame is already buffered; +these appear as requested 0 and do not call the kernel. Readiness waits and +would-block results are not visible through Asio and are reported as 0. The option +cannot be combined with `--socket-operation-trace`, and the wrapper itself is part +of the measured configuration. The poll control always reports the same object, +including `EAGAIN` counts, waits and returned events. + The default active capacity is 5000, connect parallelism is 32, data inflight capacity is 20000 and the send/data byte limit is 64 MiB. These are declared limits, not measured capacity. `--max-inflight` adds an explicit global DATA count limit; @@ -456,3 +475,96 @@ not the older published IO snapshot time. Validate wall/steady offset stability across anchors; a clock step, large bracket or missing cancellation anchor makes precise cross-process timing inconclusive. Earlier output without these anchors cannot be precisely aligned to packet wall timestamps from its steady epoch alone. + +## POSIX poll control without Asio + +`arknet_poll_connection_control` is a second diagnostic TCP peer, built with the +benchmark option on POSIX systems. It uses neither Asio nor strands: by default one +thread drives non-blocking IPv4 sockets with `poll(2)`. It speaks the same +length-prefixed HELLO, START, DATA, FIN and acknowledgement protocol as +`arknet_asio_connection_control`, so either role can be paired with the native +Asio control. Use the four pairings to separate the event loop of each peer from +the transport underneath. + +```sh +cmake --build build --target arknet_poll_connection_control --parallel 2 +build/benchmarks/arknet_poll_connection_control --self-test +build/benchmarks/arknet_poll_connection_control --role server --listen 127.0.0.1 \ + --port 21002 --run-id 71 --expected-connections 8 --window 4 \ + --hold-seconds 0.1 --deadline-seconds 5 +# Separate terminal/process; the server may also be arknet_asio_connection_control +# with --scenario traffic: +build/benchmarks/arknet_poll_connection_control --role client --target 127.0.0.1 \ + --port 21002 --run-id 71 --client-id 91 --connections 8 --window 4 \ + --hold-seconds 0.1 --deadline-seconds 5 +``` + +The client paces connects with `--connect-rate` and `--connect-parallel`, sends +HELLO with `TCP_NODELAY`, starts traffic after every connection is ready and keeps +`--window` DATA requests outstanding until `--hold-seconds` elapses. Each peer tries +`send` immediately and waits for `POLLOUT` only for the unsent remainder; input +uses the same 4096-byte bound and canonical length rules. Payload is limited to +1–4023 bytes. The traffic scenario is the only scenario. + +On Apple and BSD systems `--events kqueue` replaces `poll(2)` with a loop that +registers events the way the Boost.Asio kqueue reactor does, leaving the protocol, +socket calls and diagnostics unchanged. Sends, receives, accepts and connects are +attempted first; only one that must wait registers its descriptor with +`EV_ADD | EV_CLEAR`, `EVFILT_READ` for reads and `EVFILT_READ` plus `EVFILT_WRITE` +once a write or connect must wait. A write event that arrives with nothing left to +write deletes the `EVFILT_WRITE` registration, and one `kevent` call returns at +most 128 events. Other platforms reject `kqueue`; the `result` event records +`events`. Pair both peers in the same mode to compare the two notification models. + +`--write-pacing serial` reproduces the native control's write cadence in either +event mode: each session writes one frame at a time, trying `send` immediately +when nothing is outstanding. A fully written frame posts its completion to the +next loop pass, which then starts the next queued frame; posted completions make +that pass's wait non-blocking. The default `burst` appends each frame and sends +at once, so a window of requests usually leaves in consecutive calls. The `result` +event records `write_pacing`. + +`--read-pacing serial` reproduces `async_read_until` on the native control's +4096-byte stream buffer. A frame already buffered completes through a posted step +without `recv`. Otherwise one `recv` is attempted and its completion is posted to a +later pass, where framing is checked again; a would-block result waits for one +readiness event, which performs one `recv`. Each completion handles one frame and +then starts the next read; readiness that arrives while no read waits is dropped. +The default `drain` reads until `EAGAIN` and handles every buffered frame at once. +The `result` event records `read_pacing`. + +`--socket-setup asio` prepares sockets the way Boost.Asio's `socket_ops` does on +Apple and FreeBSD: `SO_NOSIGPIPE` on the listener and on every created and +accepted socket, and non-blocking mode through `ioctl(FIONBIO)`. The default sets +`O_NONBLOCK` with `fcntl` and relies on an ignored `SIGPIPE`. Single-buffer reads +and writes already use `recv` and `send` in both programs. The `result` event +records `socket_setup`. + +`--send-queue-watch on` makes either role query every open socket's kernel send +queue every 250 ms. A queue that stays non-empty and unchanged for one second has +made no delivery progress; the peer writes a `send_queue_stalled` event with +`phase` `detected`, then `sample` once per further second (at most 20 per +connection and 64 connections), and `progressed` once the queue changes. Each +event carries ports, sequence state, application and kernel queue bytes and +`tcp_info` from Apple `TCP_CONNECTION_INFO` (state, flags, RTO, congestion window, +slow-start threshold, peer window, transmitted, retransmitted and out-of-order +bytes); other platforms report `"status":"unsupported"`. `stalled_connection` +events include the same `tcp_info`. The scan adds one `getsockopt` per socket per +interval and is part of the measured configuration; the `result` event records +`send_queue_watch`. + +The first request older than `--request-timeout-ms` fails the client. It writes one +`stalled_connection` event per affected connection with `phase` `at_failure`, +then stops admitting requests. `--observe-seconds` keeps reading for that interval +and reports the same connections again with `phase` `after_observation`. The server +reports, at the first premature EOF, deadline or signal, unfinished connections +whose reads stopped at least `--request-timeout-ms` before its latest read on any +connection. Each event contains the local and peer ports, sequence state, bytes +handed to and read from the kernel, bytes still queued in the application, and the +kernel readable and send-queue byte counts (`FIONREAD`; `SO_NWRITE` on macOS, +`SIOCOUTQ` on Linux, otherwise -1). Match peers by reversed ports. + +Stdout ends with one `result` event (`success`, `failure_scope`, counters). Exit +status 0 means success, 1 a failed run and 2 an argument or startup error. This +control has no operation trace, kernel TCP samples or resource events; it does not +replace the arknet benchmark or the native Asio control. diff --git a/benchmarks/CONNECTIONS_CN.md b/benchmarks/CONNECTIONS_CN.md index 0571f9f..425d8d8 100644 --- a/benchmarks/CONNECTIONS_CN.md +++ b/benchmarks/CONNECTIONS_CN.md @@ -185,6 +185,19 @@ build/benchmarks/arknet_asio_connection_control --role client --target 127.0.0.1 序号规则与原 benchmark 一致。payload 限制为 1–4023 字节，保持原程序对 4096 字节 接收上限的保守校验。 +两个诊断隔离开关各只改变一种机制；默认值保持上述行为，最终 `result` 会记录两者。 +`--session-strands off` 把每个会话的回调绑定到 context executor，而不是每会话 strand； +只有一个 IO 工作线程时，所有回调仍然串行执行。`--write-buffers joined` 在每次写入时 +把长度前缀和协议帧复制到同一个 buffer，而不是传入两个 buffer；这份额外副本不计入声明的 +buffer 预算。隔离某种机制时，两端应使用相同开关，并与交替运行的默认配置对比。 + +`--io-histogram on` 用一个计数 stream 包装每个 socket，并在 `result` 中增加 `io_histogram`： +按读写方向记录成功的 `async_read_some`/`async_write_some` 调用次数、字节数、未完成全部请求的 +次数，以及请求字节数和完成字节数的 log2 直方图。帧已在缓冲区中时，`async_read_until` 会发起 +零字节读取；它们记为请求 0 字节，不会调用内核。Asio 不暴露就绪等待和 would-block 结果， +这两项报告为 0。该选项不能与 `--socket-operation-trace` 同时使用，包装层本身也属于被测配置。 +poll 对照总是输出同样的对象，并包含 `EAGAIN` 次数、等待次数和返回的事件数。 + 默认活动连接容量为 5000，最多同时建连 32 个，DATA 在途容量为 20000，发送和 DATA 字节限制为 64 MiB。这些是配置边界，不是已验证容量。`--max-inflight` 增加显式全局 DATA 数量限制；声明的连接窗口必须同时满足数量和字节限制。控制帧也占用发送字节额度。 @@ -374,3 +387,73 @@ OS 缺陷。首次发布仍依赖 IO worker，发布前 worker 已阻塞时不 `[w+q-b, w+q-a]`，再纳入PCAP时间戳量化误差。应使用query区间，不用较旧的IO状态快照时间。 须检查anchor间wall/steady偏移稳定性；时钟跳变、过宽区间或缺取消anchor时，精确跨进程 对齐不可判定。旧输出仅有steady epoch，无法精确对齐包的wall时间戳。 + +## 不使用 Asio 的 POSIX poll 对照 + +`arknet_poll_connection_control` 是第二个诊断用 TCP 对端，在 POSIX 系统上随 benchmark +选项构建。它不使用 Asio 或 strand：默认由一个线程用 `poll(2)` 驱动非阻塞 +IPv4 socket。它与 `arknet_asio_connection_control` 使用相同的长度前缀 HELLO、START、 +DATA、FIN 及确认协议，因此任一角色都可以与原生 Asio 对照配对。用四种配对区分两端各自的 +事件循环与其下的传输层。 + +```sh +cmake --build build --target arknet_poll_connection_control --parallel 2 +build/benchmarks/arknet_poll_connection_control --self-test +build/benchmarks/arknet_poll_connection_control --role server --listen 127.0.0.1 \ + --port 21002 --run-id 71 --expected-connections 8 --window 4 \ + --hold-seconds 0.1 --deadline-seconds 5 +# 在另一个终端或进程中运行；服务端也可以换成带 --scenario traffic 的 +# arknet_asio_connection_control： +build/benchmarks/arknet_poll_connection_control --role client --target 127.0.0.1 \ + --port 21002 --run-id 71 --client-id 91 --connections 8 --window 4 \ + --hold-seconds 0.1 --deadline-seconds 5 +``` + +客户端按 `--connect-rate` 和 `--connect-parallel` 控制建连节奏，开启 `TCP_NODELAY` 后发送 +HELLO，所有连接就绪后开始流量，并在 `--hold-seconds` 内保持 `--window` 个 DATA 请求在途。 +两端都先立即调用 `send`，只对未发出的剩余部分等待 `POLLOUT`；输入使用相同的 4096 字节上限 +和规范长度规则。payload 限制为 1–4023 字节。只支持 traffic 场景。 + +在 Apple 和 BSD 系统上，`--events kqueue` 用一个按 Boost.Asio kqueue reactor 方式注册事件 +的循环替换 `poll(2)`，协议、socket 调用和诊断输出保持不变。send、recv、accept 和 connect +都先直接尝试；只有需要等待的操作才用 `EV_ADD | EV_CLEAR` 注册描述符：读取只注册 +`EVFILT_READ`，写入或建连需要等待时同时注册 `EVFILT_READ` 和 `EVFILT_WRITE`。写事件到达时 +若已无待写数据，就删除 `EVFILT_WRITE` 注册；每次 `kevent` 调用最多返回 128 个事件。其他 +平台会拒绝 `kqueue`；`result` 事件记录 `events`。比较两种通知模型时，两端应使用相同模式。 + +`--write-pacing serial` 在两种事件模式下都能复现原生对照的写入节奏：每个会话同一时间只写一帧， +没有在途写入时立即尝试 `send`。一帧完全写出后，把完成通知投递到下一轮循环，由那一轮再开始写 +队列中的下一帧；有待处理的完成时，那一轮的等待不会阻塞。默认的 `burst` 会追加每一帧并立即 +发送，因此同一窗口的请求通常在连续的调用中发出。`result` 事件记录 `write_pacing`。 + +`--read-pacing serial` 复现原生对照在 4096 字节 stream buffer 上的 `async_read_until`： +缓冲区里已有完整帧时，不调用 `recv`，通过投递的一步完成；否则尝试一次 `recv`，并把完成投递到 +下一轮，在那一轮重新检查帧边界；遇到 would-block 时等待一次就绪事件，由它执行一次 `recv`。 +每次完成只处理一帧，然后开始下一次读取；没有读取在等待时到达的就绪事件会被丢弃。默认的 +`drain` 一直读到 `EAGAIN`，并一次处理缓冲区里的全部帧。`result` 事件记录 `read_pacing`。 + +`--socket-setup asio` 按 Boost.Asio 的 `socket_ops` 在 Apple 和 FreeBSD 上的做法准备 socket： +监听 socket 以及每个创建和接受的 socket 都设置 `SO_NOSIGPIPE`，并用 `ioctl(FIONBIO)` 设置非阻塞。 +默认方式用 `fcntl` 设置 `O_NONBLOCK`，并依靠忽略 `SIGPIPE`。两个程序的单缓冲区读写本来就都使用 +`recv` 和 `send`。`result` 事件记录 `socket_setup`。 + +`--send-queue-watch on` 让任一角色每 250 ms 查询一次每个打开 socket 的内核发送队列。发送队列 +连续一秒非空且没有变化，说明没有任何投递进展；此时输出 `phase` 为 `detected` 的 +`send_queue_stalled` 事件，之后每多一秒输出一次 `sample`（每个连接最多 20 次，最多跟踪 64 个 +连接），队列一旦变化就输出 `progressed`。每条事件包含端口、序号状态、应用层与内核队列字节数， +以及来自 Apple `TCP_CONNECTION_INFO` 的 `tcp_info`（状态、标志、RTO、拥塞窗口、慢启动阈值、 +对端窗口、已发送、重传和乱序字节数）；其他平台报告 `"status":"unsupported"`。 +`stalled_connection` 事件也包含同样的 `tcp_info`。每个间隔对每个 socket 多一次 `getsockopt`， +这属于被测配置的一部分；`result` 事件记录 `send_queue_watch`。 + +第一个超过 `--request-timeout-ms` 的请求会使客户端失败。客户端为每个受影响的连接输出一条 +`phase` 为 `at_failure` 的 `stalled_connection` 事件，然后停止发出新请求。 +`--observe-seconds` 会在该时长内继续读取，并以 `phase` `after_observation` 再次报告同一批 +连接。服务端在第一次提前 EOF、到达截止时间或收到信号时，报告读取停止时间比任一连接最近一次 +读取早至少 `--request-timeout-ms` 的未完成连接。每条事件包含本端与对端端口、序号状态、交给 +内核及从内核读取的字节数、仍在应用层排队的字节数，以及内核可读字节数和发送队列字节数 +（`FIONREAD`；macOS 用 `SO_NWRITE`，Linux 用 `SIOCOUTQ`，其他平台为 -1）。按端口互换匹配两端。 + +stdout 以一条 `result` 事件结束（`success`、`failure_scope` 及计数）。退出码 0 表示成功， +1 表示运行失败，2 表示参数或启动错误。该对照不提供操作历史、内核 TCP 采样或资源事件， +不能替代 arknet benchmark 或原生 Asio 对照。 diff --git a/benchmarks/asio_connections.cpp b/benchmarks/asio_connections.cpp index 9818105..24d3c96 100644 --- a/benchmarks/asio_connections.cpp +++ b/benchmarks/asio_connections.cpp @@ -75,6 +75,8 @@ struct options std::size_t inflight_bytes = 64 * 1024 * 1024, budget = 512 * 1024 * 1024; double hold = 1, deadline = 30, cleanup = 5; bool operation_trace{}, socket_operation_trace{}, tcp_kernel_samples{}, socket_observer{}; + // Diagnostic isolation switches; the defaults preserve the original control. + bool session_strands = true, joined_writes{}, io_histogram{}; }; std::uint64_t integer(std::string_view text) { @@ -116,6 +118,10 @@ void validate(const options& o) { { throw std::invalid_argument("socket-operation-trace requires operation-trace on"); } + if (o.io_histogram && o.socket_operation_trace) + { + throw std::invalid_argument("io-histogram and socket-operation-trace both wrap the socket"); + } if ((o.role != "server" && o.role != "client") || (o.scenario != "idle" && o.scenario != "traffic") || !o.run || !o.client || o.host.empty() || o.port > 65535 || (o.role == "client" && !o.port)) { @@ -165,8 +171,11 @@ void help() { " --socket-operation-trace off|on (default off; requires operation-trace on)\n" " --tcp-kernel-samples off|on (default off; requires operation-trace on; Apple TCP info)\n" " --socket-observer off|on (default off; Unix only; non-consuming controller observations)\n" + " --session-strands on|off (default on; off binds sessions to the single-worker context executor)\n" + " --write-buffers split|joined (default split; joined sends prefix and frame as one buffer)\n" + " --io-histogram off|on (default off; per-call read/write byte histograms in the result)\n" " --self-test (unit checks only; no sockets)\n" - "One native Asio context/worker, session strands, TCP_NODELAY, 4096B read_until cap.\n" + "One native Asio context/worker, session strands by default, TCP_NODELAY, 4096B read_until cap.\n" "Defaults permit 5000 active connections, 32 connecting and 20000 data inflight.\n" "Server port 0 reports its selected port; expected count must be nonzero.\n" "Unsupported options (including sample-ms and multiple workers) are rejected.\n"; @@ -287,6 +296,18 @@ options arguments(int argc, char** argv) { { o.socket_observer = value == "on"; } + else if (key == "--session-strands" && (value == "on" || value == "off")) + { + o.session_strands = value == "on"; + } + else if (key == "--io-histogram" && (value == "on" || value == "off")) + { + o.io_histogram = value == "on"; + } + else if (key == "--write-buffers" && (value == "split" || value == "joined")) + { + o.joined_writes = value == "joined"; + } else if (key == "--io-threads" && integer(value) == 1) { } @@ -448,6 +469,7 @@ struct wire_frame std::array prefix{}; std::size_t prefix_size{}; std::string body; + std::string joined; // Prefix and body copy for --write-buffers joined only. bool is_data() const { return connection_load::read_integer(body, 8) == static_cast(frame_kind::data); @@ -516,7 +538,9 @@ struct session : std::enable_shared_from_this void trace(connection_load::operation_phase phase, std::uint64_t bytes, std::uint64_t sequence, int error = 0); application& app; - asio::strand strand; + // A per-session strand, or the context executor when --session-strands off. + // Only one IO worker exists, so both serialize every session handler. + asio::any_io_executor strand; tcp::socket socket; asio::streambuf input{4096}; protocol_state state; @@ -529,6 +553,7 @@ struct session : std::enable_shared_from_this bool read_pending{}; connection_load::operation_ring* operations{}; std::unique_ptr> socket_trace; + std::unique_ptr> io_count; std::uint64_t read_completions{}, write_completions{}, last_write_kind{}, last_write_sequence{}; clock_type::time_point last_read_at{}, last_write_at{}; }; @@ -566,6 +591,7 @@ struct application connection_load::operation_recorder operations; connection_load::tcp_kernel_samples kernel_samples; connection_load::socket_observer socket_observer; + connection_load::io_histogram histogram; // IO worker only. std::atomic marker_completed_us{}; bool trace_emitted{}; asio::io_context io{1}; @@ -596,7 +622,8 @@ struct application session::session(application& owner, asio::io_context& context) : app(owner) - , strand(asio::make_strand(context)) + , strand(owner.o.session_strands ? asio::any_io_executor(asio::make_strand(context)) + : asio::any_io_executor(context.get_executor())) , socket(strand) { // Match the arknet driver's initial receive reservation as well as its cap. input.prepare(4096); @@ -737,8 +764,11 @@ void session::write() { writing = queued.front(); queued.pop_front(); const auto value = writing; - const std::array buffers{ - asio::buffer(value->prefix.data(), value->prefix_size), asio::buffer(value->body)}; + if (app.o.joined_writes) + { + value->joined.assign(value->prefix.data(), value->prefix_size); + value->joined += value->body; + } ++app.callbacks; if (operations) { @@ -796,14 +826,30 @@ void session::write() { } a.drained(); }); - if (socket_trace) + // Exactly one branch runs, so the completion is moved once. + const auto start = [&](const auto& buffers) { + if (socket_trace) + { + socket_trace->write_sequence(connection_load::read_integer(value->body, 40)); + asio::async_write(*socket_trace, buffers, std::move(completion)); + } + else if (io_count) + { + asio::async_write(*io_count, buffers, std::move(completion)); + } + else + { + asio::async_write(socket, buffers, std::move(completion)); + } + }; + if (app.o.joined_writes) { - socket_trace->write_sequence(connection_load::read_integer(value->body, 40)); - asio::async_write(*socket_trace, buffers, std::move(completion)); + start(asio::buffer(value->joined)); } else { - asio::async_write(socket, buffers, std::move(completion)); + start(std::array{ + asio::buffer(value->prefix.data(), value->prefix_size), asio::buffer(value->body)}); } } @@ -866,6 +912,10 @@ void session::read() { socket_trace->read_sequence(state.next_receive); asio::async_read_until(*socket_trace, input, dgram_match{}, std::move(completion)); } + else if (io_count) + { + asio::async_read_until(*io_count, input, dgram_match{}, std::move(completion)); + } else { asio::async_read_until(socket, input, dgram_match{}, std::move(completion)); @@ -880,6 +930,10 @@ void session::enable_trace() { socket_trace = std::make_unique>(socket, *socket_operations, app.began); } + else if (app.o.io_histogram) + { + io_count = std::make_unique>(socket, app.histogram); + } } void session::trace(connection_load::operation_phase phase, std::uint64_t bytes, std::uint64_t sequence, int error) { @@ -1508,7 +1562,14 @@ void application::result() const { << ",\"io_contexts\":1,\"io_workers\":1,\"controller_on_io_worker\":false,\"initial_buffer_bytes\":4096," "\"tcp_nodelay\":true,\"max_" "buffer_bytes\":4096" - << ",\"role\":" << std::quoted(o.role) + << ",\"session_strands\":" << (o.session_strands ? "true" : "false") + << ",\"write_buffers\":" << (o.joined_writes ? "\"joined\"" : "\"split\""); + if (o.io_histogram) + { + std::cout << ",\"io_histogram\":"; + histogram.write_json(std::cout); + } + std::cout << ",\"role\":" << std::quoted(o.role) << ",\"success\":" << (failure.empty() && clean && complete ? "true" : "false") << ",\"cleanup_complete\":" << (clean ? "true" : "false") << ",\"error\":" << std::quoted(failure) << ",\"attempted\":" << attempted << ",\"accepted\":" << accepted << ",\"established\":" << established @@ -1586,6 +1647,26 @@ void self_test() { require(kernel_enabled.deadline == enabled.deadline && kernel_enabled.request_ms == enabled.request_ms && kernel_enabled.connections == enabled.connections && kernel_enabled.window == enabled.window && kernel_enabled.payload == enabled.payload); + const auto parse_values = [](std::vector values) { + values.insert( + values.begin(), {"control", "--role", "client", "--target", "127.0.0.1", "--port", "1", "--run-id", "1"}); + std::vector argv; + for (auto& value : values) + { + argv.push_back(value.data()); + } + return arguments(static_cast(argv.size()), argv.data()); + }; + const auto switched = + parse_values({"--session-strands", "off", "--write-buffers", "joined", "--io-histogram", "on"}); + require(disabled.session_strands && !disabled.joined_writes && !disabled.io_histogram && + !switched.session_strands && switched.joined_writes && switched.io_histogram); + bool double_wrap_rejected = false; + try + { parse_values({"--operation-trace", "on", "--socket-operation-trace", "on", "--io-histogram", "on"}); } + catch (const std::invalid_argument&) + { double_wrap_rejected = true; } + require(double_wrap_rejected); // Synthetic queries only: exercise retention/JSON without sockets or load. connection_load::tcp_kernel_samples retained(true); unsigned kernel_queries{}; diff --git a/benchmarks/connections_io_histogram.hpp b/benchmarks/connections_io_histogram.hpp new file mode 100644 index 0000000..3d50f4e --- /dev/null +++ b/benchmarks/connections_io_histogram.hpp @@ -0,0 +1,77 @@ +// Copyright (c) 2026 OpenArkStudio. Distributed under BSL-1.0. See LICENSE. +#pragma once + +#include +#include +#include +#include +#include +#include + +namespace connection_load { + +// Per-call socket read/write sizes for comparing IO cadence between diagnostic +// peers. Log2 byte buckets: 0, 1, 2-3, ..., 2048-4095 and 4096 or more. +// Updated only by the single IO thread; read after that thread has stopped. +struct io_histogram +{ + static constexpr std::size_t buckets = 14; + + static std::size_t bucket(std::size_t bytes) noexcept { + return (std::min)(static_cast(std::bit_width(bytes)), buckets - 1); + } + + struct direction + { + std::array requested{}, completed{}; + std::uint64_t calls{}, bytes{}, short_calls{}, would_block{}; + + void record(std::size_t requested_bytes, std::size_t completed_bytes) noexcept { + ++calls; + bytes += completed_bytes; + ++requested[bucket(requested_bytes)]; + ++completed[bucket(completed_bytes)]; + if (completed_bytes < requested_bytes) + { + ++short_calls; + } + } + + void write_json(std::ostream& out) const { + out << "{\"calls\":" << calls << ",\"bytes\":" << bytes << ",\"short_calls\":" << short_calls + << ",\"would_block\":" << would_block << ",\"requested\":"; + write_buckets(out, requested); + out << ",\"completed\":"; + write_buckets(out, completed); + out << '}'; + } + }; + + direction read, write; + std::uint64_t waits{}, events{}; + + void write_json(std::ostream& out) const { + out << "{\"bucket_upper_bytes\":[0"; + for (std::size_t index = 1; index + 1 < buckets; ++index) + { + out << ',' << ((std::uint64_t{1} << index) - 1); + } + out << ",null],\"read\":"; + read.write_json(out); + out << ",\"write\":"; + write.write_json(out); + out << ",\"waits\":" << waits << ",\"events\":" << events << '}'; + } + +private: + static void write_buckets(std::ostream& out, const std::array& values) { + out << '['; + for (std::size_t index = 0; index < buckets; ++index) + { + out << (index ? "," : "") << values[index]; + } + out << ']'; + } +}; + +} // namespace connection_load diff --git a/benchmarks/connections_socket_trace.hpp b/benchmarks/connections_socket_trace.hpp index e43bed8..2bf6885 100644 --- a/benchmarks/connections_socket_trace.hpp +++ b/benchmarks/connections_socket_trace.hpp @@ -14,6 +14,7 @@ #endif +#include "connections_io_histogram.hpp" #include "connections_trace.hpp" namespace connection_load { @@ -108,6 +109,54 @@ class socket_trace_stream std::uint64_t read_sequence_{}, write_sequence_{}; }; +template +struct io_count_completion +{ + Handler handler; + io_histogram::direction* direction; + std::size_t requested; + + void operator()(socket_trace_error error, std::size_t bytes) { + if (!error) + { + direction->record(requested, bytes); + } + handler(error, bytes); + } +}; + +// Native-control cadence diagnostics only: counts each successful stream call. +template +class io_count_stream +{ +public: + using executor_type = typename Stream::executor_type; + + io_count_stream(Stream& stream, io_histogram& histogram) + : stream_(stream) + , histogram_(histogram) {} + + executor_type get_executor() noexcept { + return stream_.get_executor(); + } + + template + void async_read_some(const Buffers& buffers, Handler&& handler) { + stream_.async_read_some(buffers, io_count_completion>{std::forward(handler), + &histogram_.read, socket_trace_asio::buffer_size(buffers)}); + } + + template + void async_write_some(const Buffers& buffers, Handler&& handler) { + stream_.async_write_some(buffers, io_count_completion>{std::forward(handler), + &histogram_.write, socket_trace_asio::buffer_size(buffers)}); + } + +private: + Stream& stream_; + io_histogram& histogram_; +}; + } // namespace connection_load #if ARKNET_CONTROL_BOOST @@ -136,4 +185,20 @@ struct associator, } }; +template class Associator, class Handler, class DefaultCandidate> +struct associator, DefaultCandidate> + : Associator +{ + static typename Associator::type get( + const connection_load::io_count_completion& value) noexcept { + return Associator::get(value.handler); + } + + static auto get( + const connection_load::io_count_completion& value, const DefaultCandidate& candidate) noexcept + -> decltype(Associator::get(value.handler, candidate)) { + return Associator::get(value.handler, candidate); + } +}; + } // namespace asio diff --git a/benchmarks/poll_connections.cpp b/benchmarks/poll_connections.cpp new file mode 100644 index 0000000..060688a --- /dev/null +++ b/benchmarks/poll_connections.cpp @@ -0,0 +1,1588 @@ +// Copyright (c) 2026 OpenArkStudio. Distributed under BSL-1.0. See LICENSE. +// +// Diagnostic TCP peer without Asio or strands. One thread drives non-blocking +// sockets with poll(2), or with --events kqueue a loop that registers events the +// way Boost.Asio's kqueue reactor does. It speaks the native Asio control's wire +// protocol, so either role can be paired with arknet_asio_connection_control. + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#if defined(__linux__) +#include + +#endif + +#if defined(__APPLE__) || defined(__FreeBSD__) || defined(__OpenBSD__) || defined(__DragonFly__) +#include + +#define ARKNET_POLL_CONTROL_HAS_KQUEUE 1 +#else +#define ARKNET_POLL_CONTROL_HAS_KQUEUE 0 +#endif + +#include "connections_io_histogram.hpp" +#include "connections_kernel_samples.hpp" +#include "connections_protocol.hpp" + +namespace { + +using clock_type = std::chrono::steady_clock; +using connection_load::frame_kind; +using connection_load::header_bytes; + +constexpr std::size_t max_wire_bytes = 4096; + +volatile std::sig_atomic_t signalled = 0; + +struct options +{ + std::string role, host = "127.0.0.1"; + int port = -1; + std::uint64_t run = 1, client = 1, connections = 0, expected = 0, rate = 1000, parallel = 32, payload = 1024, + window = 4, connect_ms = 5000, request_ms = 5000; + double hold = 3, deadline = 40, observe = 0; + bool kqueue_events{}, serial_writes{}, serial_reads{}, asio_socket_setup{}, send_queue_watch{}; +}; + +std::uint64_t integer(const std::string& value) { + if (value.empty() || value.find_first_not_of("0123456789") != std::string::npos) + { + throw std::invalid_argument("expected a decimal integer: " + value); + } + return std::stoull(value); +} + +double seconds(const std::string& value) { + std::size_t used{}; + const auto result = std::stod(value, &used); + if (used != value.size() || !(result >= 0)) + { + throw std::invalid_argument("expected non-negative seconds: " + value); + } + return result; +} + +options parse(const std::vector& arguments) { + options o; + std::set seen; + if (arguments.size() % 2) + { + throw std::invalid_argument("options must be --key value pairs"); + } + for (std::size_t index = 0; index < arguments.size(); index += 2) + { + const auto& key = arguments[index]; + const auto& value = arguments[index + 1]; + if (!seen.insert(key).second) + { + throw std::invalid_argument("duplicate option " + key); + } + if (key == "--role") + { + o.role = value; + } + else if (key == "--listen" || key == "--target") + { + o.host = value; + } + else if (key == "--port") + { + const auto port = integer(value); + if (port > 65535) + { + throw std::invalid_argument("port out of range"); + } + o.port = static_cast(port); + } + else if (key == "--run-id") + { + o.run = integer(value); + } + else if (key == "--client-id") + { + o.client = integer(value); + } + else if (key == "--connections") + { + o.connections = integer(value); + } + else if (key == "--expected-connections") + { + o.expected = integer(value); + } + else if (key == "--connect-rate") + { + o.rate = integer(value); + } + else if (key == "--connect-parallel") + { + o.parallel = integer(value); + } + else if (key == "--payload") + { + o.payload = integer(value); + } + else if (key == "--window") + { + o.window = integer(value); + } + else if (key == "--connect-timeout-ms") + { + o.connect_ms = integer(value); + } + else if (key == "--request-timeout-ms") + { + o.request_ms = integer(value); + } + else if (key == "--hold-seconds") + { + o.hold = seconds(value); + } + else if (key == "--deadline-seconds") + { + o.deadline = seconds(value); + } + else if (key == "--observe-seconds") + { + o.observe = seconds(value); + } + else if (key == "--send-queue-watch" && (value == "off" || value == "on")) + { + o.send_queue_watch = value == "on"; + } + else if (key == "--socket-setup" && (value == "default" || value == "asio")) + { + o.asio_socket_setup = value == "asio"; + } + else if (key == "--read-pacing" && (value == "drain" || value == "serial")) + { + o.serial_reads = value == "serial"; + } + else if (key == "--write-pacing" && (value == "burst" || value == "serial")) + { + o.serial_writes = value == "serial"; + } + else if (key == "--events" && (value == "poll" || value == "kqueue")) + { + if (value == "kqueue" && !ARKNET_POLL_CONTROL_HAS_KQUEUE) + { + throw std::invalid_argument("--events kqueue is unavailable on this platform"); + } + o.kqueue_events = value == "kqueue"; + } + else + { + throw std::invalid_argument("unknown option " + key); + } + } + const bool server = o.role == "server"; + const bool client = o.role == "client"; + in_addr address{}; + if ((!server && !client) || o.port < 0 || (client && o.port == 0) || + (server && (seen.contains("--target") || seen.contains("--connections") || seen.contains("--client-id") || + seen.contains("--observe-seconds"))) || + (client && (seen.contains("--listen") || seen.contains("--expected-connections"))) || + inet_pton(AF_INET, o.host.c_str(), &address) != 1) + { + throw std::invalid_argument("missing/invalid role, IPv4 address or port, or role-inappropriate option"); + } + if (client) + { + o.expected = o.connections; + } + if (!o.run || !o.client || !o.expected || !o.rate || !o.parallel || !o.window || !o.connect_ms || !o.request_ms || + !o.payload || o.payload > 4023 || !(o.hold > 0) || !(o.deadline > o.hold)) + { + throw std::invalid_argument("invalid count, payload, timeout, hold or deadline"); + } + return o; +} + +void append_frame(std::string& output, const std::string& body) { + const auto size = body.size(); + if (size < 254) + { + output.push_back(static_cast(size)); + } + else + { + output.push_back(static_cast(254)); + output.push_back(static_cast(size & 0xff)); + output.push_back(static_cast(size >> 8)); + } + output += body; +} + +enum class frame_status +{ + incomplete, + invalid, + complete +}; + +struct frame_view +{ + frame_status status = frame_status::incomplete; + std::size_t prefix{}, size{}; +}; + +// Same canonical-length rules and 4096-byte wire cap as the native control. +frame_view inspect(std::string_view bytes) { + if (bytes.empty()) + { + return {}; + } + const auto marker = static_cast(bytes[0]); + const std::size_t width = marker == 255 ? 8 : marker == 254 ? 2 : 0; + if (bytes.size() < 1 + width) + { + return {}; + } + std::uint64_t size = width ? 0 : marker; + for (std::size_t byte = 0; byte < width; ++byte) + { + size |= std::uint64_t(static_cast(bytes[1 + byte])) << (byte * 8); + } + if ((width == 2 && size < 254) || (width == 8 && size < 65536) || size > max_wire_bytes - 1 - width) + { + return {frame_status::invalid}; + } + if (bytes.size() - 1 - width < size) + { + return {frame_status::incomplete, 1 + width, static_cast(size)}; + } + return {frame_status::complete, 1 + width, static_cast(size)}; +} + +std::uint64_t elapsed_ms(clock_type::time_point from, clock_type::time_point to) { + return static_cast(std::chrono::duration_cast(to - from).count()); +} + +struct session +{ + int fd = -1; + connection_load::identity owner{}; + bool connecting{}, hello{}, started{}, finish_sent{}, finished{}, peer_eof{}; + std::uint64_t next_send = 1, next_receive = 1; + std::vector sent_at; + clock_type::time_point pending_since{}, connect_since{}, last_read_at{}; + std::string output, input; + std::size_t output_offset{}; + std::uint64_t bytes_written{}, bytes_read{}; + bool watched{}; + int kevents{}; // kqueue filters registered: 0, READ, or READ and WRITE. + // --write-pacing serial: frames waiting behind the one being written, and + // whether the written frame's completion still awaits the next loop pass. + std::deque queued; + bool writing{}, completion_posted{}; + // --read-pacing serial: waiting for readiness, or a read completion awaits the next pass. + bool read_waiting{}, read_posted{}; + // --send-queue-watch: last kernel send-queue size, when it last changed or + // emptied, and how many stall events this session has emitted. + long long send_queue_last{}; + clock_type::time_point send_queue_since{}; + int send_queue_reports{}; +}; + +class peer +{ +public: + peer(options value, std::ostream& out, std::atomic* published_port = nullptr) + : o_(std::move(value)) + , out_(out) + , published_port_(published_port) {} + + peer(const peer&) = delete; + peer& operator=(const peer&) = delete; + + ~peer() { + close_all(); + if (listener_ >= 0) + { + ::close(listener_); + } + if (kqueue_ >= 0) + { + ::close(kqueue_); + } + } + + int run(); + +private: + bool client() const { + return o_.role == "client"; + } + + void open_listener(); + void launch(); + void connected(session& s); + void connect_ready(session& s); + void wait_poll(); +#if ARKNET_POLL_CONTROL_HAS_KQUEUE + void register_events(std::uintptr_t id, int fd, int count, int& registered); + void wait_kqueue(); +#endif + void accept_all(); + void readable(session& s); + void flush(session& s); + void start_write(session& s); + void run_posted(); + void read_step(session& s); + void read_resume(session& s); + void post(session& s, bool read); + void read_ready(session& s); + static bool output_idle(const session& s); + void send(session& s, const std::string& body); + void frame(session& s, std::string_view body); + void client_frame(session& s, const connection_load::frame_header& h); + void server_frame(session& s, const connection_load::frame_header& h); + void fill(session& s); + void finish(session& s); + void eof(session& s); + void close(session& s); + void close_all(); + void fail(const std::string& scope, const std::string& reason); + void advance(clock_type::time_point now); + void snapshot(const char* phase, clock_type::time_point now); + void watch_send_queues(clock_type::time_point now); + void emit_send_queue(const session& s, const char* phase, clock_type::time_point now, long long queued); + void result(clock_type::time_point now) const; + bool complete() const; + + options o_; + std::ostream& out_; + std::atomic* published_port_; + int listener_ = -1, kqueue_ = -1; + bool listener_registered_{}; + std::vector descriptors_; + // Posted completions: session index and whether it is a read (else a write). + std::vector> posted_, posted_ready_; + std::vector owners_; + std::vector sessions_; + std::set> identities_; + connection_load::io_histogram histogram_; + clock_type::time_point began_ = clock_type::now(), next_launch_ = began_, until_{}, failed_at_{}, latest_read_{}, + next_watch_ = began_; + std::size_t send_queue_tracked_{}; + std::uint64_t attempted_{}, connecting_{}, established_{}, accepted_{}, ready_{}, started_{}, finished_{}, + disconnected_{}, outstanding_{}, data_sent_{}, data_received_{}, timeouts_{}; + bool starting_{}, traffic_{}, draining_{}, failed_{}, snapshot_taken_{}; + std::string failure_scope_, failure_reason_; +}; + +// --socket-setup asio follows Boost.Asio's socket_ops on Apple/FreeBSD: SO_NOSIGPIPE +// on every created and accepted socket, and non-blocking mode through +// ioctl(FIONBIO). The default sets O_NONBLOCK with fcntl and relies on SIGPIPE +// being ignored. +void set_nonblocking(int fd, bool asio_setup) { + if (asio_setup) + { +#if defined(SO_NOSIGPIPE) + int enabled = 1; + if (::setsockopt(fd, SOL_SOCKET, SO_NOSIGPIPE, &enabled, sizeof(enabled)) != 0) + { + throw std::runtime_error(std::string("SO_NOSIGPIPE: ") + std::strerror(errno)); + } +#endif + int value = 1; + if (::ioctl(fd, FIONBIO, &value) != 0) + { + throw std::runtime_error(std::string("FIONBIO: ") + std::strerror(errno)); + } + return; + } + const auto flags = ::fcntl(fd, F_GETFL, 0); + if (flags < 0 || ::fcntl(fd, F_SETFL, flags | O_NONBLOCK) != 0) + { + throw std::runtime_error(std::string("O_NONBLOCK: ") + std::strerror(errno)); + } +} + +void set_no_delay(int fd) { + int enabled = 1; + if (::setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &enabled, sizeof(enabled)) != 0) + { + throw std::runtime_error(std::string("TCP_NODELAY: ") + std::strerror(errno)); + } +} + +int port_of(int fd, bool remote) { + sockaddr_in address{}; + socklen_t length = sizeof(address); + const auto result = remote ? ::getpeername(fd, reinterpret_cast(&address), &length) + : ::getsockname(fd, reinterpret_cast(&address), &length); + return result == 0 ? ntohs(address.sin_port) : -1; +} + +long long readable_bytes(int fd) { + int value{}; + return ::ioctl(fd, FIONREAD, &value) == 0 ? value : -1; +} + +// Bytes the kernel still holds in the send queue (unsent or unacknowledged). +long long send_queue_bytes(int fd) { +#if defined(SO_NWRITE) + int value{}; + socklen_t length = sizeof(value); + return ::getsockopt(fd, SOL_SOCKET, SO_NWRITE, &value, &length) == 0 ? value : -1; +#elif defined(SIOCOUTQ) + int value{}; + return ::ioctl(fd, SIOCOUTQ, &value) == 0 ? value : -1; +#else + (void)fd; + return -1; +#endif +} + +// Apple TCP_CONNECTION_INFO through the shared kernel-sample decoder; elsewhere +// the status is "unsupported" and no counters are written. +void write_tcp_info(std::ostream& out, int fd) { + const auto info = connection_load::query_tcp_kernel_info(fd); + using status = connection_load::tcp_kernel_info::status; + if (info.result != status::available) + { + out << "{\"status\":\"" + << (info.result == status::unsupported ? "unsupported" + : info.result == status::closed ? "closed" + : info.result == status::short_response ? "short_response" + : "query_failed") + << "\",\"error\":" << info.error << '}'; + return; + } + out << "{\"status\":\"available\",\"state\":" << info.state << ",\"flags\":" << info.flags + << ",\"rto_ms\":" << info.rto_ms << ",\"send_buffer_bytes\":" << info.send_buffer_bytes + << ",\"send_window_bytes\":" << info.send_window_bytes + << ",\"receive_window_bytes\":" << info.receive_window_bytes + << ",\"congestion_window_bytes\":" << info.congestion_window_bytes + << ",\"slow_start_threshold_bytes\":" << info.slow_start_threshold_bytes << ",\"tx_bytes\":" << info.tx_bytes + << ",\"rx_bytes\":" << info.rx_bytes << ",\"retransmit_bytes\":" << info.retransmit_bytes + << ",\"out_of_order_bytes\":" << info.out_of_order_bytes << '}'; +} + +void peer::open_listener() { + listener_ = ::socket(AF_INET, SOCK_STREAM, 0); + if (listener_ < 0) + { + throw std::runtime_error(std::string("socket: ") + std::strerror(errno)); + } + int enabled = 1; + ::setsockopt(listener_, SOL_SOCKET, SO_REUSEADDR, &enabled, sizeof(enabled)); + sockaddr_in address{}; + address.sin_family = AF_INET; + address.sin_port = htons(static_cast(o_.port)); + inet_pton(AF_INET, o_.host.c_str(), &address.sin_addr); + if (::bind(listener_, reinterpret_cast(&address), sizeof(address)) != 0 || + ::listen(listener_, SOMAXCONN) != 0) + { + throw std::runtime_error(std::string("bind/listen: ") + std::strerror(errno)); + } + set_nonblocking(listener_, o_.asio_socket_setup); + const auto port = port_of(listener_, false); + if (published_port_) + { + published_port_->store(port); + } + out_ << "{\"event\":\"ready\",\"tool\":\"poll_control\",\"port\":" << port << "}\n" << std::flush; +} + +void peer::launch() { + session s; + s.owner = {o_.run, o_.client, ++attempted_}; + s.sent_at.resize(o_.window); + s.connect_since = clock_type::now(); + s.fd = ::socket(AF_INET, SOCK_STREAM, 0); + if (s.fd < 0) + { + fail("socket", std::strerror(errno)); + return; + } + set_nonblocking(s.fd, o_.asio_socket_setup); + sockaddr_in address{}; + address.sin_family = AF_INET; + address.sin_port = htons(static_cast(o_.port)); + inet_pton(AF_INET, o_.host.c_str(), &address.sin_addr); + const auto result = ::connect(s.fd, reinterpret_cast(&address), sizeof(address)); + if (result != 0 && errno != EINPROGRESS) + { + const auto reason = std::string(std::strerror(errno)); + ::close(s.fd); + fail("connect", reason); + return; + } + sessions_.push_back(std::move(s)); + auto& value = sessions_.back(); + if (result == 0) + { + connected(value); + } + else + { + value.connecting = true; + ++connecting_; + } +} + +void peer::connected(session& s) { + if (s.connecting) + { + s.connecting = false; + --connecting_; + } + ++established_; + set_no_delay(s.fd); + s.pending_since = clock_type::now(); + // The native control starts its read before sending HELLO. + if (o_.serial_reads) + { + read_step(s); + } + send(s, connection_load::make_frame(frame_kind::hello, s.owner, 1)); +} + +void peer::connect_ready(session& s) { + int error{}; + socklen_t length = sizeof(error); + ::getsockopt(s.fd, SOL_SOCKET, SO_ERROR, &error, &length); + if (error) + { + fail("connect callback", std::strerror(error)); + close(s); + return; + } + connected(s); +} + +void peer::accept_all() { + while (!failed_) + { + const auto fd = ::accept(listener_, nullptr, nullptr); + if (fd < 0) + { + if (errno != EAGAIN && errno != EWOULDBLOCK && errno != ECONNABORTED && errno != EINTR) + { + fail("accept", std::strerror(errno)); + } + return; + } + if (++accepted_ > o_.expected) + { + ::close(fd); + fail("accept admission", "expected connection count exceeded"); + return; + } + set_nonblocking(fd, o_.asio_socket_setup); + set_no_delay(fd); + session s; + s.fd = fd; + s.last_read_at = clock_type::now(); + sessions_.push_back(std::move(s)); + if (o_.serial_reads) + { + read_step(sessions_.back()); + } + } +} + +void peer::flush(session& s) { + while (s.fd >= 0 && s.output_offset < s.output.size()) + { + const auto requested = s.output.size() - s.output_offset; + const auto sent = ::send(s.fd, s.output.data() + s.output_offset, requested, 0); + if (sent < 0) + { + if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) + { + ++histogram_.write.would_block; + return; + } + fail("send", std::strerror(errno)); + return; + } + histogram_.write.record(requested, static_cast(sent)); + s.output_offset += static_cast(sent); + s.bytes_written += static_cast(sent); + } + s.output.clear(); + s.output_offset = 0; + if (s.writing) + { + // Like an Asio write completion: posted, never run inline. + s.writing = false; + s.completion_posted = true; + post(s, false); + return; + } + if (s.peer_eof && s.fd >= 0) + { + close(s); + } +} + +// Serial pacing mirrors the native control's one async_write per session: a new +// frame is attempted immediately only when no write is outstanding; the next +// queued frame starts when the previous completion runs on a later loop pass. +void peer::start_write(session& s) { + if (s.fd < 0 || s.writing || s.completion_posted || s.queued.empty()) + { + return; + } + s.output = std::move(s.queued.front()); + s.queued.pop_front(); + s.output_offset = 0; + s.writing = true; + flush(s); +} + +void peer::post(session& s, bool read) { + posted_.emplace_back(static_cast(&s - sessions_.data()), read); +} + +void peer::run_posted() { + posted_ready_.clear(); + posted_ready_.swap(posted_); + for (const auto& [index, read] : posted_ready_) + { + auto& s = sessions_[index]; + if (read) + { + read_resume(s); + continue; + } + s.completion_posted = false; + if (s.fd < 0) + { + continue; + } + if (!s.queued.empty()) + { + start_write(s); + } + else if (s.peer_eof) + { + close(s); + } + } +} + +bool peer::output_idle(const session& s) { + return s.output_offset >= s.output.size() && s.queued.empty() && !s.writing && !s.completion_posted; +} + +void peer::send(session& s, const std::string& body) { + if (!o_.serial_writes) + { + append_frame(s.output, body); + flush(s); + return; + } + std::string wire; + append_frame(wire, body); + s.queued.push_back(std::move(wire)); + start_write(s); +} + +// Serial read pacing mirrors async_read_until on a 4096-byte streambuf: a frame +// already buffered completes through a posted bounce without recv; otherwise one +// recv is attempted, and its completion is posted to a later pass, where the +// matcher runs again. Would-block waits for one readiness event, which performs +// one recv. Each completion handles at most one frame, then starts the next read. +void peer::read_step(session& s) { + if (s.fd < 0 || s.read_waiting || s.read_posted) + { + return; + } + const auto view = inspect(s.input); + if (view.status == frame_status::invalid) + { + fail("read frame", "invalid or oversized dgram prefix"); + return; + } + if (view.status == frame_status::complete) + { + s.read_posted = true; + post(s, true); + return; + } + std::array buffer{}; + const auto room = max_wire_bytes - s.input.size(); + const auto received = ::recv(s.fd, buffer.data(), room, 0); + if (received < 0) + { + if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) + { + ++histogram_.read.would_block; + s.read_waiting = true; + return; + } + if (!failed_) + { + fail("recv", std::strerror(errno)); + } + close(s); + return; + } + if (received == 0) + { + eof(s); + return; + } + histogram_.read.record(room, static_cast(received)); + s.bytes_read += static_cast(received); + s.last_read_at = latest_read_ = clock_type::now(); + s.input.append(buffer.data(), static_cast(received)); + s.read_posted = true; + post(s, true); +} + +void peer::read_resume(session& s) { + s.read_posted = false; + if (s.fd < 0) + { + return; + } + const auto view = inspect(s.input); + if (view.status == frame_status::complete) + { + frame(s, std::string_view(s.input).substr(view.prefix, view.size)); + if (s.fd < 0) + { + return; + } + s.input.erase(0, view.prefix + view.size); + } + read_step(s); +} + +// A readiness event with no read waiting is consumed, as with an empty Asio op queue. +void peer::read_ready(session& s) { + if (!o_.serial_reads) + { + readable(s); + return; + } + if (s.read_waiting) + { + s.read_waiting = false; + read_step(s); + } +} + +void peer::readable(session& s) { + // Mirrors the native control's 4096-byte bounded input buffer. + std::array buffer{}; + while (s.fd >= 0) + { + const auto room = max_wire_bytes - s.input.size(); + if (!room) + { + fail("read frame", "input buffer full without a complete frame"); + return; + } + const auto received = ::recv(s.fd, buffer.data(), room, 0); + if (received < 0) + { + if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) + { + ++histogram_.read.would_block; + return; + } + if (!failed_) + { + fail("recv", std::strerror(errno)); + } + close(s); + return; + } + if (received == 0) + { + eof(s); + return; + } + histogram_.read.record(room, static_cast(received)); + s.bytes_read += static_cast(received); + s.last_read_at = latest_read_ = clock_type::now(); + s.input.append(buffer.data(), static_cast(received)); + std::size_t consumed{}; + while (s.fd >= 0) + { + const auto view = inspect(std::string_view(s.input).substr(consumed)); + if (view.status == frame_status::invalid) + { + fail("read frame", "invalid or oversized dgram prefix"); + return; + } + if (view.status == frame_status::incomplete) + { + break; + } + frame(s, std::string_view(s.input).substr(consumed + view.prefix, view.size)); + consumed += view.prefix + view.size; + } + s.input.erase(0, consumed); + } +} + +void peer::frame(session& s, std::string_view body) { + connection_load::frame_header h; + if (!connection_load::decode(body, h)) + { + fail("protocol", "frame decode failed"); + return; + } + if (client()) + { + client_frame(s, h); + } + else + { + server_frame(s, h); + } +} + +void peer::client_frame(session& s, const connection_load::frame_header& h) { + if (h.owner != s.owner) + { + fail("protocol", "identity mismatch"); + } + else if (h.kind == frame_kind::hello_ack && !s.hello && !h.sequence) + { + s.hello = true; + s.pending_since = {}; + ++ready_; + } + else if (h.kind == frame_kind::start_ack && s.hello && !s.started && !h.sequence) + { + s.started = true; + s.pending_since = {}; + ++started_; + } + else if (h.kind == frame_kind::data && s.started && !s.finished && h.body_bytes == o_.payload && + h.sequence == s.next_receive && h.sequence < s.next_send) + { + ++s.next_receive; + ++data_received_; + --outstanding_; + s.pending_since = + s.next_receive < s.next_send ? s.sent_at[(s.next_receive - 1) % o_.window] : clock_type::time_point{}; + fill(s); + finish(s); + } + else if (h.kind == frame_kind::finish_ack && s.started && s.finish_sent && !s.finished && + h.sequence == s.next_send - 1) + { + s.finished = true; + s.pending_since = {}; + ++finished_; + } + else + { + fail("protocol", "unexpected phase or sequence"); + } +} + +void peer::server_frame(session& s, const connection_load::frame_header& h) { + if (h.kind == frame_kind::hello && !s.hello && h.owner.run == o_.run && h.owner.client && h.owner.connection && + h.sequence == 1) + { + if (!identities_.emplace(h.owner.client, h.owner.connection).second) + { + fail("protocol", "duplicate client/connection identity"); + return; + } + s.owner = h.owner; + s.hello = true; + ++ready_; + send(s, connection_load::make_frame(frame_kind::hello_ack, s.owner)); + } + else if (h.owner != s.owner || !s.hello) + { + fail("protocol", "identity mismatch"); + } + else if (h.kind == frame_kind::start && !s.started && !h.sequence) + { + s.started = true; + ++started_; + send(s, connection_load::make_frame(frame_kind::start_ack, s.owner)); + } + else if (h.kind == frame_kind::data && s.started && !s.finished && h.body_bytes == o_.payload && + h.sequence == s.next_receive) + { + ++s.next_receive; + ++data_received_; + send(s, connection_load::make_frame(frame_kind::data, s.owner, h.sequence, o_.payload)); + ++data_sent_; + } + else if (h.kind == frame_kind::finish && s.started && !s.finished && h.sequence == s.next_receive - 1) + { + s.finished = true; + ++finished_; + send(s, connection_load::make_frame(frame_kind::finish_ack, s.owner, h.sequence)); + } + else + { + fail("protocol", "unexpected phase or sequence"); + } +} + +void peer::fill(session& s) { + while (traffic_ && !draining_ && !failed_ && s.fd >= 0 && s.next_send - s.next_receive < o_.window) + { + const auto sequence = s.next_send++; + const auto now = clock_type::now(); + s.sent_at[(sequence - 1) % o_.window] = now; + if (sequence == s.next_receive) + { + s.pending_since = now; + } + ++outstanding_; + ++data_sent_; + send(s, connection_load::make_frame(frame_kind::data, s.owner, sequence, o_.payload)); + } +} + +void peer::finish(session& s) { + if (draining_ && !failed_ && s.fd >= 0 && !s.finish_sent && s.next_receive == s.next_send) + { + s.finish_sent = true; + s.pending_since = clock_type::now(); + send(s, connection_load::make_frame(frame_kind::finish, s.owner, s.next_send - 1)); + } +} + +void peer::eof(session& s) { + if (!client() && s.finished) + { + s.peer_eof = true; + if (output_idle(s)) + { + close(s); + } + return; + } + if (!failed_) + { + // Capture every unfinished peer before this EOF closes anything else. + snapshot("before_unfinished_close", clock_type::now()); + fail("read", "peer closed before finish"); + } + close(s); +} + +void peer::close(session& s) { + if (s.fd < 0) + { + return; + } + if (s.connecting) + { + s.connecting = false; + --connecting_; + } + ::close(s.fd); + s.fd = -1; + s.kevents = 0; + s.queued.clear(); + s.writing = false; + ++disconnected_; +} + +void peer::close_all() { + for (auto& s : sessions_) + { + close(s); + } +} + +void peer::fail(const std::string& scope, const std::string& reason) { + if (failed_) + { + return; + } + failed_ = true; + failed_at_ = clock_type::now(); + failure_scope_ = scope; + failure_reason_ = reason; +} + +void peer::snapshot(const char* phase, clock_type::time_point now) { + snapshot_taken_ = true; + for (auto& s : sessions_) + { + const bool stalled = s.pending_since != clock_type::time_point{} && + now - s.pending_since >= std::chrono::milliseconds(o_.request_ms); + // The server cannot see request age; it reports unfinished peers whose reads + // stopped at least request-timeout-ms before its most recent read anywhere. + const bool quiet = latest_read_ - s.last_read_at >= std::chrono::milliseconds(o_.request_ms); + if (client() ? !(s.watched || stalled) : (s.finished || !quiet)) + { + continue; + } + s.watched = true; + const bool open = s.fd >= 0; + out_ << "{\"event\":\"stalled_connection\",\"tool\":\"poll_control\",\"role\":\"" << o_.role + << "\",\"phase\":\"" << phase << "\",\"elapsed_ms\":" << elapsed_ms(began_, now) + << ",\"client\":" << s.owner.client << ",\"connection\":" << s.owner.connection + << ",\"open\":" << (open ? "true" : "false") << ",\"local_port\":" << (open ? port_of(s.fd, false) : -1) + << ",\"peer_port\":" << (open ? port_of(s.fd, true) : -1) << ",\"next_send\":" << s.next_send + << ",\"next_receive\":" << s.next_receive << ",\"hello\":" << (s.hello ? "true" : "false") + << ",\"started\":" << (s.started ? "true" : "false") + << ",\"finish_sent\":" << (s.finish_sent ? "true" : "false") << ",\"pending_ms\":" + << (s.pending_since == clock_type::time_point{} ? -1 + : static_cast(elapsed_ms(s.pending_since, now))) + << ",\"app_unsent_bytes\":" << (s.output.size() - s.output_offset) + << ",\"app_buffered_input_bytes\":" << s.input.size() << ",\"bytes_written\":" << s.bytes_written + << ",\"bytes_read\":" << s.bytes_read << ",\"read_idle_ms\":" + << (s.last_read_at == clock_type::time_point{} ? -1 + : static_cast(elapsed_ms(s.last_read_at, now))) + << ",\"kernel_readable_bytes\":" << (open ? readable_bytes(s.fd) : -1) + << ",\"kernel_send_queue_bytes\":" << (open ? send_queue_bytes(s.fd) : -1) << ",\"tcp_info\":"; + write_tcp_info(out_, s.fd); + out_ << "}\n"; + } + out_ << std::flush; +} + +// Opt-in: every 250 ms query each open socket's kernel send queue. A queue that +// stays non-empty and unchanged for one second has made no delivery progress on +// loopback; report it then once per second (at most 20 reports per session and 64 +// sessions), and report recovery once it changes. Reports carry Apple TCP info. +void peer::watch_send_queues(clock_type::time_point now) { + if (!o_.send_queue_watch || now < next_watch_) + { + return; + } + next_watch_ = now + std::chrono::milliseconds(250); + for (auto& s : sessions_) + { + if (s.fd < 0 || s.connecting) + { + continue; + } + const auto queued = send_queue_bytes(s.fd); + const bool reported = s.send_queue_reports > 0; + if (queued <= 0 || queued != s.send_queue_last || s.send_queue_since == clock_type::time_point{}) + { + if (reported && queued != s.send_queue_last) + { + emit_send_queue(s, "progressed", now, queued); + s.send_queue_reports = 0; + --send_queue_tracked_; + } + s.send_queue_last = queued; + s.send_queue_since = queued > 0 ? now : clock_type::time_point{}; + continue; + } + const auto unchanged = now - s.send_queue_since; + if (unchanged < std::chrono::seconds(1) || s.send_queue_reports >= 20 || + (!reported && send_queue_tracked_ >= 64)) + { + continue; + } + // Report on the first detection, then once per further second unchanged. + if (reported && unchanged < std::chrono::seconds(1 + s.send_queue_reports)) + { + continue; + } + if (!reported) + { + ++send_queue_tracked_; + } + ++s.send_queue_reports; + emit_send_queue(s, reported ? "sample" : "detected", now, queued); + } + out_ << std::flush; +} + +void peer::emit_send_queue(const session& s, const char* phase, clock_type::time_point now, long long queued) { + out_ << "{\"event\":\"send_queue_stalled\",\"tool\":\"poll_control\",\"role\":\"" << o_.role << "\",\"phase\":\"" + << phase << "\",\"elapsed_ms\":" << elapsed_ms(began_, now) << ",\"client\":" << s.owner.client + << ",\"connection\":" << s.owner.connection << ",\"local_port\":" << port_of(s.fd, false) + << ",\"peer_port\":" << port_of(s.fd, true) << ",\"unchanged_ms\":" + << (s.send_queue_since == clock_type::time_point{} ? 0 : elapsed_ms(s.send_queue_since, now)) + << ",\"kernel_send_queue_bytes\":" << queued << ",\"previous_send_queue_bytes\":" << s.send_queue_last + << ",\"kernel_readable_bytes\":" << readable_bytes(s.fd) << ",\"next_send\":" << s.next_send + << ",\"next_receive\":" << s.next_receive << ",\"finished\":" << (s.finished ? "true" : "false") + << ",\"app_unsent_bytes\":" << (s.output.size() - s.output_offset) + << ",\"app_queued_frames\":" << s.queued.size() << ",\"bytes_written\":" << s.bytes_written + << ",\"bytes_read\":" << s.bytes_read << ",\"tcp_info\":"; + write_tcp_info(out_, s.fd); + out_ << "}\n"; +} + +bool peer::complete() const { + if (client()) + { + return finished_ == o_.connections && !outstanding_ && + std::all_of(sessions_.begin(), sessions_.end(), [](const session& s) { return output_idle(s); }); + } + return finished_ == o_.expected && disconnected_ == o_.expected; +} + +void peer::advance(clock_type::time_point now) { + if (signalled && !failed_) + { + snapshot("on_signal", now); + fail("signal", "SIGINT or SIGTERM"); + } + if (failed_) + { + return; + } + if (now - began_ >= std::chrono::duration(o_.deadline)) + { + snapshot("at_deadline", now); + fail("deadline", "workload exceeded deadline-seconds"); + return; + } + if (!client()) + { + return; + } + const auto interval = std::chrono::nanoseconds(1000000000ULL / o_.rate); + while (!failed_ && attempted_ < o_.connections && connecting_ < o_.parallel && now >= next_launch_) + { + launch(); + next_launch_ += interval; + } + if (!starting_ && ready_ == o_.connections) + { + starting_ = true; + for (auto& s : sessions_) + { + s.pending_since = now; + send(s, connection_load::make_frame(frame_kind::start, s.owner)); + } + } + if (!traffic_ && started_ == o_.connections) + { + traffic_ = true; + until_ = now + std::chrono::duration_cast(std::chrono::duration(o_.hold)); + for (auto& s : sessions_) + { + fill(s); + } + } + if (traffic_ && !draining_ && now >= until_) + { + draining_ = true; + for (auto& s : sessions_) + { + finish(s); + } + } + for (auto& s : sessions_) + { + if (s.connecting && now - s.connect_since >= std::chrono::milliseconds(o_.connect_ms)) + { + ++timeouts_; + fail("connect timeout", "pending connect exceeded connect-timeout-ms"); + return; + } + if (s.pending_since != clock_type::time_point{} && + now - s.pending_since >= std::chrono::milliseconds(o_.request_ms)) + { + ++timeouts_; + snapshot("at_failure", now); + fail("request timeout", "pending response exceeded request-timeout-ms"); + return; + } + } +} + +void peer::result(clock_type::time_point now) const { + out_ << "{\"event\":\"result\",\"tool\":\"poll_control\",\"role\":\"" << o_.role << "\",\"events\":\"" + << (o_.kqueue_events ? "kqueue" : "poll") << "\",\"socket_setup\":\"" + << (o_.asio_socket_setup ? "asio" : "default") + << "\",\"send_queue_watch\":" << (o_.send_queue_watch ? "true" : "false") << ",\"read_pacing\":\"" + << (o_.serial_reads ? "serial" : "drain") << "\",\"write_pacing\":\"" + << (o_.serial_writes ? "serial" : "burst") << "\",\"success\":" << (failed_ ? "false" : "true") + << ",\"failure_scope\":\"" << failure_scope_ << "\",\"failure_reason\":\"" << failure_reason_ + << "\",\"elapsed_ms\":" << elapsed_ms(began_, now) << ",\"expected\":" << o_.expected + << ",\"attempted\":" << attempted_ << ",\"established\":" << established_ << ",\"accepted\":" << accepted_ + << ",\"ready\":" << ready_ << ",\"started\":" << started_ << ",\"finished\":" << finished_ + << ",\"disconnected\":" << disconnected_ << ",\"data_sent\":" << data_sent_ + << ",\"data_received\":" << data_received_ << ",\"outstanding\":" << outstanding_ + << ",\"timeouts\":" << timeouts_ << ",\"io_histogram\":"; + histogram_.write_json(out_); + out_ << "}\n" << std::flush; +} + +void peer::wait_poll() { + descriptors_.clear(); + owners_.clear(); + if (listener_ >= 0 && !failed_) + { + descriptors_.push_back({listener_, POLLIN, 0}); + owners_.push_back(sessions_.size()); + } + for (std::size_t index = 0; index < sessions_.size(); ++index) + { + const auto& s = sessions_[index]; + if (s.fd < 0) + { + continue; + } + // Serial reads ask for POLLIN only while waiting, so level triggering cannot spin. + short events = s.connecting ? POLLOUT : (!o_.serial_reads || s.read_waiting ? POLLIN : 0); + if (s.output_offset < s.output.size()) + { + events |= POLLOUT; + } + descriptors_.push_back({s.fd, events, 0}); + owners_.push_back(index); + } + // Posted completions make the wait non-blocking, like Asio's reactor with ready handlers. + const auto ready = ::poll(descriptors_.data(), static_cast(descriptors_.size()), posted_.empty() ? 5 : 0); + if (ready < 0) + { + if (errno != EINTR) + { + fail("poll", std::strerror(errno)); + } + return; + } + ++histogram_.waits; + histogram_.events += static_cast(ready); + // Accept last: it may append to sessions_ and owners_ index the old vector. + bool accept_ready = false; + for (std::size_t index = 0; index < descriptors_.size() && ready > 0; ++index) + { + const auto revents = descriptors_[index].revents; + if (!revents) + { + continue; + } + if (owners_[index] == sessions_.size()) + { + accept_ready = true; + continue; + } + auto& s = sessions_[owners_[index]]; + if (s.connecting) + { + connect_ready(s); + continue; + } + if (revents & POLLOUT) + { + flush(s); + } + if (s.fd >= 0 && (revents & (POLLIN | POLLHUP | POLLERR))) + { + read_ready(s); + } + } + if (accept_ready) + { + accept_all(); + } +} + +#if ARKNET_POLL_CONTROL_HAS_KQUEUE +// Mirrors Boost.Asio's kqueue reactor rather than an idealized kqueue loop: +// send/recv/accept/connect are attempted first; only an operation that must wait +// registers its descriptor, with EV_ADD | EV_CLEAR, READ alone for reads and READ +// plus WRITE once a write or connect must wait. A WRITE event that arrives with +// nothing left to write deletes the WRITE registration. One kevent call returns at +// most 128 events. udata is the session index plus one; zero is the listener. +void peer::register_events(std::uintptr_t id, int fd, int count, int& registered) { + struct kevent changes[2]; + EV_SET(&changes[0], fd, EVFILT_READ, EV_ADD | EV_CLEAR, 0, 0, reinterpret_cast(id)); + EV_SET(&changes[1], fd, EVFILT_WRITE, EV_ADD | EV_CLEAR, 0, 0, reinterpret_cast(id)); + if (::kevent(kqueue_, changes, count, nullptr, 0, nullptr) != 0) + { + fail("kevent register", std::strerror(errno)); + return; + } + registered = count; +} + +void peer::wait_kqueue() { + if (kqueue_ < 0) + { + kqueue_ = ::kqueue(); + if (kqueue_ < 0) + { + fail("kqueue", std::strerror(errno)); + return; + } + } + if (listener_ >= 0 && !listener_registered_) + { + int registered{}; + register_events(0, listener_, 1, registered); + listener_registered_ = registered != 0; + } + for (std::size_t index = 0; index < sessions_.size() && !failed_; ++index) + { + auto& s = sessions_[index]; + const int wanted = s.connecting || s.output_offset < s.output.size() ? 2 : 1; + if (s.fd >= 0 && s.kevents < wanted) + { + register_events(index + 1, s.fd, wanted, s.kevents); + } + } + std::array events{}; + const timespec timeout{0, posted_.empty() ? 5000000 : 0}; + const auto count = ::kevent(kqueue_, nullptr, 0, events.data(), static_cast(events.size()), &timeout); + if (count < 0) + { + if (errno != EINTR) + { + fail("kevent wait", std::strerror(errno)); + } + return; + } + ++histogram_.waits; + histogram_.events += static_cast(count); + bool accept_ready = false; + for (int event = 0; event < count; ++event) + { + const auto& value = events[static_cast(event)]; + const auto id = reinterpret_cast(value.udata); + if (!id) + { + accept_ready = true; + continue; + } + // Closing a descriptor removes its registrations; skip events already fetched. + auto& s = sessions_[id - 1]; + if (s.fd < 0) + { + continue; + } + if (value.flags & EV_ERROR) + { + fail("kevent event", std::strerror(static_cast(value.data))); + continue; + } + if (value.filter == EVFILT_WRITE && s.kevents == 2 && !s.connecting && s.output_offset >= s.output.size()) + { + struct kevent change; + EV_SET(&change, s.fd, EVFILT_WRITE, EV_DELETE, 0, 0, nullptr); + ::kevent(kqueue_, &change, 1, nullptr, 0, nullptr); + s.kevents = 1; + } + if (s.connecting) + { + connect_ready(s); + } + else if (value.filter == EVFILT_WRITE) + { + flush(s); + } + else if (value.filter == EVFILT_READ) + { + read_ready(s); + } + } + if (accept_ready) + { + accept_all(); + } +} +#endif + +int peer::run() { + if (!client()) + { + open_listener(); + } + while (true) + { + auto now = clock_type::now(); + advance(now); + if (!failed_ && complete()) + { + break; + } + // After the first request timeout the client admits nothing new but keeps + // reading for --observe-seconds, then reports the same connections again. + if (failed_ && (!client() || failure_scope_ != "request timeout" || + now - failed_at_ >= std::chrono::duration(o_.observe))) + { + if (client() && failure_scope_ == "request timeout" && o_.observe > 0) + { + snapshot("after_observation", now); + } + break; + } + watch_send_queues(now); + run_posted(); +#if ARKNET_POLL_CONTROL_HAS_KQUEUE + if (o_.kqueue_events) + { + wait_kqueue(); + continue; + } +#endif + wait_poll(); + } + const auto now = clock_type::now(); + close_all(); + result(now); + return failed_ ? 1 : 0; +} + +void raise_descriptor_limit(std::uint64_t connections) { + rlimit limit{}; + if (::getrlimit(RLIMIT_NOFILE, &limit) != 0) + { + return; + } + const auto wanted = static_cast(connections + 64); + if (limit.rlim_cur < wanted) + { + limit.rlim_cur = (std::min)(wanted, limit.rlim_max); + ::setrlimit(RLIMIT_NOFILE, &limit); + } +} + +void require(bool condition, const char* message) { + if (!condition) + { + throw std::runtime_error(std::string("self-test failed: ") + message); + } +} + +void self_test() { + std::string wire; + append_frame(wire, std::string(1088, 'x')); + require(wire.size() == 1091 && static_cast(wire[0]) == 254, "three-byte prefix"); + require(inspect(wire).status == frame_status::complete && inspect(wire).size == 1088, "complete frame"); + require(inspect(std::string_view(wire).substr(0, 2)).status == frame_status::incomplete, "partial prefix"); + require(inspect(std::string_view(wire).substr(0, 100)).status == frame_status::incomplete, "partial body"); + require(inspect(std::string("\xfe\x05\x00", 3)).status == frame_status::invalid, "non-canonical prefix"); + require(inspect(std::string("\xfe\xff\x0f", 3)).status == frame_status::invalid, "oversized frame"); + bool rejected = false; + try + { parse({"--role", "server", "--listen", "127.0.0.1", "--port", "1", "--connections", "1"}); } + catch (const std::invalid_argument&) + { rejected = true; } + require(rejected, "role-inappropriate option"); + + std::vector modes{"poll"}; + if (ARKNET_POLL_CONTROL_HAS_KQUEUE) + { + modes.push_back("kqueue"); + } + std::size_t runs{}; + for (const auto& mode : modes) + { + for (const std::string pacing : {"burst", "serial"}) + { + // Serial reads are exercised with serial writes, the native-control-like pairing. + const std::string reads = pacing == "serial" ? "serial" : "drain"; + const std::string setup = pacing == "serial" ? "asio" : "default"; + const std::string watch = pacing == "serial" ? "on" : "off"; + ++runs; + std::atomic port{0}; + std::ostringstream server_out, client_out; + int server_status = -1; + std::thread server([&] { + auto o = parse({"--role", "server", "--listen", "127.0.0.1", "--port", "0", "--expected-connections", + "8", "--deadline-seconds", "10", "--events", mode, "--write-pacing", pacing, "--read-pacing", reads, + "--socket-setup", setup, "--send-queue-watch", watch}); + server_status = peer(o, server_out, &port).run(); + }); + while (!port.load()) + { + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + } + auto o = parse( + {"--role", "client", "--target", "127.0.0.1", "--port", std::to_string(port.load()), "--connections", + "8", "--hold-seconds", "0.2", "--deadline-seconds", "10", "--events", mode, "--write-pacing", + pacing, "--read-pacing", reads, "--socket-setup", setup, "--send-queue-watch", watch}); + const auto client_status = peer(o, client_out).run(); + server.join(); + require(client_status == 0 && server_status == 0, "loopback traffic run"); + require(client_out.str().find("\"success\":true") != std::string::npos, "client result"); + require(client_out.str().find("\"events\":\"" + mode + "\",\"socket_setup\":\"" + setup + + "\",\"send_queue_watch\":" + (watch == "on" ? "true" : "false") + + ",\"read_pacing\":\"" + reads + "\",\"write_pacing\":\"" + pacing + "\"") != + std::string::npos, + "event mode and pacing result"); + // A healthy loopback run drains every send queue well within the one-second threshold. + require(server_out.str().find("send_queue_stalled") == std::string::npos && + client_out.str().find("send_queue_stalled") == std::string::npos, + "no send-queue stall on a healthy run"); + } + } + std::cout << "poll control self-test passed (" << runs << " event/pacing combinations)\n"; +} + +void usage() { + std::cout << "arknet_poll_connection_control: poll(2) or kqueue TCP peer without Asio or strands.\n" + "Speaks the arknet_asio_connection_control traffic protocol; either role may be paired with it.\n\n" + " --role server --listen IPV4 --port N --expected-connections N [--payload N]\n" + " --role client --target IPV4 --port N --connections N [--client-id N] [--payload N]\n" + "Common: --run-id N --window N --connect-rate N --connect-parallel N --connect-timeout-ms N\n" + " --request-timeout-ms N --hold-seconds S --deadline-seconds S\n" + "Client: --observe-seconds S keeps reading after the first request timeout, then reports again.\n" + " --events poll|kqueue default poll; kqueue (Apple/BSD) mirrors Boost.Asio's kqueue reactor\n" + " --write-pacing burst|serial default burst; serial writes one frame per session at a time and\n" + " starts the next on a later loop pass, like the native control's serialized async_write\n" + " --socket-setup default|asio asio sets SO_NOSIGPIPE and FIONBIO like Boost.Asio's socket_ops\n" + " --send-queue-watch off|on default off; on reports kernel send queues unchanged for 1 s,\n" + " with Apple TCP_CONNECTION_INFO, every 250 ms scan\n" + " --read-pacing drain|serial default drain; serial performs one recv per step and handles one\n" + " frame per posted completion, like the native control's async_read_until\n" + " --self-test run codec, option and eight-connection loopback checks in each mode\n"; +} + +} // namespace + +int main(int argc, char** argv) { + try + { + if (argc == 2 && std::string_view(argv[1]) == "--help") + { + usage(); + return 0; + } + std::signal(SIGPIPE, SIG_IGN); + if (argc == 2 && std::string_view(argv[1]) == "--self-test") + { + self_test(); + return 0; + } + const auto o = parse(std::vector(argv + 1, argv + argc)); + raise_descriptor_limit(o.expected); + std::signal(SIGINT, [](int) { signalled = 1; }); + std::signal(SIGTERM, [](int) { signalled = 1; }); + return peer(o, std::cout).run(); + } + catch (const std::exception& error) + { + std::cerr << error.what() << '\n'; + return 2; + } +} diff --git a/docs/project/testing.md b/docs/project/testing.md index fb4c038..54136f5 100644 --- a/docs/project/testing.md +++ b/docs/project/testing.md @@ -317,6 +317,7 @@ the unresolved 5,000-connection root cause or capacity. See the | Redis | RESP2/RESP3, authentication, TCP/TLS/mTLS, ordered pipelines, cancellation reply draining, bounded admission, unsupported replication commands and exclusive pool reset/authentication/database isolation | | Installed consumer | Provider/TLS configuration, installed public headers and optional C++ compiler launcher forwarding; empty launcher, paths with spaces, launcher argument lists and real compiler failure propagation | | Connection-load process | Raw TCP peers split the frame prefix/body and combine eight requests across a 4 KiB receive capacity against both arknet and native Asio servers; ordered byte-exact replies and FIN exchange, not a high-concurrency or TCP-loss test | +| Poll connection control | `--self-test` checks the canonical length prefix, role-specific option rejection and eight-connection loopback traffic of the POSIX peer without Asio: default `poll(2)` pacing, and serial read/write pacing with Asio socket setup and send-queue watching that must report no stall; each also under `kqueue` on Apple/BSD; not a high-concurrency or TCP-loss test | Sustained stress validates content/sequence, completion counts, connection and server restarts, pending writes and queue cleanup. Each protocol splits diff --git a/docs/project/testing_CN.md b/docs/project/testing_CN.md index 46ee528..f13e3bd 100644 --- a/docs/project/testing_CN.md +++ b/docs/project/testing_CN.md @@ -249,6 +249,7 @@ build/tests/arknet_tls_test --reporters=junit --out=build/tls.xml | Redis | RESP2/RESP3、认证、TCP/TLS/mTLS、有序流水线、取消后的回复排空、有界准入、拒绝复制命令，以及独占连接池的重置、认证与数据库隔离 | | 安装包消费项目 | 依赖方案／TLS、公开头文件和可选 C++ compiler launcher 转发；空 launcher、带空格路径、launcher 参数列表与真实编译错误传播 | | 连接负载进程 | 原始 TCP 对端拆分帧前缀／消息体，并跨 4 KiB 接收容量合并八个请求，分别对接 arknet 与原生 Asio 服务端；检查按序逐字节回复和 FIN，不代表高并发或 TCP 丢包验收 | +| poll 连接对照 | `--self-test` 检查规范长度前缀、按角色拒绝不适用选项，以及不使用 Asio 的 POSIX 对端在八个连接上的 loopback 流量：默认 `poll(2)` 节奏，以及逐帧读写 + Asio 式 socket 设置 + 发送队列监视（必须不报告卡住）；在 Apple/BSD 上两者还各用 `kqueue` 跑一遍；不代表高并发或 TCP 丢包验收 | 持续压力测试验证数据与序号、完成计数、连接／服务端重启、未完成写入和队列清理。 每种协议将 `--seconds` 分配给四种配置：库内置调度器、单 context 单线程、