diff --git a/benchmarks/CMakeLists.txt b/benchmarks/CMakeLists.txt index ee5ab47..a7ea70f 100644 --- a/benchmarks/CMakeLists.txt +++ b/benchmarks/CMakeLists.txt @@ -122,6 +122,15 @@ if(ARKNET_BUILD_TESTS AND UNIX) target_compile_features(arknet_asio_control_process_test PRIVATE cxx_std_20) add_dependencies(arknet_asio_control_process_test arknet_asio_connection_control arknet_connection_benchmark) target_link_libraries(arknet_asio_control_process_test PRIVATE Threads::Threads) + arknet_add_check(benchmark_asio_socket_observer 10 arknet_asio_control_process_test + "$" --observer-only) arknet_add_check(benchmark_asio_connections_processes 90 arknet_asio_control_process_test "$" "$") endif() + +if(ARKNET_BUILD_TESTS) + add_executable(arknet_connection_socket_observer_test + ../tests/connection_socket_observer.cpp ../tests/test_main.cpp) + target_link_libraries(arknet_connection_socket_observer_test PRIVATE arknet_test_framework Threads::Threads) + arknet_add_check(connection_socket_observer 10 arknet_connection_socket_observer_test) +endif() diff --git a/benchmarks/CONNECTIONS.md b/benchmarks/CONNECTIONS.md index 36eaeb8..a026497 100644 --- a/benchmarks/CONNECTIONS.md +++ b/benchmarks/CONNECTIONS.md @@ -416,3 +416,40 @@ deterministic fake stream without sockets to check partial writes, buffered read zero-byte operations, move-only completions, forwarded associations, cancellation and frozen history. These are observer correctness checks, not a reproduced stall or performance/capacity acceptance. + +### Independent socket observation (native control) + +`--socket-observer on` is an optional Unix-only diagnostic for +`arknet_asio_connection_control`; it defaults to `off`, requires +`--operation-trace on`, and is rejected on Windows. After the IO worker publishes +a session idle for at least one second, the existing controller samples at most +16 pinned descriptors, at most eight times per session, at least 100 ms apart. +`poll(..., 0)`, `FIONREAD` and `recv(MSG_PEEK | MSG_DONTWAIT)` inspect readiness and +up to 67 bytes without consuming application data, sending traffic or running the +Asio reactor. A slot generation and owned duplicate prevent descriptor reuse from +changing identity; the duplicate is retired before the original socket closes. +Recovered sessions release slots. Closed/still-stalled histories are printed after +worker cleanup, including on successful runs; omissions and pin failures are counted. + +`socket_observation` schema 1 records controller time, the IO-owned immutable +buffer/read snapshot, endpoints, the global marker and a per-session strand probe. +Apple queries also retain requested/returned TCP info lengths, bounded raw bytes +and decoded fields; the summary records struct size/field offsets. TCP status +values are 0=available, 1=unsupported, 2=closed, 3=query failed, 4=short response. +Unavailable TCP info is `null`; failed FIONREAD/peek use -1 plus errno, not a false +zero. Retain compiler/SDK identity with the raw output. The collector does not +calibrate kernel counter semantics or prove an OS fault. Its initial publication +still depends on IO-worker progress; a worker blocked before publication leaves +no independent per-socket history. Diagnostic overhead and `dup` lifetime are +part of the measured configuration, not a transparent production measurement. + +Each independent query includes `clock_anchor`: steady-before, Unix wall and +steady-after times in nanoseconds, plus query-completed steady time. Nanoseconds +are JSON strings to preserve precision. A retained session also emits +`socket_observer_before_cancel` from the IO worker before cancelling the original +socket. Map a steady time `q` through anchor `(a,w,b)` to wall interval +`[w+q-b, w+q-a]`, then include PCAP timestamp quantization. Use the query interval, +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. diff --git a/benchmarks/CONNECTIONS_CN.md b/benchmarks/CONNECTIONS_CN.md index 72145a4..e2fc024 100644 --- a/benchmarks/CONNECTIONS_CN.md +++ b/benchmarks/CONNECTIONS_CN.md @@ -346,3 +346,28 @@ Echo，完成 FIN 确认并正常退出。这项检查在不反复执行高连 在原有完成回调体内记录事件。`connection_socket_operation_trace` 使用无需 socket 的确定性模拟流， 检查部分写入、已有缓冲读取、零字节操作、仅移动完成处理器、关联转发、取消与历史冻结。 这些检查验证观察器本身，不代表重现停顿，也不构成性能或容量验收。 + +### 独立 socket 观察（原生对照） + +`arknet_asio_connection_control` 的 `--socket-observer on` 仅支持 Unix，默认关闭， +要求 `--operation-trace on`；Windows 明确拒绝。IO worker 发布静止至少一秒的会话后， +既有 controller 对最多16个固定身份的 fd 各观察最多8次，间隔至少100ms。 +`poll(..., 0)`、`FIONREAD` 和 `recv(MSG_PEEK | MSG_DONTWAIT)` 检查可读性及最多67字节， +不消费业务数据、不发送流量、不运行 Asio reactor。槽位代次和持有的 fd 副本防止复用误配； +原 socket 关闭前先释放副本。恢复进展的会话释放槽位；关闭或持续停滞的历史在 worker +清理后输出，成功运行也保留，并统计遗漏和固定 fd 失败次数。 + +`socket_observation` schema 1 包含 controller 时间、IO worker 发布的不可变缓冲/读状态、 +端点、全局 marker 及单会话 strand probe。Apple 查询另存 TCP info 请求/返回长度、 +有界原始字节和解码字段，摘要记录结构大小/字段偏移。TCP status：0可用、1平台不支持、 +2已关闭、3查询失败、4响应过短；不可用信息为 `null`，FIONREAD/peek 失败用 -1及errno， +不伪装为零。原始输出须同时保留 compiler/SDK 身份；采集器不校准内核计数语义，不能证明 +OS 缺陷。首次发布仍依赖 IO worker，发布前 worker 已阻塞时不会有独立会话历史。 +诊断开销与 fd 副本生存期属于本次观测配置，不能当作无扰动的生产测量。 + +每次独立查询增加 `clock_anchor`：steady-before、Unix wall、steady-after纳秒值， +以及query完成的steady时间，均用JSON字符串保留精度。IO worker还在取消原socket前输出 +`socket_observer_before_cancel`。由anchor `(a,w,b)`把steady时间`q`映射为wall区间 +`[w+q-b, w+q-a]`，再纳入PCAP时间戳量化误差。应使用query区间，不用较旧的IO状态快照时间。 +须检查anchor间wall/steady偏移稳定性；时钟跳变、过宽区间或缺取消anchor时，精确跨进程 +对齐不可判定。旧输出仅有steady epoch，无法精确对齐包的wall时间戳。 diff --git a/benchmarks/asio_connections.cpp b/benchmarks/asio_connections.cpp index 8cedaf6..8a9b673 100644 --- a/benchmarks/asio_connections.cpp +++ b/benchmarks/asio_connections.cpp @@ -53,6 +53,7 @@ constexpr auto dependency_version = ASIO_VERSION; #include "connections_kernel_samples.hpp" #include "connections_protocol.hpp" #include "connections_resources.hpp" +#include "connections_socket_observer.hpp" #include "connections_socket_trace.hpp" #include "connections_trace.hpp" @@ -73,7 +74,7 @@ struct options std::size_t connect_ms = 5000, request_ms = 5000, inflight = 20000; 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{}; + bool operation_trace{}, socket_operation_trace{}, tcp_kernel_samples{}, socket_observer{}; }; std::uint64_t integer(std::string_view text) { @@ -97,6 +98,16 @@ double seconds(const std::string& text) { } void validate(const options& o) { + if (o.socket_observer && !o.operation_trace) + { + throw std::invalid_argument("socket-observer requires operation-trace on"); + } +#if defined(_WIN32) + if (o.socket_observer) + { + throw std::invalid_argument("socket-observer is unavailable on Windows"); + } +#endif if (o.tcp_kernel_samples && !o.operation_trace) { throw std::invalid_argument("tcp-kernel-samples requires operation-trace on"); @@ -153,6 +164,7 @@ void help() { " --operation-trace off|on (default off; bounded composed-operation diagnostics)\n" " --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" " --self-test (unit checks only; no sockets)\n" "One native Asio context/worker, session strands, TCP_NODELAY, 4096B read_until cap.\n" "Defaults permit 5000 active connections, 32 connecting and 20000 data inflight.\n" @@ -271,6 +283,10 @@ options arguments(int argc, char** argv) { { o.tcp_kernel_samples = value == "on"; } + else if (key == "--socket-observer" && (value == "on" || value == "off")) + { + o.socket_observer = value == "on"; + } else if (key == "--io-threads" && integer(value) == 1) { } @@ -525,6 +541,7 @@ struct application sizeof(connection_load::socket_trace_stream) + 2 * connection_load::socket_handler_reservation) , kernel_samples(o.tcp_kernel_samples) + , socket_observer(o.socket_observer) , acceptor(io) , cleanup(io) , signals(io, SIGINT, SIGTERM) {} @@ -548,6 +565,8 @@ struct application options o; connection_load::operation_recorder operations; connection_load::tcp_kernel_samples kernel_samples; + connection_load::socket_observer socket_observer; + std::atomic marker_completed_us{}; bool trace_emitted{}; asio::io_context io{1}; tcp::acceptor acceptor; @@ -605,6 +624,7 @@ void session::close() { return; } closed = true; + app.socket_observer.close(state.owner); error_code ignored; socket.cancel(ignored); socket.close(ignored); @@ -1012,25 +1032,25 @@ void session::diagnose() { << ",\"available_error\":" << available_error.value() << ",\"receive_buffer_bytes\":" << (receive_error ? -1 : receive_buffer.value()) << ",\"send_buffer_bytes\":" << (send_error ? -1 : send_buffer.value()); -#if defined(__APPLE__) && defined(TCP_CONNECTION_INFO) - tcp_connection_info info{}; - socklen_t length = sizeof(info); - if (socket.is_open() && getsockopt(socket.native_handle(), IPPROTO_TCP, TCP_CONNECTION_INFO, &info, &length) == 0) - { - std::cout << ",\"tcp_info\":{\"state\":" << unsigned(info.tcpi_state) - << ",\"send_buffer_bytes\":" << info.tcpi_snd_sbbytes - << ",\"send_window_bytes\":" << info.tcpi_snd_wnd << ",\"receive_window_bytes\":" << info.tcpi_rcv_wnd - << ",\"rto_ms\":" << info.tcpi_rto << ",\"tx_bytes\":" << info.tcpi_txbytes - << ",\"rx_bytes\":" << info.tcpi_rxbytes << ",\"retransmit_bytes\":" << info.tcpi_txretransmitbytes - << ",\"out_of_order_bytes\":" << info.tcpi_rxoutoforderbytes << '}'; +#if defined(__APPLE__) + const auto info = connection_load::query_tcp_kernel_info(socket.is_open() ? socket.native_handle() : -1); +#else + const auto info = connection_load::query_tcp_kernel_info(-1); +#endif + if (info.result == connection_load::tcp_kernel_info::status::available) + { + std::cout << ",\"tcp_info\":{\"state\":" << info.state << ",\"send_buffer_bytes\":" << info.send_buffer_bytes + << ",\"send_window_bytes\":" << info.send_window_bytes + << ",\"receive_window_bytes\":" << info.receive_window_bytes << ",\"rto_ms\":" << info.rto_ms + << ",\"tx_bytes\":" << info.tx_bytes << ",\"rx_bytes\":" << info.rx_bytes + << ",\"retransmit_bytes\":" << info.retransmit_bytes + << ",\"out_of_order_bytes\":" << info.out_of_order_bytes << '}'; } else { - std::cout << ",\"tcp_info\":null,\"tcp_info_error\":" << (socket.is_open() ? errno : -1); + std::cout << ",\"tcp_info\":null,\"tcp_info_error\":" << info.error + << ",\"tcp_info_status\":" << static_cast(info.result); } -#else - std::cout << ",\"tcp_info\":null"; -#endif std::cout << ",\"frame_boundary\":"; boundary.write_json(std::cout); std::cout << "}\n"; @@ -1342,11 +1362,39 @@ void application::sample_kernel() { for (const auto& value : sessions) { kernel_samples.prune(value->state.owner, stalled(*value), value->read_completions + value->write_completions); + socket_observer.prune(value->state.owner, stalled(*value), value->read_completions + value->write_completions); } for (const auto& value : sessions) { if (stalled(*value)) { + if (o.socket_observer) + { + error_code local_error, remote_error; + const auto local = value->socket.local_endpoint(local_error); + const auto remote = value->socket.remote_endpoint(remote_error); + const auto endpoint = [](tcp::endpoint item, const error_code& error) { + return error ? std::string{} : item.address().to_string() + ":" + std::to_string(item.port()); + }; + const auto bytes = value->input.data(); + std::string_view input; + if (value->input.size()) + { + const auto buffer = *asio::buffer_sequence_begin(bytes); + input = {static_cast(buffer.data()), buffer.size()}; + } +#if !defined(_WIN32) + const auto probe = socket_observer.publish(value->socket.native_handle(), + {value->state.owner, elapsed_us(), idle(*value), value->read_completions + value->write_completions, + value->state.next_receive, value->state.next_send, value->input.size(), value->read_pending, + inspect_frame_boundary(input).complete(), endpoint(local, local_error), + endpoint(remote, remote_error)}); + if (probe.generation) + { + asio::post(value->strand, [this, probe] { socket_observer.complete(probe, elapsed_us()); }); + } +#endif + } kernel_samples.sample(value->state.owner, elapsed_us(), idle(*value), value->read_completions + value->write_completions, value->state.next_receive, value->state.next_send, [&] { @@ -1372,7 +1420,8 @@ void application::sample() { const auto submitted = elapsed_us(); asio::post(io, [this, submitted] { operations.complete_marker(submitted, elapsed_us()); - if (o.tcp_kernel_samples) + marker_completed_us = elapsed_us(); + if (o.tcp_kernel_samples || o.socket_observer) { sample_kernel(); } @@ -1432,10 +1481,13 @@ void application::run() { if (!stopping) { sample(); + socket_observer.sample(elapsed_us(), marker_completed_us.load()); std::this_thread::sleep_for(std::chrono::milliseconds(1)); } } worker.join(); + socket_observer.write_json( + std::cout, o.role, std::chrono::duration_cast(began.time_since_epoch()).count()); } void application::result() const { @@ -1486,6 +1538,10 @@ void application::result() const { { std::cout << ",\"tcp_kernel_samples_enabled\":true"; } + if (o.socket_observer) + { + std::cout << ",\"socket_observer_enabled\":true"; + } } std::cout << "}\n"; if (!failure.empty() || !clean || !complete) diff --git a/benchmarks/connections_kernel_samples.hpp b/benchmarks/connections_kernel_samples.hpp index de7d834..58ea7ac 100644 --- a/benchmarks/connections_kernel_samples.hpp +++ b/benchmarks/connections_kernel_samples.hpp @@ -36,23 +36,15 @@ struct tcp_kernel_info int error{}; }; -inline tcp_kernel_info query_tcp_kernel_info(int descriptor) noexcept { - tcp_kernel_info result; #if defined(__APPLE__) && defined(TCP_CONNECTION_INFO) - if (descriptor < 0) - { - result.result = tcp_kernel_info::status::closed; - return result; - } - tcp_connection_info info{}; - socklen_t length = sizeof(info); - if (getsockopt(descriptor, IPPROTO_TCP, TCP_CONNECTION_INFO, &info, &length) != 0) - { - result.result = tcp_kernel_info::status::query_failed; - result.error = errno; - return result; - } - if (length < offsetof(tcp_connection_info, tcpi_rxoutoforderbytes) + sizeof(info.tcpi_rxoutoforderbytes)) +inline constexpr std::size_t tcp_kernel_info_required_length = + offsetof(tcp_connection_info, tcpi_rxoutoforderbytes) + sizeof(tcp_connection_info::tcpi_rxoutoforderbytes); +#endif + +#if defined(__APPLE__) && defined(TCP_CONNECTION_INFO) +inline tcp_kernel_info decode_tcp_kernel_info(const tcp_connection_info& info, std::size_t length) noexcept { + tcp_kernel_info result; + if (length < tcp_kernel_info_required_length) { result.result = tcp_kernel_info::status::short_response; return result; @@ -70,10 +62,31 @@ inline tcp_kernel_info query_tcp_kernel_info(int descriptor) noexcept { result.rx_bytes = info.tcpi_rxbytes; result.retransmit_bytes = info.tcpi_txretransmitbytes; result.out_of_order_bytes = info.tcpi_rxoutoforderbytes; + return result; +} +#endif + +inline tcp_kernel_info query_tcp_kernel_info(int descriptor) noexcept { + tcp_kernel_info result; +#if defined(__APPLE__) && defined(TCP_CONNECTION_INFO) + if (descriptor < 0) + { + result.result = tcp_kernel_info::status::closed; + return result; + } + tcp_connection_info info{}; + socklen_t length = sizeof(info); + if (getsockopt(descriptor, IPPROTO_TCP, TCP_CONNECTION_INFO, &info, &length) != 0) + { + result.result = tcp_kernel_info::status::query_failed; + result.error = errno; + return result; + } + return decode_tcp_kernel_info(info, length); #else (void)descriptor; -#endif return result; +#endif } // Single IO-worker ownership. No per-session allocation or work in disabled mode. diff --git a/benchmarks/connections_socket_observer.hpp b/benchmarks/connections_socket_observer.hpp new file mode 100644 index 0000000..f02a723 --- /dev/null +++ b/benchmarks/connections_socket_observer.hpp @@ -0,0 +1,394 @@ +// Copyright (c) 2026 OpenArkStudio. Distributed under BSL-1.0. See LICENSE. +#pragma once + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#if !defined(_WIN32) +#include +#include +#include +#include + +#endif + +#include "connections_kernel_samples.hpp" + +namespace connection_load { + +struct socket_observation +{ + int poll_result{-1}, poll_error{}, readable_bytes{-1}, readable_error{}, peek_bytes{-1}, peek_error{}; + short events{}; + std::array peek{}; + tcp_kernel_info tcp; + std::uint32_t tcp_requested{}, tcp_returned{}; + std::array tcp_raw{}; +}; + +// No consuming read, registration with Asio, or change to file status flags. +inline socket_observation observe_socket(int descriptor) noexcept { + socket_observation result; +#if !defined(_WIN32) + if (descriptor < 0) + { + result.poll_error = result.readable_error = result.peek_error = EBADF; + result.tcp = query_tcp_kernel_info(-1); + return result; + } + pollfd item{descriptor, POLLIN, 0}; + result.poll_result = ::poll(&item, 1, 0); + result.poll_error = result.poll_result < 0 ? errno : 0; + result.events = item.revents; + int available{}; + if (::ioctl(descriptor, FIONREAD, &available) == 0) + { + result.readable_bytes = available; + } + else + { + result.readable_error = errno; + } + const auto bytes = ::recv(descriptor, result.peek.data(), result.peek.size(), MSG_PEEK | MSG_DONTWAIT); + result.peek_bytes = static_cast(bytes); + result.peek_error = bytes < 0 ? errno : 0; +#if defined(__APPLE__) && defined(TCP_CONNECTION_INFO) + tcp_connection_info raw{}; + static_assert(sizeof(raw) <= result.tcp_raw.size()); + socklen_t length = sizeof(raw); + result.tcp_requested = sizeof(raw); + if (::getsockopt(descriptor, IPPROTO_TCP, TCP_CONNECTION_INFO, &raw, &length) == 0) + { + result.tcp_returned = length; + result.tcp = decode_tcp_kernel_info(raw, length); + const auto* begin = reinterpret_cast(&raw); + std::copy_n(begin, (std::min)(std::size_t(length), sizeof(raw)), result.tcp_raw.begin()); + } + else + { + result.tcp.result = tcp_kernel_info::status::query_failed; + result.tcp.error = errno; + } +#endif +#else + (void)descriptor; +#endif + return result; +} + +struct observation_clock_anchor +{ + std::int64_t steady_before_ns{}, unix_ns{}, steady_after_ns{}; +}; + +inline std::int64_t observation_steady_ns() noexcept { + return std::chrono::duration_cast(std::chrono::steady_clock::now().time_since_epoch()) + .count(); +} + +inline observation_clock_anchor observation_anchor() noexcept { + observation_clock_anchor value; + value.steady_before_ns = observation_steady_ns(); + value.unix_ns = + std::chrono::duration_cast(std::chrono::system_clock::now().time_since_epoch()) + .count(); + value.steady_after_ns = observation_steady_ns(); + return value; +} + +// Single IO worker publishes snapshots; the existing controller observes pinned +// descriptors. The mutex also serializes dup retirement before socket.close(). +class socket_observer +{ +public: + static constexpr std::size_t session_capacity = 16, sample_capacity = 8; + static constexpr std::int64_t interval_us = 100000; + + struct snapshot + { + identity owner{}; + std::int64_t time_us{}, idle_us{}; + std::uint64_t completions{}, next_receive{}, next_send{}, input_bytes{}; + bool read_pending{}, frame_ready{}; + std::string local, remote; + }; + + struct token + { + identity owner{}; + std::uint64_t generation{}; + }; + + explicit socket_observer(bool enabled) + : data_(enabled ? std::make_unique() : nullptr) {} + + ~socket_observer() { + if (data_) + { + for (auto& value : data_->slots) + { + retire(value); + } + } + } + + token publish(int descriptor, snapshot state) { + if (!data_) + { + return {}; + } + std::lock_guard lock(mutex_); + auto* value = find(state.owner); + const bool created = !value; + if (!value) + { + for (auto& candidate : data_->slots) + { + if (!candidate.generation) + { + value = &candidate; + break; + } + } + if (!value) + { + ++omitted_; + return {}; + } +#if !defined(_WIN32) + value->descriptor = ::dup(descriptor); + if (value->descriptor < 0) + { + ++pin_failures_; + return {}; + } +#else + return {}; +#endif + value->generation = ++generation_; + value->probe_submitted = state.time_us; + } + value->state = std::move(state); + return created ? token{value->state.owner, value->generation} : token{}; + } + + // Recovered sessions free capacity; histories for still-stalled/closed peers + // survive final cleanup. Never confuse a recycled fd/slot with its predecessor. + void prune(identity owner, bool eligible, std::uint64_t completions) { + if (!data_) + { + return; + } + std::lock_guard lock(mutex_); + if (auto* value = find(owner); + value && value->descriptor >= 0 && (!eligible || value->state.completions != completions)) + { + retire(*value); + *value = {}; + ++released_; + } + } + + void close(identity owner) { + if (data_) + { + std::lock_guard lock(mutex_); + if (auto* value = find(owner); value && value->descriptor >= 0) + { + // This runs before the caller cancels/closes the original socket. + value->before_cancel = observation_anchor(); + retire(*value); + } + } + } + + void complete(token expected, std::int64_t time_us) { + if (data_) + { + std::lock_guard lock(mutex_); + if (auto* value = find(expected.owner); value && value->generation == expected.generation) + { + value->probe_completed = time_us; + } + } + } + + void sample(std::int64_t time_us, std::int64_t marker_completed) { + if (!data_) + { + return; + } + std::lock_guard lock(mutex_); + for (auto& value : data_->slots) + { + if (value.descriptor >= 0 && value.count < sample_capacity && + (!value.count || time_us - value.samples[value.count - 1].time_us >= interval_us)) + { + auto& entry = value.samples[value.count++]; + entry.time_us = time_us; + entry.marker_completed = marker_completed; + entry.probe_completed = value.probe_completed; + entry.state = value.state; + entry.anchor = observation_anchor(); + entry.query = observe_socket(value.descriptor); + entry.query_completed_steady_ns = observation_steady_ns(); + } + } + } + + void write_json(std::ostream& out, const std::string& role, std::int64_t clock_origin_us = 0) const { + if (!data_) + { + return; + } + for (const auto& value : data_->slots) + { + for (std::size_t index = 0; index < value.count; ++index) + { + const auto& entry = value.samples[index]; + const auto& q = entry.query; + out << "{\"event\":\"socket_observation\",\"schema\":1,\"role\":" << std::quoted(role) + << ",\"clock_origin_us\":" << clock_origin_us << ",\"run_id\":" << entry.state.owner.run + << ",\"client_id\":" << entry.state.owner.client + << ",\"connection_id\":" << entry.state.owner.connection << ",\"generation\":" << value.generation + << ",\"sample\":" << index << ",\"time_us\":" << entry.time_us + << ",\"snapshot_time_us\":" << entry.state.time_us << ",\"idle_us\":" << entry.state.idle_us + << ",\"local_endpoint\":" << std::quoted(entry.state.local) + << ",\"remote_endpoint\":" << std::quoted(entry.state.remote) + << ",\"next_receive\":" << entry.state.next_receive << ",\"next_send\":" << entry.state.next_send + << ",\"read_pending\":" << (entry.state.read_pending ? "true" : "false") + << ",\"input_bytes\":" << entry.state.input_bytes + << ",\"input_frame_ready\":" << (entry.state.frame_ready ? "true" : "false") + << ",\"global_marker_completed_us\":" << entry.marker_completed + << ",\"strand_probe_submitted_us\":" << value.probe_submitted + << ",\"strand_probe_completed_us\":" << entry.probe_completed << ",\"clock_anchor\":"; + write_anchor(out, entry.anchor); + out << ",\"query_completed_steady_ns\":" << std::quoted(std::to_string(entry.query_completed_steady_ns)) + << ",\"poll_result\":" << q.poll_result << ",\"poll_error\":" << q.poll_error + << ",\"poll_events\":" << q.events << ",\"readable_bytes\":" << q.readable_bytes + << ",\"readable_error\":" << q.readable_error << ",\"peek_bytes\":" << q.peek_bytes + << ",\"peek_error\":" << q.peek_error << ",\"peek_hex\":\""; + hex(out, q.peek.data(), q.peek_bytes > 0 ? std::size_t(q.peek_bytes) : 0); + out << "\",\"tcp_status\":" << static_cast(q.tcp.result) << ",\"tcp_error\":" << q.tcp.error + << ",\"tcp_requested_length\":" << q.tcp_requested << ",\"tcp_returned_length\":" << q.tcp_returned + << ",\"tcp_raw_hex\":\""; + hex(out, q.tcp_raw.data(), (std::min)(std::size_t(q.tcp_returned), q.tcp_raw.size())); + out << "\",\"tcp_info\":"; + if (q.tcp.result == tcp_kernel_info::status::available) + { + out << "{\"state\":" << q.tcp.state << ",\"flags\":" << q.tcp.flags + << ",\"rto_ms\":" << q.tcp.rto_ms + << ",\"congestion_window_bytes\":" << q.tcp.congestion_window_bytes + << ",\"send_buffer_bytes\":" << q.tcp.send_buffer_bytes + << ",\"send_window_bytes\":" << q.tcp.send_window_bytes + << ",\"receive_window_bytes\":" << q.tcp.receive_window_bytes + << ",\"tx_bytes\":" << q.tcp.tx_bytes << ",\"rx_bytes\":" << q.tcp.rx_bytes + << ",\"retransmit_bytes\":" << q.tcp.retransmit_bytes << '}'; + } + else + { + out << "null"; + } + out << "}\n"; + } + if (value.before_cancel.unix_ns) + { + out << "{\"event\":\"socket_observer_before_cancel\",\"schema\":1,\"role\":" << std::quoted(role) + << ",\"run_id\":" << value.state.owner.run << ",\"client_id\":" << value.state.owner.client + << ",\"connection_id\":" << value.state.owner.connection << ",\"generation\":" << value.generation + << ",\"clock_anchor\":"; + write_anchor(out, value.before_cancel); + out << "}\n"; + } + } + out << "{\"event\":\"socket_observer_summary\",\"role\":" << std::quoted(role) + << ",\"clock_domain\":\"local_steady_clock\",\"clock_origin_us\":" << clock_origin_us + << ",\"capacity\":16,\"samples_per_session\":8,\"interval_us\":100000,\"released\":" << released_ + << ",\"omitted\":" << omitted_ << ",\"pin_failures\":" << pin_failures_ + << ",\"storage_bytes\":" << sizeof(storage); +#if defined(__APPLE__) && defined(TCP_CONNECTION_INFO) + out << ",\"tcp_struct_size\":" << sizeof(tcp_connection_info) + << ",\"tcp_rto_offset\":" << offsetof(tcp_connection_info, tcpi_rto) + << ",\"tcp_cwnd_offset\":" << offsetof(tcp_connection_info, tcpi_snd_cwnd) + << ",\"tcp_last_required_end\":" << tcp_kernel_info_required_length; +#endif + out << "}\n"; + } + +private: + struct sample_entry + { + snapshot state; + socket_observation query; + std::int64_t time_us{}, marker_completed{}, probe_completed{}, query_completed_steady_ns{}; + observation_clock_anchor anchor; + }; + + struct slot + { + int descriptor{-1}; + std::uint64_t generation{}; + snapshot state; + std::int64_t probe_submitted{}, probe_completed{}; + observation_clock_anchor before_cancel; + std::size_t count{}; + std::array samples{}; + }; + + struct storage + { + std::array slots{}; + }; + + static_assert(sizeof(storage) < 128 * 1024); + + slot* find(identity owner) { + for (auto& value : data_->slots) + { + if (value.generation && value.state.owner == owner) + { + return &value; + } + } + return nullptr; + } + + static void retire(slot& value) { +#if !defined(_WIN32) + if (value.descriptor >= 0) + { + ::close(value.descriptor); + value.descriptor = -1; + } +#endif + } + + static void write_anchor(std::ostream& out, observation_clock_anchor value) { + // Strings preserve Unix nanoseconds in JSON readers with binary64 numbers. + out << "{\"steady_before_ns\":" << std::quoted(std::to_string(value.steady_before_ns)) + << ",\"unix_ns\":" << std::quoted(std::to_string(value.unix_ns)) + << ",\"steady_after_ns\":" << std::quoted(std::to_string(value.steady_after_ns)) << '}'; + } + + static void hex(std::ostream& out, const unsigned char* bytes, std::size_t count) { + constexpr char digits[] = "0123456789abcdef"; + for (std::size_t index = 0; index < count; ++index) + { + out << digits[bytes[index] >> 4] << digits[bytes[index] & 15]; + } + } + + std::unique_ptr data_; + std::mutex mutex_; + std::uint64_t generation_{}, omitted_{}, released_{}, pin_failures_{}; +}; +} // namespace connection_load diff --git a/tests/connection_load_native_control.cpp b/tests/connection_load_native_control.cpp index d0c5e81..1fc7563 100644 --- a/tests/connection_load_native_control.cpp +++ b/tests/connection_load_native_control.cpp @@ -404,7 +404,7 @@ void progress_then_stall(const std::string& executable) { {"--role", "client", "--target", "127.0.0.1", "--port", std::to_string(ntohs(address.sin_port)), "--run-id", "97", "--client-id", "3", "--connections", "1", "--scenario", "traffic", "--payload", "1", "--window", "1", "--hold-seconds", "4", "--deadline-seconds", "6", "--request-timeout-ms", "3200", "--cleanup-seconds", "1", - "--operation-trace", "on", "--tcp-kernel-samples", "on"}); + "--operation-trace", "on", "--tcp-kernel-samples", "on", "--socket-observer", "on"}); const auto deadline = clock_type::now() + std::chrono::seconds(6); auto wait_ready = [&](int fd, short events) { while (clock_type::now() < deadline) @@ -550,6 +550,43 @@ void progress_then_stall(const std::string& executable) { { throw std::runtime_error("unsupported periodic TCP info reason missing: " + client.log); } +#endif + const auto observation = output.find("\"event\":\"socket_observation\""); + if (observation == std::string::npos || output.find("\"event\":\"socket_observer_summary\"") == std::string::npos || + output.find("\"readable_bytes\":0,\"readable_error\":0,\"peek_bytes\":-1,\"peek_error\":" + + std::to_string(EAGAIN)) == std::string::npos || + output.find("\"read_pending\":true,\"input_bytes\":0,\"input_frame_ready\":false", observation) == + std::string::npos || + counter(output.substr(observation), "global_marker_completed_us") < 1 || + counter(output.substr(observation), "strand_probe_submitted_us") < 1) + { + throw std::runtime_error("independent controller observation missing: " + client.log); + } + if (output.find("\"event\":\"socket_observer_before_cancel\"") == std::string::npos || + output.find("\"clock_anchor\":{\"steady_before_ns\":\"", observation) == std::string::npos || + output.find("\"query_completed_steady_ns\":\"", observation) == std::string::npos) + { + throw std::runtime_error("observer wall/steady anchors missing: " + client.log); + } + bool strand_completed = false; + for (auto position = observation; position != std::string::npos; + position = output.find("\"event\":\"socket_observation\"", position + 1)) + { + const auto line = output.substr(position, output.find('\n', position) - position); + strand_completed = strand_completed || + counter(line, "strand_probe_completed_us") >= counter(line, "strand_probe_submitted_us"); + } + if (!strand_completed) + { + throw std::runtime_error("session strand probe did not complete: " + client.log); + } +#if defined(__APPLE__) + if (counter(output.substr(observation), "tcp_requested_length") < 100 || + counter(output.substr(observation), "tcp_returned_length") < 100 || + output.find("\"tcp_raw_hex\":\"\"", observation) != std::string::npos) + { + throw std::runtime_error("raw TCP ABI provenance missing: " + client.log); + } #endif std::cout << "controlled post-DATA stall retained periodic samples before timeout cleanup: " << client.log << '\n'; } @@ -746,6 +783,11 @@ int main(int argc, char** argv) { } directory = result; std::cout << "child logs and exits: " << directory << '\n'; + if (std::string_view(argv[2]) == "--observer-only") + { + progress_then_stall(argv[1]); + return 0; + } invalid_arguments(argv[1]); silent_peer(argv[1]); silent_peer(argv[1], true); diff --git a/tests/connection_socket_observer.cpp b/tests/connection_socket_observer.cpp new file mode 100644 index 0000000..ef3e10c --- /dev/null +++ b/tests/connection_socket_observer.cpp @@ -0,0 +1,132 @@ +// Copyright (c) 2026 OpenArkStudio. Distributed under BSL-1.0. See LICENSE. +#include +#include +#include + +#include "vendor/doctest/doctest.h" + +#include "../benchmarks/connections_socket_observer.hpp" + +using connection_load::socket_observer; + +DOCTEST_TEST_CASE("socket observer distinguishes unavailable from a valid zero") { + const auto invalid = connection_load::observe_socket(-1); + DOCTEST_CHECK(invalid.readable_bytes == -1); + DOCTEST_CHECK(invalid.peek_bytes == -1); +#if !defined(_WIN32) + DOCTEST_CHECK(invalid.readable_error == EBADF); + DOCTEST_CHECK(invalid.peek_error == EBADF); +#endif + socket_observer disabled(false); + DOCTEST_CHECK(disabled.publish(-1, {}).generation == 0); + disabled.sample(1, 0); + std::ostringstream out; + disabled.write_json(out, "client"); + DOCTEST_CHECK(out.str().empty()); +} + +DOCTEST_TEST_CASE("socket observation anchors bracket the wall query without losing JSON precision") { + const auto before = connection_load::observation_steady_ns(); + const auto anchor = connection_load::observation_anchor(); + const auto after = connection_load::observation_steady_ns(); + DOCTEST_CHECK(before <= anchor.steady_before_ns); + DOCTEST_CHECK(anchor.steady_before_ns <= anchor.steady_after_ns); + DOCTEST_CHECK(anchor.steady_after_ns <= after); + DOCTEST_CHECK(anchor.unix_ns > 0); +} + +#if !defined(_WIN32) +DOCTEST_TEST_CASE("peek preserves bytes and reports EAGAIN before data arrives") { + int pair[2]; + DOCTEST_REQUIRE(::socketpair(AF_UNIX, SOCK_STREAM, 0, pair) == 0); + const auto empty = connection_load::observe_socket(pair[1]); + DOCTEST_CHECK(empty.readable_bytes == 0); + DOCTEST_CHECK(empty.peek_bytes == -1); + DOCTEST_CHECK((empty.peek_error == EAGAIN || empty.peek_error == EWOULDBLOCK)); + DOCTEST_REQUIRE(::send(pair[0], "abc", 3, 0) == 3); + for (unsigned index = 0; index < 2; ++index) + { + const auto observed = connection_load::observe_socket(pair[1]); + DOCTEST_CHECK(observed.poll_result == 1); + DOCTEST_CHECK((observed.events & POLLIN) != 0); + DOCTEST_CHECK(observed.readable_bytes == 3); + DOCTEST_CHECK(observed.peek_bytes == 3); + DOCTEST_CHECK(std::string(reinterpret_cast(observed.peek.data()), 3) == "abc"); + } + char consumed[3]; + DOCTEST_CHECK(::recv(pair[1], consumed, 3, MSG_DONTWAIT) == 3); + DOCTEST_CHECK(std::string(consumed, 3) == "abc"); + ::close(pair[0]); + ::close(pair[1]); +} + +DOCTEST_TEST_CASE("observer pins identities and retires duplicate before close") { + int pair[2]; + DOCTEST_REQUIRE(::socketpair(AF_UNIX, SOCK_STREAM, 0, pair) == 0); + socket_observer observer(true); + socket_observer::snapshot state; + state.owner = {7, 1, 1}; + state.time_us = 1000000; + const auto first = observer.publish(pair[1], state); + DOCTEST_REQUIRE(first.generation != 0); + // Reusing the original numeric descriptor cannot redirect the held duplicate. + const int original = pair[1]; + ::close(original); + int replacement[2]; + DOCTEST_REQUIRE(::socketpair(AF_UNIX, SOCK_STREAM, 0, replacement) == 0); + DOCTEST_REQUIRE((replacement[0] == original || replacement[1] == original)); + DOCTEST_REQUIRE(::send(pair[0], "old", 3, 0) == 3); + observer.complete(first, 1000001); + observer.sample(1000002, 1000000); + observer.close(state.owner); + char byte{}; + const auto closed = ::recv(pair[0], &byte, 1, MSG_DONTWAIT); + DOCTEST_CHECK((closed == 0 || (closed < 0 && errno == ECONNRESET))); + std::ostringstream out; + observer.write_json(out, "client"); + DOCTEST_CHECK(out.str().find("\"event\":\"socket_observer_before_cancel\"") != std::string::npos); + DOCTEST_CHECK(out.str().find("\"clock_anchor\":{\"steady_before_ns\":\"") != std::string::npos); + DOCTEST_CHECK(out.str().find("\"query_completed_steady_ns\":\"") != std::string::npos); + DOCTEST_CHECK(out.str().find("\"peek_hex\":\"6f6c64\"") != std::string::npos); + DOCTEST_CHECK(out.str().find("\"strand_probe_completed_us\":1000001") != std::string::npos); + ::close(pair[0]); + ::close(replacement[0]); + ::close(replacement[1]); +} + +DOCTEST_TEST_CASE("recovered slots reject stale generation markers and bound observations") { + int pair[2]; + DOCTEST_REQUIRE(::socketpair(AF_UNIX, SOCK_STREAM, 0, pair) == 0); + socket_observer observer(true); + socket_observer::snapshot state; + state.owner = {7, 1, 1}; + const auto old = observer.publish(pair[1], state); + observer.prune(state.owner, false, 0); + const auto current = observer.publish(pair[1], state); + DOCTEST_CHECK(current.generation > old.generation); + observer.complete(old, 999); + for (unsigned index = 0; index < 20; ++index) + { + observer.sample(index * socket_observer::interval_us, 1); + } + observer.close(state.owner); + std::ostringstream out; + observer.write_json(out, "client"); + DOCTEST_CHECK(out.str().find("\"strand_probe_completed_us\":999") == std::string::npos); + DOCTEST_CHECK(out.str().find("\"sample\":7,") != std::string::npos); + DOCTEST_CHECK(out.str().find("\"sample\":8,") == std::string::npos); + ::close(pair[0]); + ::close(pair[1]); +} +#endif + +#if defined(__APPLE__) && defined(TCP_CONNECTION_INFO) +DOCTEST_TEST_CASE("short TCP responses never decode zero initialized counters as available") { + tcp_connection_info raw{}; + const auto short_info = + connection_load::decode_tcp_kernel_info(raw, connection_load::tcp_kernel_info_required_length - 1); + DOCTEST_CHECK(short_info.result == connection_load::tcp_kernel_info::status::short_response); + const auto full = connection_load::decode_tcp_kernel_info(raw, sizeof(raw)); + DOCTEST_CHECK(full.result == connection_load::tcp_kernel_info::status::available); +} +#endif