124 collect all ping values and count statistics

.github/workflows/ci-cd-multi-platforms.yml

@@ -82,13 +82,13 @@ jobs:
         platforms:
           - {
               name: "Hydrogen",
-              user_config: "../config/user_config_hydrogen.h",
-              fs_init: "../../../../config/file_system_init.h",
+              user_config: "./config/user_config_hydrogen.h",
+              fs_init: "./config/file_system_init.h",
             }
           - {
               name: "Sodium",
-              user_config: "../config/user_config_sodium.h",
-              fs_init: "../../../../config/file_system_init.h",
+              user_config: "./config/user_config_sodium.h",
+              fs_init: "./config/file_system_init.h",
             }
 
     steps:
@@ -138,6 +138,8 @@ jobs:
           git apply "../libhydrogen.patch"
           cd ../libsodium
           git apply "../libsodium.patch"
+          cd ../etl
+          git apply "../etl.patch"
           cd ../
           cp "CMakeLists.libbcrypt" "libbcrypt/CMakeLists.txt"
           cp "CMakeLists.libhydrogen" "libhydrogen/CMakeLists.txt"
@@ -192,8 +194,8 @@ jobs:
         platforms:
           - {
               name: "Optimized",
-              user_config: "../config/user_config_optimized.h",
-              fs_init: "../../../../config/file_system_init.h",
+              user_config: "./config/user_config_optimized.h",
+              fs_init: "./config/file_system_init.h",
             }
 
     steps:
@@ -212,6 +214,8 @@ jobs:
           git apply "../libhydrogen.patch"
           cd ../libsodium
           git apply "../libsodium.patch"
+          cd ../etl
+          git apply "../etl.patch"
           cd ../
           cp "CMakeLists.libbcrypt" "libbcrypt/CMakeLists.txt"
           cp "CMakeLists.libhydrogen" "libhydrogen/CMakeLists.txt"
@@ -227,8 +231,8 @@ jobs:
           path: "projects/xtensa_lx6/vscode/aether-client-cpp"
           command: idf.py build
             -DCOMPILE_EXAMPLE=${{ matrix.compile_example }}
-            -DUSER_CONFIG=${{ matrix.platforms.user_config }}
-            -DFS_INIT="${{ matrix.platforms.fs_init }}"
+            -DUSER_CONFIG=../../../../../${{ matrix.platforms.user_config }}
+            -DFS_INIT="../../../../../${{ matrix.platforms.fs_init }}"
             -DAE_DISTILLATION=Off
 
       - name: Rename artifact

.gitmodules

@@ -50,4 +50,10 @@
 	path = third_party/ini.h
 	url = https://github.com/giosali/ini.h
 	branch = main
-  ignore = dirty
+  ignore = dirty
+
+[submodule "third_party/etl"]
+	path = third_party/etl
+	url = https://github.com/ETLCPP/etl.git
+	branch = master
+  ignore = dirty

aether/CMakeLists.txt

@@ -19,31 +19,22 @@ set(AE_PROJECT_VERSION "1.0.0")
 option(AE_NO_STRIP_ALL "Do not apply --strip_all, useful for bloaty and similar tools " Off)
 
 option(AE_DISTILLATION "Build aether in distillation mode" OFF)
-option(USER_CONFIG "Path to user provided configuration header file" "")
 
-find_program(GIT_COMMAND git)
-if ( GIT_COMMAND )
-  # get current git version
-  execute_process(
-        COMMAND ${GIT_COMMAND} -C ${CMAKE_CURRENT_SOURCE_DIR} rev-parse --verify HEAD
-        OUTPUT_VARIABLE GIT_VERSION
-    )
-  if ( NOT GIT_VERSION )
-    message(WARNING "Not a git repo")
-  else()
-    string(STRIP ${GIT_VERSION} GIT_VERSION)
-    message(STATUS "get aether git version ${GIT_VERSION}")
-  endif()
-endif()
+set(USER_CONFIG "" CACHE PATH "Path to user provided configuration header file")
+set(FS_INIT "" CACHE PATH "Path to user provided saved state header file")
 
 # list a common dependencies
 list(APPEND common_dependencies
     "../third_party/libbcrypt"
     "../third_party/libhydrogen"
     "../third_party/libsodium"
     "../third_party/gcem"
+    "../third_party/etl"
 )
 
+# for etl
+set(GIT_DIR_LOOKUP_POLICY ALLOW_LOOKING_ABOVE_CMAKE_SOURCE_DIR)
+
 list(APPEND aether_srcs
             "aether_app.cpp"
             "aether.cpp"
@@ -55,7 +46,6 @@ list(APPEND aether_srcs
             "registration_cloud.cpp"
             "work_cloud.cpp"
             "server.cpp"
-            "statistics.cpp"
             "uid.cpp"
             "proof_of_work.cpp"
             "server_keys.cpp"
@@ -206,20 +196,21 @@ list(APPEND port_srcs
             "port/file_systems/drivers/driver_header.cpp"
             "port/file_systems/drivers/driver_spifs.cpp")
 
+list(APPEND statistics_srcs
+      "statistics/channel_statistics.cpp")
+
 list(APPEND tele_srcs
             "tele/traps/io_stream_traps.cpp"
             "tele/traps/statistics_trap.cpp"
             "tele/traps/tele_statistics.cpp")
 
 message(STATUS "Cmake system name is ${CMAKE_SYSTEM_NAME}")
 
-if(NOT CM_PLATFORM AND (
-  CMAKE_SYSTEM_NAME STREQUAL "Darwin" OR
-  CMAKE_SYSTEM_NAME STREQUAL "Linux" OR
-  CMAKE_SYSTEM_NAME STREQUAL "Windows" OR
-  CMAKE_SYSTEM_NAME MATCHES ".*BSD"))
-  message(STATUS "Build for regular cmake project")
+if(NOT CM_PLATFORM)
+  message(STATUS "Aether build for regular cmake project")
   set(REGULAR_CMAKE_PROJECT On)
+else()
+  message(STATUS "Aether build for CM_PLATFORM=${CM_PLATFORM}")
 endif()
 
 if(REGULAR_CMAKE_PROJECT)
@@ -252,6 +243,7 @@ if(REGULAR_CMAKE_PROJECT)
                             ${client_messages_srcs}
                             ${client_connections_srcs}
                             ${port_srcs}
+                            ${statistics_srcs}
                             ${tele_srcs}
                             )
 
@@ -264,7 +256,8 @@ if(REGULAR_CMAKE_PROJECT)
                                         bcrypt
                                         sodium
                                         hydrogen
-                                        gcem)
+                                        gcem
+                                        etl)
   target_link_libraries(${PROJECT_NAME} PRIVATE c-ares )
 
   set(TARGET_NAME "${PROJECT_NAME}")
@@ -292,6 +285,7 @@ else()
                             ${client_messages_srcs}
                             ${client_connections_srcs}
                             ${port_srcs}
+                            ${statistics_srcs}
                             ${tele_srcs}
                             INCLUDE_DIRS ${include_dirs}
                             REQUIRES esp_wifi nvs_flash spiffs)
@@ -307,7 +301,8 @@ else()
                                          bcrypt
                                          sodium
                                          hydrogen
-                                         gcem)
+                                         gcem
+                                         etl)
   else()
     #ERROR
     message(SEND_ERROR "You must specify the CMAKE version!")
@@ -325,24 +320,40 @@ if (CMAKE_SYSTEM_NAME STREQUAL "Windows")
   target_link_libraries(${TARGET_NAME} PRIVATE ws2_32)
 endif()
 
+find_program(GIT_COMMAND git)
+if ( GIT_COMMAND )
+  # get current git version
+  execute_process(
+        COMMAND ${GIT_COMMAND} -C ${CMAKE_CURRENT_SOURCE_DIR} rev-parse --verify HEAD
+        OUTPUT_VARIABLE GIT_VERSION
+    )
+  if ( NOT GIT_VERSION )
+    message(WARNING "Not a git repo")
+  else()
+    string(STRIP ${GIT_VERSION} GIT_VERSION)
+    message(STATUS "get aether git version ${GIT_VERSION}")
+  endif()
+endif()
+
 if (GIT_VERSION)
   target_compile_definitions(${TARGET_NAME} PUBLIC "AE_GIT_VERSION=\"${GIT_VERSION}\"")
 endif()
 target_compile_definitions(${TARGET_NAME} PUBLIC "AE_PROJECT_VERSION=\"${AE_PROJECT_VERSION}\"")
 
-if (USER_CONFIG)
+if (NOT "${USER_CONFIG}" STREQUAL "")
   target_compile_definitions(${TARGET_NAME} PUBLIC "USER_CONFIG=\"${USER_CONFIG}\"")
 endif()
 
-if (FS_INIT)
+if (NOT "${FS_INIT}" STREQUAL "")
   target_compile_definitions(${TARGET_NAME} PUBLIC "FS_INIT=\"${FS_INIT}\"")
 endif()
 
 if (AE_DISTILLATION)
   target_compile_definitions(${TARGET_NAME} PUBLIC "AE_DISTILLATION=1")
 endif()
 
-if(_AE_REG_CLOUD_IP)
+# for debug purposes only, set registration server ip address
+if(NOT "${_AE_REG_CLOUD_IP}" STREQUAL "")
   target_compile_definitions(${TARGET_NAME} PUBLIC "_AE_REG_CLOUD_IP=\"${_AE_REG_CLOUD_IP}\"")
 endif()
 

aether/ae_actions/ae_actions_tele.h

@@ -56,5 +56,7 @@ AE_TAG(kGetClientCloudConnectionServerListIsOver, kAeActions)
 AE_TAG(kPing, kAeActions)
 AE_TAG(kPingSend, kAeActions)
 AE_TAG(kPingWriteError, kAeActions)
+AE_TAG(kPingTimeout, kAeActions)
+AE_TAG(kPingTimeoutError, kAeActions)
 
 #endif  // AETHER_AE_ACTIONS_AE_ACTIONS_TELE_H_

aether/ae_actions/ping.cpp

@@ -17,6 +17,7 @@
 #include "aether/ae_actions/ping.h"
 
 #include <utility>
+#include <optional>
 
 #include "aether/api_protocol/packet_builder.h"
 #include "aether/methods/work_server_api/authorized_api.h"
@@ -33,6 +34,7 @@ Ping::Ping(ActionContext action_context, Server::ptr const& server,
       server_stream_{&server_stream},
       ping_interval_{ping_interval},
       read_client_safe_api_gate_{protocol_context_, client_safe_api_},
+      repeat_count_{},
       state_{State::kWaitLink},
       state_changed_sub_{state_.changed_event().Subscribe(
           [this](auto) { Action::Trigger(); })},
@@ -56,8 +58,7 @@ TimePoint Ping::Update(TimePoint current_time) {
       case State::kSendPing:
         SendPing(current_time);
         break;
-      case State::kWaitResponse:  // TODO: response timeout and response
-                                  // statistics
+      case State::kWaitResponse:
       case State::kWaitInterval:
         break;
       case State::kError:
@@ -66,21 +67,20 @@ TimePoint Ping::Update(TimePoint current_time) {
     }
   }
 
+  if (state_.get() == State::kWaitResponse) {
+    return WaitResponse(current_time);
+  }
   if (state_.get() == State::kWaitInterval) {
-    if ((last_ping_time_ + ping_interval_) <= current_time) {
-      state_ = State::kSendPing;
-    } else {
-      return last_ping_time_ + ping_interval_;
-    }
+    return WaitInterval(current_time);
   }
 
   return current_time;
 }
 
 void Ping::SendPing(TimePoint current_time) {
   AE_TELE_DEBUG(kPingSend, "Send ping");
-  last_ping_time_ = current_time;
   auto request_id = RequestId::GenRequestId();
+  ping_times_.push(std::make_pair(request_id, current_time));
   auto packet = PacketBuilder{
       protocol_context_,
       PackMessage{AuthorizedApi{},
@@ -100,18 +100,69 @@ void Ping::SendPing(TimePoint current_time) {
   });
 
   // Wait for response
-  SendResult::OnResponse(protocol_context_, request_id,
-                         [&](ApiParser&) { PingResponse(); });
+  SendResult::OnResponse(
+      protocol_context_, request_id,
+      [&, req_id{request_id}](ApiParser&) { PingResponse(req_id); });
 
   state_ = State::kWaitResponse;
 }
 
-void Ping::PingResponse() {
+TimePoint Ping::WaitInterval(TimePoint current_time) {
+  auto const& ping_time = ping_times_.back().second;
+  if ((ping_time + ping_interval_) > current_time) {
+    return ping_time + ping_interval_;
+  }
+  state_ = State::kSendPing;
+  return current_time;
+}
+
+TimePoint Ping::WaitResponse(TimePoint current_time) {
+  auto channel_ptr = channel_.Lock();
+  assert(channel_ptr);
+  auto timeout = channel_ptr->expected_ping_time();
+
+  auto const& ping_time = ping_times_.back().second;
+  if ((ping_time + timeout) > current_time) {
+    return ping_time + timeout;
+  }
+  // timeout
+  AE_TELE_INFO(kPingTimeout, "Timeout is {:%S}", timeout);
+  if (repeat_count_ >= kMaxRepeatPingCount) {
+    AE_TELE_ERROR(kPingTimeoutError, "Ping repeat count exceeded");
+    state_ = State::kError;
+  } else {
+    repeat_count_++;
+    state_ = State::kSendPing;
+  }
+
+  return current_time;
+}
+
+void Ping::PingResponse(RequestId request_id) {
+  auto request_time = [&]() -> std::optional<TimePoint> {
+    for (auto const& [req_id, time] : ping_times_) {
+      if (req_id == request_id) {
+        return time;
+      }
+    }
+    return std::nullopt;
+  }();
+
+  if (!request_time) {
+    AE_TELED_DEBUG("Got lost, or not our ping request");
+    return;
+  }
+
+  repeat_count_ = 0;
   auto current_time = Now();
   auto ping_duration =
-      std::chrono::duration_cast<Duration>(current_time - last_ping_time_);
+      std::chrono::duration_cast<Duration>(current_time - *request_time);
 
   AE_TELED_DEBUG("Ping received by {:%S} s", ping_duration);
+  auto channel_ptr = channel_.Lock();
+  assert(channel_ptr);
+  channel_ptr->AddPingTime(ping_duration);
+
   state_ = State::kWaitInterval;
 }