From 9202ffed3337720f5afa65dff3ff8b94cc67d5f6 Mon Sep 17 00:00:00 2001 From: BartolomeyKant Date: Thu, 16 Jul 2026 15:32:01 +0500 Subject: [PATCH 1/3] update info about dependancies --- .gitmodules | 0 AGENTS.md | 16 ++++++++++++++++ README.md | 2 +- 3 files changed, 17 insertions(+), 1 deletion(-) delete mode 100644 .gitmodules diff --git a/.gitmodules b/.gitmodules deleted file mode 100644 index e69de29b..00000000 diff --git a/AGENTS.md b/AGENTS.md index 6ee89322..bf045824 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -93,6 +93,22 @@ To run tests, go into `` and run `ctest . --progress -j -E "((sodium)|(hydro)|(bcrypt)).*" --output-on-failure`. Or run specific test by name from `/tests/run/`. +## Dependencies + +- Dependencies are managed by CMake through `cmake/CPM.cmake` and `CPMAddPackage` calls in the root `CMakeLists.txt`. +- Do not add conan, vcpkg, git submodules, or vendored dependency copies unless explicitly requested. +- The root dependency list is the source of truth. Test-only dependencies live in `tests/CMakeLists.txt`; tool-only dependencies may live in the tool's own `CMakeLists.txt`. +- CPM downloads dependencies during CMake configure. If `CPM_SOURCE_CACHE` is not set, the project uses `${CMAKE_CURRENT_BINARY_DIR}/cpm.cache`. +- Prefer setting the `CPM_SOURCE_CACHE` environment variable for repeated local builds to avoid re-downloading dependencies. +- To use locally checked-out dependencies, pass `-DCPM__SOURCE=/absolute/path`. +- `CPM_USE_LOCAL_PACKAGES` may be used to let CPM try `find_package` before downloading from source. +- Several dependencies require local patches from `third_party/*.patch`. Keep patch files in sync when changing dependency versions or repositories. +- Root dependencies should use `EXCLUDE_FROM_ALL FALSE` so they can be installed together with `aether`. +- Install options are propagated through dependency-specific CMake options such as `ENABLE_INSTALL`, `AE_INSTALL`, `AE_NUMERIC_INSTALL`, and `STDEXEC_INSTALL`. +- Do not casually update dependency tags pinned to branches such as `master` or `main`; check patches, build behavior, and compatibility first. In particular, `libhydrogen` is pinned to `bbca575` because newer versions are noted as incompatible with the Aether server. +- Desktop builds additionally fetch and link `c-ares`. +- ESP-IDF builds do not fetch ESP components through CPM; required IDF targets must already exist: `idf::esp_wifi`, `idf::esp_netif`, `idf::nvs_flash`, `idf::spiffs`, and `idf::esp_driver_uart`. + ### Smoke test The first smoke test is a `/aether-client-cpp-cloud`. diff --git a/README.md b/README.md index 4bce94a0..0966aa4c 100644 --- a/README.md +++ b/README.md @@ -1,5 +1,5 @@ # Aethernet C++ client -Run git_init script suitable for particular platform. It clones submodules and applies patches. +Dependencies are managed by CMake through CPM.cmake. Configure the project with CMake to download dependencies and apply local patches from `third_party`. **Projects** folder contains cmake files to build axamples for target platforms. VSCode + Platformio + ESP-IDF is a toolset that is currently supported. From 9acb5013a66850b034e4a1f95bf5a93c04d82940 Mon Sep 17 00:00:00 2001 From: BartolomeyKant Date: Thu, 16 Jul 2026 15:34:19 +0500 Subject: [PATCH 2/3] make store pending responses in protocol context instead of api_method --- aether/ae_actions/ae_actions_tele.h | 2 +- aether/ae_actions/ping.cpp | 82 ++++-- aether/ae_actions/ping.h | 5 +- aether/api_protocol/api_method.h | 59 +---- aether/api_protocol/api_promise.h | 77 +++--- aether/api_protocol/protocol_context.cpp | 151 ++++++++--- aether/api_protocol/protocol_context.h | 100 ++++++-- .../cloud_connections/ping_cloud_servers.cpp | 24 +- aether/config.h | 22 +- aether/registration/registration.cpp | 40 +-- aether/tele/env/compilation_options.h | 12 +- .../send_messages_bandwidth/sender/sender.cpp | 2 +- examples/cloud/cloud_test.cpp | 12 +- tests/test-api-protocol/test-method-call.cpp | 234 +++++++++++++++++- 14 files changed, 603 insertions(+), 219 deletions(-) diff --git a/aether/ae_actions/ae_actions_tele.h b/aether/ae_actions/ae_actions_tele.h index 96c82690..9dbe7e02 100644 --- a/aether/ae_actions/ae_actions_tele.h +++ b/aether/ae_actions/ae_actions_tele.h @@ -56,7 +56,7 @@ AE_TAG(kGetClientCloudConnectionServerListIsOver, kAeActions) AE_TAG(kPing, kAeActions) AE_TAG(kPingSend, kAeActions) AE_TAG(kPingWriteError, kAeActions) -AE_TAG(kPingTimeout, kAeActions) +AE_TAG(kPingResponseError, kAeActions) AE_TAG(kPingTimeoutError, kAeActions) AE_TAG(TelemetryCreated, kAeActions) diff --git a/aether/ae_actions/ping.cpp b/aether/ae_actions/ping.cpp index b53f1029..1a380d45 100644 --- a/aether/ae_actions/ping.cpp +++ b/aether/ae_actions/ping.cpp @@ -18,6 +18,9 @@ #if AE_ENABLE_PING # include +# include +# include +# include # include "aether/server.h" @@ -27,6 +30,35 @@ # include "aether/ae_actions/ae_actions_tele.h" namespace ae { +namespace { + +constexpr int kPingPromiseEvictedError{-1}; + +constexpr int PromiseErrorCodeToPingErrorCode( + std::uint32_t error_code) noexcept { + if (error_code == std::numeric_limits::max()) { + return kPingPromiseEvictedError; + } + if (error_code <= + static_cast(std::numeric_limits::max())) { + return static_cast(error_code); + } + return kPingPromiseEvictedError; +} + +static_assert(PromiseErrorCodeToPingErrorCode( + std::numeric_limits::max()) == + kPingPromiseEvictedError); +static_assert(PromiseErrorCodeToPingErrorCode(0) == 0); +static_assert(PromiseErrorCodeToPingErrorCode(static_cast( + std::numeric_limits::max())) == + std::numeric_limits::max()); +static_assert(PromiseErrorCodeToPingErrorCode( + static_cast(std::numeric_limits::max()) + + 1U) == kPingPromiseEvictedError); + +} // namespace + Ping::Ping(AeContext const& ae_context, CloudServerConnection& cloud_server_connection, Duration next_ping_hint, Duration rx_window, Duration timeout) @@ -56,8 +88,6 @@ void Ping::Start(TimePoint current_time) { } started_ = true; - AE_TELE_DEBUG(kPingSend, "Send ping"); - auto& write_action = cc->AuthorizedApiCall( SubApi{[this, current_time](ApiContext& auth_api) { auto next_ping_hint_ms = static_cast( @@ -68,23 +98,28 @@ void Ping::Start(TimePoint current_time) { std::chrono::duration_cast(rx_window_) .count()); - auto& pong_promise = auth_api->ping(next_ping_hint_ms, rx_window_ms); + auto pong_promise = auth_api->ping(next_ping_hint_ms, rx_window_ms); auto req_id = pong_promise.request_id(); - AE_TELED_DEBUG("Ping server id {}, request {} expected time {:%S}s", - server_id_, req_id, timeout_); + AE_TELE_DEBUG(kPingSend, + "Ping server id {}, request {} expected time {}", + server_id_, req_id, timeout_); request_start_ = current_time; - request_id_ = req_id; - auto wait_result_sub = pong_promise.Subscribe( - [this, req_id](auto&&...) { PingResponse(req_id); }); + auto wait_result_sub = + pong_promise.Subscribe([this, req_id](auto&& res) { + if (res) { + PingResponse(req_id); + } else { + PingResponseError(req_id, res.error()); + } + }); wait_result_sub_ = std::move(wait_result_sub); - auto timeout_sub = ae_context_.scheduler().DelayedTask( + timeout_sub_ = ae_context_.scheduler().DelayedTask( [this, req_id]() { PingResponseTimeout(req_id); }, current_time + timeout_); - timeout_sub_ = std::move(timeout_sub); }}); write_sub_ = write_action.status_event().Subscribe([this](auto status) { @@ -112,28 +147,29 @@ void Ping::Start(TimePoint current_time) { } void Ping::PingResponse(RequestId request_id) { - if (!HasActiveRequest() || request_id_ != request_id) { - AE_TELED_WARNING("Got lost, or not our pong response"); - return; - } - auto current_time = Now(); auto ping_duration = std::chrono::duration_cast(current_time - request_start_); - AE_TELED_DEBUG("Ping server id {} request {} received by {:%S} s", server_id_, + AE_TELED_DEBUG("Ping server id {} request {} received by {}", server_id_, request_id, ping_duration); ResetRequestSubscriptions(); result_event_.Emit(Ok{ping_duration}); } -void Ping::PingResponseTimeout(RequestId request_id) { - if (!HasActiveRequest() || request_id_ != request_id) { - AE_TELED_WARNING("Timeout for lost, or not our pong response"); - return; +void Ping::PingResponseError(RequestId request_id, std::int32_t error_code) { + AE_TELE_ERROR(kPingResponseError, "Ping server id {} request {} error {}", + server_id_, request_id, error_code); + ResetRequestSubscriptions(); + if (error_code == -1) { + result_event_.Emit(Error{4}); + } else { + result_event_.Emit(Error{3}); } +} - AE_TELE_ERROR(kPingTimeout, "Ping server id {} request {} timeout", +void Ping::PingResponseTimeout(RequestId request_id) { + AE_TELE_ERROR(kPingTimeoutError, "Ping server id {} request {} timeout", server_id_, request_id); ResetRequestSubscriptions(); result_event_.Emit(Error{2}); @@ -145,9 +181,5 @@ void Ping::ResetRequestSubscriptions() { write_sub_.Reset(); } -bool Ping::HasActiveRequest() const noexcept { - return static_cast(wait_result_sub_); -} - } // namespace ae #endif // AE_ENABLE_PING diff --git a/aether/ae_actions/ping.h b/aether/ae_actions/ping.h index d94eaaf5..1be813ed 100644 --- a/aether/ae_actions/ping.h +++ b/aether/ae_actions/ping.h @@ -50,9 +50,9 @@ class Ping { private: void PingResponse(RequestId request_id); + void PingResponseError(RequestId request_id, std::int32_t error_code); void PingResponseTimeout(RequestId request_id); void ResetRequestSubscriptions(); - bool HasActiveRequest() const noexcept; AeContext ae_context_; CloudServerConnection* cloud_server_connection_; @@ -61,8 +61,7 @@ class Ping { Duration timeout_; ServerId server_id_; - TimePoint request_start_{}; - RequestId request_id_{}; + TimePoint request_start_; Subscription wait_result_sub_; TaskSubscription timeout_sub_; Subscription write_sub_; diff --git a/aether/api_protocol/api_method.h b/aether/api_protocol/api_method.h index a252c25f..418502b5 100644 --- a/aether/api_protocol/api_method.h +++ b/aether/api_protocol/api_method.h @@ -17,13 +17,6 @@ #ifndef AETHER_API_PROTOCOL_API_METHOD_H_ #define AETHER_API_PROTOCOL_API_METHOD_H_ -#include "aether/warning_disable.h" - -DISABLE_WARNING_PUSH() -IGNORE_IMPLICIT_CONVERSION() -#include -DISABLE_WARNING_POP() - #include "aether/api_protocol/api_message.h" #include "aether/api_protocol/api_context.h" #include "aether/api_protocol/api_promise.h" @@ -60,7 +53,8 @@ struct Method { void operator()(Args... args) { auto* packet_stack = protocol_context_->packet_stack(); - assert(packet_stack); + assert(packet_stack != nullptr && + "Method::operator() requires active ApiContext"); packet_stack->Push(*this, arg_proc_(std::forward(args)...)); } @@ -77,27 +71,21 @@ struct Method { /** * \brief Specialization for method with return value. * A GenericMessage generated with request id and list of args. - * return PromiseView for waiting the result or error. + * return ApiPromise for waiting the result or error. */ template struct Method(Args...), ArgProc> { explicit Method(ProtocolContext& protocol_context, ArgProc arg_proc = {}) : protocol_context_{&protocol_context}, arg_proc_{std::move(arg_proc)} {} - ~Method() noexcept { - for (auto i = api_promise_pool_.begin(); i != api_promise_pool_.end(); - ++i) { - api_promise_pool_.destroy(&i.template get>()); - } - } - - ApiPromise& operator()(Args... args) { + ApiPromise operator()(Args... args) { auto request_id = RequestId::GenRequestId(); auto* packet_stack = protocol_context_->packet_stack(); - assert(packet_stack); + assert(packet_stack != nullptr && + "Method::operator() requires active ApiContext"); packet_stack->Push(*this, arg_proc_(request_id, std::forward(args)...)); - return UpdateRequestCb(request_id); + return ApiPromise{*protocol_context_, request_id}; } template @@ -106,37 +94,7 @@ struct Method(Args...), ArgProc> { } private: - ApiPromise& UpdateRequestCb(RequestId request_id) { - auto* api_promise = api_promise_pool_.create(request_id); - assert(api_promise != nullptr); - - if constexpr (!std::is_same_v) { - protocol_context_->AddSendResultCallback( - request_id, [this, api_promise]() { - api_promise->SetValue( - protocol_context_->parser()->template Extract()); - - api_promise_pool_.destroy(api_promise); - }); - } else { - protocol_context_->AddSendResultCallback( - request_id, [this, api_promise]() { - api_promise->SetValue(); - api_promise_pool_.destroy(api_promise); - }); - } - protocol_context_->AddSendErrorCallback( - request_id, [this, api_promise](auto, std::uint32_t err) { - api_promise->SetError(std::move(err)); - api_promise_pool_.destroy(api_promise); - }); - - return *api_promise; - } - ProtocolContext* protocol_context_; - // TODO: configure pool size - etl::pool, 10> api_promise_pool_; ArgProc arg_proc_; }; @@ -159,7 +117,8 @@ struct Method(Args...), ArgProc> { SubContext context{*api_, *child_stack}; auto* packet_stack = protocol_context_->packet_stack(); - assert(packet_stack); + assert(packet_stack != nullptr && + "Method::operator() requires active ApiContext"); packet_stack->Push( *this, arg_proc_(std::forward(args)..., std::move(child_stack))); return context; diff --git a/aether/api_protocol/api_promise.h b/aether/api_protocol/api_promise.h index aa037b81..c14c23df 100644 --- a/aether/api_protocol/api_promise.h +++ b/aether/api_protocol/api_promise.h @@ -18,74 +18,67 @@ #define AETHER_API_PROTOCOL_API_PROMISE_H_ #include +#include +#include +#include "aether-miscpp/meta/ignore_t.h" #include "aether-miscpp/types/result.h" -#include "aether/events/events.h" +#include "aether/api_protocol/api_pack_parser.h" +#include "aether/api_protocol/protocol_context.h" #include "aether/api_protocol/request_id.h" +#include "aether/events/events.h" namespace ae { -namespace api_promise_action_internal { -struct Empty {}; -}; // namespace api_promise_action_internal - -template -class ApiPromise { +namespace api_promise_internal { +template +class PendingResponseEntry final : public ProtocolContext::PendingResponse { public: - using value_type = Value; - using error_type = Error; - using result_type = Result; - - constexpr explicit ApiPromise(RequestId req_id) noexcept - : request_id_{req_id} {} - - constexpr void SetValue(value_type&& val) noexcept { - event_.Emit(result_type{Ok{std::move(val)}}); + using value_type = std::conditional_t, Ignore, Value>; + using result_type = Result; + + void OnResult(ApiParser& parser) override { + if constexpr (std::is_void_v) { + event.Emit(result_type{Ok{kIgnore}}); + } else { + event.Emit(result_type{Ok{parser.template Extract()}}); + } } - constexpr void SetError(error_type&& err) noexcept { - event_.Emit(result_type{Error{std::move(err)}}); + void OnError(std::uint8_t, std::int32_t error_code) override { + event.Emit(result_type{Error{static_cast(error_code)}}); } - constexpr RequestId request_id() const { return request_id_; } - - template - constexpr decltype(auto) Subscribe(Fn&& fn) noexcept { - return EventSubscriber{event_}.Subscribe(std::forward(fn)); + void OnEvicted() override { + event.Emit(result_type{Error{static_cast(-1)}}); } - private: - RequestId request_id_; - Event event_; + Event event; }; +} // namespace api_promise_internal -template -class ApiPromise { +template +class ApiPromise { public: - using value_type = api_promise_action_internal::Empty; using error_type = Error; + using value_type = std::conditional_t, Ignore, Value>; using result_type = Result; - constexpr explicit ApiPromise(RequestId req_id) noexcept - : request_id_{req_id} {} - - constexpr void SetValue() noexcept { - event_.Emit(result_type{api_promise_action_internal::Empty{}}); - } - - constexpr void SetError(error_type&& err) noexcept { - event_.Emit(result_type{Error{std::move(err)}}); - } + ApiPromise(ProtocolContext& protocol_context, RequestId req_id) noexcept + : protocol_context_{&protocol_context}, request_id_{req_id} {} constexpr RequestId request_id() const { return request_id_; } template - [[nodiscard]] constexpr decltype(auto) Subscribe(Fn&& fn) noexcept { - return EventSubscriber{event_}.Subscribe(std::forward(fn)); + constexpr decltype(auto) Subscribe(Fn&& fn) { + using Entry = api_promise_internal::PendingResponseEntry; + auto& entry = + protocol_context_->template CreatePendingResponse(request_id_); + return EventSubscriber{entry.event}.Subscribe(std::forward(fn)); } private: + ProtocolContext* protocol_context_; RequestId request_id_; - Event event_; }; } // namespace ae diff --git a/aether/api_protocol/protocol_context.cpp b/aether/api_protocol/protocol_context.cpp index d78525a8..3539411e 100644 --- a/aether/api_protocol/protocol_context.cpp +++ b/aether/api_protocol/protocol_context.cpp @@ -17,7 +17,8 @@ #include "aether/api_protocol/protocol_context.h" #include -#include +#include +#include #include "aether/api_protocol/api_protocol.h" @@ -25,81 +26,159 @@ namespace ae { ProtocolContext::ProtocolContext() = default; -ProtocolContext::~ProtocolContext() = default; - -void ProtocolContext::AddSendResultCallback(RequestId request_id, - SendResultCb callback) { - send_result_events_.emplace(request_id, std::move(callback)); -} - -void ProtocolContext::AddSendErrorCallback(RequestId request_id, - SendErrorCb callback) { - send_error_events_.emplace(request_id, std::move(callback)); +ProtocolContext::~ProtocolContext() { + while (!pending_responses_.empty()) { + auto entry = TakeOldestPending(); + DestroyPending(entry); + } } void ProtocolContext::SetSendResultResponse(RequestId request_id) { - auto it = send_result_events_.find(request_id); - if (it != send_result_events_.end()) { - it->second(); - send_result_events_.erase(it); - } else { + auto entry = TakePending(request_id); + if (entry.response == nullptr) { AE_TELED_DEBUG("No callback for request id {} cancel parse", request_id); parser()->Cancel(); } - send_error_events_.erase(request_id); + auto* p = parser(); + assert(p != nullptr && "Parser shouldn't be null"); + entry.response->OnResult(*p); + DestroyPending(entry); } void ProtocolContext::SetSendErrorResponse(RequestId req_id, std::uint8_t error_type, std::uint32_t error_code) { - auto it = send_error_events_.find(req_id); - if (it != std::end(send_error_events_)) { - it->second(error_type, error_code); - send_error_events_.erase(it); - } else { + auto entry = TakePending(req_id); + if (entry.response == nullptr) { AE_TELED_DEBUG("No callback for error with request id {}", req_id); - fprintf(stderr, "SendError: id %zu, type: %d, error_code %zu\n", - static_cast(req_id.id), static_cast(error_type), - static_cast(error_code)); parser()->Cancel(); } - send_result_events_.erase(req_id); + + entry.response->OnError(error_type, static_cast(error_code)); + DestroyPending(entry); +} + +void ProtocolContext::EvictPending(PendingEntry const& entry) { + assert(entry.response != nullptr && + "EvictPending requires a pending response entry"); + entry.response->OnEvicted(); + DestroyPending(entry); +} + +void ProtocolContext::DestroyPending(PendingEntry const& entry) { + assert(entry.response != nullptr && + "DestroyPending requires a pending response entry"); + pending_response_pool_.destroy(entry.response); +} + +void ProtocolContext::PreparePendingResponseSlot(RequestId request_id) { + // ensure there is one pending response for request_id + auto existing_entry = TakePending(request_id); + if (existing_entry.response != nullptr) { + EvictPending(existing_entry); + return; + } + + // ensure there is enough in pool for new pending response + // oldest pending should be evicted + if (pending_responses_.full()) { + auto oldest_entry = TakeOldestPending(); + EvictPending(oldest_entry); + } + + assert(!pending_responses_.full() && + "Pending response registry must have a free slot"); + assert(pending_response_pool_.available() != 0U && + "Pending response pool must have a free slot"); +} + +ProtocolContext::PendingEntry ProtocolContext::TakePending( + RequestId request_id) { + auto* it = + std::find_if(std::begin(pending_responses_), std::end(pending_responses_), + [request_id](auto const& pe) noexcept { + return pe.request_id == request_id; + }); + if (it == std::end(pending_responses_)) { + return PendingEntry{RequestId{}, nullptr}; + } + + auto found = *it; + pending_responses_.erase(it); + return found; +} + +ProtocolContext::PendingEntry ProtocolContext::TakeOldestPending() { + assert(!pending_responses_.empty() && + "TakeOldestPending requires a pending response"); + + // oldest is the first + auto* it = pending_responses_.begin(); + auto found = *it; + pending_responses_.erase(it); + + return found; } -void ProtocolContext::PushPacketStack(class PacketStack& packet_stack) { +void ProtocolContext::PushPacketStack(PacketStack& packet_stack) { + assert(!packet_stacks_.full() && + "AE_API_PROTOCOL_MAX_PACKET_STACK_DEPTH exceeded"); packet_stacks_.push(&packet_stack); } -void ProtocolContext::PopPacketStack() { packet_stacks_.pop(); } +void ProtocolContext::PopPacketStack() { + assert(!packet_stacks_.empty() && "Packet stack context is empty"); + packet_stacks_.pop(); +} -class PacketStack* ProtocolContext::packet_stack() { +PacketStack* ProtocolContext::packet_stack() { if (packet_stacks_.empty()) { return nullptr; } - return packet_stacks_.top(); + auto* packet_stack = packet_stacks_.top(); + assert(packet_stack != nullptr && "Packet stack context contains nullptr"); + return packet_stack; } -void ProtocolContext::PushParser(ApiParser& parser) { parsers_.push(&parser); } +void ProtocolContext::PushParser(ApiParser& parser) { + assert(!parsers_.full() && + "AE_API_PROTOCOL_MAX_PARSER_PACKER_DEPTH exceeded for parser"); + parsers_.push(&parser); +} -void ProtocolContext::PopParser() { parsers_.pop(); } +void ProtocolContext::PopParser() { + assert(!parsers_.empty() && "Parser context is empty"); + parsers_.pop(); +} ApiParser* ProtocolContext::parser() { if (parsers_.empty()) { return nullptr; } - return parsers_.top(); + auto* parser = parsers_.top(); + assert(parser != nullptr && "Parser context contains nullptr"); + return parser; } -void ProtocolContext::PushPacker(ApiPacker& packer) { packers_.push(&packer); } +void ProtocolContext::PushPacker(ApiPacker& packer) { + assert(!packers_.full() && + "AE_API_PROTOCOL_MAX_PARSER_PACKER_DEPTH exceeded for packer"); + packers_.push(&packer); +} -void ProtocolContext::PopPacker() { packers_.pop(); } +void ProtocolContext::PopPacker() { + assert(!packers_.empty() && "Packer context is empty"); + packers_.pop(); +} ApiPacker* ProtocolContext::packer() { if (packers_.empty()) { return nullptr; } - return packers_.top(); + auto* packer = packers_.top(); + assert(packer != nullptr && "Packer context contains nullptr"); + return packer; } } // namespace ae diff --git a/aether/api_protocol/protocol_context.h b/aether/api_protocol/protocol_context.h index 051935bf..a75e1427 100644 --- a/aether/api_protocol/protocol_context.h +++ b/aether/api_protocol/protocol_context.h @@ -17,37 +17,85 @@ #ifndef AETHER_API_PROTOCOL_PROTOCOL_CONTEXT_H_ #define AETHER_API_PROTOCOL_PROTOCOL_CONTEXT_H_ -#include -#include +#include +#include #include -#include "aether-miscpp/types/small_function.h" +#include + +#include "aether/warning_disable.h" +DISABLE_WARNING_PUSH() +IGNORE_IMPLICIT_CONVERSION() +#include +#include +DISABLE_WARNING_POP() + #include "aether/api_protocol/request_id.h" +#include "aether/config.h" namespace ae { +class PacketStack; class ApiParser; class ApiPacker; class ProtocolContext { public: - using SendResultCb = SmallFunction; - using SendErrorCb = - SmallFunction; + class PendingResponse { + public: + virtual ~PendingResponse() = default; + + virtual void OnResult(ApiParser& parser) = 0; + virtual void OnError(std::uint8_t error_type, std::int32_t error_code) = 0; + virtual void OnEvicted() = 0; + }; + + static constexpr auto kMaxPendingResponses = + AE_API_PROTOCOL_MAX_PENDING_RESPONSES; + static constexpr auto kPendingResponseMaxSize = + AE_API_PROTOCOL_PENDING_RESPONSE_MAX_SIZE; + static constexpr auto kPendingResponseAlign = + AE_API_PROTOCOL_PENDING_RESPONSE_ALIGN; + static constexpr auto kMaxPacketStackDepth = + AE_API_PROTOCOL_MAX_PACKET_STACK_DEPTH; + static constexpr auto kMaxParserPackerDepth = + AE_API_PROTOCOL_MAX_PARSER_PACKER_DEPTH; + static_assert(kMaxPendingResponses > 0, + "AE_API_PROTOCOL_MAX_PENDING_RESPONSES must be greater than 0"); + static_assert( + kMaxPacketStackDepth > 0, + "AE_API_PROTOCOL_MAX_PACKET_STACK_DEPTH must be greater than 0"); + static_assert( + kMaxParserPackerDepth > 0, + "AE_API_PROTOCOL_MAX_PARSER_PACKER_DEPTH must be greater than 0"); ProtocolContext(); ~ProtocolContext(); - void AddSendResultCallback(RequestId request_id, SendResultCb callback); - void AddSendErrorCallback(RequestId request_id, SendErrorCb callback); + template + Entry& CreatePendingResponse(RequestId request_id) { + static_assert(sizeof(Entry) <= kPendingResponseMaxSize, + "Pending response entry exceeds " + "AE_API_PROTOCOL_PENDING_RESPONSE_MAX_SIZE"); + static_assert(alignof(Entry) <= kPendingResponseAlign, + "Pending response entry alignment exceeds " + "AE_API_PROTOCOL_PENDING_RESPONSE_ALIGN"); + + PreparePendingResponseSlot(request_id); + + auto* entry_ptr = pending_response_pool_.template create(); + assert(entry_ptr != nullptr && + "Pending response pool allocation failed after slot preparation"); + pending_responses_.emplace_back(PendingEntry{request_id, entry_ptr}); + return *entry_ptr; + } void SetSendResultResponse(RequestId request_id); void SetSendErrorResponse(RequestId req_id, std::uint8_t error_type, std::uint32_t error_code); - void PushPacketStack(class PacketStack& packet_stack); + void PushPacketStack(PacketStack& packet_stack); void PopPacketStack(); - class PacketStack* packet_stack(); + PacketStack* packet_stack(); void PushParser(ApiParser& parser); void PopParser(); @@ -58,12 +106,32 @@ class ProtocolContext { ApiPacker* packer(); private: - std::map send_result_events_; - std::map send_error_events_; + using PendingResponsePool = + etl::generic_pool; + + struct PendingEntry { + RequestId request_id; + PendingResponse* response; + }; + + using PendingList = etl::vector; + using PacketStackStack = etl::stack; + using ParserStack = etl::stack; + using PackerStack = etl::stack; + + void EvictPending(PendingEntry const& entry); + void DestroyPending(PendingEntry const& entry); + void PreparePendingResponseSlot(RequestId request_id); + PendingEntry TakePending(RequestId request_id); + PendingEntry TakeOldestPending(); + + PendingResponsePool pending_response_pool_; + PendingList pending_responses_; - std::stack packet_stacks_; - std::stack parsers_; - std::stack packers_; + PacketStackStack packet_stacks_; + ParserStack parsers_; + PackerStack packers_; }; } // namespace ae diff --git a/aether/cloud_connections/ping_cloud_servers.cpp b/aether/cloud_connections/ping_cloud_servers.cpp index 61480d8a..b27676b7 100644 --- a/aether/cloud_connections/ping_cloud_servers.cpp +++ b/aether/cloud_connections/ping_cloud_servers.cpp @@ -134,7 +134,7 @@ auto PingCloudServers::ServerPing::MakePing() { policy_->ReportNextServiceTime(priority_, next_ping_time_); AE_TELED_DEBUG( "Next ping time for priority {} at {:%H:%M:%S} after {:%S}", - next_ping_time_, timing_conf_.interval); + priority_, next_ping_time_, timing_conf_.interval); return ex::set_value(std::move(ctx.receiver)); }); @@ -143,16 +143,18 @@ auto PingCloudServers::ServerPing::MakePing() { void PingCloudServers::ServerPing::Start() { waiter_.emplace( - ae_context_, EnsureLinked() | - ex::let_value([&]() noexcept - -> ex::variant_sender { - if (stop_) { - return ex::just_stopped(); - } - return ex::just(); - }) | - MakePing() | + ae_context_, + EnsureLinked() | + ex::let_value( + [&]() noexcept + -> ex::variant_sender { + if (stop_) { + return ex::just_stopped(); + } + return ex::just(); + }) | + MakePing() | // track Stop command ex::let_value( [&]() noexcept diff --git a/aether/config.h b/aether/config.h index 58ad3113..0dd682be 100644 --- a/aether/config.h +++ b/aether/config.h @@ -18,6 +18,7 @@ #define AETHER_CONFIG_H_ // IWYU pragma: begin_exports +#include #include #include @@ -41,8 +42,25 @@ # define AE_TASK_ALIGN alignof(std::max_align_t) #endif -#ifndef AE_INDEX_REGISTRY_CAPACITY -# define AE_INDEX_REGISTRY_CAPACITY 128 +#ifndef AE_API_PROTOCOL_MAX_PENDING_RESPONSES +# define AE_API_PROTOCOL_MAX_PENDING_RESPONSES 10 +#endif + +#ifndef AE_API_PROTOCOL_MAX_PACKET_STACK_DEPTH +# define AE_API_PROTOCOL_MAX_PACKET_STACK_DEPTH 4 +#endif + +#ifndef AE_API_PROTOCOL_MAX_PARSER_PACKER_DEPTH +# define AE_API_PROTOCOL_MAX_PARSER_PACKER_DEPTH 4 +#endif + +// Maximum size in bytes for a single pending response pool element. +#ifndef AE_API_PROTOCOL_PENDING_RESPONSE_MAX_SIZE +# define AE_API_PROTOCOL_PENDING_RESPONSE_MAX_SIZE 2 * sizeof(void*) +#endif + +#ifndef AE_API_PROTOCOL_PENDING_RESPONSE_ALIGN +# define AE_API_PROTOCOL_PENDING_RESPONSE_ALIGN alignof(std::max_align_t) #endif #ifndef AE_SUPPORT_IPV4 diff --git a/aether/registration/registration.cpp b/aether/registration/registration.cpp index a6d06194..06130e4a 100644 --- a/aether/registration/registration.cpp +++ b/aether/registration/registration.cpp @@ -20,14 +20,14 @@ # include +# include "aether-miscpp/misc/override.h" # include "aether/aether.h" +# include "aether/api_protocol/api_context.h" +# include "aether/api_protocol/make_api_call_sender.h" # include "aether/crypto.h" +# include "aether/crypto/crypto_definitions.h" # include "aether/crypto/sign.h" -# include "aether-miscpp/misc/override.h" # include "aether/executors/executors.h" -# include "aether/api_protocol/api_context.h" -# include "aether/crypto/crypto_definitions.h" -# include "aether/api_protocol/make_api_call_sender.h" # include "aether/registration/proof_of_work.h" # include "aether/registration/registration_crypto_provider.h" @@ -93,7 +93,7 @@ void Registration::InitConnection() { void Registration::Restream() { root_server_select_stream_.Restream(); } auto Registration::GetKeys() { - return make_api_call( + return make_api_call( server_reg_root_api_, root_server_select_stream_, [&, s{Subscription{}}](ApiContext& api_call, auto& r) mutable noexcept { @@ -101,10 +101,10 @@ auto Registration::GetKeys() { s = api_call->get_asymmetric_public_key(kDefaultCryptoLibProfile) .Subscribe([&](auto&& res) mutable noexcept { if (!res) { - ex::set_error(std::move(r), std::move(res).error()); + ex::set_error(std::move(r), res.error()); return; } - auto signed_key = std::move(res).value(); + auto signed_key = std::forward(res).value(); if (!CryptoSignVerify(signed_key.sign, signed_key.key, sign_pk_)) { AE_TELE_ERROR(RegisterGetKeysVerificationFailed, @@ -135,7 +135,7 @@ auto Registration::GetKeys() { } auto Registration::RequestPowParams() { - return make_api_call( + return make_api_call( server_reg_root_api_, root_server_select_stream_, [&, s{Subscription{}}](ApiContext& api_call, auto& r) mutable noexcept { @@ -150,11 +150,12 @@ auto Registration::RequestPowParams() { PowMethod::kBCryptCrc32) .Subscribe([&](auto&& res) mutable noexcept { if (!res) { - ex::set_error(std::move(r), std::move(res).error()); + ex::set_error(std::move(r), res.error()); return; } - auto pow_params = std::move(res).value(); + auto pow_params = + std::forward(res).value(); if (!CryptoSignVerify(pow_params.global_key.sign, pow_params.global_key.key, sign_pk_)) { @@ -184,7 +185,7 @@ auto Registration::RequestPowParams() { } auto Registration::MakeRegistration() { - return make_api_call( + return make_api_call( server_reg_root_api_, root_server_select_stream_, [&, s{Subscription{}}](ApiContext& api_call, auto& r) mutable noexcept { @@ -213,15 +214,17 @@ auto Registration::MakeRegistration() { [&, r{std::forward(r)}]( auto&& res) mutable noexcept { if (!res) { - ex::set_error(std::move(r), std::move(res).error()); + ex::set_error(std::move(r), res.error()); return; } AE_TELE_INFO(RegisterRegistrationConfirmed, "Registration confirmed"); - client_uid_ = res.value().uid; - ephemeral_uid_ = res.value().ephemeral_uid; + auto reg_info = + std::forward(res).value(); + client_uid_ = reg_info.uid; + ephemeral_uid_ = reg_info.ephemeral_uid; // use client cloud_ for cloud resolve request - client_cloud_ = res.value().cloud; + client_cloud_ = std::move(reg_info.cloud); assert((client_cloud_.size() > 0) && "Client's cloud is empty"); ex::set_value(std::move(r), Ignore{}); @@ -233,7 +236,7 @@ auto Registration::MakeRegistration() { auto Registration::ResolveCloud() { return make_api_call( + ex::set_error_t(std::int32_t)>( server_reg_root_api_, root_server_select_stream_, [&, s{Subscription{}}](ApiContext& api_call, auto& r) mutable noexcept { @@ -249,9 +252,10 @@ auto Registration::ResolveCloud() { if (res) { ex::set_value( std::move(r), - OnCloudResolved(std::move(res).value())); + OnCloudResolved( + std::forward(res).value())); } else { - ex::set_error(std::move(r), std::move(res).error()); + ex::set_error(std::move(r), res.error()); } }); }}); diff --git a/aether/tele/env/compilation_options.h b/aether/tele/env/compilation_options.h index e7bbad0e..81e991b0 100644 --- a/aether/tele/env/compilation_options.h +++ b/aether/tele/env/compilation_options.h @@ -18,13 +18,13 @@ #define AETHER_TELE_ENV_COMPILATION_OPTIONS_H_ #include -#include #include -#include +#include #include +#include -#include "aether/config.h" #include "aether/common.h" +#include "aether/config.h" #define _OPTION_VALUE(option, value) \ std::pair { \ @@ -81,7 +81,11 @@ constexpr inline auto _compile_options_list = std::array{ _OPTION(AE_TASK_MAX_COUNT), _OPTION(AE_TASK_MAX_SIZE), _OPTION(AE_TASK_ALIGN), - _OPTION(AE_INDEX_REGISTRY_CAPACITY), + _OPTION(AE_API_PROTOCOL_MAX_PENDING_RESPONSES), + _OPTION(AE_API_PROTOCOL_MAX_PACKET_STACK_DEPTH), + _OPTION(AE_API_PROTOCOL_MAX_PARSER_PACKER_DEPTH), + _OPTION(AE_API_PROTOCOL_PENDING_RESPONSE_MAX_SIZE), + _OPTION(AE_API_PROTOCOL_PENDING_RESPONSE_ALIGN), _OPTION(AE_SUPPORT_IPV4), _OPTION(AE_SUPPORT_IPV6), _OPTION(AE_SUPPORT_UDP), diff --git a/examples/benches/send_messages_bandwidth/sender/sender.cpp b/examples/benches/send_messages_bandwidth/sender/sender.cpp index 9a2923c8..8c1c5ba8 100644 --- a/examples/benches/send_messages_bandwidth/sender/sender.cpp +++ b/examples/benches/send_messages_bandwidth/sender/sender.cpp @@ -46,7 +46,7 @@ void Sender::Disconnect() { message_stream_.reset(); } EventSubscriber Sender::Handshake() { auto api = ApiCallAdapter{ApiContext{bandwidth_api_}, *message_stream_}; - auto& res = api->handshake(); + auto res = api->handshake(); handshake_sub_ = res.Subscribe([this](auto const& res) { AE_TELED_DEBUG("Handshake received"); if (res && res.value()) { diff --git a/examples/cloud/cloud_test.cpp b/examples/cloud/cloud_test.cpp index 0fd3eeac..809c6763 100644 --- a/examples/cloud/cloud_test.cpp +++ b/examples/cloud/cloud_test.cpp @@ -37,6 +37,10 @@ // IWYU pragma: end_keeps namespace ae::cloud_test { + +static constexpr inline auto kParentUid = + ae::Uid::FromString("3ac93165-3d37-4970-87a6-fa4ee27744e4"); + constexpr SafeStreamConfig kSafeStreamConfig{ .window_size = AE_SAFE_STREAM_CAPACITY / 2 - 1, .max_packet_size = AE_SAFE_STREAM_CAPACITY / 2 - 1, @@ -65,8 +69,8 @@ int AetherCloudExample() { ae::Client::ptr client_a; ae::Client::ptr client_b; - auto& select_client_a = aether_app->aether()->SelectClient( - ae::Uid::FromString("3ac93165-3d37-4970-87a6-fa4ee27744e4"), "A"); + auto& select_client_a = + aether_app->aether()->SelectClient(ae::cloud_test::kParentUid, "A"); select_client_a.result_event().Subscribe([&](auto const& res) { if (res) { client_a = res.value(); @@ -75,8 +79,8 @@ int AetherCloudExample() { } }); - auto& select_client_b = aether_app->aether()->SelectClient( - ae::Uid::FromString("3ac93165-3d37-4970-87a6-fa4ee27744e4"), "B"); + auto& select_client_b = + aether_app->aether()->SelectClient(ae::cloud_test::kParentUid, "B"); select_client_b.result_event().Subscribe([&](auto const& res) { if (res) { client_b = res.value(); diff --git a/tests/test-api-protocol/test-method-call.cpp b/tests/test-api-protocol/test-method-call.cpp index c7099e82..7b57d11d 100644 --- a/tests/test-api-protocol/test-method-call.cpp +++ b/tests/test-api-protocol/test-method-call.cpp @@ -17,9 +17,11 @@ #include #include +#include +#include -#include "aether/events/events.h" #include "aether/api_protocol/api_protocol.h" +#include "aether/events/events.h" #include "aether/types/data_buffer.h" @@ -27,6 +29,11 @@ namespace ae::test_method_call { +struct Number { + AE_REFLECT_MEMBERS(value) + int value; +}; + class ApiLevel1 : public ApiClassImpl { public: explicit ApiLevel1(ProtocolContext& protocol_context) @@ -66,7 +73,7 @@ class ApiLevel0 : public ApiClassImpl { // methods invoked then packet received void Method3Impl(int a, std::string b) { method_3_event.Emit(a, b); } - void Method4Impl(PromiseResult promise, int a) { + void Method4Impl(PromiseResult promise, int a) { method_4_event.Emit(promise.request_id, a); } void Method5Impl(SubContextImpl sub_api, int a) { @@ -86,7 +93,7 @@ class ApiLevel0 : public ApiClassImpl { // call methods to make packet Method<03, void(int a, std::string b)> method_3; - Method<04, ApiPromise(int a)> method_4; + Method<04, ApiPromise(int a)> method_4; Method<05, SubContext(int a)> method_5; Method<06, void(int a, SubApi sub), Method6Proc> method_6; @@ -137,10 +144,10 @@ void test_ReturnResult() { auto api_level0 = ApiLevel0{pc}; auto call_context = ApiContext{api_level0}; - auto& promise = call_context->method_4(42); + auto promise = call_context->method_4(42); promise.Subscribe([&](auto const& res) { if (res.IsOk()) { - auto value = res.value(); + auto value = res.value().value; TEST_ASSERT_EQUAL(78, value); promise_get_value = true; } else { @@ -158,7 +165,7 @@ void test_ReturnResult() { [&](RequestId req_id, int) { level0_method4_called = true; auto response_context = ApiContext{api_level0}; - response_context->return_result_api.SendResult(req_id, 78); + response_context->return_result_api.SendResult(req_id, Number{78}); auto data = std::move(response_context).Pack(); auto parser = ApiParser{pc, data}; // send/receive @@ -197,6 +204,212 @@ void test_MethodWithSubApi() { TEST_ASSERT_TRUE(level1_method3_called); } +void test_ProtocolContextStackAccess() { + ProtocolContext pc; + + TEST_ASSERT_NULL(pc.packet_stack()); + TEST_ASSERT_NULL(pc.parser()); + TEST_ASSERT_NULL(pc.packer()); + + auto packet_stack0 = PacketStack{}; + auto packet_stack1 = PacketStack{}; + pc.PushPacketStack(packet_stack0); + TEST_ASSERT_EQUAL_PTR(&packet_stack0, pc.packet_stack()); + pc.PushPacketStack(packet_stack1); + TEST_ASSERT_EQUAL_PTR(&packet_stack1, pc.packet_stack()); + pc.PopPacketStack(); + TEST_ASSERT_EQUAL_PTR(&packet_stack0, pc.packet_stack()); + pc.PopPacketStack(); + TEST_ASSERT_NULL(pc.packet_stack()); + + auto data0 = std::vector{}; + auto data1 = std::vector{}; + { + auto parser0 = ApiParser{pc, data0}; + TEST_ASSERT_EQUAL_PTR(&parser0, pc.parser()); + { + auto parser1 = ApiParser{pc, data1}; + TEST_ASSERT_EQUAL_PTR(&parser1, pc.parser()); + } + TEST_ASSERT_EQUAL_PTR(&parser0, pc.parser()); + } + TEST_ASSERT_NULL(pc.parser()); + + { + auto packer0 = ApiPacker{pc, data0}; + TEST_ASSERT_EQUAL_PTR(&packer0, pc.packer()); + { + auto packer1 = ApiPacker{pc, data1}; + TEST_ASSERT_EQUAL_PTR(&packer1, pc.packer()); + } + TEST_ASSERT_EQUAL_PTR(&packer0, pc.packer()); + } + TEST_ASSERT_NULL(pc.packer()); +} + +void test_PendingResponseCapacityEvictsOldest() { + ProtocolContext pc; + + bool first_promise_evicted = false; + + auto api_level0 = ApiLevel0{pc}; + auto call_context = ApiContext{api_level0}; + auto subscriptions = std::vector{}; + subscriptions.reserve(AE_API_PROTOCOL_MAX_PENDING_RESPONSES + 1); + + for (auto i = 0U; i < AE_API_PROTOCOL_MAX_PENDING_RESPONSES + 1; ++i) { + auto promise = call_context->method_4(static_cast(i)); + auto subscription = promise.Subscribe([&, i](auto const& res) { + if (i == 0U) { + TEST_ASSERT_FALSE(res.IsOk()); + TEST_ASSERT_EQUAL(-1, res.error()); + first_promise_evicted = true; + } + }); + subscriptions.emplace_back(std::move(subscription)); + } + + TEST_ASSERT_TRUE(first_promise_evicted); +} + +void test_PendingResponseDuplicateRequestIdReplacesOld() { + ProtocolContext pc; + + auto request_id = RequestId{42}; + auto first_promise = ApiPromise{pc, request_id}; + auto first_evicted = false; + auto first_subscription = first_promise.Subscribe([&](auto const& res) { + TEST_ASSERT_FALSE(res.IsOk()); + TEST_ASSERT_EQUAL(-1, res.error()); + first_evicted = true; + }); + + auto second_promise = ApiPromise{pc, request_id}; + auto second_subscription = second_promise.Subscribe([](auto const&) {}); + + static_cast(first_subscription); + static_cast(second_subscription); + TEST_ASSERT_TRUE(first_evicted); +} + +void test_PendingResponseFreshSameTypeReplacesOld() { + ProtocolContext pc; + + auto request_id = RequestId{42}; + auto first_promise = ApiPromise{pc, request_id}; + auto first_evicted = false; + auto first_subscription = first_promise.Subscribe([&](auto const& res) { + TEST_ASSERT_FALSE(res.IsOk()); + TEST_ASSERT_EQUAL(-1, res.error()); + first_evicted = true; + }); + + auto second_promise = ApiPromise{pc, request_id}; + auto second_subscription = second_promise.Subscribe([](auto const&) {}); + + static_cast(first_subscription); + static_cast(second_subscription); + TEST_ASSERT_TRUE(first_evicted); +} + +void test_PendingResponseCompletionFreesPoolSlot() { + ProtocolContext pc; + + auto second_promise_evicted = false; + auto subscriptions = std::vector{}; + subscriptions.reserve(AE_API_PROTOCOL_MAX_PENDING_RESPONSES + 1); + + for (auto i = 0U; i < AE_API_PROTOCOL_MAX_PENDING_RESPONSES; ++i) { + auto promise = ApiPromise{pc, RequestId{i + 1}}; + auto subscription = promise.Subscribe([&, i](auto const& res) { + if (i == 1U) { + TEST_ASSERT_FALSE(res.IsOk()); + TEST_ASSERT_EQUAL(static_cast(-1), res.error()); + second_promise_evicted = true; + } + }); + subscriptions.emplace_back(std::move(subscription)); + } + + pc.SetSendErrorResponse(RequestId{1}, 0, 7); + + auto extra_promise = ApiPromise{pc, RequestId{1000}}; + subscriptions.emplace_back(extra_promise.Subscribe([](auto const&) {})); + + TEST_ASSERT_FALSE(second_promise_evicted); +} + +void test_PendingResponseFifoAfterMiddleRemoval() { + ProtocolContext pc; + + auto first_evicted = false; + auto second_evicted = false; + auto subscriptions = std::vector{}; + subscriptions.reserve(AE_API_PROTOCOL_MAX_PENDING_RESPONSES + 2); + + for (auto i = 0U; i < AE_API_PROTOCOL_MAX_PENDING_RESPONSES; ++i) { + auto id = i + 1U; + auto promise = ApiPromise{pc, RequestId{id}}; + auto subscription = promise.Subscribe([&, id](auto const& res) { + if (!res.IsOk() && (res.error() == static_cast(-1))) { + if (id == 1U) { + first_evicted = true; + } else if (id == 2U) { + second_evicted = true; + } + } + }); + subscriptions.emplace_back(std::move(subscription)); + } + + pc.SetSendErrorResponse(RequestId{3}, 0, 7); + + auto fill_promise = ApiPromise{pc, RequestId{1000}}; + subscriptions.emplace_back(fill_promise.Subscribe([](auto const&) {})); + auto overflow_promise = ApiPromise{pc, RequestId{1001}}; + subscriptions.emplace_back(overflow_promise.Subscribe([](auto const&) {})); + + TEST_ASSERT_TRUE(first_evicted); + TEST_ASSERT_FALSE(second_evicted); +} + +void test_PendingResponseDuplicateReplacementPreservesFifo() { + ProtocolContext pc; + + auto first_evicted = false; + auto second_evicted = false; + auto third_evicted = false; + auto subscriptions = std::vector{}; + subscriptions.reserve(AE_API_PROTOCOL_MAX_PENDING_RESPONSES + 2); + + for (auto i = 0U; i < AE_API_PROTOCOL_MAX_PENDING_RESPONSES; ++i) { + auto id = i + 1U; + auto promise = ApiPromise{pc, RequestId{id}}; + auto subscription = promise.Subscribe([&, id](auto const& res) { + if (!res.IsOk() && (res.error() == static_cast(-1))) { + if (id == 1U) { + first_evicted = true; + } else if (id == 2U) { + second_evicted = true; + } else if (id == 3U) { + third_evicted = true; + } + } + }); + subscriptions.emplace_back(std::move(subscription)); + } + + auto replacement_promise = ApiPromise{pc, RequestId{3}}; + subscriptions.emplace_back(replacement_promise.Subscribe([](auto const&) {})); + TEST_ASSERT_TRUE(third_evicted); + + auto overflow_promise = ApiPromise{pc, RequestId{1000}}; + subscriptions.emplace_back(overflow_promise.Subscribe([](auto const&) {})); + + TEST_ASSERT_TRUE(first_evicted); + TEST_ASSERT_FALSE(second_evicted); +} + } // namespace ae::test_method_call int test_method_call() { @@ -204,5 +417,14 @@ int test_method_call() { RUN_TEST(ae::test_method_call::test_ApiMethodInvoke); RUN_TEST(ae::test_method_call::test_ReturnResult); RUN_TEST(ae::test_method_call::test_MethodWithSubApi); + RUN_TEST(ae::test_method_call::test_ProtocolContextStackAccess); + RUN_TEST(ae::test_method_call::test_PendingResponseCapacityEvictsOldest); + RUN_TEST( + ae::test_method_call::test_PendingResponseDuplicateRequestIdReplacesOld); + RUN_TEST(ae::test_method_call::test_PendingResponseFreshSameTypeReplacesOld); + RUN_TEST(ae::test_method_call::test_PendingResponseCompletionFreesPoolSlot); + RUN_TEST(ae::test_method_call::test_PendingResponseFifoAfterMiddleRemoval); + RUN_TEST(ae::test_method_call:: + test_PendingResponseDuplicateReplacementPreservesFifo); return UNITY_END(); } From 147e63eca480defe588b7d693587c39b459140f6 Mon Sep 17 00:00:00 2001 From: BartolomeyKant Date: Thu, 16 Jul 2026 18:34:58 +0500 Subject: [PATCH 3/3] prevent double call for channel error --- aether/channels/ethernet_channel.cpp | 8 +- aether/channels/wifi_channel.cpp | 10 ++- .../server_connections/server_connection.cpp | 88 ++++++++++++++----- aether/server_connections/server_connection.h | 34 ++++++- 4 files changed, 109 insertions(+), 31 deletions(-) diff --git a/aether/channels/ethernet_channel.cpp b/aether/channels/ethernet_channel.cpp index 06293323..ecff57ab 100644 --- a/aether/channels/ethernet_channel.cpp +++ b/aether/channels/ethernet_channel.cpp @@ -19,16 +19,18 @@ #include #include -#include "aether/memory.h" #include "aether/events/event_subscription.h" +#include "aether/memory.h" #include "aether/aether.h" -#include "aether/poller/poller.h" #include "aether/dns/dns_resolve.h" #include "aether/executors/executors.h" +#include "aether/poller/poller.h" #include "aether/channels/ethernet_transport_factory.h" +#include "aether/tele/tele.h" + namespace ae { namespace ethernet_access_point_internal { using ResolveSender = @@ -165,6 +167,8 @@ TransportBuildSender EthernetChannel::TransportBuilder() { auto poller = poller_.Load(); assert(poller && "Poller is not loaded"); + AE_TELED_DEBUG("Make transport builder for {}", address); + return ethernet_access_point_internal::MakeTransportBuilder( AeContext{*aether_.Load().as()}, resolver, poller, address); } diff --git a/aether/channels/wifi_channel.cpp b/aether/channels/wifi_channel.cpp index 2806c713..c33f5fae 100644 --- a/aether/channels/wifi_channel.cpp +++ b/aether/channels/wifi_channel.cpp @@ -17,17 +17,17 @@ #include "aether/channels/wifi_channel.h" #if AE_SUPPORT_WIFIS +# include # include # include -# include -# include "aether/ptr/ptr_view.h" -# include "aether/executors/executors.h" # include "aether/events/event_subscription.h" +# include "aether/executors/executors.h" +# include "aether/ptr/ptr_view.h" # include "aether/aether.h" -# include "aether/poller/poller.h" # include "aether/dns/dns_resolve.h" +# include "aether/poller/poller.h" # include "aether/channels/ethernet_transport_factory.h" @@ -206,6 +206,8 @@ TransportBuildSender WifiChannel::TransportBuilder() { assert(poller && "Poller is not loaded"); assert(access_point && "Access point is not loaded"); + AE_TELED_DEBUG("Make transport builder for {}", address); + return wifi_channel_internal::MakeTransportBuilder( *aether_.Load().as(), resolver, poller, access_point, address); } diff --git a/aether/server_connections/server_connection.cpp b/aether/server_connections/server_connection.cpp index cd0100df..b63e07e1 100644 --- a/aether/server_connections/server_connection.cpp +++ b/aether/server_connections/server_connection.cpp @@ -16,16 +16,50 @@ #include "aether/server_connections/server_connection.h" -#include #include +#include #include "aether/aether.h" -#include "aether/server.h" #include "aether/channels/channel.h" +#include "aether/server.h" #include "aether/tele/tele.h" namespace ae { +ServerConnection::ChannelSelectAction::ChannelSelectAction( + AeContext const& ae_context, ChannelEntry& top_channel) noexcept + : ae_context_{ae_context}, top_channel_{&top_channel} { + auto channel = top_channel_->channel.Lock(); + assert(channel && "Channel is null"); + + top_channel_->connection.BuildTransport(channel, [this](auto&& res) { + if (res) { + ChannelSelected(); + } else { + ChannelFailed(); + } + }); +} + +auto ServerConnection::ChannelSelectAction::result_event() noexcept + -> ResultEvent::Subscriber { + return EventSubscriber{result_event_}; +} + +void ServerConnection::ChannelSelectAction::ChannelSelected() { + task_sub_ = ae_context_.scheduler().Task([&]() noexcept { + result_event_.Emit(Ok{*top_channel_}); + Finish(); + }); +} + +void ServerConnection::ChannelSelectAction::ChannelFailed() { + task_sub_ = ae_context_.scheduler().Task([&]() noexcept { + result_event_.Emit(Error{1}); + Finish(); + }); +} + ServerConnection::ServerConnection(AeContext const& ae_context, Ptr const& server) : ae_context_{ae_context}, server_{server}, full_connected_{false} { @@ -34,6 +68,8 @@ ServerConnection::ServerConnection(AeContext const& ae_context, } WriteAction& ServerConnection::Write(DataBuffer&& in_data) { + // Write allowed only if stream_info.is_writable == true + assert(stream_info_.is_writable && "Channel is not writable"); assert((top_channel_ != nullptr) && "channel connection is not available"); auto* stream = top_channel_->connection.stream(); @@ -72,10 +108,11 @@ Ptr ServerConnection::current_channel() const { if (top_channel_ == nullptr) { return {}; } + // ensure server is alive [[maybe_unused]] auto server = server_.Lock(); assert(server && "Server is null"); - // ensure channel is loaded + // ensure channel is alive return top_channel_->channel.Lock(); } @@ -125,30 +162,25 @@ ServerConnection::ChannelEntry* ServerConnection::TopChannel() { } void ServerConnection::SelectChannel() { - AE_TELED_DEBUG("Select channel"); - top_channel_ = TopChannel(); - if (top_channel_ == nullptr) { - DeferServerError(); + // prevent new channel selection while one is active + if (channel_select_action_ && !channel_select_action_->is_finished()) { + AE_TELED_DEBUG("Repeated select channel"); return; } - auto server = server_.Lock(); - assert(server && "Server is null"); - auto channel = top_channel_->channel.Lock(); - if (!channel) { - DeferChannelError(); + AE_TELED_DEBUG("Select channel"); + auto* top = TopChannel(); + if (top == nullptr) { + DeferServerError(); return; } - auto channel_props = channel->transport_properties(); - stream_info_.is_reliable = - (channel_props.reliability == Reliability::kReliable); - stream_info_.rec_element_size = channel_props.rec_packet_size; - stream_info_.max_element_size = channel_props.max_packet_size; stream_info_.link_state = LinkState::kUnlinked; stream_info_.is_writable = false; + stream_update_event_.Emit(); - top_channel_->connection.BuildTransport(channel, [this](auto&& res) { + channel_select_action_.emplace(ae_context_, *top); + channel_select_action_->result_event().Subscribe([this](auto&& res) noexcept { if (res) { ChannelUpdated(res.value()); } else { @@ -161,21 +193,35 @@ void ServerConnection::SelectChannel() { stream_update_event_.Emit(); } -void ServerConnection::ChannelUpdated(ByteIStream& stream) { +void ServerConnection::ChannelUpdated(ChannelEntry& new_channel) { AE_TELED_DEBUG("Channel updated"); + top_channel_ = &new_channel; + auto& stream = *top_channel_->connection.stream(); + assert(stream.stream_info().link_state == LinkState::kLinked && + "New channel should be linked"); + + // track channel stream link error channel_stream_update_sub_ = stream.stream_update_event().Subscribe([this, s_ = &stream]() { auto info = s_->stream_info(); if (info.link_state == LinkState::kLinkError) { ChannelError(); - } else { - stream_update_event_.Emit(); } }); channel_stream_out_data_sub_ = stream.out_data_event().Subscribe( MethodPtr<&ServerConnection::OnRead>{this}); + auto channel = top_channel_->channel.Lock(); + assert(channel && "Cahnel is null"); + + auto channel_props = channel->transport_properties(); + stream_info_.is_reliable = + (channel_props.reliability == Reliability::kReliable); + stream_info_.rec_element_size = channel_props.rec_packet_size; + stream_info_.max_element_size = channel_props.max_packet_size; + + // now it's safe to write to server stream stream_info_.link_state = LinkState::kLinked; stream_info_.is_writable = true; stream_update_event_.Emit(); diff --git a/aether/server_connections/server_connection.h b/aether/server_connections/server_connection.h index 367cd0bb..3898d2fd 100644 --- a/aether/server_connections/server_connection.h +++ b/aether/server_connections/server_connection.h @@ -17,14 +17,19 @@ #ifndef AETHER_SERVER_CONNECTIONS_SERVER_CONNECTION_H_ #define AETHER_SERVER_CONNECTIONS_SERVER_CONNECTION_H_ +#include #include +#include "aether-miscpp/types/result.h" + +#include "aether/actions/action.h" #include "aether/ae_context.h" +#include "aether/events/events.h" #include "aether/ptr/ptr.h" #include "aether/ptr/ptr_view.h" -#include "aether/events/events.h" -#include "aether/stream_api/istream.h" + #include "aether/server_connections/channel_connection.h" +#include "aether/stream_api/istream.h" namespace ae { class Server; @@ -32,13 +37,33 @@ class Channel; class ServerConnection final : public ByteIStream { struct ChannelEntry { - ChannelEntry(AeContext const& ae_context, PtrView const& c): channel{c}, connection{ae_context} {} + ChannelEntry(AeContext const& ae_context, PtrView const& c) + : channel{c}, connection{ae_context} {} PtrView channel; ChannelConnection connection; bool failed = false; }; + class ChannelSelectAction final : public Action { + public: + using ResultEvent = Event)>; + + ChannelSelectAction(AeContext const& ae_context, + ChannelEntry& top_channel) noexcept; + + ResultEvent::Subscriber result_event() noexcept; + + private: + void ChannelSelected(); + void ChannelFailed(); + + AeContext ae_context_; + ChannelEntry* top_channel_; + TaskSubscription task_sub_; + ResultEvent result_event_; + }; + public: using ServerErrorEvent = Event; using ChannelChangedEvent = Event; @@ -61,7 +86,7 @@ class ServerConnection final : public ByteIStream { // return top not failed channel or null if nothing was selected ChannelEntry* TopChannel(); void SelectChannel(); - void ChannelUpdated(ByteIStream& stream); + void ChannelUpdated(ChannelEntry& new_channel); void ServerError(); void ChannelError(); @@ -76,6 +101,7 @@ class ServerConnection final : public ByteIStream { bool full_connected_; ChannelEntry* top_channel_; std::vector> channels_; + std::optional channel_select_action_; StreamInfo stream_info_; OutDataEvent out_data_event_;