test(benchmarks): observe stalled sockets outside the IO worker

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
+        "$<TARGET_FILE:arknet_asio_connection_control>" --observer-only)
     arknet_add_check(benchmark_asio_connections_processes 90 arknet_asio_control_process_test
         "$<TARGET_FILE:arknet_asio_connection_control>" "$<TARGET_FILE:arknet_connection_benchmark>")
 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()

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.

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时间戳。

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<tcp::socket>) +
                   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<std::int64_t> 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<unsigned>(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<const char*>(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<std::chrono::microseconds>(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)

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.