Preserve UDP benchmark diagnostics and opt-in ping timeout controls

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"

aether/ae_actions/ping.cpp

@@ -23,7 +23,10 @@
 #  include <utility>
 
 #  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,18 +58,59 @@ static_assert(PromiseErrorCodeToPingErrorCode(
                   static_cast<std::uint32_t>(std::numeric_limits<int>::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 original_timeout_ms,
+                    std::uint32_t applied_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,
+                     original_timeout_ms, applied_timeout_ms, write_status,
+                     error_code, -1);
+}
+#  endif
+
 }  // namespace
 
 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::uint32_t>(
+      std::chrono::duration_cast<std::chrono::milliseconds>(response_timeout)
+          .count());
+  diag_original_timeout_ms_ = static_cast<std::uint32_t>(
+      std::chrono::duration_cast<std::chrono::milliseconds>(
+          channel_response_timeout)
+          .count());
+#endif
   AE_TELE_INFO(
       kPing,
       "Ping action created to server id: {}, interval: {:%S}s, rx_window: "
@@ -87,6 +131,12 @@ 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;
+  RecordPingDiag(cloud_server_connection_, "ping_start", 0,
+                 diag_original_timeout_ms_, diag_applied_timeout_ms_, -1, 0);
+#  endif
 
   auto& write_action = cc->AuthorizedApiCall(
       [this, current_time](ApiContext<AuthorizedApi>& auth_api) {
@@ -106,6 +156,12 @@ 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_original_timeout_ms_, diag_applied_timeout_ms_, -1,
+                       0);
+#  endif
 
         auto wait_result_sub =
             pong_promise.Subscribe([this, req_id](auto&& res) {
@@ -132,6 +188,20 @@ 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_original_timeout_ms_, diag_applied_timeout_ms_,
+                     WriteStatusToInt(status), 0);
+    }
+#  endif
     if (status == WriteAction::Status::kFail) {
       if (state_ != RequestState::kPending) {
         return;
@@ -170,10 +240,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_original_timeout_ms_, diag_applied_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_original_timeout_ms_, diag_applied_timeout_ms_, -1, 0);
+#  endif
     wait_result_sub_.Reset();
     state_ = RequestState::kFinished;
     result_event_.Emit(PingResult{LateDuration{ping_duration}});
@@ -185,6 +263,11 @@ 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_original_timeout_ms_, diag_applied_timeout_ms_, -1,
+                   static_cast<int>(error_code));
+#  endif
     state_ = RequestState::kFinished;
     ResetRequestSubscriptions();
     if (error_code == -1) {
@@ -204,13 +287,31 @@ 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_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_original_timeout_ms_,
+                   diag_applied_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_original_timeout_ms_,
+                   diag_applied_timeout_ms_, -1, 0);
+  }
+#  endif
   wait_result_sub_.Reset();
   timeout_sub_.Reset();
   write_sub_.Reset();

aether/ae_actions/ping.h

@@ -47,14 +47,19 @@ 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);
 
   ResultEvent const& result_event() const;
 
   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);
@@ -82,6 +87,12 @@ 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_original_timeout_ms_{0};
+  std::uint32_t diag_applied_timeout_ms_{0};
+  bool diag_write_terminal_{false};
+#endif
 };
 }  // namespace ae
 #endif  // AE_ENABLE_PING

aether/channels/ethernet_channel.cpp

@@ -177,8 +177,16 @@ EthernetChannel::EthernetChannel(ObjProp prop, ObjPtr<Aether> 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<std::uint32_t>(AE_BENCH_UDP_MAX_PACKET_BYTES);
+      transport_properties_.rec_packet_size =
+          static_cast<std::uint32_t>(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;
     }

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<AeMessage, kBufferCapacity>;

aether/cloud_connections/cloud_server_connection.cpp

@@ -20,6 +20,7 @@
 #include <cstdint>
 #include <limits>
 
+#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_) {

aether/cloud_connections/cloud_server_connection.h

@@ -18,6 +18,7 @@
 #define AETHER_CLOUD_CONNECTIONS_CLOUD_SERVER_CONNECTION_H_
 
 #include <cstddef>
+#include <cstdint>
 #include <memory>
 
 #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<ClientServerConnection> client_connection_;
   bool is_quarantined_ = false;
+  std::uint64_t bench_cloud_instance_id_{0};
 };
 }  // namespace ae
 

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<CloudServerConnection*> const& selected) {
+  if (selected.empty()) {
+    return -1;
+  }
+  return static_cast<std::int32_t>(selected.front()->server_id());
+}
+
+void RecordCloudDiag(char const* event, CloudServerConnection& sc,
+                     std::vector<CloudServerConnection*> 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, 0, -1, 0, MainSelectedServerId(selected));
+}
+
+}  // namespace
 #if DEBUG
 auto ServerList(std::vector<CloudServerConnection*> const& servers) {
   std::vector<ServerId> 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,13 +336,18 @@ 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;
   }
   if (emplaced) {
     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();
   }
 }