From ad93aac9a6fc58393cb1e44bcc18d1ea7c6c5c97 Mon Sep 17 00:00:00 2001 From: aethernet-io Date: Wed, 28 Oct 2026 01:16:34 -0700 Subject: [PATCH 1/8] Dispatch failed IOCP completions when OVERLAPPED is set. Route errored GetQueuedCompletionStatus results to the registered socket callback instead of dropping them in the poller loop. --- aether/poller/win_poller.cpp | 25 +++++++++++-------------- 1 file changed, 11 insertions(+), 14 deletions(-) diff --git a/aether/poller/win_poller.cpp b/aether/poller/win_poller.cpp index f79f8b0be..75c875cb8 100644 --- a/aether/poller/win_poller.cpp +++ b/aether/poller/win_poller.cpp @@ -84,25 +84,22 @@ void IoCpPoller::Loop() { DWORD bytes_transferred; ULONG_PTR completion_key; LPOVERLAPPED overlapped; - if (!GetQueuedCompletionStatus(iocp_, &bytes_transferred, &completion_key, - &overlapped, INFINITE)) { - auto error = GetLastError(); - if (error == ERROR_CONNECTION_ABORTED) { - // socket closed - continue; - } - AE_TELE_ERROR(kWinpollWaitFailed, "GetQueuedCompletionStatus error {}", - error); - continue; - } + BOOL const gqcs_ok = GetQueuedCompletionStatus( + iocp_, &bytes_transferred, &completion_key, &overlapped, INFINITE); if (overlapped == nullptr) { + if (!gqcs_ok) { + auto const error = GetLastError(); + if (error == ERROR_CONNECTION_ABORTED) { + continue; + } + AE_TELE_ERROR(kWinpollWaitFailed, "GetQueuedCompletionStatus error {}", + error); + } continue; } - HANDLE descriptor; - // check if descriptor still in handles_ + HANDLE const descriptor = reinterpret_cast(completion_key); auto lock = std::scoped_lock{poller_mutex_}; - descriptor = reinterpret_cast(completion_key); auto it = event_map_.find(descriptor); if (it != event_map_.end()) { it->second(overlapped); From d4280f49983895b4b34a19903d766cf1899b0999 Mon Sep 17 00:00:00 2001 From: aethernet-io Date: Wed, 28 Oct 2026 09:36:26 -0700 Subject: [PATCH 2/8] Increase WinUdpSocket IOCP recv buffer to 64 KiB. Use a 65536-byte receive buffer so overlapped UDP recv can hold full datagrams beyond the 1200 B send MTU preference without WSAEMSGSIZE disconnects. Co-authored-by: Cursor --- .../system_sockets/sockets/win_udp_socket.cpp | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/aether/transport/system_sockets/sockets/win_udp_socket.cpp b/aether/transport/system_sockets/sockets/win_udp_socket.cpp index 86489d8ae..fd3205255 100644 --- a/aether/transport/system_sockets/sockets/win_udp_socket.cpp +++ b/aether/transport/system_sockets/sockets/win_udp_socket.cpp @@ -27,8 +27,17 @@ # include "aether/tele.h" namespace ae { + +// Outbound UDP payloads should stay near MTU to avoid IP fragmentation on send. +// This does not cap WinSocket receive capacity below. +static constexpr std::size_t kWinUdpPreferredSendDatagramBytes = 1200; + +// IOCP recv buffer must hold one complete datagram; cloud servers may aggregate +// payloads larger than the preferred send size without changing StreamInfo MTU. +static constexpr std::size_t kWinUdpRecvDatagramBufferBytes = 65536; + WinUdpSocket::WinUdpSocket(Ptr const& poller) - : WinSocket{*poller, 1200} { + : WinSocket{*poller, kWinUdpRecvDatagramBufferBytes} { bool created = false; // ::socket() sets WSA_FLAG_OVERLAPPED by default that allows us to use iocp From 001ff4354c0f73e7c89d0ea8ce25474302615315 Mon Sep 17 00:00:00 2001 From: aethernet-io Date: Thu, 29 Oct 2026 11:03:29 -0700 Subject: [PATCH 3/8] Add optional AE_BENCH_WIN_UDP_SNDBUF_BYTES SO_SNDBUF tuning for bench. --- .../system_sockets/sockets/win_udp_socket.cpp | 52 +++++++++++++++++++ 1 file changed, 52 insertions(+) diff --git a/aether/transport/system_sockets/sockets/win_udp_socket.cpp b/aether/transport/system_sockets/sockets/win_udp_socket.cpp index fd3205255..af14b839a 100644 --- a/aether/transport/system_sockets/sockets/win_udp_socket.cpp +++ b/aether/transport/system_sockets/sockets/win_udp_socket.cpp @@ -26,6 +26,32 @@ # include "aether/tele.h" +# ifdef AE_BENCH_WIN_UDP_SNDBUF_BYTES +# include +# include + +namespace { + +void WriteBenchUdpSndbufDiag(int requested_bytes, int actual_bytes, int error, + bool sndbuf_ok) { + char const* path = "win_udp_sndbuf_diag.txt"; + char* env_path = nullptr; + size_t env_len = 0; + if (_dupenv_s(&env_path, &env_len, "AE_BENCH_UDP_SNDBUF_DIAG") == 0 && + env_path != nullptr && env_path[0] != '\0') { + path = env_path; + } + std::ofstream out(path, std::ios::trunc); + out << "requested_bytes=" << requested_bytes << "\n"; + out << "actual_bytes=" << actual_bytes << "\n"; + out << "setsockopt_wsa_error=" << error << "\n"; + out << "sndbuf_ok=" << (sndbuf_ok ? "true" : "false") << "\n"; + free(env_path); +} + +} // namespace +# endif + namespace ae { // Outbound UDP payloads should stay near MTU to avoid IP fragmentation on send. @@ -61,6 +87,32 @@ WinUdpSocket::WinUdpSocket(Ptr const& poller) return; } +# ifdef AE_BENCH_WIN_UDP_SNDBUF_BYTES + { + int const requested = AE_BENCH_WIN_UDP_SNDBUF_BYTES; + int actual = 0; + int wsa_error = 0; + bool ok = false; + if (::setsockopt(sock, SOL_SOCKET, SO_SNDBUF, + reinterpret_cast(&requested), + sizeof(requested)) == SOCKET_ERROR) { + wsa_error = static_cast(WSAGetLastError()); + WriteBenchUdpSndbufDiag(requested, -1, wsa_error, false); + } else { + int len = sizeof(actual); + if (::getsockopt(sock, SOL_SOCKET, SO_SNDBUF, + reinterpret_cast(&actual), &len) == + SOCKET_ERROR) { + wsa_error = static_cast(WSAGetLastError()); + WriteBenchUdpSndbufDiag(requested, -1, wsa_error, false); + } else { + ok = actual >= (requested / 2); + WriteBenchUdpSndbufDiag(requested, actual, 0, ok); + } + } + } +# endif + created = true; socket_ = sock; } From f903c722d218d9aed53c6ab2b56cc246eda852d9 Mon Sep 17 00:00:00 2001 From: aethernet-io Date: Thu, 29 Oct 2026 14:57:37 -0700 Subject: [PATCH 4/8] Add read-only P2pStream destination cloud diagnostic accessor. --- aether/client_messages/p2p_message_stream.h | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/aether/client_messages/p2p_message_stream.h b/aether/client_messages/p2p_message_stream.h index 519ae1d05..5e443cd6a 100644 --- a/aether/client_messages/p2p_message_stream.h +++ b/aether/client_messages/p2p_message_stream.h @@ -60,6 +60,12 @@ class P2pStream final : public ByteIStream { void WriteOut(DataBuffer const& data); Uid const& destination() const; + // Read-only diagnostic view of the destination-cloud server pool (no I/O). + CloudServerConnections const* DiagnosticDestinationCloudConnections() + const { + return dest_cloud_conn_.get(); + } + private: static constexpr auto kBufferCapacity = AE_P2P_MESSAGE_STREAM_BUFFER_CAPACITY; using Buffer = BufferWrite; From 28bd7bed27646d235bbe8df25e3ba22081ed205a Mon Sep 17 00:00:00 2001 From: aethernet-io Date: Thu, 29 Oct 2026 17:13:24 -0700 Subject: [PATCH 5/8] Add bench connection-event diagnostic ring buffer (AE_BENCH_CONNECTION_EVENT_DIAG). --- aether/CMakeLists.txt | 3 + aether/ae_actions/ping.cpp | 86 +++++++++++ aether/ae_actions/ping.h | 5 + .../cloud_server_connection.cpp | 10 +- .../cloud_server_connection.h | 6 + .../cloud_server_connections.cpp | 34 +++++ .../cloud_connections/ping_cloud_servers.cpp | 28 +++- aether/config.h | 5 + .../bench_connection_event_diag.cpp | 143 ++++++++++++++++++ .../diagnostics/bench_connection_event_diag.h | 40 +++++ .../client_server_connection.cpp | 15 +- .../client_server_connection.h | 10 +- .../server_connections/server_connection.cpp | 27 +++- aether/server_connections/server_connection.h | 15 ++ 14 files changed, 420 insertions(+), 7 deletions(-) create mode 100644 aether/diagnostics/bench_connection_event_diag.cpp create mode 100644 aether/diagnostics/bench_connection_event_diag.h diff --git a/aether/CMakeLists.txt b/aether/CMakeLists.txt index a2919fa78..468ae3d73 100644 --- a/aether/CMakeLists.txt +++ b/aether/CMakeLists.txt @@ -169,6 +169,9 @@ list(APPEND aether_srcs "work_cloud_api/client_api/client_api_safe.cpp") +list(APPEND aether_srcs + "diagnostics/bench_connection_event_diag.cpp") + list(APPEND aether_srcs "cloud_connections/cloud_server_connection.cpp" "cloud_connections/cloud_server_connections.cpp" diff --git a/aether/ae_actions/ping.cpp b/aether/ae_actions/ping.cpp index e9eb0785c..7a594220f 100644 --- a/aether/ae_actions/ping.cpp +++ b/aether/ae_actions/ping.cpp @@ -23,7 +23,10 @@ # include # include "aether/cloud_connections/cloud_server_connection.h" +# include "aether/diagnostics/bench_connection_event_diag.h" +# include "aether/server_connections/client_server_connection.h" # include "aether/work_cloud_api/work_server_api/authorized_api.h" +# include "aether/write_action/write_action.h" # include "aether/ae_actions/ae_actions_tele.h" @@ -55,6 +58,35 @@ static_assert(PromiseErrorCodeToPingErrorCode( static_cast(std::numeric_limits::max()) + 1U) == kPingPromiseEvictedError); +# if AE_BENCH_CONNECTION_EVENT_DIAG +int WriteStatusToInt(WriteAction::Status status) { + switch (status) { + case WriteAction::Status::kSuccess: + return 0; + case WriteAction::Status::kFail: + return 1; + case WriteAction::Status::kStop: + return 2; + } + return -1; +} + +void RecordPingDiag(CloudServerConnection* cloud, char const* event, + RequestId request_id, std::uint32_t timeout_ms, + int write_status, int error_code) { + if (cloud == nullptr) { + return; + } + std::uint64_t const cloud_id = cloud->bench_cloud_instance_id(); + std::uint64_t client_id = 0; + if (auto* cc = cloud->client_connection()) { + client_id = cc->bench_client_instance_id(); + } + AE_BENCH_CONN_DIAG(event, cloud->server_id(), cloud_id, client_id, request_id, + timeout_ms, write_status, error_code, -1); +} +# endif + } // namespace Ping::Ping(AeContext const& ae_context, @@ -87,6 +119,14 @@ void Ping::Start(TimePoint current_time) { return; } state_ = RequestState::kPending; +# if AE_BENCH_CONNECTION_EVENT_DIAG + diag_write_terminal_ = false; + diag_active_request_id_ = 0; + diag_timeout_ms_ = static_cast( + std::chrono::duration_cast(timeout_).count()); + RecordPingDiag(cloud_server_connection_, "ping_start", 0, diag_timeout_ms_, + -1, 0); +# endif auto& write_action = cc->AuthorizedApiCall( [this, current_time](ApiContext& auth_api) { @@ -106,6 +146,11 @@ void Ping::Start(TimePoint current_time) { server_id_, req_id, timeout_); request_start_ = current_time; +# if AE_BENCH_CONNECTION_EVENT_DIAG + diag_active_request_id_ = req_id; + RecordPingDiag(cloud_server_connection_, "ping_request", req_id, + diag_timeout_ms_, -1, 0); +# endif auto wait_result_sub = pong_promise.Subscribe([this, req_id](auto&& res) { @@ -132,6 +177,19 @@ void Ping::Start(TimePoint current_time) { }); write_sub_ = write_action.status_event().Subscribe([this](auto status) { +# if AE_BENCH_CONNECTION_EVENT_DIAG + if (!diag_write_terminal_) { + diag_write_terminal_ = true; + char const* ev = "ping_write_success"; + if (status == WriteAction::Status::kFail) { + ev = "ping_write_fail"; + } else if (status == WriteAction::Status::kStop) { + ev = "ping_write_stop"; + } + RecordPingDiag(cloud_server_connection_, ev, diag_active_request_id_, + diag_timeout_ms_, WriteStatusToInt(status), 0); + } +# endif if (status == WriteAction::Status::kFail) { if (state_ != RequestState::kPending) { return; @@ -170,10 +228,18 @@ void Ping::PingResponse(RequestId request_id) { AE_TELED_DEBUG("Ping received server id {} request {} duration {}", server_id_, request_id, ping_duration); if (state_ == RequestState::kPending) { +# if AE_BENCH_CONNECTION_EVENT_DIAG + RecordPingDiag(cloud_server_connection_, "ping_pong", request_id, + diag_timeout_ms_, -1, 0); +# endif state_ = RequestState::kFinished; ResetRequestSubscriptions(); result_event_.Emit(PingResult{Ok{ping_duration}}); } else if (state_ == RequestState::kTimedOut) { +# if AE_BENCH_CONNECTION_EVENT_DIAG + RecordPingDiag(cloud_server_connection_, "ping_pong_late", request_id, + diag_timeout_ms_, -1, 0); +# endif wait_result_sub_.Reset(); state_ = RequestState::kFinished; result_event_.Emit(PingResult{LateDuration{ping_duration}}); @@ -185,6 +251,10 @@ void Ping::PingResponseError(RequestId request_id, std::int32_t error_code) { "Ping error server id {} request {} code {}", server_id_, request_id, error_code); if (state_ == RequestState::kPending) { +# if AE_BENCH_CONNECTION_EVENT_DIAG + RecordPingDiag(cloud_server_connection_, "ping_error", request_id, + diag_timeout_ms_, -1, static_cast(error_code)); +# endif state_ = RequestState::kFinished; ResetRequestSubscriptions(); if (error_code == -1) { @@ -204,13 +274,29 @@ void Ping::PingResponseTimeout(RequestId request_id) { } AE_TELE_ERROR(kPingTimeoutError, "Ping timeout server id {} request {}", server_id_, request_id); +# if AE_BENCH_CONNECTION_EVENT_DIAG + RecordPingDiag(cloud_server_connection_, "ping_timeout", request_id, + diag_timeout_ms_, -1, 0); +# endif state_ = RequestState::kTimedOut; +# if AE_BENCH_CONNECTION_EVENT_DIAG + if (!diag_write_terminal_ && diag_active_request_id_.id != 0) { + RecordPingDiag(cloud_server_connection_, "ping_write_unobserved", + diag_active_request_id_, diag_timeout_ms_, -1, 0); + } +# endif timeout_sub_.Reset(); write_sub_.Reset(); result_event_.Emit(PingResult{Error{2}}); } void Ping::ResetRequestSubscriptions() { +# if AE_BENCH_CONNECTION_EVENT_DIAG + if (!diag_write_terminal_ && diag_active_request_id_.id != 0) { + RecordPingDiag(cloud_server_connection_, "ping_write_unobserved", + diag_active_request_id_, diag_timeout_ms_, -1, 0); + } +# endif wait_result_sub_.Reset(); timeout_sub_.Reset(); write_sub_.Reset(); diff --git a/aether/ae_actions/ping.h b/aether/ae_actions/ping.h index fbd29f8f0..285d05d14 100644 --- a/aether/ae_actions/ping.h +++ b/aether/ae_actions/ping.h @@ -82,6 +82,11 @@ class Ping { ResultEvent result_event_; RequestState state_{RequestState::kCreated}; +#if AE_BENCH_CONNECTION_EVENT_DIAG + RequestId diag_active_request_id_{0}; + std::uint32_t diag_timeout_ms_{0}; + bool diag_write_terminal_{false}; +#endif }; } // namespace ae #endif // AE_ENABLE_PING diff --git a/aether/cloud_connections/cloud_server_connection.cpp b/aether/cloud_connections/cloud_server_connection.cpp index aaaf578e2..fea0a4301 100644 --- a/aether/cloud_connections/cloud_server_connection.cpp +++ b/aether/cloud_connections/cloud_server_connection.cpp @@ -20,6 +20,7 @@ #include #include +#include "aether/diagnostics/bench_connection_event_diag.h" #include "aether/server.h" namespace ae { @@ -29,7 +30,14 @@ CloudServerConnection::CloudServerConnection( IServerConnectionFactory& connection_factory) : cloud_{cloud}, server_id_{server_id}, - connection_factory_{&connection_factory} {} + connection_factory_{&connection_factory} +#if AE_BENCH_CONNECTION_EVENT_DIAG + , + bench_cloud_instance_id_{ + bench_conn_event_diag::AllocateCloudServerConnectionInstanceId()} +#endif +{ +} void CloudServerConnection::Restream() { if (client_connection_) { diff --git a/aether/cloud_connections/cloud_server_connection.h b/aether/cloud_connections/cloud_server_connection.h index cb761a71d..b298627aa 100644 --- a/aether/cloud_connections/cloud_server_connection.h +++ b/aether/cloud_connections/cloud_server_connection.h @@ -18,6 +18,7 @@ #define AETHER_CLOUD_CONNECTIONS_CLOUD_SERVER_CONNECTION_H_ #include +#include #include #include "aether-objects/ptr/ptr_view.h" @@ -50,6 +51,10 @@ class CloudServerConnection { std::size_t priority() const; void SetPriority(std::size_t priority); + std::uint64_t bench_cloud_instance_id() const { + return bench_cloud_instance_id_; + } + private: CloudServer& cloud_server() const; @@ -58,6 +63,7 @@ class CloudServerConnection { IServerConnectionFactory* connection_factory_; std::shared_ptr client_connection_; bool is_quarantined_ = false; + std::uint64_t bench_cloud_instance_id_{0}; }; } // namespace ae diff --git a/aether/cloud_connections/cloud_server_connections.cpp b/aether/cloud_connections/cloud_server_connections.cpp index e6ff53d4a..803666338 100644 --- a/aether/cloud_connections/cloud_server_connections.cpp +++ b/aether/cloud_connections/cloud_server_connections.cpp @@ -25,8 +25,30 @@ #include "aether/server_connections/server_connection.h" #include "aether/cloud_connections/cloud_connections_tele.h" // IWYU pragma: keep +#include "aether/diagnostics/bench_connection_event_diag.h" namespace ae { +namespace { + +std::int32_t MainSelectedServerId( + std::vector const& selected) { + if (selected.empty()) { + return -1; + } + return static_cast(selected.front()->server_id()); +} + +void RecordCloudDiag(char const* event, CloudServerConnection& sc, + std::vector const& selected) { + std::uint64_t client_id = 0; + if (auto* cc = sc.client_connection()) { + client_id = cc->bench_client_instance_id(); + } + AE_BENCH_CONN_DIAG(event, sc.server_id(), sc.bench_cloud_instance_id(), + client_id, 0, 0, -1, 0, MainSelectedServerId(selected)); +} + +} // namespace #if DEBUG auto ServerList(std::vector const& servers) { std::vector sids; @@ -209,6 +231,11 @@ bool CloudServerConnections::QuarantineServer( if (was_selected) { servers_update_event_.Emit(); } + RecordCloudDiag("cloud_quarantine", server_connection, selected_servers_); + if (was_selected) { + RecordCloudDiag("cloud_selected_changed", server_connection, + selected_servers_); + } server_quarantined_event_.Emit(&server_connection); // One delayed release: Disconnect + clear quarantine + reconcile. Do not @@ -253,6 +280,8 @@ void CloudServerConnections::ReleaseQuarantinedServer( std::rotate(first_quarantined, server_it, server_it + 1); server_connection.SetQuarantine(false); NormalizeServerPriorities(); + RecordCloudDiag("cloud_quarantine_release", server_connection, + selected_servers_); server_quarantine_release_event_.Emit(&server_connection); ServerEntryFor(server_connection).quarantine_sub.Reset(); @@ -307,6 +336,7 @@ void CloudServerConnections::ReconcileServers() { assert(candidate_index == selected_servers_.size() && "candidate must follow selected prefix"); selected_servers_.emplace_back(candidate); + RecordCloudDiag("cloud_selected_append", *candidate, selected_servers_); candidate_index = selected_servers_.size(); emplaced = true; } @@ -314,6 +344,10 @@ void CloudServerConnections::ReconcileServers() { NormalizeServerPriorities(); AE_TELED_DEBUG("Selected servers reconciled selected_count={}, list={}", selected_servers_.size(), ServerList(selected_servers_)); + if (!selected_servers_.empty()) { + RecordCloudDiag("cloud_selected_reconciled", *selected_servers_.front(), + selected_servers_); + } servers_update_event_.Emit(); } } diff --git a/aether/cloud_connections/ping_cloud_servers.cpp b/aether/cloud_connections/ping_cloud_servers.cpp index b1128e142..a36baca0c 100644 --- a/aether/cloud_connections/ping_cloud_servers.cpp +++ b/aether/cloud_connections/ping_cloud_servers.cpp @@ -26,6 +26,7 @@ # include "aether/executors/executors.h" # include "aether/cloud_connections/cloud_connections_tele.h" +# include "aether/diagnostics/bench_connection_event_diag.h" namespace ae { @@ -185,7 +186,15 @@ void PingCloudServers::ServerPing::Start() { } void PingCloudServers::ServerPing::OnPingResult(Ping::PingResult const& res) { + std::uint64_t const cloud_id = cloud_sc_->bench_cloud_instance_id(); + std::uint64_t client_id = 0; auto* cc = cloud_sc_->client_connection(); + if (cc != nullptr) { + client_id = cc->bench_client_instance_id(); + } + AE_BENCH_CONN_DIAG("ping_on_result", cloud_sc_->server_id(), cloud_id, + client_id, 0, 0, -1, 0, -1); + cc = cloud_sc_->client_connection(); if (cc == nullptr) { AE_TELED_ERROR("Client connection is null"); return; @@ -198,7 +207,7 @@ void PingCloudServers::ServerPing::OnPingResult(Ping::PingResult const& res) { } std::visit( - [this, c](auto const& value) { + [this, c, cloud_id, client_id](auto const& value) { using T = std::decay_t; if constexpr (std::is_same_v>) { c->channel_statistics().AddResponseTime(value.value); @@ -207,6 +216,8 @@ void PingCloudServers::ServerPing::OnPingResult(Ping::PingResult const& res) { c->channel_statistics().AddResponseTime(value.duration); } else { AE_TELED_ERROR("Ping error!"); + AE_BENCH_CONN_DIAG("ping_on_result_restream", cloud_sc_->server_id(), + cloud_id, client_id, 0, 0, -1, 0, -1); ScheduleRestream(); } }, @@ -226,10 +237,25 @@ void PingCloudServers::ServerPing::ScheduleRestream() { return; } + std::uint64_t const cloud_id = cloud_sc_->bench_cloud_instance_id(); + std::uint64_t client_id = 0; + if (auto* cc = cloud_sc_->client_connection()) { + client_id = cc->bench_client_instance_id(); + } + AE_BENCH_CONN_DIAG("schedule_restream", cloud_sc_->server_id(), cloud_id, + client_id, 0, 0, -1, 0, -1); + // TODO: should we block till restream? restream_blocker_ = policy_->AcquireSuspendBlock(); restream_sub_ = ae_context_.scheduler().Task([this]() { + std::uint64_t const cloud_id = cloud_sc_->bench_cloud_instance_id(); + std::uint64_t client_id = 0; auto* cc = cloud_sc_->client_connection(); + if (cc != nullptr) { + client_id = cc->bench_client_instance_id(); + } + AE_BENCH_CONN_DIAG("schedule_restream_run", cloud_sc_->server_id(), + cloud_id, client_id, 0, 0, -1, 0, -1); if (cc != nullptr) { cc->Restream(); } diff --git a/aether/config.h b/aether/config.h index 70b757fd0..6972303e1 100644 --- a/aether/config.h +++ b/aether/config.h @@ -528,6 +528,11 @@ # define AE_STATISTICS_MAX_SIZE (10 * 1024) // 10 KB #endif +// Bench-only connection/ping rotation trace (heap ring buffer, no per-event I/O). +#ifndef AE_BENCH_CONNECTION_EVENT_DIAG +# define AE_BENCH_CONNECTION_EVENT_DIAG 0 +#endif + #ifndef NDEBUG # define DEBUG 1 #endif diff --git a/aether/diagnostics/bench_connection_event_diag.cpp b/aether/diagnostics/bench_connection_event_diag.cpp new file mode 100644 index 000000000..fbeb99173 --- /dev/null +++ b/aether/diagnostics/bench_connection_event_diag.cpp @@ -0,0 +1,143 @@ +#include "aether/diagnostics/bench_connection_event_diag.h" + +#if AE_BENCH_CONNECTION_EVENT_DIAG + +# include +# include +# include +# include + +# if defined(_WIN32) +# ifndef NOMINMAX +# define NOMINMAX +# endif +# include +# endif + +namespace ae::bench_conn_event_diag { +namespace { + +struct Row { + std::int64_t qpc_ns{0}; + char const* event{nullptr}; + std::uint32_t server_id{0}; + std::uint64_t cloud_conn_instance_id{0}; + std::uint64_t client_conn_instance_id{0}; + std::uint32_t ping_request_id{0}; + std::uint32_t timeout_ms{0}; + int write_status{-1}; + int error_code{0}; + std::int32_t main_server_id{-1}; +}; + +std::atomic g_cloud_id{1}; +std::atomic g_client_id{1}; +std::atomic g_server_id{1}; +std::atomic g_dropped{0}; + +std::mutex g_mu; +std::vector g_rows; +std::size_t g_max{65536}; + +std::int64_t QpcNowNs() { +# if defined(_WIN32) + LARGE_INTEGER c{}; + LARGE_INTEGER f{}; + if (::QueryPerformanceCounter(&c) == 0 || + ::QueryPerformanceFrequency(&f) == 0 || f.QuadPart == 0) { + return 0; + } + std::int64_t const q = c.QuadPart; + std::int64_t const fr = f.QuadPart; + return (q / fr) * 1000000000LL + ((q % fr) * 1000000000LL) / fr; +# else + return 0; +# endif +} + +} // namespace + +std::uint64_t AllocateCloudServerConnectionInstanceId() { + return g_cloud_id.fetch_add(1, std::memory_order_relaxed); +} +std::uint64_t AllocateClientServerConnectionInstanceId() { + return g_client_id.fetch_add(1, std::memory_order_relaxed); +} +std::uint64_t AllocateServerConnectionInstanceId() { + return g_server_id.fetch_add(1, std::memory_order_relaxed); +} + +void Configure(std::size_t max_records) { + std::lock_guard lock(g_mu); + g_max = max_records > 0 ? max_records : 65536; + g_rows.clear(); + g_rows.reserve(g_max); + g_dropped.store(0, std::memory_order_relaxed); +} + +void ResetForTest() { Configure(4096); } + +void Record(char const* event, std::uint32_t server_id, + std::uint64_t cloud_conn_instance_id, + std::uint64_t client_conn_instance_id, std::uint32_t ping_request_id, + std::uint32_t timeout_ms, int write_status, int error_code, + std::int32_t main_server_id) { + Row r; + r.qpc_ns = QpcNowNs(); + r.event = event; + r.server_id = server_id; + r.cloud_conn_instance_id = cloud_conn_instance_id; + r.client_conn_instance_id = client_conn_instance_id; + r.ping_request_id = ping_request_id; + r.timeout_ms = timeout_ms; + r.write_status = write_status; + r.error_code = error_code; + r.main_server_id = main_server_id; + std::lock_guard lock(g_mu); + if (g_rows.size() >= g_max) { + g_dropped.fetch_add(1, std::memory_order_relaxed); + return; + } + g_rows.push_back(r); +} + +std::uint64_t DroppedCount() { + return g_dropped.load(std::memory_order_relaxed); +} + +void WriteTsv(char const* path) { + std::vector copy; + { + std::lock_guard lock(g_mu); + copy = g_rows; + } + FILE* f = nullptr; +# if defined(_WIN32) + if (fopen_s(&f, path, "w") != 0 || f == nullptr) { + return; + } +# else + f = std::fopen(path, "w"); + if (f == nullptr) { + return; + } +# endif + std::fprintf( + f, + "qpc_ns\tevent\tserver_id\tcloud_conn_instance_id\t" + "client_conn_instance_id\tping_request_id\ttimeout_ms\twrite_status\t" + "error_code\tmain_server_id\n"); + for (auto const& r : copy) { + std::fprintf(f, "%lld\t%s\t%u\t%llu\t%llu\t%u\t%u\t%d\t%d\t%d\n", + static_cast(r.qpc_ns), r.event, r.server_id, + static_cast(r.cloud_conn_instance_id), + static_cast(r.client_conn_instance_id), + r.ping_request_id, r.timeout_ms, r.write_status, r.error_code, + static_cast(r.main_server_id)); + } + std::fclose(f); +} + +} // namespace ae::bench_conn_event_diag + +#endif diff --git a/aether/diagnostics/bench_connection_event_diag.h b/aether/diagnostics/bench_connection_event_diag.h new file mode 100644 index 000000000..4d376bf85 --- /dev/null +++ b/aether/diagnostics/bench_connection_event_diag.h @@ -0,0 +1,40 @@ +#ifndef AETHER_DIAGNOSTICS_BENCH_CONNECTION_EVENT_DIAG_H_ +#define AETHER_DIAGNOSTICS_BENCH_CONNECTION_EVENT_DIAG_H_ + +#include "aether/config.h" + +#if AE_BENCH_CONNECTION_EVENT_DIAG + +# include +# include + +namespace ae::bench_conn_event_diag { + +std::uint64_t AllocateCloudServerConnectionInstanceId(); +std::uint64_t AllocateClientServerConnectionInstanceId(); +std::uint64_t AllocateServerConnectionInstanceId(); + +void Configure(std::size_t max_records); +void ResetForTest(); + +void Record(char const* event, std::uint32_t server_id, + std::uint64_t cloud_conn_instance_id, + std::uint64_t client_conn_instance_id, std::uint32_t ping_request_id, + std::uint32_t timeout_ms, int write_status, int error_code, + std::int32_t main_server_id); + +std::uint64_t DroppedCount(); +void WriteTsv(char const* path); + +} // namespace ae::bench_conn_event_diag + +# define AE_BENCH_CONN_DIAG(...) \ + ::ae::bench_conn_event_diag::Record(__VA_ARGS__) + +#else + +# define AE_BENCH_CONN_DIAG(...) ((void)0) + +#endif + +#endif diff --git a/aether/server_connections/client_server_connection.cpp b/aether/server_connections/client_server_connection.cpp index 3ff926ef7..76f9d50f9 100644 --- a/aether/server_connections/client_server_connection.cpp +++ b/aether/server_connections/client_server_connection.cpp @@ -17,6 +17,7 @@ #include "aether/server_connections/client_server_connection.h" #include "aether/aether.h" +#include "aether/diagnostics/bench_connection_event_diag.h" #include "aether/api_protocol/api_protocol.h" #include "aether/client.h" #include "aether/crypto/ikey_provider.h" @@ -91,8 +92,9 @@ class ClientCryptoProvider final : public ICryptoProvider { SyncDecryptProvider decryptor_; }; -BufferedServerConnection::BufferedServerConnection(AeContext const& ae_context, - Ptr const& server) +BufferedServerConnection::BufferedServerConnection( + AeContext const& ae_context, Ptr const& server, + std::uint64_t bench_client_instance_id) : buffer_write{ae_context, [&](DataBuffer&& in_data) -> WriteAction* { if (server_connection.stream_info().is_writable) { @@ -101,6 +103,9 @@ BufferedServerConnection::BufferedServerConnection(AeContext const& ae_context, return nullptr; }}, server_connection{ae_context, server} { +#if AE_BENCH_CONNECTION_EVENT_DIAG + server_connection.SetBenchDiagClientInstanceId(bench_client_instance_id); +#endif if (server_connection.stream_info().is_writable) { buffer_write.buffer_off(); } @@ -153,7 +158,11 @@ ClientServerConnection::ClientServerConnection(AeContext const& ae_context, client_api_unsafe_{ae_context.event_system(), *crypto_provider_->decryptor()}, login_api_{*crypto_provider_->encryptor()}, - server_connection_{ae_context_, server} { +#if AE_BENCH_CONNECTION_EVENT_DIAG + bench_client_instance_id_{ + bench_conn_event_diag::AllocateClientServerConnectionInstanceId()}, +#endif + server_connection_{ae_context_, server, bench_client_instance_id_} { AE_TELED_DEBUG("Client server connection from {}:e-{} to {}", uid_, ephemeral_uid_, server->server_id); diff --git a/aether/server_connections/client_server_connection.h b/aether/server_connections/client_server_connection.h index 1b0264664..cb0d4e8e0 100644 --- a/aether/server_connections/client_server_connection.h +++ b/aether/server_connections/client_server_connection.h @@ -19,6 +19,8 @@ #include "aether/config.h" +#include + #include "aether/ae_context.h" #include "aether/common.h" #include "aether/crypto/icrypto_provider.h" @@ -44,7 +46,8 @@ class BufferedServerConnection : public ByteIStream { using Buffer = BufferWrite; BufferedServerConnection(AeContext const& ae_context, - Ptr const& server); + Ptr const& server, + std::uint64_t bench_client_instance_id); WriteAction& Write(DataBuffer&& in_data) override; StreamUpdateEvent const& stream_update_event() override; @@ -78,6 +81,10 @@ class ClientServerConnection { ServerConnection& server_connection(); + std::uint64_t bench_client_instance_id() const { + return bench_client_instance_id_; + } + private: void OutData(DataBuffer const& data); @@ -92,6 +99,7 @@ class ClientServerConnection { ClientApiUnsafe client_api_unsafe_; LoginApi login_api_; + std::uint64_t bench_client_instance_id_{0}; client_server_connection_internal ::BufferedServerConnection server_connection_; }; diff --git a/aether/server_connections/server_connection.cpp b/aether/server_connections/server_connection.cpp index be042d6f7..0bc7a4075 100644 --- a/aether/server_connections/server_connection.cpp +++ b/aether/server_connections/server_connection.cpp @@ -22,6 +22,7 @@ #include "aether-miscpp/types/method_ptr.h" #include "aether/aether.h" +#include "aether/diagnostics/bench_connection_event_diag.h" #include "aether/channels/channel.h" #include "aether/server.h" @@ -38,7 +39,13 @@ ServerConnection::ServerConnection(AeContext const& ae_context, out_data_event_{ae_context_}, stream_update_event_{ae_context_}, server_error_{ae_context_}, - channel_changed_{ae_context_} { + channel_changed_{ae_context_} +#if AE_BENCH_CONNECTION_EVENT_DIAG + , + bench_server_instance_id_{ + bench_conn_event_diag::AllocateServerConnectionInstanceId()} +#endif +{ InitChannels(); SelectChannel(); } @@ -65,6 +72,12 @@ ServerConnection::OutDataEvent const& ServerConnection::out_data_event() { } void ServerConnection::Restream() { + ServerId sid{}; + if (auto server = server_.Lock()) { + sid = server->server_id; + } + AE_BENCH_CONN_DIAG("server_restream", sid, 0, bench_client_instance_id_, 0, 0, + -1, 0, -1); // restream means something wrong with current channel ChannelError(); } @@ -236,6 +249,12 @@ void ServerConnection::ChannelBuildFailed(ChannelEntry& attempted_channel) { } void ServerConnection::ServerError() { + ServerId sid{}; + if (auto server = server_.Lock()) { + sid = server->server_id; + } + AE_BENCH_CONN_DIAG("server_error", sid, 0, bench_client_instance_id_, 0, 0, -1, + 0, -1); AE_TELED_ERROR("SERVER_CONNECTION_ERROR"); channel_stream_update_sub_.Reset(); channel_stream_out_data_sub_.Reset(); @@ -249,6 +268,12 @@ void ServerConnection::ServerError() { } void ServerConnection::ChannelError() { + ServerId sid{}; + if (auto server = server_.Lock()) { + sid = server->server_id; + } + AE_BENCH_CONN_DIAG("channel_error", sid, 0, bench_client_instance_id_, 0, 0, + -1, 0, -1); AE_TELED_ERROR("Channel error"); channel_stream_update_sub_.Reset(); channel_stream_out_data_sub_.Reset(); diff --git a/aether/server_connections/server_connection.h b/aether/server_connections/server_connection.h index 993723926..78e8ff778 100644 --- a/aether/server_connections/server_connection.h +++ b/aether/server_connections/server_connection.h @@ -17,10 +17,13 @@ #ifndef AETHER_SERVER_CONNECTIONS_SERVER_CONNECTION_H_ #define AETHER_SERVER_CONNECTIONS_SERVER_CONNECTION_H_ +#include #include #include #include +#include "aether/config.h" + #include "aether-objects/ptr/ptr.h" #include "aether-objects/ptr/ptr_view.h" #include "aether/ae_context.h" @@ -56,6 +59,16 @@ class ServerConnection final : public ByteIStream { Ptr current_channel() const; + std::uint64_t bench_server_instance_id() const { + return bench_server_instance_id_; + } + +#if AE_BENCH_CONNECTION_EVENT_DIAG + void SetBenchDiagClientInstanceId(std::uint64_t id) { + bench_client_instance_id_ = id; + } +#endif + private: friend struct ServerConnectionTestAccess; @@ -78,6 +91,8 @@ class ServerConnection final : public ByteIStream { PtrView server_; bool full_connected_; + std::uint64_t bench_server_instance_id_{0}; + std::uint64_t bench_client_instance_id_{0}; ChannelEntry* top_channel_; std::unique_ptr stream_; std::vector channels_; From e7aad30ce9f024122dd2a0f4c19f531ad19716cf Mon Sep 17 00:00:00 2001 From: aethernet-io Date: Thu, 29 Oct 2026 17:28:18 -0700 Subject: [PATCH 6/8] Add optional AE_BENCH_PING_MIN_TIMEOUT_MS floor and dual timeout diag fields. --- aether/ae_actions/ping.cpp | 49 ++++++++++++------- aether/ae_actions/ping.h | 6 ++- .../cloud_server_connections.cpp | 2 +- .../cloud_connections/ping_cloud_servers.cpp | 15 ++++-- aether/config.h | 5 ++ .../bench_connection_event_diag.cpp | 19 ++++--- .../diagnostics/bench_connection_event_diag.h | 4 +- aether/diagnostics/bench_ping_min_timeout.h | 32 ++++++++++++ .../server_connections/server_connection.cpp | 8 +-- 9 files changed, 101 insertions(+), 39 deletions(-) create mode 100644 aether/diagnostics/bench_ping_min_timeout.h diff --git a/aether/ae_actions/ping.cpp b/aether/ae_actions/ping.cpp index 7a594220f..39b964ff6 100644 --- a/aether/ae_actions/ping.cpp +++ b/aether/ae_actions/ping.cpp @@ -72,8 +72,9 @@ int WriteStatusToInt(WriteAction::Status status) { } void RecordPingDiag(CloudServerConnection* cloud, char const* event, - RequestId request_id, std::uint32_t timeout_ms, - int write_status, int error_code) { + RequestId request_id, std::uint32_t original_timeout_ms, + std::uint32_t applied_timeout_ms, int write_status, + int error_code) { if (cloud == nullptr) { return; } @@ -83,7 +84,8 @@ void RecordPingDiag(CloudServerConnection* cloud, char const* event, client_id = cc->bench_client_instance_id(); } AE_BENCH_CONN_DIAG(event, cloud->server_id(), cloud_id, client_id, request_id, - timeout_ms, write_status, error_code, -1); + original_timeout_ms, applied_timeout_ms, write_status, + error_code, -1); } # endif @@ -91,14 +93,24 @@ void RecordPingDiag(CloudServerConnection* cloud, char const* event, Ping::Ping(AeContext const& ae_context, CloudServerConnection& cloud_server_connection, - Duration next_ping_hint, Duration rx_window, Duration timeout) + Duration next_ping_hint, Duration rx_window, + Duration response_timeout, Duration channel_response_timeout) : ae_context_{ae_context}, cloud_server_connection_{&cloud_server_connection}, next_ping_hint_{next_ping_hint}, rx_window_{rx_window}, - timeout_{timeout}, + timeout_{response_timeout}, server_id_{cloud_server_connection_->server_id()}, result_event_{ae_context_} { +#if AE_BENCH_CONNECTION_EVENT_DIAG + diag_applied_timeout_ms_ = static_cast( + std::chrono::duration_cast(response_timeout) + .count()); + diag_original_timeout_ms_ = static_cast( + std::chrono::duration_cast( + channel_response_timeout) + .count()); +#endif AE_TELE_INFO( kPing, "Ping action created to server id: {}, interval: {:%S}s, rx_window: " @@ -122,10 +134,8 @@ void Ping::Start(TimePoint current_time) { # if AE_BENCH_CONNECTION_EVENT_DIAG diag_write_terminal_ = false; diag_active_request_id_ = 0; - diag_timeout_ms_ = static_cast( - std::chrono::duration_cast(timeout_).count()); - RecordPingDiag(cloud_server_connection_, "ping_start", 0, diag_timeout_ms_, - -1, 0); + RecordPingDiag(cloud_server_connection_, "ping_start", 0, + diag_original_timeout_ms_, diag_applied_timeout_ms_, -1, 0); # endif auto& write_action = cc->AuthorizedApiCall( @@ -149,7 +159,8 @@ void Ping::Start(TimePoint current_time) { # if AE_BENCH_CONNECTION_EVENT_DIAG diag_active_request_id_ = req_id; RecordPingDiag(cloud_server_connection_, "ping_request", req_id, - diag_timeout_ms_, -1, 0); + diag_original_timeout_ms_, diag_applied_timeout_ms_, -1, + 0); # endif auto wait_result_sub = @@ -187,7 +198,8 @@ void Ping::Start(TimePoint current_time) { ev = "ping_write_stop"; } RecordPingDiag(cloud_server_connection_, ev, diag_active_request_id_, - diag_timeout_ms_, WriteStatusToInt(status), 0); + diag_original_timeout_ms_, diag_applied_timeout_ms_, + WriteStatusToInt(status), 0); } # endif if (status == WriteAction::Status::kFail) { @@ -230,7 +242,7 @@ void Ping::PingResponse(RequestId request_id) { if (state_ == RequestState::kPending) { # if AE_BENCH_CONNECTION_EVENT_DIAG RecordPingDiag(cloud_server_connection_, "ping_pong", request_id, - diag_timeout_ms_, -1, 0); + diag_original_timeout_ms_, diag_applied_timeout_ms_, -1, 0); # endif state_ = RequestState::kFinished; ResetRequestSubscriptions(); @@ -238,7 +250,7 @@ void Ping::PingResponse(RequestId request_id) { } else if (state_ == RequestState::kTimedOut) { # if AE_BENCH_CONNECTION_EVENT_DIAG RecordPingDiag(cloud_server_connection_, "ping_pong_late", request_id, - diag_timeout_ms_, -1, 0); + diag_original_timeout_ms_, diag_applied_timeout_ms_, -1, 0); # endif wait_result_sub_.Reset(); state_ = RequestState::kFinished; @@ -253,7 +265,8 @@ void Ping::PingResponseError(RequestId request_id, std::int32_t error_code) { if (state_ == RequestState::kPending) { # if AE_BENCH_CONNECTION_EVENT_DIAG RecordPingDiag(cloud_server_connection_, "ping_error", request_id, - diag_timeout_ms_, -1, static_cast(error_code)); + diag_original_timeout_ms_, diag_applied_timeout_ms_, -1, + static_cast(error_code)); # endif state_ = RequestState::kFinished; ResetRequestSubscriptions(); @@ -276,13 +289,14 @@ void Ping::PingResponseTimeout(RequestId request_id) { server_id_, request_id); # if AE_BENCH_CONNECTION_EVENT_DIAG RecordPingDiag(cloud_server_connection_, "ping_timeout", request_id, - diag_timeout_ms_, -1, 0); + diag_original_timeout_ms_, diag_applied_timeout_ms_, -1, 0); # endif state_ = RequestState::kTimedOut; # if AE_BENCH_CONNECTION_EVENT_DIAG if (!diag_write_terminal_ && diag_active_request_id_.id != 0) { RecordPingDiag(cloud_server_connection_, "ping_write_unobserved", - diag_active_request_id_, diag_timeout_ms_, -1, 0); + diag_active_request_id_, diag_original_timeout_ms_, + diag_applied_timeout_ms_, -1, 0); } # endif timeout_sub_.Reset(); @@ -294,7 +308,8 @@ void Ping::ResetRequestSubscriptions() { # if AE_BENCH_CONNECTION_EVENT_DIAG if (!diag_write_terminal_ && diag_active_request_id_.id != 0) { RecordPingDiag(cloud_server_connection_, "ping_write_unobserved", - diag_active_request_id_, diag_timeout_ms_, -1, 0); + diag_active_request_id_, diag_original_timeout_ms_, + diag_applied_timeout_ms_, -1, 0); } # endif wait_result_sub_.Reset(); diff --git a/aether/ae_actions/ping.h b/aether/ae_actions/ping.h index 285d05d14..8174876a3 100644 --- a/aether/ae_actions/ping.h +++ b/aether/ae_actions/ping.h @@ -47,7 +47,8 @@ class Ping { Ping(AeContext const& ae_context, CloudServerConnection& cloud_server_connection, Duration next_ping_hint, - Duration rx_window, Duration timeout); + Duration rx_window, Duration response_timeout, + Duration channel_response_timeout); AE_CLASS_NO_COPY_MOVE(Ping); @@ -84,7 +85,8 @@ class Ping { RequestState state_{RequestState::kCreated}; #if AE_BENCH_CONNECTION_EVENT_DIAG RequestId diag_active_request_id_{0}; - std::uint32_t diag_timeout_ms_{0}; + std::uint32_t diag_original_timeout_ms_{0}; + std::uint32_t diag_applied_timeout_ms_{0}; bool diag_write_terminal_{false}; #endif }; diff --git a/aether/cloud_connections/cloud_server_connections.cpp b/aether/cloud_connections/cloud_server_connections.cpp index 803666338..61078ca0e 100644 --- a/aether/cloud_connections/cloud_server_connections.cpp +++ b/aether/cloud_connections/cloud_server_connections.cpp @@ -45,7 +45,7 @@ void RecordCloudDiag(char const* event, CloudServerConnection& sc, client_id = cc->bench_client_instance_id(); } AE_BENCH_CONN_DIAG(event, sc.server_id(), sc.bench_cloud_instance_id(), - client_id, 0, 0, -1, 0, MainSelectedServerId(selected)); + client_id, 0, 0, 0, -1, 0, MainSelectedServerId(selected)); } } // namespace diff --git a/aether/cloud_connections/ping_cloud_servers.cpp b/aether/cloud_connections/ping_cloud_servers.cpp index a36baca0c..231964dcf 100644 --- a/aether/cloud_connections/ping_cloud_servers.cpp +++ b/aether/cloud_connections/ping_cloud_servers.cpp @@ -27,6 +27,7 @@ # include "aether/cloud_connections/cloud_connections_tele.h" # include "aether/diagnostics/bench_connection_event_diag.h" +# include "aether/diagnostics/bench_ping_min_timeout.h" namespace ae { @@ -120,8 +121,12 @@ auto PingCloudServers::ServerPing::MakePing() { return ex::set_error(std::move(ctx.receiver), 2); } + Duration const channel_timeout = c->ResponseTimeout(); + Duration const applied_timeout = + bench_ping_min_timeout::ApplyBenchPingMinTimeout(channel_timeout); ping_.emplace(ae_context_, *cloud_sc_, timing_conf_.interval, - timing_conf_.rx_window, c->ResponseTimeout()); + timing_conf_.rx_window, applied_timeout, + channel_timeout); ping_blocker_ = policy_->AcquireSuspendBlock(); ping_->result_event().Subscribe( @@ -193,7 +198,7 @@ void PingCloudServers::ServerPing::OnPingResult(Ping::PingResult const& res) { client_id = cc->bench_client_instance_id(); } AE_BENCH_CONN_DIAG("ping_on_result", cloud_sc_->server_id(), cloud_id, - client_id, 0, 0, -1, 0, -1); + client_id, 0, 0, 0, -1, 0, -1); cc = cloud_sc_->client_connection(); if (cc == nullptr) { AE_TELED_ERROR("Client connection is null"); @@ -217,7 +222,7 @@ void PingCloudServers::ServerPing::OnPingResult(Ping::PingResult const& res) { } else { AE_TELED_ERROR("Ping error!"); AE_BENCH_CONN_DIAG("ping_on_result_restream", cloud_sc_->server_id(), - cloud_id, client_id, 0, 0, -1, 0, -1); + cloud_id, client_id, 0, 0, 0, -1, 0, -1); ScheduleRestream(); } }, @@ -243,7 +248,7 @@ void PingCloudServers::ServerPing::ScheduleRestream() { client_id = cc->bench_client_instance_id(); } AE_BENCH_CONN_DIAG("schedule_restream", cloud_sc_->server_id(), cloud_id, - client_id, 0, 0, -1, 0, -1); + client_id, 0, 0, 0, -1, 0, -1); // TODO: should we block till restream? restream_blocker_ = policy_->AcquireSuspendBlock(); @@ -255,7 +260,7 @@ void PingCloudServers::ServerPing::ScheduleRestream() { client_id = cc->bench_client_instance_id(); } AE_BENCH_CONN_DIAG("schedule_restream_run", cloud_sc_->server_id(), - cloud_id, client_id, 0, 0, -1, 0, -1); + cloud_id, client_id, 0, 0, 0, -1, 0, -1); if (cc != nullptr) { cc->Restream(); } diff --git a/aether/config.h b/aether/config.h index 6972303e1..e5cdf98c7 100644 --- a/aether/config.h +++ b/aether/config.h @@ -533,6 +533,11 @@ # define AE_BENCH_CONNECTION_EVENT_DIAG 0 #endif +// Optional UDP-sender ping response timeout floor (ms); 0 = use channel timeout only. +#ifndef AE_BENCH_PING_MIN_TIMEOUT_MS +# define AE_BENCH_PING_MIN_TIMEOUT_MS 0 +#endif + #ifndef NDEBUG # define DEBUG 1 #endif diff --git a/aether/diagnostics/bench_connection_event_diag.cpp b/aether/diagnostics/bench_connection_event_diag.cpp index fbeb99173..3dc5b81e6 100644 --- a/aether/diagnostics/bench_connection_event_diag.cpp +++ b/aether/diagnostics/bench_connection_event_diag.cpp @@ -24,7 +24,8 @@ struct Row { std::uint64_t cloud_conn_instance_id{0}; std::uint64_t client_conn_instance_id{0}; std::uint32_t ping_request_id{0}; - std::uint32_t timeout_ms{0}; + std::uint32_t original_timeout_ms{0}; + std::uint32_t applied_timeout_ms{0}; int write_status{-1}; int error_code{0}; std::int32_t main_server_id{-1}; @@ -80,8 +81,8 @@ void ResetForTest() { Configure(4096); } void Record(char const* event, std::uint32_t server_id, std::uint64_t cloud_conn_instance_id, std::uint64_t client_conn_instance_id, std::uint32_t ping_request_id, - std::uint32_t timeout_ms, int write_status, int error_code, - std::int32_t main_server_id) { + std::uint32_t original_timeout_ms, std::uint32_t applied_timeout_ms, + int write_status, int error_code, std::int32_t main_server_id) { Row r; r.qpc_ns = QpcNowNs(); r.event = event; @@ -89,7 +90,8 @@ void Record(char const* event, std::uint32_t server_id, r.cloud_conn_instance_id = cloud_conn_instance_id; r.client_conn_instance_id = client_conn_instance_id; r.ping_request_id = ping_request_id; - r.timeout_ms = timeout_ms; + r.original_timeout_ms = original_timeout_ms; + r.applied_timeout_ms = applied_timeout_ms; r.write_status = write_status; r.error_code = error_code; r.main_server_id = main_server_id; @@ -125,14 +127,15 @@ void WriteTsv(char const* path) { std::fprintf( f, "qpc_ns\tevent\tserver_id\tcloud_conn_instance_id\t" - "client_conn_instance_id\tping_request_id\ttimeout_ms\twrite_status\t" - "error_code\tmain_server_id\n"); + "client_conn_instance_id\tping_request_id\toriginal_timeout_ms\t" + "applied_timeout_ms\twrite_status\terror_code\tmain_server_id\n"); for (auto const& r : copy) { - std::fprintf(f, "%lld\t%s\t%u\t%llu\t%llu\t%u\t%u\t%d\t%d\t%d\n", + std::fprintf(f, "%lld\t%s\t%u\t%llu\t%llu\t%u\t%u\t%u\t%d\t%d\t%d\n", static_cast(r.qpc_ns), r.event, r.server_id, static_cast(r.cloud_conn_instance_id), static_cast(r.client_conn_instance_id), - r.ping_request_id, r.timeout_ms, r.write_status, r.error_code, + r.ping_request_id, r.original_timeout_ms, r.applied_timeout_ms, + r.write_status, r.error_code, static_cast(r.main_server_id)); } std::fclose(f); diff --git a/aether/diagnostics/bench_connection_event_diag.h b/aether/diagnostics/bench_connection_event_diag.h index 4d376bf85..0910ea491 100644 --- a/aether/diagnostics/bench_connection_event_diag.h +++ b/aether/diagnostics/bench_connection_event_diag.h @@ -20,8 +20,8 @@ void ResetForTest(); void Record(char const* event, std::uint32_t server_id, std::uint64_t cloud_conn_instance_id, std::uint64_t client_conn_instance_id, std::uint32_t ping_request_id, - std::uint32_t timeout_ms, int write_status, int error_code, - std::int32_t main_server_id); + std::uint32_t original_timeout_ms, std::uint32_t applied_timeout_ms, + int write_status, int error_code, std::int32_t main_server_id); std::uint64_t DroppedCount(); void WriteTsv(char const* path); diff --git a/aether/diagnostics/bench_ping_min_timeout.h b/aether/diagnostics/bench_ping_min_timeout.h new file mode 100644 index 000000000..d50811659 --- /dev/null +++ b/aether/diagnostics/bench_ping_min_timeout.h @@ -0,0 +1,32 @@ +#ifndef AETHER_DIAGNOSTICS_BENCH_PING_MIN_TIMEOUT_H_ +#define AETHER_DIAGNOSTICS_BENCH_PING_MIN_TIMEOUT_H_ + +#include "aether/clock.h" +#include "aether/config.h" + +#include + +namespace ae::bench_ping_min_timeout { + +inline std::uint32_t ChannelTimeoutMs(Duration channel_timeout) { + return static_cast( + std::chrono::duration_cast(channel_timeout) + .count()); +} + +inline std::uint32_t AppliedTimeoutMs(std::uint32_t channel_timeout_ms) { +#if AE_BENCH_PING_MIN_TIMEOUT_MS > 0 + if (channel_timeout_ms < static_cast(AE_BENCH_PING_MIN_TIMEOUT_MS)) { + return static_cast(AE_BENCH_PING_MIN_TIMEOUT_MS); + } +#endif + return channel_timeout_ms; +} + +inline Duration ApplyBenchPingMinTimeout(Duration channel_timeout) { + return std::chrono::milliseconds(AppliedTimeoutMs(ChannelTimeoutMs(channel_timeout))); +} + +} // namespace ae::bench_ping_min_timeout + +#endif diff --git a/aether/server_connections/server_connection.cpp b/aether/server_connections/server_connection.cpp index 0bc7a4075..4afc5cb19 100644 --- a/aether/server_connections/server_connection.cpp +++ b/aether/server_connections/server_connection.cpp @@ -77,7 +77,7 @@ void ServerConnection::Restream() { sid = server->server_id; } AE_BENCH_CONN_DIAG("server_restream", sid, 0, bench_client_instance_id_, 0, 0, - -1, 0, -1); + 0, -1, 0, -1); // restream means something wrong with current channel ChannelError(); } @@ -253,8 +253,8 @@ void ServerConnection::ServerError() { if (auto server = server_.Lock()) { sid = server->server_id; } - AE_BENCH_CONN_DIAG("server_error", sid, 0, bench_client_instance_id_, 0, 0, -1, - 0, -1); + AE_BENCH_CONN_DIAG("server_error", sid, 0, bench_client_instance_id_, 0, 0, 0, + -1, 0, -1); AE_TELED_ERROR("SERVER_CONNECTION_ERROR"); channel_stream_update_sub_.Reset(); channel_stream_out_data_sub_.Reset(); @@ -272,7 +272,7 @@ void ServerConnection::ChannelError() { if (auto server = server_.Lock()) { sid = server->server_id; } - AE_BENCH_CONN_DIAG("channel_error", sid, 0, bench_client_instance_id_, 0, 0, + AE_BENCH_CONN_DIAG("channel_error", sid, 0, bench_client_instance_id_, 0, 0, 0, -1, 0, -1); AE_TELED_ERROR("Channel error"); channel_stream_update_sub_.Reset(); From 64eadbc4fa64a2b178d2a40214ac91ce17e24cf9 Mon Sep 17 00:00:00 2001 From: aethernet-io Date: Thu, 29 Oct 2026 19:26:15 -0700 Subject: [PATCH 7/8] Defer periodic ServerPing Start while bench ping is pending. When AE_BENCH_PING_MIN_TIMEOUT_MS is positive, avoid emplace that cancels in-flight pings; reschedule Start after success via scheduler. Co-authored-by: Cursor --- aether/ae_actions/ping.h | 4 ++ .../cloud_connections/ping_cloud_servers.cpp | 30 ++++++++++ aether/cloud_connections/ping_cloud_servers.h | 4 ++ .../bench_ping_server_start_defer.h | 55 +++++++++++++++++++ 4 files changed, 93 insertions(+) create mode 100644 aether/diagnostics/bench_ping_server_start_defer.h diff --git a/aether/ae_actions/ping.h b/aether/ae_actions/ping.h index 8174876a3..2873c1779 100644 --- a/aether/ae_actions/ping.h +++ b/aether/ae_actions/ping.h @@ -56,6 +56,10 @@ class Ping { void Start(TimePoint current_time); + bool IsPending() const noexcept { + return state_ == RequestState::kPending; + } + private: void PingResponse(RequestId request_id); void PingResponseError(RequestId request_id, std::int32_t error_code); diff --git a/aether/cloud_connections/ping_cloud_servers.cpp b/aether/cloud_connections/ping_cloud_servers.cpp index 231964dcf..1755bad00 100644 --- a/aether/cloud_connections/ping_cloud_servers.cpp +++ b/aether/cloud_connections/ping_cloud_servers.cpp @@ -64,6 +64,9 @@ PingCloudServers::ServerPing::~ServerPing() = default; void PingCloudServers::ServerPing::Stop() { stop_ = true; +#if AE_BENCH_PING_MIN_TIMEOUT_MS > 0 + bench_ping_server_start::ClearDeferredStart(defer_state_); +#endif waiter_.reset(); start_sub_.Reset(); rx_window_sub_.Reset(); @@ -153,6 +156,16 @@ void PingCloudServers::ServerPing::Start() { if (stop_) { return; } +#if AE_BENCH_PING_MIN_TIMEOUT_MS > 0 + if (bench_ping_server_start::TryDeferPeriodicStart( + defer_state_, ping_.has_value(), + ping_ && ping_->IsPending())) { + AE_TELED_DEBUG( + "Defer periodic Start while ping pending (bench min timeout {} ms)", + AE_BENCH_PING_MIN_TIMEOUT_MS); + return; + } +#endif waiter_.emplace( ae_context_, EnsureLinked() | @@ -216,11 +229,28 @@ void PingCloudServers::ServerPing::OnPingResult(Ping::PingResult const& res) { using T = std::decay_t; if constexpr (std::is_same_v>) { c->channel_statistics().AddResponseTime(value.value); +#if AE_BENCH_PING_MIN_TIMEOUT_MS > 0 + if (bench_ping_server_start::TakeDeferredStartAfterSuccess( + defer_state_)) { + start_sub_ = ae_context_.scheduler().DelayedTask( + [&]() noexcept { Start(); }, next_ping_time_); + } +#endif } else if constexpr (std::is_same_v) { AE_TELED_DEBUG("Got late ping duration"); c->channel_statistics().AddResponseTime(value.duration); +#if AE_BENCH_PING_MIN_TIMEOUT_MS > 0 + if (bench_ping_server_start::TakeDeferredStartAfterSuccess( + defer_state_)) { + start_sub_ = ae_context_.scheduler().DelayedTask( + [&]() noexcept { Start(); }, next_ping_time_); + } +#endif } else { AE_TELED_ERROR("Ping error!"); +#if AE_BENCH_PING_MIN_TIMEOUT_MS > 0 + bench_ping_server_start::ClearDeferredStart(defer_state_); +#endif AE_BENCH_CONN_DIAG("ping_on_result_restream", cloud_sc_->server_id(), cloud_id, client_id, 0, 0, 0, -1, 0, -1); ScheduleRestream(); diff --git a/aether/cloud_connections/ping_cloud_servers.h b/aether/cloud_connections/ping_cloud_servers.h index 3f28e0f94..e44db6779 100644 --- a/aether/cloud_connections/ping_cloud_servers.h +++ b/aether/cloud_connections/ping_cloud_servers.h @@ -31,6 +31,7 @@ # include "aether/ae_actions/ping.h" # include "aether/client_connectivity_policy.h" +# include "aether/diagnostics/bench_ping_server_start_defer.h" # include "aether/cloud_connections/cloud_server_connections.h" namespace ae { @@ -81,6 +82,9 @@ class PingCloudServers { ClientConnectivityPolicy::SuspendBlocker rx_window_blocker_; ClientConnectivityPolicy::SuspendBlocker restream_blocker_; TimePoint next_ping_time_; +#if AE_BENCH_PING_MIN_TIMEOUT_MS > 0 + bench_ping_server_start::DeferState defer_state_; +#endif }; public: diff --git a/aether/diagnostics/bench_ping_server_start_defer.h b/aether/diagnostics/bench_ping_server_start_defer.h new file mode 100644 index 000000000..08caa3031 --- /dev/null +++ b/aether/diagnostics/bench_ping_server_start_defer.h @@ -0,0 +1,55 @@ +#ifndef AETHER_DIAGNOSTICS_BENCH_PING_SERVER_START_DEFER_H_ +#define AETHER_DIAGNOSTICS_BENCH_PING_SERVER_START_DEFER_H_ + +#include "aether/config.h" + +namespace ae::bench_ping_server_start { + +#if AE_BENCH_PING_MIN_TIMEOUT_MS > 0 + +struct DeferState { + bool deferred_start_after_ping{false}; +}; + +inline bool ShouldDeferPeriodicStart(bool has_ping, bool ping_is_pending) noexcept { + return has_ping && ping_is_pending; +} + +inline bool TryDeferPeriodicStart(DeferState& state, bool has_ping, + bool ping_is_pending) noexcept { + if (!ShouldDeferPeriodicStart(has_ping, ping_is_pending)) { + return false; + } + state.deferred_start_after_ping = true; + return true; +} + +inline bool TakeDeferredStartAfterSuccess(DeferState& state) noexcept { + if (!state.deferred_start_after_ping) { + return false; + } + state.deferred_start_after_ping = false; + return true; +} + +inline void ClearDeferredStart(DeferState& state) noexcept { + state.deferred_start_after_ping = false; +} + +#else + +struct DeferState {}; + +inline bool TryDeferPeriodicStart(DeferState&, bool, bool) noexcept { + return false; +} + +inline bool TakeDeferredStartAfterSuccess(DeferState&) noexcept { return false; } + +inline void ClearDeferredStart(DeferState&) noexcept {} + +#endif + +} // namespace ae::bench_ping_server_start + +#endif From 9468f5682e718a5af00dc193150a0f8885626455 Mon Sep 17 00:00:00 2001 From: aethernet-io Date: Thu, 29 Oct 2026 22:36:18 -0700 Subject: [PATCH 8/8] Bench opt-in AE_BENCH_UDP_MAX_PACKET_BYTES and Windows IP fragmentation. --- aether/channels/ethernet_channel.cpp | 8 +++ .../system_sockets/sockets/win_udp_socket.cpp | 59 +++++++++++++++++++ 2 files changed, 67 insertions(+) diff --git a/aether/channels/ethernet_channel.cpp b/aether/channels/ethernet_channel.cpp index 20eb78bf6..fb92141f5 100644 --- a/aether/channels/ethernet_channel.cpp +++ b/aether/channels/ethernet_channel.cpp @@ -177,8 +177,16 @@ EthernetChannel::EthernetChannel(ObjProp prop, ObjPtr aether, } case Protocol::kUdp: { transport_properties_.connection_type = ConnectionType::kConnectionLess; +#if defined(AE_BENCH_UDP_MAX_PACKET_BYTES) && \ + (AE_BENCH_UDP_MAX_PACKET_BYTES > 0) + transport_properties_.max_packet_size = + static_cast(AE_BENCH_UDP_MAX_PACKET_BYTES); + transport_properties_.rec_packet_size = + static_cast(AE_BENCH_UDP_MAX_PACKET_BYTES); +#else transport_properties_.max_packet_size = 1200; transport_properties_.rec_packet_size = 1200; +#endif transport_properties_.reliability = Reliability::kUnreliable; break; } diff --git a/aether/transport/system_sockets/sockets/win_udp_socket.cpp b/aether/transport/system_sockets/sockets/win_udp_socket.cpp index af14b839a..e9f027436 100644 --- a/aether/transport/system_sockets/sockets/win_udp_socket.cpp +++ b/aether/transport/system_sockets/sockets/win_udp_socket.cpp @@ -20,6 +20,7 @@ # include # include # include +# include # include "aether-miscpp/misc/defer.h" # include "aether/transport/system_sockets/sockets/get_sock_addr.h" @@ -52,6 +53,30 @@ void WriteBenchUdpSndbufDiag(int requested_bytes, int actual_bytes, int error, } // namespace # endif +# if defined(AE_BENCH_UDP_MAX_PACKET_BYTES) && \ + (AE_BENCH_UDP_MAX_PACKET_BYTES > 0) +namespace { + +void WriteBenchUdpIpFragDiag(int requested_pmtu, int read_back_pmtu, int error, + bool ip_frag_ok) { + char const* path = "win_udp_ip_frag_diag.txt"; + char* env_path = nullptr; + size_t env_len = 0; + if (_dupenv_s(&env_path, &env_len, "AE_BENCH_UDP_IP_FRAG_DIAG") == 0 && + env_path != nullptr && env_path[0] != '\0') { + path = env_path; + } + std::ofstream out(path, std::ios::trunc); + out << "requested_ip_mtu_discover=" << requested_pmtu << "\n"; + out << "read_back_ip_mtu_discover=" << read_back_pmtu << "\n"; + out << "setsockopt_wsa_error=" << error << "\n"; + out << "ip_frag_ok=" << (ip_frag_ok ? "true" : "false") << "\n"; + free(env_path); +} + +} // namespace +# endif + namespace ae { // Outbound UDP payloads should stay near MTU to avoid IP fragmentation on send. @@ -113,6 +138,40 @@ WinUdpSocket::WinUdpSocket(Ptr const& poller) } # endif +# if defined(AE_BENCH_UDP_MAX_PACKET_BYTES) && \ + (AE_BENCH_UDP_MAX_PACKET_BYTES > 0) + { + DWORD const requested = IP_PMTUDISC_DONT; + DWORD read_back = 0; + int wsa_error = 0; + bool ok = false; + if (::setsockopt(sock, IPPROTO_IP, IP_MTU_DISCOVER, + reinterpret_cast(&requested), + sizeof(requested)) == SOCKET_ERROR) { + wsa_error = static_cast(WSAGetLastError()); + WriteBenchUdpIpFragDiag(static_cast(requested), -1, wsa_error, false); + } else { + int len = sizeof(read_back); + if (::getsockopt(sock, IPPROTO_IP, IP_MTU_DISCOVER, + reinterpret_cast(&read_back), &len) == + SOCKET_ERROR) { + wsa_error = static_cast(WSAGetLastError()); + WriteBenchUdpIpFragDiag(static_cast(requested), -1, wsa_error, + false); + } else { + ok = read_back == IP_PMTUDISC_DONT; + WriteBenchUdpIpFragDiag(static_cast(requested), + static_cast(read_back), 0, ok); + } + } + if (!ok) { + AE_TELED_ERROR("IP_MTU_DISCOVER IP_PMTUDISC_DONT failed wsa={}", + wsa_error); + return; + } + } +# endif + created = true; socket_ = sock; }