diff --git a/CMakeLists.txt b/CMakeLists.txt index 64b73e0..ce3daff 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -15,10 +15,21 @@ set (CMAKE_POLICY_VERSION_MINIMUM 3.5) ### Define the project name and version ### Version will be added to version output of apps and library project(laps - VERSION 0.18.12 + VERSION 0.18.15 DESCRIPTION "Latency Aware Publish/Subscribe" LANGUAGES CXX) +## +## CPM +## +if (NOT DEFINED ENV{IDF_PATH}) + set (CMAKE_MODULE_PATH ${CMAKE_CURRENT_SOURCE_DIR}/cmake) +endif() + +include(CPM) + +## + configure_file( version_config.h.in ${CMAKE_BINARY_DIR}/generated/version_config.h ) include_directories( ${CMAKE_BINARY_DIR}/generated/ ${CMAKE_CURRENT_SOURCE_DIR}/src) @@ -43,6 +54,20 @@ endif() ### add_subdirectory(dependencies) +# quicr/log.h picks between std::format and fmt on this definition, which libquicr only defines for its own +# sources. The relay logs through quicr::Logger, so it has to make the same choice libquicr did. +if(HAS_STD_VFORMAT) + add_compile_definitions(QUICR_HAVE_STD_FORMAT) +endif() + +# spdlog compiles a bundled copy of fmt by default. libquicr also links fmt::fmt +# when std::vformat is unavailable (typical on libstdc++ / GCC < 13), so linking +# both into lapsRelay produces duplicate fmt::v12 symbols on Linux. +if(TARGET fmt::fmt) + set(SPDLOG_FMT_EXTERNAL ON CACHE BOOL "Use external fmt; avoid duplicate symbols with libquicr" FORCE) +endif() +CPMAddPackage("gh:gabime/spdlog@1.17.0") + ### ### Build relay ### diff --git a/Dockerfile b/Dockerfile index b6c1057..c52f7a0 100644 --- a/Dockerfile +++ b/Dockerfile @@ -13,6 +13,7 @@ WORKDIR /ws COPY ./Makefile ./ COPY ./CMakeLists.txt ./ +COPY ./cmake ./cmake COPY ./version_config.h.in ./ COPY ./dependencies ./dependencies COPY ./src ./src diff --git a/cmake/CPM.cmake b/cmake/CPM.cmake new file mode 100644 index 0000000..a4d7934 --- /dev/null +++ b/cmake/CPM.cmake @@ -0,0 +1,24 @@ +# SPDX-License-Identifier: MIT +# +# SPDX-FileCopyrightText: Copyright (c) 2019-2023 Lars Melchior and contributors + +set(CPM_DOWNLOAD_VERSION 0.42.3) +set(CPM_HASH_SUM "a609e875fd532b067174250f6abbc3dac22fe2d64869783fb1e80bda1625c844") + +if(CPM_SOURCE_CACHE) + set(CPM_DOWNLOAD_LOCATION "${CPM_SOURCE_CACHE}/cpm/CPM_${CPM_DOWNLOAD_VERSION}.cmake") +elseif(DEFINED ENV{CPM_SOURCE_CACHE}) + set(CPM_DOWNLOAD_LOCATION "$ENV{CPM_SOURCE_CACHE}/cpm/CPM_${CPM_DOWNLOAD_VERSION}.cmake") +else() + set(CPM_DOWNLOAD_LOCATION "${CMAKE_BINARY_DIR}/cmake/CPM_${CPM_DOWNLOAD_VERSION}.cmake") +endif() + +# Expand relative path. This is important if the provided path contains a tilde (~) +get_filename_component(CPM_DOWNLOAD_LOCATION ${CPM_DOWNLOAD_LOCATION} ABSOLUTE) + +file(DOWNLOAD + https://github.com/cpm-cmake/CPM.cmake/releases/download/v${CPM_DOWNLOAD_VERSION}/CPM.cmake + ${CPM_DOWNLOAD_LOCATION} EXPECTED_HASH SHA256=${CPM_HASH_SUM} +) + +include(${CPM_DOWNLOAD_LOCATION}) diff --git a/debian.Dockerfile b/debian.Dockerfile index e99ed96..b0d9855 100644 --- a/debian.Dockerfile +++ b/debian.Dockerfile @@ -19,6 +19,7 @@ WORKDIR /ws COPY ./Makefile ./ COPY ./CMakeLists.txt ./ +COPY ./cmake ./cmake COPY ./version_config.h.in ./ COPY ./dependencies ./dependencies COPY ./src ./src diff --git a/debian.clang.Dockerfile b/debian.clang.Dockerfile index a3ca8bc..1b4d0da 100644 --- a/debian.clang.Dockerfile +++ b/debian.clang.Dockerfile @@ -21,6 +21,7 @@ WORKDIR /ws COPY ./Makefile ./ COPY ./CMakeLists.txt ./ +COPY ./cmake ./cmake COPY ./version_config.h.in ./ COPY ./dependencies ./dependencies COPY ./src ./src diff --git a/dependencies/libquicr b/dependencies/libquicr index 2345e4b..15766e6 160000 --- a/dependencies/libquicr +++ b/dependencies/libquicr @@ -1 +1 @@ -Subproject commit 2345e4b42bba612e3c461236938468f7f28a7bb0 +Subproject commit 15766e6b407a0a07ca9ec0a75c04d137101cafbf diff --git a/healthcheck.Dockerfile b/healthcheck.Dockerfile index a936140..661ca02 100644 --- a/healthcheck.Dockerfile +++ b/healthcheck.Dockerfile @@ -20,6 +20,7 @@ RUN apk add --no-cache \ WORKDIR /ws COPY ./CMakeLists.txt ./ +COPY ./cmake ./cmake COPY ./version_config.h.in ./ COPY ./dependencies ./dependencies COPY ./src ./src diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index 237c83d..20bd703 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -29,7 +29,7 @@ add_executable(lapsRelay peering/info_base.cc ) -target_link_libraries(lapsRelay PRIVATE quicr) +target_link_libraries(lapsRelay PRIVATE quicr spdlog::spdlog) target_include_directories(lapsRelay PRIVATE ${PROJECT_SOURCE_DIR}/dependencies/oss ) target_compile_options(lapsRelay @@ -54,7 +54,7 @@ set_target_properties(relay_health_check_lib CXX_EXTENSIONS OFF) add_executable(relay_health_check relay_health_check.cpp) -target_link_libraries(relay_health_check PRIVATE quicr relay_health_check_lib) +target_link_libraries(relay_health_check PRIVATE quicr spdlog::spdlog relay_health_check_lib) target_compile_options(relay_health_check PRIVATE diff --git a/src/client_manager.cc b/src/client_manager.cc index 7f85325..37c158b 100644 --- a/src/client_manager.cc +++ b/src/client_manager.cc @@ -12,6 +12,7 @@ #include "publish_handler.h" #include "publish_namespace_handler.h" #include "subscribe_handler.h" +#include "transport_helpers.h" #include @@ -21,10 +22,11 @@ namespace laps { const quicr::ServerConfig& cfg, peering::PeerManager& peer_manager, size_t cache_duration_ms) - : quicr::Session(cfg, config.tick_service_) - , state_(state) + : state_(state) , config_(config) + , server_config_(cfg) , peer_manager_(peer_manager) + , tick_service_(config.tick_service_) , metrics_publisher_(*this, config) , cache_duration_ms_(cache_duration_ms) { @@ -32,38 +34,340 @@ namespace laps { ClientManager::~ClientManager() { - metrics_publisher_.Stop(); + Stop(); } quicr::Session::Status ClientManager::Start() { /* - * Set up relay-local metrics track state before starting the transport thread. Session::Start() - * spins up the transport thread that dispatches all control/data callbacks; starting metrics - * publishing first avoids any chance of racing with that thread while touching ClientManager state. + * Set up relay-local metrics track state before the server transport is listening. The transport + * thread dispatches all control/data callbacks; starting metrics publishing first avoids any chance + * of racing with that thread while touching ClientManager state. */ metrics_publisher_.Start(); - return quicr::Session::Start(); + session_manager_ = std::make_unique( + std::static_pointer_cast(shared_from_this()), + tick_service_, + config_.quicr_logger_); + + session_manager_->AddTransport(server_config_, + std::static_pointer_cast(shared_from_this())); + + return quicr::Session::Status::kReady; } void ClientManager::Stop() { + /* + * Shut the session manager down before anything else so its transport thread stops dispatching + * callbacks into relay state that is about to go away. + */ + session_manager_.reset(); + metrics_publisher_.Stop(); - quicr::Session::Stop(); + + std::lock_guard _(sessions_mutex_); + sessions_.clear(); } - void ClientManager::NewConnectionAccepted(std::uint64_t connection_handle, const ConnectionRemoteInfo& remote) + std::shared_ptr ClientManager::GetSession(std::uint64_t connection_handle) const { - metrics_publisher_.AddConnection(connection_handle, remote); + std::lock_guard _(sessions_mutex_); - SPDLOG_LOGGER_INFO( - LOGGER, "New connection handle {0} accepted from {1}:{2}", connection_handle, remote.ip, remote.port); + const auto it = sessions_.find(connection_handle); + if (it == sessions_.end()) { + return nullptr; + } + + return it->second.lock(); + } + + void ClientManager::NewConnectionAccepted(const std::shared_ptr& session) + { + const auto connection_handle = ConnectionHandle(session); + + { + std::lock_guard _(sessions_mutex_); + sessions_[connection_handle] = session; + } + + // Peer address requires the owning transport; SessionManager does not expose it. See libquicr issue. + const auto [ip, port] = GetPeerAddress(nullptr, session->GetConnection()); + metrics_publisher_.AddConnection(connection_handle, { ip, port }); + + SPDLOG_LOGGER_INFO(LOGGER, "New connection handle {0} accepted from {1}:{2}", connection_handle, ip, port); + } + + void ClientManager::OnSessionRemoved(const std::shared_ptr& session) + { + const auto connection_handle = ConnectionHandle(session); + + /* + * Drop the session first so that every GetSession() lookup for this handle now fails. Relay teardown + * below walks state that still references the connection, and without this the teardown would call + * back into a session that libquicr is closing on another thread. + */ + { + std::lock_guard _(sessions_mutex_); + + if (sessions_.erase(connection_handle) == 0) { + return; + } + } + + SPDLOG_LOGGER_DEBUG(LOGGER, "Connection closed; connection_handle: {0} ", connection_handle); + + metrics_publisher_.RemoveConnection(connection_handle); + + // Remove all subscribe announces for this connection handle + std::vector remove_ns; + for (auto& [ns, conns] : state_.subscribes_namespaces) { + auto it = conns.find(connection_handle); + if (it != conns.end()) { + remove_ns.emplace_back(ns); + } + } + + for (auto ns : remove_ns) { + UnsubscribeNamespaceReceived(session, ns); + } + + // Clean up subscribe states + std::vector> unsub_list; + for (auto it = state_.subscribe_alias_req_id.lower_bound({ connection_handle, 0 }); + it != state_.subscribe_alias_req_id.end(); + it++) { + const auto& [key, _] = *it; + if (key.first != connection_handle) + break; + + unsub_list.push_back(key); + } + + for (auto& key : unsub_list) { + UnsubscribeReceived(session, key.second); + } + + // Cleanup publish states + PurgePublishState(connection_handle); + + // Drop any namespaces that were fanned out to this connection + for (auto it = fanout_namespaces_.begin(); it != fanout_namespaces_.end();) { + it = it->first.second == connection_handle ? fanout_namespaces_.erase(it) : std::next(it); + } + } + + // ------------------------------------------------------------------------------- + // Session API by connection handle + // ------------------------------------------------------------------------------- + + void ClientManager::SubscribeTrack(std::uint64_t connection_handle, + std::shared_ptr handler) + { + if (auto session = GetSession(connection_handle)) { + session->SubscribeTrack(std::move(handler)); + } } - std::vector ClientManager::PublishNamespaceDoneReceived(std::uint64_t connection_handle, - uint64_t request_id) + void ClientManager::UnsubscribeTrack(std::uint64_t connection_handle, + const std::shared_ptr& handler) { + if (auto session = GetSession(connection_handle)) { + session->UnsubscribeTrack(handler); + } + } + + void ClientManager::UpdateTrackSubscription(std::uint64_t connection_handle, + std::shared_ptr handler) + { + if (auto session = GetSession(connection_handle)) { + session->UpdateTrackSubscription(std::move(handler)); + } + } + + void ClientManager::PublishNamespace(std::uint64_t connection_handle, + std::shared_ptr handler, + bool passive) + { + if (auto session = GetSession(connection_handle)) { + session->PublishNamespace(std::move(handler), passive); + } + } + + void ClientManager::FetchTrack(std::uint64_t connection_handle, std::shared_ptr handler) + { + if (auto session = GetSession(connection_handle)) { + session->FetchTrack(std::move(handler)); + } + } + + void ClientManager::CancelFetchTrack(std::uint64_t connection_handle, + std::shared_ptr handler) + { + if (auto session = GetSession(connection_handle)) { + session->CancelFetchTrack(std::move(handler)); + } + } + + void ClientManager::BindPublisherTrack(std::uint64_t connection_handle, + std::uint64_t src_id, + uint64_t request_id, + const std::shared_ptr& handler, + bool ephemeral) + { + if (auto session = GetSession(connection_handle)) { + session->BindPublisherTrack(src_id, request_id, handler, ephemeral); + } + } + + void ClientManager::UnbindPublisherTrack(std::uint64_t connection_handle, + std::uint64_t src_id, + const std::shared_ptr& handler, + bool send_publish_done) + { + if (auto session = GetSession(connection_handle)) { + session->UnbindPublisherTrack(src_id, handler, send_publish_done); + } + } + + void ClientManager::BindFetchTrack(std::uint64_t connection_handle, + std::shared_ptr handler) + { + if (auto session = GetSession(connection_handle)) { + session->BindFetchTrack(std::move(handler)); + } + } + + void ClientManager::UnbindFetchTrack(std::uint64_t connection_handle, + const std::shared_ptr& handler) + { + if (auto session = GetSession(connection_handle)) { + session->UnbindFetchTrack(handler); + } + } + + // ------------------------------------------------------------------------------- + // SessionManager lifecycle + // ------------------------------------------------------------------------------- + + void ClientManager::OnNewServerSession(const std::shared_ptr& session) + { + NewConnectionAccepted(session); + } + + // ------------------------------------------------------------------------------- + // Session request callbacks + // ------------------------------------------------------------------------------- + + std::uint64_t ClientManager::ConnectionHandle(const std::shared_ptr& session) + { + if (session == nullptr) { + return 0; + } + + const auto& connection = session->GetConnection(); + return connection != nullptr ? connection->GetID() : 0; + } + + quicr::Reply ClientManager::NewGroupRequested(const quicr::FullTrackName& track_full_name, + std::uint64_t group_id) + { + auto th = quicr::TrackHash(track_full_name); + SPDLOG_INFO("New group requested received track_alais: {} group_id: {} ", th.track_fullname_hash, group_id); + + // Update peering subscribe info - This will update existing instead of creating new + peer_manager_.ClientSubscribeUpdate(track_full_name, + { + kDefaultPriority, + quicr::messages::GroupOrder::kAscending, + quicr::messages::GroupOrder::kAscending, + std::chrono::milliseconds(kDefaultObjectTtl), + std::chrono::milliseconds(0), + std::monostate{}, + 1, + true, + }); + + // Notify all publishers that there is a new group request + for (auto it = state_.pub_subscribes.lower_bound({ th.track_fullname_hash, 0 }); + it != state_.pub_subscribes.end(); + ++it) { + auto& track_alias = it->first.first; + auto& pub_conn_id = it->first.second; + + if (track_alias != th.track_fullname_hash) { + break; + } + + if (!it->second->GetPendingNewRquestId().has_value() || + (group_id == 0 && *it->second->GetPendingNewRquestId()) || + *it->second->GetPendingNewRquestId() < group_id) { + + it->second->SetNewGroupRequestId(group_id); + DampenOrUpdateTrackSubscription(it->second, true); + } + } + + return {}; + } + + void ClientManager::FanoutPublishNamespace(const quicr::TrackNamespace& track_namespace, + std::uint64_t exclude_connection) + { + // TODO: Fix O(prefix namespaces) matching + for (const auto& [prefix, conns] : state_.subscribes_namespaces) { + if (!prefix.HasSamePrefix(track_namespace)) { + continue; + } + + for (const auto connection_handle : std::views::keys(conns)) { + if (connection_handle == exclude_connection) { + continue; + } + + auto session = GetSession(connection_handle); + if (session == nullptr) { + continue; + } + + auto [it, is_new] = fanout_namespaces_.try_emplace({ track_namespace, connection_handle }); + if (!is_new) { + continue; + } + + it->second = quicr::PublishNamespaceHandler::Create(track_namespace); + session->PublishNamespace(it->second); + + SPDLOG_LOGGER_DEBUG(LOGGER, + "Publish namespace fanned out to prefix subscriber connection handle: {} for " + "namespace hash: {}", + connection_handle, + quicr::TrackHash({ track_namespace, {} }).track_namespace_hash); + } + } + } + + void ClientManager::FanoutPublishNamespaceDone(const quicr::TrackNamespace& track_namespace) + { + for (auto it = fanout_namespaces_.lower_bound({ track_namespace, 0 }); it != fanout_namespaces_.end();) { + if (it->first.first != track_namespace) { + break; + } + + if (auto session = GetSession(it->first.second)) { + session->PublishNamespaceDone(it->second); + } + + it = fanout_namespaces_.erase(it); + } + } + + quicr::Reply ClientManager::PublishNamespaceDoneReceived( + const std::shared_ptr& session, + std::uint64_t request_id) + { + const auto connection_handle = ConnectionHandle(session); auto it = state_.requests.find({ connection_handle, request_id }); if (it == state_.requests.end()) { @@ -86,24 +390,7 @@ namespace laps { peer_manager_.ClientAnnounce({ track_namespace, {} }, {}, true, true); - // TODO: Fix O(prefix namespaces) matching - std::vector sub_namespace_connections; - for (const auto& [ns, conns] : state_.subscribes_namespaces) { - if (!ns.HasSamePrefix(track_namespace)) { - continue; - } - - for (auto conn_handle : std::views::keys(conns)) { - SPDLOG_DEBUG("Received publish namespace done matches prefix subscribed from connection handle: {} for " - "namespace hash: {}", - conn_handle, - th.track_namespace_hash); - - sub_namespace_connections.emplace_back(conn_handle); - } - } - - ResolvePublishNamespaceDone(connection_handle, request_id, sub_namespace_connections); + FanoutPublishNamespaceDone(track_namespace); for (auto track_alias : state_.pub_namespace_active[{ track_namespace, connection_handle }]) { auto ptd = state_.pub_subscribes[{ track_alias, connection_handle }]; @@ -126,7 +413,7 @@ namespace laps { if (connection_handle) peer_manager_.ClientUnannounce({ track_namespace, {} }); - return sub_namespace_connections; + return {}; } void ClientManager::PurgePublishState(std::uint64_t connection_handle) @@ -189,10 +476,12 @@ namespace laps { } } - void ClientManager::PublishNamespaceReceived(std::uint64_t connection_handle, - const quicr::TrackNamespace& track_namespace, - const quicr::PublishNamespaceAttributes& attrs) + quicr::Reply ClientManager::PublishNamespaceReceived( + const std::shared_ptr& session, + const quicr::TrackNamespace& track_namespace, + const quicr::PublishNamespaceAttributes& attrs) { + const auto connection_handle = ConnectionHandle(session); auto [req_it, _] = state_.requests.try_emplace({ connection_handle, attrs.request_id }, State::RequestTransaction::Type::kPublishNamespace, @@ -267,34 +556,17 @@ namespace laps { */ if (connection_handle) subscribe_to_publisher(); - return; + return {}; } state_.pub_namespace_active.try_emplace(std::make_pair(track_namespace, connection_handle)); - PublishNamespaceResponse announce_response; - announce_response.reason_code = PublishNamespaceResponse::ReasonCode::kOk; - - std::vector sub_annos_connections; - - // TODO: Fix O(prefix namespaces) matching - for (const auto& [ns, conns] : state_.subscribes_namespaces) { - if (!ns.HasSamePrefix(track_namespace)) { - continue; - } - - for (auto& [conn_handle, _] : conns) { - SPDLOG_DEBUG("Received publish namespace matches prefix subscribed namespace from connection handle: " - "{} for namespace hash: {}", - conn_handle, - th.track_namespace_hash); - - sub_annos_connections.emplace_back(conn_handle); - } - } - - ResolvePublishNamespace( - connection_handle, attrs.request_id, track_namespace, sub_annos_connections, announce_response); + /* + * The announcing connection is acknowledged by libquicr once this callback returns, which also mirrors + * the namespace back to a matching prefix subscription on that same connection. Every other prefix + * subscriber is the relay's own work. + */ + FanoutPublishNamespace(track_namespace, connection_handle); if (connection_handle) { subscribe_to_publisher(); @@ -305,14 +577,19 @@ namespace laps { */ peer_manager_.ClientAnnounce({ track_namespace, {} }, attrs, false, true); } + + return {}; } - void ClientManager::PublishReceived(std::uint64_t connection_handle, - uint64_t request_id, - const quicr::PublishAttributes& publish_attributes, - [[maybe_unused]] std::weak_ptr sub_ns_handler) + quicr::Reply ClientManager::PublishReceived( + const std::shared_ptr& session, + uint64_t request_id, + const quicr::PublishAttributes& publish_attributes, + [[maybe_unused]] std::weak_ptr sub_ns_handler) { - PublishReceivedInternal(connection_handle, request_id, publish_attributes, !connection_handle && !request_id); + const auto connection_handle = ConnectionHandle(session); + return PublishReceivedInternal( + connection_handle, request_id, publish_attributes, !connection_handle && !request_id); } void ClientManager::RegisterLocalPublish(const quicr::PublishAttributes& publish_attributes) @@ -320,10 +597,11 @@ namespace laps { PublishReceivedInternal(0, 0, publish_attributes, false); } - void ClientManager::PublishReceivedInternal(std::uint64_t connection_handle, - uint64_t request_id, - const quicr::PublishAttributes& publish_attributes, - bool is_from_peer) + quicr::Reply ClientManager::PublishReceivedInternal( + std::uint64_t connection_handle, + uint64_t request_id, + const quicr::PublishAttributes& publish_attributes, + bool is_from_peer) { /* * Peer-originated publishes and relay-local publishes (e.g. the metrics track) both use the @@ -344,9 +622,6 @@ namespace laps { publish_attributes.track_full_name.NamespaceStr(), publish_attributes.track_full_name.NameStr()); - quicr::PublishResponse publish_response; - publish_response.reason_code = quicr::PublishResponse::ReasonCode::kOk; - // PublishTrack within publish namespace handler if matched for (auto& [tn, conns] : state_.subscribes_namespaces) { const auto prefix_match = tn.IsPrefixOf(publish_attributes.track_full_name.name_space); @@ -391,8 +666,8 @@ namespace laps { true); if (bypass_resolve) { - // ResolvePublish() normally sets the track alias against a real connection; set it explicitly - // since that call is skipped below. + // Accepting the publish normally binds the handler to its connection, which is what assigns the + // track alias. There is no connection here, so set it explicitly. sub_track_handler->SetTrackAlias(publish_attributes.track_alias); } @@ -409,9 +684,6 @@ namespace laps { if (!bypass_resolve) { state_.pub_subscribes_by_req_id[{ request_id, connection_handle }] = sub_track_handler; - // Do this before pause to maintain MOQT message sequence order - ResolvePublish(connection_handle, request_id, publish_attributes, publish_response, sub_track_handler); - /* * Always send publish as an announcement to peer manager so new clients can trigger subscribe matching and * data forwarding path creation. This needs to be done last to all other client work. @@ -454,23 +726,45 @@ namespace laps { sub_track_handler->SetTrackRanking(rank_it->second); } + /* + * Accept the publish, but only ask the publisher to forward data if the relay has somewhere to send it. + * The handler is not bound to the connection until the accept is processed, so Pause() cannot send the + * update itself; answering the publish with forward off does the same thing in one less message. The + * handler is still marked paused so that gaining a subscriber resumes forwarding. + */ + std::optional forward = std::nullopt; if (!has_subs) { SPDLOG_LOGGER_INFO(LOGGER, "No subscribers, pause publish connection handle: {0} using track alias: {1}", connection_handle, th.track_fullname_hash); - sub_track_handler->Pause(); + if (bypass_resolve) { + sub_track_handler->Pause(); + } else { + forward = false; + sub_track_handler->MarkPaused(); + } } state_.conn_state_metrics[connection_handle].published_tracks++; + + return quicr::PublishResponse{ { .subscriber_priority = std::nullopt, + .group_order = std::nullopt, + .filter = {}, + .forward = forward, + .subgroup_delivery_timeout = std::nullopt, + .object_delivery_timeout = std::nullopt, + .new_group_request_id = std::nullopt }, + sub_track_handler }; } - void ClientManager::SubscribeTracksReceived(std::uint64_t connection_handle, - std::uint64_t data_ctx_id, - const quicr::TrackNamespace& prefix_namespace, - const quicr::SubscribeNamespaceAttributes& attributes) + quicr::Reply, quicr::RequestErrorCode> ClientManager::SubscribeTracksReceived( + const std::shared_ptr& session, + const quicr::TrackNamespace& prefix_namespace, + const quicr::SubscribeNamespaceAttributes& attributes) { + const auto connection_handle = ConnectionHandle(session); auto th = quicr::TrackHash({ prefix_namespace, {} }); auto [it, is_new] = state_.subscribes_namespaces.try_emplace(prefix_namespace); @@ -520,11 +814,6 @@ namespace laps { } } - const quicr::SubscribeNamespaceResponse response = { .reason_code = - quicr::SubscribeNamespaceResponse::ReasonCode::kOk, - .namespaces = std::move(matched_ns) }; - ResolveSubscribeTracks(connection_handle, data_ctx_id, attributes.request_id, prefix_namespace, response); - // TODO: Need to change this to use what peering is using to prefix match instead of O(n) over all publish // subscribes for (const auto& [ta_conn, handler] : state_.pub_subscribes) { @@ -566,14 +855,18 @@ namespace laps { quicr::TrackHash(track_full_name).track_fullname_hash); } } + + return matched_ns; } - void ClientManager::UnsubscribeNamespaceReceived(std::uint64_t connection_handle, - const quicr::TrackNamespace& prefix_namespace) + quicr::Reply ClientManager::UnsubscribeNamespaceReceived( + const std::shared_ptr& session, + const quicr::TrackNamespace& prefix_namespace) { + const auto connection_handle = ConnectionHandle(session); auto it = state_.subscribes_namespaces.find(prefix_namespace); if (it == state_.subscribes_namespaces.end()) { - return; + return {}; } auto th = quicr::TrackHash({ prefix_namespace, {} }); @@ -587,7 +880,7 @@ namespace laps { if (it->second.empty()) { state_.subscribes_namespaces.erase(it); } - return; + return {}; } // Loop through publishes and remove subscribe namespace @@ -612,73 +905,26 @@ namespace laps { state_.subscribes_namespaces.erase(it); track_rankings_.erase(th.track_namespace_hash); } - } - void ClientManager::ConnectionStatusChanged(std::uint64_t connection_handle, ConnectionStatus status) - { - switch (status) { - case ConnectionStatus::kConnected: - SPDLOG_LOGGER_DEBUG(LOGGER, "Connection ready; connection_handle: {0} ", connection_handle); - return; - case ConnectionStatus::kConnecting: - return; - - case ConnectionStatus::kNotConnected: - SPDLOG_LOGGER_DEBUG(LOGGER, "Connection not connected; connection_handle: {0} ", connection_handle); - break; - case ConnectionStatus::kClosedByRemote: - SPDLOG_LOGGER_DEBUG(LOGGER, "Connection closed by remote; connection_handle: {0} ", connection_handle); - break; - case ConnectionStatus::kIdleTimeout: - SPDLOG_LOGGER_DEBUG(LOGGER, "Connection idle timeout; connection_handle: {0} ", connection_handle); - break; - } - - metrics_publisher_.RemoveConnection(connection_handle); - - // Remove all subscribe announces for this connection handle - std::vector remove_ns; - for (auto& [ns, conns] : state_.subscribes_namespaces) { - auto it = conns.find(connection_handle); - if (it != conns.end()) { - remove_ns.emplace_back(ns); - } - } - - for (auto ns : remove_ns) { - UnsubscribeNamespaceReceived(connection_handle, ns); - } - - // Clean up subscribe states - std::vector> unsub_list; - for (auto it = state_.subscribe_alias_req_id.lower_bound({ connection_handle, 0 }); - it != state_.subscribe_alias_req_id.end(); - it++) { - const auto& [key, _] = *it; - if (key.first != connection_handle) - break; - - unsub_list.push_back(key); - } - - for (auto& key : unsub_list) { - UnsubscribeReceived(key.first, key.second); - } - - // Cleanup publish states - PurgePublishState(connection_handle); + return {}; } - void ClientManager::ClientSetupReceived(std::uint64_t connection_handle, - const quicr::ClientSetupAttributes& client_setup_attributes) + quicr::Reply ClientManager::ClientSetupReceived( + const std::shared_ptr& session, + const quicr::ClientSetupAttributes& client_setup_attributes) { + const auto connection_handle = ConnectionHandle(session); metrics_publisher_.SetConnectionEndpointId(connection_handle, client_setup_attributes.endpoint_id); SPDLOG_LOGGER_INFO(LOGGER, "Client setup received from endpoint_id: {0}", client_setup_attributes.endpoint_id); + + return {}; } - void ClientManager::PublishDoneReceived(std::uint64_t connection_handle, uint64_t request_id) + quicr::Reply ClientManager::PublishDoneReceived(const std::shared_ptr& session, + uint64_t request_id) { + const auto connection_handle = ConnectionHandle(session); SPDLOG_LOGGER_INFO( LOGGER, "Publish Done connection handle: {0} request_id: {1}", connection_handle, request_id); @@ -688,7 +934,7 @@ namespace laps { "Unable to find subscribe by request id for connection handle: {0} request_id: {1}", connection_handle, request_id); - return; + return {}; } if (connection_handle) @@ -746,7 +992,9 @@ namespace laps { lock.unlock(); for (auto& [c_handle, req_id] : unsub_list) { - UnsubscribeReceived(c_handle, req_id); + if (auto sub_session = GetSession(c_handle)) { + UnsubscribeReceived(sub_session, req_id); + } } lock.lock(); @@ -759,17 +1007,21 @@ namespace laps { if (!have_publishers) { publisher_handler->RemoveFromTrackRanking(); } + + return {}; } - void ClientManager::UnsubscribeReceived(std::uint64_t connection_handle, uint64_t request_id) + quicr::Reply ClientManager::UnsubscribeReceived(const std::shared_ptr& session, + uint64_t request_id) { + const auto connection_handle = ConnectionHandle(session); SPDLOG_LOGGER_INFO(LOGGER, "Unsubscribe connection handle: {0} request_id: {1}", connection_handle, request_id); const auto ta_it = state_.subscribe_alias_req_id.find({ connection_handle, request_id }); if (ta_it == state_.subscribe_alias_req_id.end()) { SPDLOG_WARN( "Unable to find track alias for connection handle: {0} request_id: {1}", connection_handle, request_id); - return; + return {}; } std::lock_guard _(state_.state_mutex); @@ -782,7 +1034,7 @@ namespace laps { "Unsubscribe unable to find track handler for connection handle: {0} request_id: {1}", connection_handle, request_id); - return; + return {}; } auto& metric_sub_count = state_.conn_state_metrics[connection_handle].subscribed_tracks; @@ -819,6 +1071,8 @@ namespace laps { state_.subscribes.erase(sub_it); RemoveOrPausePublisherSubscribe(th.track_fullname_hash); + + return {}; } void ClientManager::RemoveOrPausePublisherSubscribe(std::uint64_t track_fullname_hash) @@ -872,10 +1126,12 @@ namespace laps { } } - void ClientManager::TrackStatusReceived(std::uint64_t connection_handle, - uint64_t request_id, - const quicr::FullTrackName& track_full_name) + quicr::Reply ClientManager::TrackStatusReceived( + const std::shared_ptr& session, + uint64_t request_id, + const quicr::FullTrackName& track_full_name) { + const auto connection_handle = ConnectionHandle(session); auto th = quicr::TrackHash(track_full_name); SPDLOG_LOGGER_INFO(LOGGER, @@ -895,44 +1151,28 @@ namespace laps { } if (it->first.second != connection_handle) { - ResolveTrackStatus(connection_handle, - request_id, - { - quicr::RequestResponse::ReasonCode::kOk, - it->second->IsPublisherInitiated(), - std::nullopt, - largest, - }); - return; + return quicr::RequestResponse{ + .is_publisher_initiated = it->second->IsPublisherInitiated(), + .largest_location = largest, + }; } } - ResolveTrackStatus(connection_handle, - request_id, - { - quicr::RequestResponse::ReasonCode::kDoesNotExist, - false, - "Track does not exist", - std::nullopt, - }); + return quicr::Unexpected>(quicr::RequestErrorCode::kDoesNotExist, + "Track does not exist"); } - void ClientManager::SubscribeReceived(std::uint64_t connection_handle, - uint64_t request_id, - const quicr::FullTrackName& track_full_name, - const quicr::SubscribeAttributes& attrs) + quicr::Reply ClientManager::SubscribeReceived( + const std::shared_ptr& session, + uint64_t request_id, + const quicr::FullTrackName& track_full_name, + const quicr::SubscribeAttributes& attrs) { + const auto connection_handle = ConnectionHandle(session); auto th = quicr::TrackHash(track_full_name); auto it = state_.subscribes.find({ th.track_fullname_hash, connection_handle }); if (it != state_.subscribes.end()) { - ResolveSubscribe(connection_handle, - request_id, - th.track_fullname_hash, - { .reason_code = quicr::RequestResponse::ReasonCode::kNotSupported, - .is_publisher_initiated = false, - .error_reason = "Duplicate subscribe" }); - SPDLOG_LOGGER_INFO( LOGGER, "Duplicate subscribe connection handle: {0} request_id: {1} track alias: {2} priority: {3}", @@ -941,7 +1181,8 @@ namespace laps { th.track_fullname_hash, attrs.priority); - return; + return quicr::Unexpected>( + quicr::RequestErrorCode::kDuplicateSubscription, "Duplicate subscribe"); } SPDLOG_LOGGER_INFO(LOGGER, @@ -956,18 +1197,18 @@ namespace laps { track_full_name.NameStr()); auto largest = GetLargestAvailable(track_full_name); - if (largest.has_value()) { - ResolveSubscribe( - connection_handle, - request_id, - th.track_fullname_hash, - { quicr::RequestResponse::ReasonCode::kOk, attrs.is_publisher_initiated, std::nullopt, largest }); - } else { - ResolveSubscribe( - connection_handle, request_id, th.track_fullname_hash, { quicr::RequestResponse::ReasonCode::kOk }); - } + /* + * libquicr sends the SUBSCRIBE_OK once this returns, so the subscriber cannot be forwarded anything + * before it is acknowledged: the work below only arranges for forwarding, and the objects themselves + * are not available until the publisher sends them. + */ ProcessSubscribe(connection_handle, request_id, th, track_full_name, attrs, largest); + + return quicr::RequestResponse{ + .is_publisher_initiated = attrs.is_publisher_initiated, + .largest_location = largest, + }; } std::optional ClientManager::GetLargestAvailable(const quicr::FullTrackName& track_name) @@ -992,51 +1233,39 @@ namespace laps { return quicr::messages::Location{ .group = largest_group_id.value(), .object = largest_object_id.value() }; } - void ClientManager::FetchReceived(std::uint64_t connection_handle, - uint64_t request_id, - const quicr::FullTrackName& track_full_name, - uint8_t priority, - std::optional group_order, - quicr::messages::Location start, - quicr::messages::FetchEndLocation end) + quicr::Reply ClientManager::FetchReceived( + std::uint64_t connection_handle, + uint64_t request_id, + const quicr::FullTrackName& track_full_name, + uint8_t priority, + std::optional group_order, + quicr::messages::Location start, + quicr::messages::FetchEndLocation end) { - auto reason_code = quicr::FetchResponse::ReasonCode::kOk; - std::optional largest_location = std::nullopt; - auto th = quicr::TrackHash(track_full_name); + using Reply = quicr::Reply; + using Result = Reply::ResultType; + using Failure = quicr::Unexpected>; - auto cache_entry_it = cache_.find(th.track_fullname_hash); - if (cache_entry_it != cache_.end()) { - auto& [_, cache] = *cache_entry_it; - if (const auto& latest_group = cache.Last(); latest_group && !latest_group->empty()) { - const auto& latest_object = *std::prev(latest_group->end()); - largest_location = { latest_object.headers.group_id, latest_object.headers.object_id }; - } - } + const auto th = quicr::TrackHash(track_full_name); + const auto resolved_group_order = group_order.value_or(quicr::messages::GroupOrder::kAscending); + + const auto largest_location = GetLargestAvailable(track_full_name); + + const auto cache_entry_it = cache_.find(th.track_fullname_hash); + auto cache_entries = cache_entry_it != cache_.end() ? cache_entry_it->second.Get(start.group, end.group) + : std::vector>>{}; + // Why the cache cannot answer the fetch, which is the error reported if the publisher cannot either + std::optional cache_miss = std::nullopt; if (!largest_location.has_value()) { // TODO: This changes to send an empty object instead of REQUEST_ERROR - reason_code = quicr::FetchResponse::ReasonCode::kNoObjects; + cache_miss = quicr::FetchErrorCode::kNoObjects; } else if (start.group > end.group || largest_location->group < start.group) { - reason_code = quicr::FetchResponse::ReasonCode::kInvalidRange; - } - - const auto& cache_entries = cache_entry_it != cache_.end() - ? cache_entry_it->second.Get(start.group, end.group) - : std::vector>>{}; - - if (cache_entries.empty() && reason_code == quicr::FetchResponse::ReasonCode::kOk) { - reason_code = quicr::FetchResponse::ReasonCode::kNoObjects; + cache_miss = quicr::FetchErrorCode::kInvalidRange; + } else if (cache_entries.empty()) { + cache_miss = quicr::FetchErrorCode::kNoObjects; } - const auto resolved_group_order = group_order.value_or(quicr::messages::GroupOrder::kAscending); - - // TODO: Adjust the TTL to allow more time for transmission - auto pub_fetch_h = quicr::PublishFetchHandler::Create( - track_full_name, priority, request_id, resolved_group_order, config_.object_ttl_); - BindFetchTrack(connection_handle, pub_fetch_h); - - stop_fetch_.try_emplace({ connection_handle, request_id }, false); - SPDLOG_LOGGER_DEBUG(LOGGER, "Fetch received conn_id: {} request_id: {} range start group: {} start object: {} end " "group: {} end object: {} largest_location: {}", @@ -1048,13 +1277,20 @@ namespace laps { end.object.value_or(0), largest_location.has_value() ? largest_location.value().group : 0); - std::thread retrieve_cache_thread([=, cache_entries = std::move(cache_entries), this] { - defer(UnbindFetchTrack(connection_handle, pub_fetch_h)); + // TODO: Adjust the TTL to allow more time for transmission + auto pub_fetch_h = quicr::PublishFetchHandler::Create( + track_full_name, priority, request_id, resolved_group_order, config_.object_ttl_); + BindFetchTrack(connection_handle, pub_fetch_h); - auto rc = reason_code; + stop_fetch_.try_emplace({ connection_handle, request_id }, false); - if (rc != quicr::FetchResponse::ReasonCode::kOk) { - // Try to see if original publisher can provide the data + if (cache_miss.has_value()) { + /* + * Nothing usable is cached, so the publisher is asked for the range instead. Whether the fetch can + * be answered at all is only known once the publisher responds, so the reply is deferred to keep + * that wait off the session's thread. + */ + return Reply::Defer([=, this]() -> Result { auto track_handler = FetchTrackHandler::Create(pub_fetch_h, track_full_name, priority, @@ -1062,91 +1298,85 @@ namespace laps { { .group = end.group, .object = end.object }, resolved_group_order); - std::uint64_t pub_connection_handle = 0; - // Find the publisher connection handle to send the fetch request // TODO: Add peering support + std::uint64_t pub_connection_handle = 0; { std::lock_guard _(state_.state_mutex); - for (auto it = state_.pub_subscribes.lower_bound({ th.track_fullname_hash, 0 }); - it != state_.pub_subscribes.end(); - ++it) { - auto& track_alias = it->first.first; - auto& pub_conn_id = it->first.second; - - if (track_alias != th.track_fullname_hash) { - rc = quicr::FetchResponse::ReasonCode::kNoObjects; - break; - } - - pub_connection_handle = pub_conn_id; - break; // TODO: Support multiple publishers + + const auto it = state_.pub_subscribes.lower_bound({ th.track_fullname_hash, 0 }); + if (it != state_.pub_subscribes.end() && it->first.first == th.track_fullname_hash) { + pub_connection_handle = it->first.second; // TODO: Support multiple publishers } } - if (pub_connection_handle) { - SPDLOG_LOGGER_DEBUG(LOGGER, - "Fetch received conn_id: {} request_id: {}, sending to publisher conn_id: {}", - connection_handle, - request_id, - pub_connection_handle); - FetchTrack(pub_connection_handle, track_handler); - - for (int to = 0; to < kFetchUpstreamMaxWaitMs; to += 5) { - if (track_handler->GetStatus() != quicr::FetchTrackHandler::Status::kPendingResponse && - track_handler->GetStatus() != quicr::FetchTrackHandler::Status::kNotSubscribed) { - break; - } - std::this_thread::sleep_for(std::chrono::milliseconds(5)); + if (pub_connection_handle == 0) { + UnbindFetchTrack(connection_handle, pub_fetch_h); + return Failure(*cache_miss, "Cannot process fetch"); + } + + SPDLOG_LOGGER_DEBUG(LOGGER, + "Fetch received conn_id: {} request_id: {}, sending to publisher conn_id: {}", + connection_handle, + request_id, + pub_connection_handle); + FetchTrack(pub_connection_handle, track_handler); + + for (int to = 0; to < kFetchUpstreamMaxWaitMs; to += 5) { + if (track_handler->GetStatus() != quicr::FetchTrackHandler::Status::kPendingResponse && + track_handler->GetStatus() != quicr::FetchTrackHandler::Status::kNotSubscribed) { + break; } + std::this_thread::sleep_for(std::chrono::milliseconds(5)); + } + + /* + * The publisher keeps forwarding through pub_fetch_h after this reply is answered, so its + * teardown waits on the upstream fetch from a thread of its own. + */ + std::thread([=, this] { + defer(UnbindFetchTrack(connection_handle, pub_fetch_h)); - switch (track_handler->GetStatus()) { - case quicr::FetchTrackHandler::Status::kDoneByFin: - [[fallthrough]]; - case quicr::FetchTrackHandler::Status::kOk: - rc = quicr::FetchResponse::ReasonCode::kOk; - break; - case quicr::FetchTrackHandler::Status::kError: - rc = quicr::FetchResponse::ReasonCode::kNoObjects; - break; - default: - rc = quicr::FetchResponse::ReasonCode::kInternalError; - break; + while (track_handler->GetStatus() == quicr::FetchTrackHandler::Status::kOk) { + std::this_thread::sleep_for(std::chrono::milliseconds(2)); } + + std::this_thread::sleep_for(std::chrono::milliseconds(2000)); + CancelFetchTrack(pub_connection_handle, track_handler); + }).detach(); + + switch (track_handler->GetStatus()) { + case quicr::FetchTrackHandler::Status::kDoneByFin: + [[fallthrough]]; + case quicr::FetchTrackHandler::Status::kOk: + break; + + case quicr::FetchTrackHandler::Status::kError: + return Failure(quicr::FetchErrorCode::kNoObjects, "Cannot process fetch"); + + default: + return Failure(quicr::FetchErrorCode::kInternalError, "Cannot process fetch"); } - ResolveFetch(connection_handle, - request_id, - priority, - group_order, - { - rc, - rc == quicr::FetchResponse::ReasonCode::kOk ? std::nullopt - : std::make_optional("Cannot process fetch"), - track_handler->GetLatestLocation(), - }); - - while (track_handler->GetStatus() == quicr::FetchTrackHandler::Status::kOk) { - std::this_thread::sleep_for(std::chrono::milliseconds(2)); + const auto upstream_largest = track_handler->GetLatestLocation(); + if (!upstream_largest.has_value()) { + return Failure(quicr::FetchErrorCode::kNoObjects, "Cannot process fetch"); } - std::this_thread::sleep_for(std::chrono::milliseconds(2000)); - CancelFetchTrack(pub_connection_handle, track_handler); - return; - } + return quicr::FetchResponse{ upstream_largest, resolved_group_order }; + }); + } - SPDLOG_LOGGER_DEBUG( - LOGGER, "Fetch received conn_id: {} request_id: {}, using cache", connection_handle, request_id); - ResolveFetch( - connection_handle, - request_id, - priority, - group_order, - { - rc, - rc == quicr::FetchResponse::ReasonCode::kOk ? std::nullopt : std::make_optional("Cannot process fetch"), - largest_location, - }); + SPDLOG_LOGGER_DEBUG( + LOGGER, "Fetch received conn_id: {} request_id: {}, using cache", connection_handle, request_id); + + /* + * The cached objects are sent from their own thread so that the reply below can be answered right away, + * which is what sends the FETCH_OK. The objects go out on a stream of their own, so they are not + * ordered against that FETCH_OK either way. + */ + std::thread retrieve_cache_thread([=, cache_entries = std::move(cache_entries), this] { + defer(UnbindFetchTrack(connection_handle, pub_fetch_h)); const auto publish_group = [&](const std::set& group) { for (const auto& object : group) { @@ -1197,41 +1427,39 @@ namespace laps { }); retrieve_cache_thread.detach(); + + return quicr::FetchResponse{ largest_location, resolved_group_order }; } - void ClientManager::StandaloneFetchReceived(std::uint64_t connection_handle, - uint64_t request_id, - const quicr::FullTrackName& track_full_name, - const quicr::StandaloneFetchAttributes& attributes) + quicr::Reply ClientManager::StandaloneFetchReceived( + const std::shared_ptr& session, + uint64_t request_id, + const quicr::FullTrackName& track_full_name, + const quicr::StandaloneFetchAttributes& attributes) { - FetchReceived(connection_handle, - request_id, - track_full_name, - attributes.priority, - attributes.group_order, - attributes.start_location, - attributes.end_location); + const auto connection_handle = ConnectionHandle(session); + return FetchReceived(connection_handle, + request_id, + track_full_name, + attributes.priority, + attributes.group_order, + attributes.start_location, + attributes.end_location); } - void ClientManager::JoiningFetchReceived(std::uint64_t connection_handle, - uint64_t request_id, - const quicr::FullTrackName& track_full_name, - const quicr::JoiningFetchAttributes& attributes) + quicr::Reply ClientManager::JoiningFetchReceived( + const std::shared_ptr& session, + uint64_t request_id, + const quicr::FullTrackName& track_full_name, + const quicr::JoiningFetchAttributes& attributes) { - std::optional largest_location = GetLargestAvailable(track_full_name); + const auto connection_handle = ConnectionHandle(session); + const auto largest_location = GetLargestAvailable(track_full_name); // No largest location is an error. if (!largest_location.has_value()) { - ResolveFetch(connection_handle, - request_id, - attributes.priority, - attributes.group_order, - { - quicr::FetchResponse::ReasonCode::kInvalidRange, - "No objects available for joining fetch", - std::nullopt, - }); - return; + return quicr::Unexpected>(quicr::FetchErrorCode::kInvalidRange, + "No objects available for joining fetch"); } uint64_t joining_start = 0; @@ -1242,60 +1470,26 @@ namespace laps { joining_start = attributes.joining_start; } - FetchReceived(connection_handle, - request_id, - track_full_name, - attributes.priority, - attributes.group_order, - { joining_start, 0 }, - { largest_location->group, std::nullopt }); + return FetchReceived(connection_handle, + request_id, + track_full_name, + attributes.priority, + attributes.group_order, + { joining_start, 0 }, + { largest_location->group, std::nullopt }); } - void ClientManager::FetchCancelReceived(std::uint64_t connection_handle, uint64_t request_id) + quicr::Reply ClientManager::FetchCancelReceived( + const std::shared_ptr& session, + uint64_t request_id) { + const auto connection_handle = ConnectionHandle(session); SPDLOG_INFO("Canceling fetch for connection_handle: {} request_id: {}", connection_handle, request_id); if (stop_fetch_.count({ connection_handle, request_id }) == 0) stop_fetch_[{ connection_handle, request_id }] = true; - } - - void ClientManager::NewGroupRequested(const quicr::FullTrackName& track_full_name, std::uint64_t group_id) - { - auto th = quicr::TrackHash(track_full_name); - SPDLOG_INFO("New group requested received track_alais: {} group_id: {} ", th.track_fullname_hash, group_id); - - // Update peering subscribe info - This will update existing instead of creating new - peer_manager_.ClientSubscribeUpdate(track_full_name, - { - kDefaultPriority, - quicr::messages::GroupOrder::kAscending, - quicr::messages::GroupOrder::kAscending, - std::chrono::milliseconds(kDefaultObjectTtl), - std::chrono::milliseconds(0), - std::monostate{}, - 1, - true, - }); - - // Notify all publishers that there is a new group request - for (auto it = state_.pub_subscribes.lower_bound({ th.track_fullname_hash, 0 }); - it != state_.pub_subscribes.end(); - ++it) { - auto& track_alias = it->first.first; - auto& pub_conn_id = it->first.second; - if (track_alias != th.track_fullname_hash) { - break; - } - - if (!it->second->GetPendingNewRquestId().has_value() || - (group_id == 0 && *it->second->GetPendingNewRquestId()) || - *it->second->GetPendingNewRquestId() < group_id) { - - it->second->SetNewGroupRequestId(group_id); - DampenOrUpdateTrackSubscription(it->second, true); - } - } + return {}; } bool ClientManager::DampenOrUpdateTrackSubscription(std::shared_ptr sub_to_pub_track_handler, @@ -1539,8 +1733,10 @@ namespace laps { it->second->StreamClosed(stream_id, reset); } - void ClientManager::MetricsSampled(const std::uint64_t connection_handle, const quicr::ConnectionMetrics& metrics) + void ClientManager::MetricsSampled(const std::shared_ptr& session, + const quicr::ConnectionMetrics& metrics) { + const auto connection_handle = ConnectionHandle(session); const auto publish_track_count = state_.PublishTrackCount(connection_handle); const auto subscribe_track_count = state_.SubscribeTrackCount(connection_handle); metrics_publisher_.QueueConnectionMetrics( diff --git a/src/client_manager.h b/src/client_manager.h index 9d72112..640f422 100644 --- a/src/client_manager.h +++ b/src/client_manager.h @@ -1,5 +1,6 @@ #pragma once +#include "client_manager.h" #include "metrics_publisher.h" #include "state.h" @@ -8,9 +9,13 @@ #include #include #include +#include +#include #include #include +#include +#include #include namespace laps { @@ -50,7 +55,10 @@ namespace laps { * @brief MoQ Server * @details Implementation of the MoQ Server */ - class ClientManager : public quicr::Session + class ClientManager + : public quicr::Session::ServerCallbacks + , public quicr::SessionManager::Callbacks + , public std::enable_shared_from_this { public: ClientManager(State& state, @@ -60,70 +68,58 @@ namespace laps { size_t cache_duration_ms = 60000); ~ClientManager(); - quicr::Session::Status Start() override; - void Stop() override; + quicr::Session::Status Start(); + void Stop(); - void NewConnectionAccepted(std::uint64_t connection_handle, const ConnectionRemoteInfo& remote) override; - - void SubscribeTracksReceived(std::uint64_t connection_handle, - std::uint64_t data_ctx_id, - const quicr::TrackNamespace& prefix_namespace, - const quicr::SubscribeNamespaceAttributes& attributes) override; - - void UnsubscribeNamespaceReceived(std::uint64_t connection_handle, - const quicr::TrackNamespace& prefix_namespace) override; - - std::vector PublishNamespaceDoneReceived(std::uint64_t connection_handle, - std::uint64_t request_id) override; + std::optional GetLargestAvailable(const quicr::FullTrackName& track_name); - void PublishNamespaceReceived(std::uint64_t connection_handle, - const quicr::TrackNamespace& track_namespace, - const quicr::PublishNamespaceAttributes&) override; + // ------------------------------------------------------------------------------- + // Session API by connection handle + // + // libquicr owns one quicr::Session per connection. These forward to the session for the given + // connection handle so relay code can stay connection-explicit. + // ------------------------------------------------------------------------------- - void ConnectionStatusChanged(std::uint64_t connection_handle, ConnectionStatus status) override; + void SubscribeTrack(std::uint64_t connection_handle, std::shared_ptr handler); - void ClientSetupReceived(std::uint64_t connection_handle, - const quicr::ClientSetupAttributes& client_setup_attributes) override; + void UnsubscribeTrack(std::uint64_t connection_handle, + const std::shared_ptr& handler); - void UnsubscribeReceived(std::uint64_t connection_handle, uint64_t request_id) override; - void PublishDoneReceived(std::uint64_t connection_handle, uint64_t request_id) override; + void UpdateTrackSubscription(std::uint64_t connection_handle, + std::shared_ptr handler); - void SubscribeReceived(std::uint64_t connection_handle, - uint64_t request_id, - const quicr::FullTrackName& track_full_name, - const quicr::SubscribeAttributes&) override; + void PublishNamespace(std::uint64_t connection_handle, + std::shared_ptr handler, + bool passive = false); - void NewGroupRequested(const quicr::FullTrackName& track_full_name, std::uint64_t group_id) override; + void FetchTrack(std::uint64_t connection_handle, std::shared_ptr handler); - void TrackStatusReceived(std::uint64_t connection_handle, - uint64_t request_id, - const quicr::FullTrackName& track_full_name) override; + void CancelFetchTrack(std::uint64_t connection_handle, std::shared_ptr handler); - std::optional GetLargestAvailable(const quicr::FullTrackName& track_name); + void BindPublisherTrack(std::uint64_t connection_handle, + std::uint64_t src_id, + uint64_t request_id, + const std::shared_ptr& handler, + bool ephemeral = false); - void FetchCancelReceived(std::uint64_t connection_handle, uint64_t request_id) override; + void UnbindPublisherTrack(std::uint64_t connection_handle, + std::uint64_t src_id, + const std::shared_ptr& handler, + bool send_publish_done = false); - void StandaloneFetchReceived(std::uint64_t connection_handle, - uint64_t request_id, - const quicr::FullTrackName& track_full_name, - const quicr::StandaloneFetchAttributes& attributes) override; + void BindFetchTrack(std::uint64_t connection_handle, std::shared_ptr handler); - void JoiningFetchReceived(std::uint64_t connection_handle, - uint64_t request_id, - const quicr::FullTrackName& track_full_name, - const quicr::JoiningFetchAttributes& attributes) override; + void UnbindFetchTrack(std::uint64_t connection_handle, + const std::shared_ptr& handler); - void PublishReceived(std::uint64_t connection_handle, - uint64_t request_id, - const quicr::PublishAttributes& publish_attributes, - std::weak_ptr ns_handler) override; + const std::shared_ptr& GetTickService() const noexcept { return tick_service_; } /** * @brief Register a relay-local publish (e.g. the internally generated metrics track) * * @details Unlike PublishReceived(), this is never treated as peer-originated even though it uses - * the same connection_handle=0/request_id=0 sentinel internally. Use this instead of calling - * PublishReceived(0, 0, ...) directly for relay-local publishes. + * a nullptr session and request_id=0 internally. Use this instead of calling + * PublishReceived(nullptr, 0, ...) directly for relay-local publishes. */ void RegisterLocalPublish(const quicr::PublishAttributes& publish_attributes); @@ -152,29 +148,153 @@ namespace laps { void RemoveOrPausePublisherSubscribe(std::uint64_t track_fullname_hash); - void MetricsSampled(const std::uint64_t connection_handle, const quicr::ConnectionMetrics& metrics) override; + friend class peering::PeerManager; private: - void PublishReceivedInternal(std::uint64_t connection_handle, - uint64_t request_id, - const quicr::PublishAttributes& publish_attributes, - bool is_from_peer); + /** + * @brief Connection handle of the session that received a request, or zero if it is already gone + */ + static std::uint64_t ConnectionHandle(const std::shared_ptr& session); + + // -- quicr::Session::ServerCallbacks ------------------------------------------------------- + + void MetricsSampled(const std::shared_ptr& session, + const quicr::ConnectionMetrics& metrics) override; + + quicr::Reply ClientSetupReceived( + const std::shared_ptr& session, + const quicr::ClientSetupAttributes& client_setup_attributes) override; + + quicr::Reply, quicr::RequestErrorCode> SubscribeTracksReceived( + const std::shared_ptr& session, + const quicr::TrackNamespace& prefix_namespace, + const quicr::SubscribeNamespaceAttributes& attributes) override; + + quicr::Reply UnsubscribeNamespaceReceived( + const std::shared_ptr& session, + const quicr::TrackNamespace& prefix_namespace) override; + + quicr::Reply PublishNamespaceReceived( + const std::shared_ptr& session, + const quicr::TrackNamespace& track_namespace, + const quicr::PublishNamespaceAttributes& attributes) override; + + quicr::Reply PublishNamespaceDoneReceived( + const std::shared_ptr& session, + std::uint64_t request_id) override; + + quicr::Reply PublishReceived( + const std::shared_ptr& session, + std::uint64_t request_id, + const quicr::PublishAttributes& publish_attributes, + std::weak_ptr sub_ns_handler) override; + + quicr::Reply PublishDoneReceived(const std::shared_ptr& session, + std::uint64_t request_id) override; + + quicr::Reply SubscribeReceived( + const std::shared_ptr& session, + std::uint64_t request_id, + const quicr::FullTrackName& track_full_name, + const quicr::SubscribeAttributes& subscribe_attributes) override; + + quicr::Reply UnsubscribeReceived(const std::shared_ptr& session, + std::uint64_t request_id) override; + + quicr::Reply NewGroupRequested(const quicr::FullTrackName& track_full_name, + std::uint64_t group_id) override; + + quicr::Reply TrackStatusReceived( + const std::shared_ptr& session, + std::uint64_t request_id, + const quicr::FullTrackName& track_full_name) override; + + quicr::Reply StandaloneFetchReceived( + const std::shared_ptr& session, + std::uint64_t request_id, + const quicr::FullTrackName& track_full_name, + const quicr::StandaloneFetchAttributes& attributes) override; + + quicr::Reply JoiningFetchReceived( + const std::shared_ptr& session, + std::uint64_t request_id, + const quicr::FullTrackName& track_full_name, + const quicr::JoiningFetchAttributes& attributes) override; + + quicr::Reply FetchCancelReceived(const std::shared_ptr& session, + std::uint64_t request_id) override; + + // -- quicr::SessionManager::Callbacks ------------------------------------------------------ + + void OnNewServerSession(const std::shared_ptr& new_session) override; + + void OnSessionRemoved(const std::shared_ptr& session) override; + + /** + * @brief Session for a connection handle, or nullptr when the connection is gone + */ + std::shared_ptr GetSession(std::uint64_t connection_handle) const; + + /** + * @brief Set up relay state for a newly accepted connection + */ + void NewConnectionAccepted(const std::shared_ptr& session); + + /** + * @brief Send a publish namespace to every connection that subscribed to a matching prefix + * + * @details Accepting the announcement only mirrors the namespace back on the announcing connection, so + * the relay fans it out to the other prefix subscribers itself. + * + * @param track_namespace Namespace that was announced + * @param exclude_connection Announcing connection, which libquicr handles when it accepts + */ + void FanoutPublishNamespace(const quicr::TrackNamespace& track_namespace, std::uint64_t exclude_connection); + + /** + * @brief Withdraw a fanned out publish namespace from every connection it was sent to + */ + void FanoutPublishNamespaceDone(const quicr::TrackNamespace& track_namespace); + + quicr::Reply PublishReceivedInternal( + std::uint64_t connection_handle, + uint64_t request_id, + const quicr::PublishAttributes& publish_attributes, + bool is_from_peer); void PurgePublishState(std::uint64_t connection_handle); - void FetchReceived(std::uint64_t connection_handle, - uint64_t request_id, - const quicr::FullTrackName& track_full_name, - uint8_t priority, - std::optional group_order, - quicr::messages::Location start, - quicr::messages::FetchEndLocation end); + quicr::Reply FetchReceived( + std::uint64_t connection_handle, + uint64_t request_id, + const quicr::FullTrackName& track_full_name, + uint8_t priority, + std::optional group_order, + quicr::messages::Location start, + quicr::messages::FetchEndLocation end); State& state_; const Config& config_; + const quicr::ServerConfig server_config_; peering::PeerManager& peer_manager_; + std::shared_ptr tick_service_; MetricsPublisher metrics_publisher_; + std::unique_ptr session_manager_; + + /** + * @brief Weak references to sessions by connection handle + * + * @details SessionManager owns the sessions. The relay only keeps weak references so it can look up a + * session for a connection it is not currently handling a request for. + */ + mutable std::mutex sessions_mutex_; + std::map> sessions_; + + /// Fanned out publish namespaces, key is announced namespace and the subscriber connection handle + std::map, std::shared_ptr> + fanout_namespaces_; + /** * @brief Map of atomic bools to mark if a fetch thread should be interrupted. */ diff --git a/src/config.cc b/src/config.cc index 76eeabb..5d98ef0 100644 --- a/src/config.cc +++ b/src/config.cc @@ -11,6 +11,7 @@ namespace laps { Config::Config() : logger_(spdlog::stderr_color_mt("lapsRelay")) + , quicr_logger_(std::make_shared(logger_)) , tick_service_(std::make_shared()) { InitDefaults(); diff --git a/src/config.h b/src/config.h index 5df368e..c9b0b31 100644 --- a/src/config.h +++ b/src/config.h @@ -1,11 +1,13 @@ #pragma once #include "peering/messages/node_info.h" -#include "spdlog/sinks/stdout_color_sinks-inl.h" +#include "spdlog_logger.h" #include "version_config.h" +#include #include #include +#include #include #include @@ -27,6 +29,9 @@ namespace laps { public: std::shared_ptr logger_ = spdlog::stderr_color_mt("LAPS"); + /// Logger handed to libquicr, which logs through its own interface + std::shared_ptr quicr_logger_; + quicr::ServerConfig server_config; bool debug{ false }; /// Debug logging/code diff --git a/src/fetch_handler.h b/src/fetch_handler.h index 1304cec..1955136 100644 --- a/src/fetch_handler.h +++ b/src/fetch_handler.h @@ -2,7 +2,6 @@ #include "client_manager.h" -#include #include namespace laps { diff --git a/src/main.cc b/src/main.cc index d063d0d..f463542 100644 --- a/src/main.cc +++ b/src/main.cc @@ -121,7 +121,6 @@ InitConfig(cxxopts::ParseResult& cli_opts, Config& cfg) config.transport_config.debug = cfg.debug; config.transport_config.tls_cert_filename = cfg.tls_cert_filename_; config.transport_config.tls_key_filename = cfg.tls_key_filename_; - config.transport_config.use_reset_wait_strategy = false; config.transport_config.quic_qlog_path = qlog_path; config.transport_config.idle_timeout_ms = 10000; config.transport_config.time_queue_rx_size = 10'000; @@ -207,8 +206,12 @@ main(int argc, char* argv[]) // Wait until told to terminate gvars::cv.wait(lock, [&]() { return gvars::terminate; }); - // Unlock the mutex lock.unlock(); + + // Stop transport before releasing external shared_ptr refs; SessionManager holds a + // shared_ptr back to ClientManager via callbacks and would otherwise prevent destruction. + server->Stop(); + peer_manager.SetClientManager(nullptr); } catch (const std::invalid_argument& e) { std::cerr << "Invalid argument: " << e.what() << std::endl; result_code = EXIT_FAILURE; diff --git a/src/metrics_publisher.cc b/src/metrics_publisher.cc index f006125..5c3693e 100644 --- a/src/metrics_publisher.cc +++ b/src/metrics_publisher.cc @@ -157,7 +157,6 @@ namespace laps { AppendUintField(out, first, "tx_dgram_ack", metrics.tx_dgram_ack); AppendUintField(out, first, "tx_dgram_lost", metrics.tx_dgram_lost); AppendUintField(out, first, "tx_dgram_spurious", metrics.tx_dgram_spurious); - AppendUintField(out, first, "tx_dgram_drops", metrics.tx_dgram_drops); out << '}'; return out.str(); } @@ -171,7 +170,6 @@ namespace laps { AppendUintField(out, first, "tx_queue_discards", metrics.tx_queue_discards); AppendUintField(out, first, "tx_queue_expired", metrics.tx_queue_expired); AppendUintField(out, first, "tx_delayed_callback", metrics.tx_delayed_callback); - AppendUintField(out, first, "tx_reset_wait", metrics.tx_reset_wait); AppendRawField(out, first, "tx_queue_size", SerializeMinMaxAvg(metrics.tx_queue_size)); AppendRawField(out, first, "tx_callback_ms", SerializeMinMaxAvg(metrics.tx_callback_ms)); AppendRawField(out, first, "tx_object_duration_us", SerializeMinMaxAvg(metrics.tx_object_duration_us)); diff --git a/src/peering/common.h b/src/peering/common.h index e8db927..dd7595d 100644 --- a/src/peering/common.h +++ b/src/peering/common.h @@ -16,6 +16,9 @@ namespace laps::peering { constexpr int kViaRelayMax = 5; ///< Maximum number of best via relays to advertise constexpr uint8_t kProtocolVersion = 1; + /// Reads per stream in a single receive callback. Supports bursts without starving other streams. + constexpr int kReadLoopMaxPerStream = 100; + using HashType = uint64_t; ///< Value data type for hashes using NamespaceTuples = std::vector; using PeerSessionId = uint64_t; diff --git a/src/peering/info_base.h b/src/peering/info_base.h index f793f7d..3fa3f98 100644 --- a/src/peering/info_base.h +++ b/src/peering/info_base.h @@ -151,8 +151,16 @@ namespace laps::peering { struct FibEntry { uint64_t update_ref{ 0 }; ///< Random reference number to detect if entry was updated or not - std::unordered_map streams; ///< Key is ingress stream Id, value is egress stream id - SubscribeNodeSetId out_sns_id; ///< Egress SNS ID + + /** + * @brief Egress stream for each ingress stream, keyed by ingress stream ID + * + * @details The transport hands out stream handles rather than IDs, and holding one keeps the + * stream usable for as long as objects are being forwarded onto it. + */ + std::unordered_map> streams; + + SubscribeNodeSetId out_sns_id; ///< Egress SNS ID decltype(nodes_best_)::mapped_type peer_session; uint64_t track_fullname_hash{ 0 }; }; diff --git a/src/peering/messages/announce_info.h b/src/peering/messages/announce_info.h index b689b60..d76d3b7 100644 --- a/src/peering/messages/announce_info.h +++ b/src/peering/messages/announce_info.h @@ -4,7 +4,7 @@ #include "node_info.h" #include "peering/common.h" -#include "quicr/handlers/base_track_handler.h" +#include "quicr/handlers/track_handler.h" #include diff --git a/src/peering/peer_manager.cc b/src/peering/peer_manager.cc index d6defb6..0451b0b 100644 --- a/src/peering/peer_manager.cc +++ b/src/peering/peer_manager.cc @@ -8,6 +8,7 @@ #include #include "subscribe_handler.h" +#include "transport_helpers.h" #include namespace laps::peering { @@ -203,7 +204,12 @@ namespace laps::peering { peer_session->GetSessionId(), subscribe_info.track_hash.track_fullname_hash); - info_base_->client_fib_.erase({ subscribe_info.track_hash.track_fullname_hash, peer_session_id }); + const auto cfib_it = + info_base_->client_fib_.find({ subscribe_info.track_hash.track_fullname_hash, peer_session_id }); + if (cfib_it != info_base_->client_fib_.end()) { + CloseFibStreams(cfib_it->second); + info_base_->client_fib_.erase(cfib_it); + } if (not HasSubscribers(subscribe_info.track_hash.track_fullname_hash)) { SPDLOG_LOGGER_INFO(LOGGER, @@ -268,9 +274,9 @@ namespace laps::peering { if (announce_info.flags & 1) { // PUBLISH_NAMESPACE if (!withdraw) { client_manager_->PublishNamespaceReceived( - 0, announce_info.name_space, { .request_id = th.track_namespace_hash }); + nullptr, announce_info.name_space, { .request_id = th.track_namespace_hash }); } else { - client_manager_->PublishNamespaceDoneReceived(0, th.track_namespace_hash); + client_manager_->PublishNamespaceDoneReceived(nullptr, th.track_namespace_hash); } } else { // PUBLISH if (!withdraw) { @@ -289,10 +295,10 @@ namespace laps::peering { .delivery_timeout = kDefaultObjectTtl, .track_properties = {} }; - client_manager_->PublishReceived(0, 0, attrs, {}); + client_manager_->PublishReceived(nullptr, 0, attrs, {}); } else { // TODO: Signal to client manager that the publish is done - client_manager_->PublishDoneReceived(0, 0); + client_manager_->PublishDoneReceived(nullptr, 0); } } @@ -394,7 +400,7 @@ namespace laps::peering { DataHeader data_header, std::shared_ptr> data, uint64_t data_offset, - quicr::ITransport::EnqueueFlags eflags) + quicr::Transport::EnqueueFlags eflags) { std::unique_lock _(info_base_->mutex_); // TODO: See about removing this lock auto it = info_base_->peer_fib_.find({ peer_session_id, data_header.sns_id }); @@ -443,7 +449,7 @@ namespace laps::peering { auto out_peer_sess = entry.peer_session.lock(); - uint64_t out_stream_id{ 0 }; + std::shared_ptr out_stream; if (eflags.use_reliable) { auto sid_it = entry.streams.find(stream_id); if (sid_it == entry.streams.end()) { @@ -452,21 +458,17 @@ namespace laps::peering { continue; // Ignore existing data and wait for start of new stream } - out_stream_id = out_peer_sess->CreateStream(entry.out_sns_id, data_header.priority); - entry.streams.emplace(stream_id, out_stream_id); + out_stream = out_peer_sess->CreateStream(data_header.priority); + entry.streams.emplace(stream_id, out_stream); } else { - out_stream_id = sid_it->second; + out_stream = sid_it->second; } } auto data_out_shared = std::make_shared>(); data_out_shared->assign(data_out.begin(), data_out.end()); - out_peer_sess->SendData(data_header.priority, - data_header.ttl, - entry.out_sns_id, - out_stream_id, - eflags, - std::move(data_out_shared)); + out_peer_sess->SendData( + data_header.priority, data_header.ttl, out_stream, eflags, std::move(data_out_shared)); } } else { SPDLOG_LOGGER_DEBUG(config_.logger_, @@ -506,8 +508,7 @@ namespace laps::peering { auto stream_it = fib_entry.streams.find(in_stream_id); if (stream_it != fib_entry.streams.end()) { - peer_sess->CloseStream(fib_entry.out_sns_id, - stream_it->second, + peer_sess->CloseStream(stream_it->second, reset ? quicr::StreamClosedFlag::kReset : quicr::StreamClosedFlag::kFin); fib_entry.streams.erase(stream_it); } @@ -532,7 +533,7 @@ namespace laps::peering { auto net_data = std::make_shared>(data_header.Serialize()); net_data->insert(net_data->end(), data->begin(), data->end()); - quicr::ITransport::EnqueueFlags eflags; + quicr::Transport::EnqueueFlags eflags; bool set_sns_id{ false }; switch (type) { @@ -571,7 +572,7 @@ namespace laps::peering { std::copy(sns_id_bytes.rbegin(), sns_id_bytes.rend(), send_data->begin() + 2); } - uint64_t out_stream_id{ 0 }; + std::shared_ptr out_stream; if (eflags.use_reliable) { uint64_t in_stream_id = group_id << 16 | static_cast(subgroup_id); @@ -582,20 +583,19 @@ namespace laps::peering { return; } - out_stream_id = peer_sess->CreateStream(fib_entry.out_sns_id, priority); - fib_entry.streams.try_emplace(in_stream_id, out_stream_id); + out_stream = peer_sess->CreateStream(priority); + fib_entry.streams.try_emplace(in_stream_id, out_stream); } else { - out_stream_id = stream_it->second; + out_stream = stream_it->second; } SPDLOG_LOGGER_TRACE( LOGGER, - "Data object send, peer_session: {} egress SNS_ID: {} in stream_id: {} out " - "stream_id: {} tfn_hash: {} group_id: {} subgroup_id: {} streams: {} data len: {}", + "Data object send, peer_session: {} egress SNS_ID: {} in stream_id: {} tfn_hash: {} " + "group_id: {} subgroup_id: {} streams: {} data len: {}", peer_sess->GetSessionId(), fib_entry.out_sns_id, in_stream_id, - out_stream_id, track_full_name_hash, group_id, subgroup_id, @@ -603,7 +603,7 @@ namespace laps::peering { net_data->size()); } - peer_sess->SendData(priority, ttl, fib_entry.out_sns_id, out_stream_id, eflags, send_data); + peer_sess->SendData(priority, ttl, out_stream, eflags, send_data); } } } @@ -911,8 +911,17 @@ namespace laps::peering { tconfig.idle_timeout_ms = 5000; tconfig.max_connections = 100; - server_transport_ = - quicr::ITransport::MakeServerTransport(std::move(server), std::move(tconfig), *this, tick_service_, LOGGER); + server_transport_ = quicr::Transport::MakeServerTransport( + std::move(server), std::move(tconfig), tick_service_, config_.quicr_logger_); + + server_transport_->OnNewConnection = [this](const std::shared_ptr& connection) { + NewPeerConnection(connection); + }; + + server_transport_->OnConnectionClosed = [this](const std::shared_ptr& connection) { + PeerConnectionClosed(connection); + }; + server_transport_->Start(); while (server_transport_->Status() == quicr::TransportStatus::kConnecting) { @@ -933,6 +942,12 @@ namespace laps::peering { SPDLOG_LOGGER_INFO(LOGGER, "Closing peer manager threads"); + if (server_transport_ != nullptr) { + server_transport_->OnNewConnection = nullptr; + server_transport_->OnConnectionClosed = nullptr; + server_transport_->Shutdown(); + } + client_peer_sessions_.clear(); server_peer_sessions_.clear(); @@ -951,10 +966,12 @@ namespace laps::peering { if (withdraw) { auto it = info_base_->peer_fib_.find({ peer_session.GetSessionId(), sns.id }); if (it != info_base_->peer_fib_.end()) { - for (const auto& [out_peer_sess_id, entry] : it->second) { + for (auto& [out_peer_sess_id, entry] : it->second) { if (out_peer_sess_id == 0) continue; + CloseFibStreams(entry); + auto out_peer_sess = entry.peer_session.lock(); out_peer_sess->RemovePeerSnsSourceNode(peer_session.GetSessionId(), sns.id, 0); } @@ -1190,7 +1207,7 @@ namespace laps::peering { void PeerManager::CreatePeerSession(const quicr::TransportRemote& peer_config) { - auto peer_sess = std::make_shared(false, 0, config_, node_info_, peer_config, *this); + auto peer_sess = std::make_shared(false, config_, node_info_, peer_config, *this); peer_sess->Connect(); client_peer_sessions_.try_emplace(peer_sess->GetSessionId(), std::move(peer_sess)); @@ -1233,107 +1250,79 @@ namespace laps::peering { } /* - * Delegate Implementations + * Server transport callbacks */ - void PeerManager::OnConnectionStatus(const std::uint64_t& conn_id, const quicr::TransportStatus status) + void PeerManager::NewPeerConnection(const std::shared_ptr& connection) { - auto peer_it = server_peer_sessions_.find(conn_id); - if (peer_it == server_peer_sessions_.end()) { - return; - } + const auto conn_id = connection->GetID(); - PeerSession::StatusValue sess_status = PeerSession::StatusValue::kConnected; - switch (status) { - case quicr::TransportStatus::kReady: { - SPDLOG_LOGGER_DEBUG(LOGGER, "Peer conn_id {0} is connected", conn_id); - break; - } - case quicr::TransportStatus::kConnecting: - break; + std::shared_ptr peer_sess; - case quicr::TransportStatus::kDisconnected: { - sess_status = PeerSession::StatusValue::kDisconnected; + { + std::lock_guard _(mutex_); - SPDLOG_LOGGER_DEBUG(LOGGER, "Peer conn_id {0} is disconnected", conn_id); - break; + if (server_peer_sessions_.contains(conn_id)) { + return; } - case quicr::TransportStatus::kRemoteRequestClose: - sess_status = PeerSession::StatusValue::kDisconnected; - - SPDLOG_LOGGER_DEBUG(LOGGER, "Peer conn_id {0} remote disconnected", conn_id); - break; + SPDLOG_LOGGER_INFO(LOGGER, "New server accepted peer, conn_id: {0}", conn_id); - case quicr::TransportStatus::kShutdown: - sess_status = PeerSession::StatusValue::kDisconnected; - SPDLOG_LOGGER_DEBUG(LOGGER, "Peer conn_id {0} shutdown", conn_id); - break; + const auto [ip, port] = GetPeerAddress(server_transport_, connection); + const quicr::TransportRemote peer{ ip, port, quicr::TransportProtocol::kQuic }; - case quicr::TransportStatus::kIdleTimeout: - sess_status = PeerSession::StatusValue::kDisconnected; - SPDLOG_LOGGER_DEBUG(LOGGER, "Peer conn_id {0} idle timeout", conn_id); - break; + peer_sess = std::make_shared(true, config_, node_info_, peer, *this); + peer_sess->SetConnection(server_transport_, connection); + peer_sess->Connect(); - case quicr::TransportStatus::kShuttingDown: - sess_status = PeerSession::StatusValue::kDisconnected; - SPDLOG_LOGGER_DEBUG(LOGGER, "Peer conn_id {0} shutdown", conn_id); - break; + server_peer_sessions_.emplace(conn_id, peer_sess); } - SessionChanged(peer_it->second->GetSessionId(), sess_status, peer_it->second->remote_node_info_); - - server_peer_sessions_.erase(peer_it); + /* + * Attach last and outside the lock. Attaching replays the current connection status, which can call + * back into the manager. + */ + connection->SetDelegate(peer_sess); } - void PeerManager::OnNewConnection(const std::uint64_t& conn_id, const quicr::TransportRemote& remote) + void PeerManager::PeerConnectionClosed(const std::shared_ptr& connection) { - auto peer_iter = server_peer_sessions_.find(conn_id); - - if (peer_iter == server_peer_sessions_.end()) { - SPDLOG_LOGGER_INFO(LOGGER, "New server accepted peer, conn_id: {0}", conn_id); + std::shared_ptr peer_sess; - quicr::TransportRemote peer = remote; - auto [iter, inserted] = server_peer_sessions_.try_emplace( - conn_id, std::make_shared(true, conn_id, config_, node_info_, std::move(peer), *this)); - - peer_iter = iter; + { + std::lock_guard _(mutex_); - auto& peer_sess = peer_iter->second; + auto peer_it = server_peer_sessions_.find(connection->GetID()); + if (peer_it == server_peer_sessions_.end()) { + return; + } - peer_sess->SetTransport(server_transport_); - peer_sess->Connect(); + peer_sess = std::move(peer_it->second); + server_peer_sessions_.erase(peer_it); } - } - void PeerManager::OnRecvStream(const std::uint64_t& conn_id, - uint64_t stream_id, - std::optional data_ctx_id, - const bool is_bidir) - { - auto peer_iter = server_peer_sessions_.find(conn_id); - if (peer_iter != server_peer_sessions_.end()) { - peer_iter->second->OnRecvStream(conn_id, stream_id, data_ctx_id, is_bidir); - } - } + SPDLOG_LOGGER_DEBUG(LOGGER, "Peer conn_id {0} closed", connection->GetID()); - void PeerManager::OnRecvDgram(const std::uint64_t& conn_id, std::optional data_ctx_id) - { - auto peer_iter = server_peer_sessions_.find(conn_id); - if (peer_iter != server_peer_sessions_.end()) { - peer_iter->second->OnRecvDgram(conn_id, data_ctx_id); + /* + * The connection holds a weak reference to the session, so the session's own disconnect notification + * is not guaranteed to be delivered before the session is dropped here. Report it only when the + * session has not already seen the status change, to avoid withdrawing the peer's info twice. + */ + if (peer_sess->Status() != PeerSession::StatusValue::kDisconnected) { + SessionChanged( + peer_sess->GetSessionId(), PeerSession::StatusValue::kDisconnected, peer_sess->remote_node_info_); } } - void PeerManager::OnStreamClosed(const std::uint64_t& connection_handle, - std::uint64_t stream_id, - std::shared_ptr rx_context, - std::optional request_id, - quicr::StreamClosedFlag flag) + void PeerManager::CloseFibStreams(InfoBase::FibEntry& entry) { - auto peer_iter = server_peer_sessions_.find(connection_handle); - if (peer_iter != server_peer_sessions_.end()) { - peer_iter->second->OnStreamClosed(connection_handle, stream_id, rx_context, request_id, flag); + const auto peer_sess = entry.peer_session.lock(); + if (peer_sess != nullptr) { + for (const auto& [_, out_stream] : entry.streams) { + peer_sess->CloseStream(out_stream, quicr::StreamClosedFlag::kFin); + } } + + entry.streams.clear(); } void PeerManager::CloseStream(PeerSessionId peer_session_id, @@ -1358,7 +1347,7 @@ namespace laps::peering { auto stream_it = entry.streams.find(stream_id); if (stream_it != entry.streams.end()) { if (auto out_peer_sess = entry.peer_session.lock()) { - out_peer_sess->CloseStream(entry.out_sns_id, stream_it->second, flag); + out_peer_sess->CloseStream(stream_it->second, flag); entry.streams.erase(stream_it); } } @@ -1372,13 +1361,11 @@ namespace laps::peering { } if (auto out_peer_sess = entry.peer_session.lock()) { - for (const auto& [in_stream_id, out_stream_id] : entry.streams) { - if (out_stream_id == stream_id) { - entry.streams.erase(out_stream_id); - client_manager_->PeerStreamClosed( - key.first, stream_id, flag == quicr::StreamClosedFlag::kReset); - break; - } + if (const auto stream_it = entry.streams.find(stream_id); stream_it != entry.streams.end()) { + out_peer_sess->CloseStream(stream_it->second, flag); + entry.streams.erase(stream_it); + client_manager_->PeerStreamClosed( + key.first, stream_id, flag == quicr::StreamClosedFlag::kReset); } } } diff --git a/src/peering/peer_manager.h b/src/peering/peer_manager.h index f9f03c9..fa5e851 100644 --- a/src/peering/peer_manager.h +++ b/src/peering/peer_manager.h @@ -3,6 +3,7 @@ #pragma once #include +#include #include #include #include @@ -21,7 +22,7 @@ namespace laps::peering { * @brief Peering manager class. Manages relay to relay (peering) forwarding of * subscriber objects. */ - class PeerManager : public quicr::ITransport::TransportDelegate + class PeerManager { public: friend class PeerSession; @@ -47,7 +48,7 @@ namespace laps::peering { DataHeader data_header, std::shared_ptr> data, uint64_t data_offset, - quicr::ITransport::EnqueueFlags eflags); + quicr::Transport::EnqueueFlags eflags); void ClientDataRecv(std::uint64_t track_full_name_hash, uint8_t priority, @@ -155,27 +156,24 @@ namespace laps::peering { bool reset = false); // ------------------------------------------------------------------------------- - // QUIC Transport callbacks + + private: + // ------------------------------------------------------------------------------- + // Server transport callbacks // ------------------------------------------------------------------------------- - void OnNewDataContext(const std::uint64_t&, const std::uint64_t&) override {} - void OnConnectionStatus(const std::uint64_t& conn_id, const quicr::TransportStatus status) override; - void OnNewConnection(const std::uint64_t& conn_id, const quicr::TransportRemote& remote) override; - void OnRecvStream(const std::uint64_t& conn_id, - uint64_t stream_id, - std::optional data_ctx_id, - const bool is_bidir = false) override; - void OnRecvDgram(const std::uint64_t& conn_id, std::optional data_ctx_id) override; - - void OnStreamClosed(const std::uint64_t& connection_handle, - std::uint64_t stream_id, - std::shared_ptr rx_context, - std::optional request_id, - quicr::StreamClosedFlag flag) override; + /** + * @brief Create an inbound peer session for a server accepted connection + */ + void NewPeerConnection(const std::shared_ptr& connection); + + /** + * @brief Remove the inbound peer session state for a closed connection + */ + void PeerConnectionClosed(const std::shared_ptr& connection); // ------------------------------------------------------------------------------- - private: /** * @brief Check Thread to perform reconnects and cleanup * @details Thread will perform various tasks each interval @@ -196,6 +194,14 @@ namespace laps::peering { void PropagateNodeInfo(PeerSessionId peer_session_id, const NodeInfo& node_info, bool withdraw = false); std::shared_ptr GetPeerSession(PeerSessionId peer_session_id); + /** + * @brief Finish the egress streams of a forwarding entry that is going away + * + * @details The entry owns the only handle to each stream, so the peer is told the flow ended before + * the entry is dropped, rather than leaving the streams open until the connection closes. + */ + static void CloseFibStreams(InfoBase::FibEntry& entry); + private: bool stop_{ false }; std::mutex mutex_; @@ -212,7 +218,7 @@ namespace laps::peering { /// Peer sessions that are initiated by the peer manager std::map> client_peer_sessions_; - std::shared_ptr server_transport_; /// Server Transport for inbound connections + std::shared_ptr server_transport_; /// Server Transport for inbound connections std::thread check_thr_; /// Check/task thread, handles reconnects diff --git a/src/peering/peer_session.cc b/src/peering/peer_session.cc index 2253aab..47d54a3 100644 --- a/src/peering/peer_session.cc +++ b/src/peering/peer_session.cc @@ -14,7 +14,6 @@ namespace laps::peering { PeerSession::PeerSession(bool is_inbound, - const std::uint64_t conn_id, const Config& cfg, const NodeInfo& node_info, const quicr::TransportRemote& remote, @@ -24,7 +23,6 @@ namespace laps::peering { , node_info_(node_info) , manager_(manager) , is_inbound_(is_inbound) - , t_conn_id_(conn_id) { if (config_.tls_cert_filename_.length() == 0) { transport_config_.tls_cert_filename = ""; @@ -39,6 +37,7 @@ namespace laps::peering { PeerSession::~PeerSession() { if (not is_inbound_) { + connection_ = nullptr; transport_ = nullptr; } @@ -60,20 +59,41 @@ namespace laps::peering { return; } - if (transport_) - transport_ = nullptr; + connection_ = nullptr; + transport_ = nullptr; + control_stream_ = nullptr; + control_stream_id_.reset(); peer_sns_.clear(); + rx_stream_headers_.clear(); + + transport_ = quicr::Transport::MakeClientTransport( + peer_config_, transport_config_, config_.tick_service_, config_.quicr_logger_); + + connection_ = transport_->Start(); + if (!connection_) { + SPDLOG_LOGGER_ERROR(LOGGER, "Failed to connect to peer {}:{}", peer_config_.host_or_ip, peer_config_.port); + status_ = StatusValue::kDisconnected; + return; + } + + connection_->SetDelegate(shared_from_this()); + + // The peer answers on the same stream, so the control exchange needs a bidirectional one + control_stream_ = transport_->CreateRequestStream(connection_); + control_stream_id_.reset(); - transport_ = - quicr::ITransport::MakeClientTransport(peer_config_, transport_config_, *this, config_.tick_service_, LOGGER); - t_conn_id_ = transport_->Start(); + SPDLOG_LOGGER_DEBUG(LOGGER, "Created control stream for peer session"); + } - // Create the control data context - control_data_ctx_id_ = transport_->CreateDataContext(t_conn_id_, true, 0, true); - control_stream_id_ = transport_->CreateStream(t_conn_id_, control_data_ctx_id_, 0); + SubscribeNodeSetId PeerSession::NextSnsId() + { + // Zero means unset, so it is skipped when the counter wraps + if (last_sns_id_ >= kMaxSnsId) { + last_sns_id_ = 0; + } - SPDLOG_LOGGER_DEBUG(LOGGER, "Control stream ID {0}", control_data_ctx_id_); + return ++last_sns_id_; } std::pair PeerSession::AddPeerSnsSourceNode(PeerSessionId in_peer_session_id, @@ -84,10 +104,9 @@ namespace laps::peering { auto [it, new_ingress] = peer_sns_.try_emplace({ in_peer_session_id, in_sns_id }); auto& sns = it->second; - if (it->second.id == 0) { // If not set, create the data context - // TODO(tievens): Add datagram support - update transport to allow changing reliable state - // TODO(tievens): Update transport to have max data context ID and to wrap if reaching max - it->second.id = transport_->CreateDataContext(t_conn_id_, true, priority, false); + if (it->second.id == 0) { + it->second.id = NextSnsId(); + it->second.priority = priority; } auto [__, is_new] = sns.nodes.emplace(sub_node_id); @@ -106,9 +125,9 @@ namespace laps::peering { auto [it, _] = sub_sns_.try_emplace(full_name_hash); auto& sns = it->second; - if (it->second.id == 0) { // If not set, create the data context - // TODO(tievens): Update transport to have max data context ID and to wrap if reaching max - it->second.id = transport_->CreateDataContext(t_conn_id_, true, priority, false); + if (it->second.id == 0) { + it->second.id = NextSnsId(); + it->second.priority = priority; } auto [__, is_new] = sns.nodes.emplace(sub_node_id); @@ -134,7 +153,6 @@ namespace laps::peering { if (sns.nodes.empty()) { sns_removed = true; - transport_->DeleteDataContext(t_conn_id_, it->second.id); SendSns(sns, true); @@ -166,7 +184,6 @@ namespace laps::peering { if (sns.nodes.empty()) { sns_removed = true; - transport_->DeleteDataContext(t_conn_id_, it->second.id); SendSns(sns, true); @@ -178,65 +195,95 @@ namespace laps::peering { return { node_removed, sns_removed }; } - uint64_t PeerSession::CreateStream(SubscribeNodeSetId sns_id, uint8_t priority) const + std::shared_ptr PeerSession::CreateStream(uint8_t priority) const { - return transport_->CreateStream(t_conn_id_, sns_id, priority); + if (!IsUsable()) { + return nullptr; + } + + return transport_->CreateDataStream(connection_, priority); } - void PeerSession::CloseStream(SubscribeNodeSetId sns_id, uint64_t stream_id, quicr::StreamClosedFlag flag) + void PeerSession::CloseStream(const std::shared_ptr& stream, quicr::StreamClosedFlag flag) { - transport_->CloseStream(t_conn_id_, sns_id, stream_id, flag == quicr::StreamClosedFlag::kReset); + if (!IsUsable() || stream == nullptr) { + return; + } + + quicr::StreamOperation operation; + switch (flag) { + case quicr::StreamClosedFlag::kFin: + operation = quicr::StreamOperation::kFin; + break; + case quicr::StreamClosedFlag::kReset: + operation = quicr::StreamOperation::kReset; + break; + case quicr::StreamClosedFlag::kStopSending: + operation = quicr::StreamOperation::kStopSending; + break; + } + transport_->CloseStream(connection_, stream, operation); } void PeerSession::SendData(uint8_t priority, uint32_t ttl, - SubscribeNodeSetId sns_id, - uint64_t stream_id, - const quicr::ITransport::EnqueueFlags& eflags, + const std::shared_ptr& stream, + const quicr::Transport::EnqueueFlags& eflags, std::shared_ptr> data) { - // SPDLOG_LOGGER_DEBUG(LOGGER, "Sending data SNS id: {} data size: {}", sns_id, data.size()); + if (status_ != StatusValue::kConnected || !IsUsable()) + return; + + if (!eflags.use_reliable) { + transport_->EnqueueDatagram(connection_, std::move(data), priority, ttl); + return; + } + + if (stream == nullptr) { + return; + } + + transport_->Enqueue(connection_, stream, std::move(data), priority, ttl, eflags); + } - if (status_ != StatusValue::kConnected) + void PeerSession::SendControl(std::vector msg) const + { + if (!IsUsable() || control_stream_ == nullptr) { + SPDLOG_LOGGER_DEBUG( + LOGGER, "No control stream for peer conn_id {}, dropping control message", GetSessionId()); return; - transport_->Enqueue(t_conn_id_, sns_id, stream_id, data, priority, ttl, 0, eflags); + } + + transport_->Enqueue( + connection_, control_stream_, std::make_shared>(std::move(msg)), 0, 1000); } void PeerSession::SendSns(const SubscribeNodeSet& sns, bool withdraw) const { - if (status_ != StatusValue::kConnected) + if (status_ != StatusValue::kConnected || !IsUsable()) return; SPDLOG_LOGGER_DEBUG(LOGGER, "Sending SNS id: {} set size: {} withdraw: {}", sns.id, sns.nodes.size(), withdraw); - transport_->Enqueue(t_conn_id_, - control_data_ctx_id_, - control_stream_id_, - std::make_shared>(sns.Serialize(true, withdraw)), - 0, - 1000); + SendControl(sns.Serialize(true, withdraw)); } void PeerSession::SendAnnounceInfo(const AnnounceInfo& announce_info, bool withdraw) { - if (status_ != StatusValue::kConnected) + if (status_ != StatusValue::kConnected || !IsUsable()) return; SPDLOG_LOGGER_DEBUG(LOGGER, "Sending announce info id: {} source_node_id: {} withdraw: {}", announce_info.fullname_hash, announce_info.source_node_id, withdraw); - transport_->Enqueue(t_conn_id_, - control_data_ctx_id_, - control_stream_id_, - std::make_shared>(announce_info.Serialize(true, withdraw)), - 0, - 1000); + + SendControl(announce_info.Serialize(true, withdraw)); } void PeerSession::SendSubscribeInfo(SubscribeInfo& subscribe_info, bool withdraw) const { - if (status_ != StatusValue::kConnected) + if (status_ != StatusValue::kConnected || !IsUsable()) return; SPDLOG_LOGGER_DEBUG(LOGGER, "Sending subscribe fullname: {} source_node_id: {} withdraw: {} sub_data_size: {}", @@ -245,26 +292,16 @@ namespace laps::peering { withdraw, subscribe_info.subscribe_data.size()); - transport_->Enqueue(t_conn_id_, - control_data_ctx_id_, - control_stream_id_, - std::make_shared>( - subscribe_info.Serialize(true, withdraw, node_info_.id == subscribe_info.source_node_id)), - 0, - 1000); + SendControl(subscribe_info.Serialize(true, withdraw, node_info_.id == subscribe_info.source_node_id)); } void PeerSession::SendNodeInfo(const NodeInfo& node_info, bool withdraw) const { - if (status_ != StatusValue::kConnected) + if (status_ != StatusValue::kConnected || !IsUsable()) return; SPDLOG_LOGGER_DEBUG(LOGGER, "Sending node info id: {}", NodeId().Value(node_info.id)); - transport_->Enqueue(t_conn_id_, - control_data_ctx_id_, - control_stream_id_, - std::make_shared>(node_info.Serialize(true, withdraw)), - 0, - 1000); + + SendControl(node_info.Serialize(true, withdraw)); } void PeerSession::SendConnect() @@ -276,12 +313,8 @@ namespace laps::peering { peer_sns_.clear(); SPDLOG_LOGGER_DEBUG(LOGGER, "Sending connect length: {}", connect.Serialize().size()); - transport_->Enqueue(t_conn_id_, - control_data_ctx_id_, - control_stream_id_, - std::make_shared>(connect.Serialize()), - 0, - 1000); + + SendControl(connect.Serialize()); } void PeerSession::SendConnectOk() const @@ -291,53 +324,60 @@ namespace laps::peering { connect_resp.node_info = node_info_; SPDLOG_LOGGER_DEBUG(LOGGER, "Sending connect ok length: {}", connect_resp.Serialize().size()); - transport_->Enqueue(t_conn_id_, - control_data_ctx_id_, - control_stream_id_, - std::make_shared>(connect_resp.Serialize()), - 0, - 1000); + SendControl(connect_resp.Serialize()); } /* * Delegate Implementations */ - void PeerSession::OnConnectionStatus(const std::uint64_t& conn_id, const quicr::TransportStatus status) + void PeerSession::OnConnectionStatus(const quicr::Connection::Status status) { + const auto conn_id = GetSessionId(); + switch (status) { - case quicr::TransportStatus::kReady: { + case quicr::Connection::Status::kReady: { + /* + * Inbound sessions do not initiate the peering handshake; they wait for the remote connect + * message and answer it with connect ok. + */ + if (is_inbound_) { + status_ = StatusValue::kConnected; + SPDLOG_LOGGER_DEBUG(LOGGER, "Peer conn_id {0} is ready", conn_id); + break; + } + SPDLOG_LOGGER_DEBUG(LOGGER, "Peer conn_id {0} is ready, sending connect message", conn_id); SendConnect(); break; } - case quicr::TransportStatus::kConnecting: + case quicr::Connection::Status::kConnecting: break; - case quicr::TransportStatus::kDisconnected: { + case quicr::Connection::Status::kDisconnected: { status_ = StatusValue::kDisconnected; SPDLOG_LOGGER_DEBUG(LOGGER, "Peer conn_id {0} is disconnected", conn_id); break; } - case quicr::TransportStatus::kRemoteRequestClose: + case quicr::Connection::Status::kRemoteRequestClose: status_ = StatusValue::kDisconnected; SPDLOG_LOGGER_DEBUG(LOGGER, "Peer conn_id {0} remote disconnected", conn_id); break; - case quicr::TransportStatus::kShutdown: + case quicr::Connection::Status::kShutdown: status_ = StatusValue::kDisconnected; SPDLOG_LOGGER_DEBUG(LOGGER, "Peer conn_id {0} shutdown", conn_id); break; - case quicr::TransportStatus::kIdleTimeout: + case quicr::Connection::Status::kIdleTimeout: status_ = StatusValue::kDisconnected; SPDLOG_LOGGER_DEBUG(LOGGER, "Peer conn_id {0} idle timeout", conn_id); break; - case quicr::TransportStatus::kShuttingDown: + case quicr::Connection::Status::kShuttingDown: status_ = StatusValue::kDisconnected; SPDLOG_LOGGER_DEBUG(LOGGER, "Peer conn_id {0} shutdown", conn_id); break; @@ -346,12 +386,6 @@ namespace laps::peering { manager_.SessionChanged(GetSessionId(), status_, remote_node_info_); } - void PeerSession::OnNewConnection([[maybe_unused]] const std::uint64_t& conn_id, - [[maybe_unused]] const quicr::TransportRemote& remote) - { - // Not used for outgoing connections. Incoming connections are handled by the server delegate - } - void PeerSession::ProcessControlMessage() try { if (controL_msg_buffer_.size() >= kCommonHeadersSize) { @@ -483,60 +517,62 @@ namespace laps::peering { controL_msg_buffer_.clear(); } - bool PeerSession::ProcessReceivedData(std::optional stream_id, - std::any& ctx, - std::shared_ptr> data) const + bool PeerSession::ProcessReceivedData(std::uint64_t stream_id, std::shared_ptr> data) { // TODO(tievens): Update to not buffer when node type is Via - quicr::ITransport::EnqueueFlags eflags; - eflags.use_reliable = stream_id.has_value(); // If stream isn't set, it's datagram - - // NEW STREAM - parse start of stream headers - if (!ctx.has_value()) { - ctx.emplace(); + constexpr quicr::Transport::EnqueueFlags eflags{ true, false, false, false }; - auto cursor_it = data->begin(); - const auto hdr_len = *cursor_it; + const auto header_it = rx_stream_headers_.find(stream_id); - if (data->size() < hdr_len) { + // NEW STREAM - parse start of stream headers + if (header_it == rx_stream_headers_.end()) { + if (data->empty() || data->size() < data->front()) { SPDLOG_LOGGER_DEBUG( LOGGER, "Received new data object stream id: {}, not enough bytes yet to read headers {} > {}", - stream_id.has_value() ? *stream_id : 0, - *cursor_it, + stream_id, + data->empty() ? 0 : data->front(), data->size()); return false; // Not enough bytes to parse the headers, wait till more arrives } - auto& data_header = std::any_cast(ctx); + const auto hdr_len = data->front(); + + auto& data_header = rx_stream_headers_[stream_id]; data_header.Deserialize(*data); // Pipeline forward to other peers. - manager_.ForwardPeerData( - GetSessionId(), true, stream_id.has_value() ? *stream_id : 0, data_header, data, hdr_len, eflags); + manager_.ForwardPeerData(GetSessionId(), true, stream_id, data_header, data, hdr_len, eflags); return true; } - // Existing data - auto& data_header = std::any_cast(ctx); - // Pipeline forward to other peers. Not all data may have been popped, so only forward popped data - manager_.ForwardPeerData( - GetSessionId(), false, stream_id.has_value() ? *stream_id : 0, data_header, data, 0, eflags); + manager_.ForwardPeerData(GetSessionId(), false, stream_id, header_it->second, data, 0, eflags); return true; } - void PeerSession::OnRecvStream(const std::uint64_t& conn_id, - uint64_t stream_id, - std::optional data_ctx_id, + void PeerSession::OnRecvStream(std::uint64_t stream_id, + const std::shared_ptr& rx_ctx, + const std::shared_ptr& stream, const bool is_bidir) { - auto rx_ctx = transport_->GetStreamRxContext(conn_id, stream_id); + if (!IsUsable() || rx_ctx == nullptr) { + return; + } + + /* + * Control messages arrive on the one bidirectional stream, which the inbound side of the session + * only learns about here. Both sides answer on it, so it is adopted either way. + */ + if (is_bidir && stream != nullptr) { + control_stream_ = stream; + control_stream_id_ = stream_id; + } - for (int i = 0; i < quicr::kReadLoopMaxPerStream; i++) { + for (int i = 0; i < kReadLoopMaxPerStream; i++) { if (rx_ctx->data_queue.Empty()) { break; } @@ -550,24 +586,27 @@ namespace laps::peering { // Get common header if (is_bidir) { // control - control_data_ctx_id_ = *data_ctx_id; controL_msg_buffer_.insert(controL_msg_buffer_.end(), data->begin(), data->end()); ProcessControlMessage(); - } else if (!ProcessReceivedData(stream_id, rx_ctx->caller_any, std::move(data))) { + } else if (!ProcessReceivedData(stream_id, std::move(data))) { i = 59; continue; // Try once more } } } - void PeerSession::OnRecvDgram(const std::uint64_t& conn_id, std::optional data_ctx_id) + void PeerSession::OnRecvDgram() { - constexpr quicr::ITransport::EnqueueFlags eflags{ false, false, false, false }; + constexpr quicr::Transport::EnqueueFlags eflags{ false, false, false, false }; + + if (!IsUsable()) { + return; + } for (int i = 0; i < 80; i++) { - auto data = transport_->Dequeue(conn_id, data_ctx_id); + auto data = transport_->Dequeue(connection_); if (!data) { return; @@ -586,34 +625,39 @@ namespace laps::peering { } void PeerSession::OnConnectionMetricsSampled([[maybe_unused]] const quicr::MetricsTimeStamp sample_time, - [[maybe_unused]] const std::uint64_t conn_id, const quicr::QuicConnectionMetrics& quic_connection_metrics) { metrics_.srtt_us = quic_connection_metrics.srtt_us.avg; } - void PeerSession::OnStreamClosed(const std::uint64_t& connection_handle, - std::uint64_t stream_id, + void PeerSession::OnStreamClosed(std::uint64_t stream_id, [[maybe_unused]] std::shared_ptr rx_context, - [[maybe_unused]] std::optional request_id, quicr::StreamClosedFlag flag) { - if (auto rx_ctx = transport_->GetStreamRxContext(connection_handle, stream_id)) { - auto& data_header = std::any_cast(rx_ctx->caller_any); + const auto conn_id = GetSessionId(); + + if (control_stream_ != nullptr && control_stream_id_.has_value() && *control_stream_id_ == stream_id) { + control_stream_ = nullptr; + control_stream_id_.reset(); + } + + if (const auto header_it = rx_stream_headers_.find(stream_id); header_it != rx_stream_headers_.end()) { + const auto& data_header = header_it->second; SPDLOG_LOGGER_DEBUG(LOGGER, "Peer stream closed conn_id {} stream id: {} flag: {} track fullname hash: {}", - connection_handle, + conn_id, stream_id, static_cast(flag), data_header.track_full_name_hash); - manager_.CloseStream( - connection_handle, data_header.sns_id, stream_id, data_header.track_full_name_hash, flag); + manager_.CloseStream(conn_id, data_header.sns_id, stream_id, data_header.track_full_name_hash, flag); + + rx_stream_headers_.erase(header_it); } else { SPDLOG_LOGGER_DEBUG(LOGGER, "Peer conn_id {} stream id: {} flag: {} closed without having existing state", - connection_handle, + conn_id, stream_id, static_cast(flag)); } diff --git a/src/peering/peer_session.h b/src/peering/peer_session.h index a56fffb..8071756 100644 --- a/src/peering/peer_session.h +++ b/src/peering/peer_session.h @@ -4,11 +4,13 @@ #include #include +#include #include #include #include "config.h" #include "messages/announce_info.h" +#include "messages/data_header.h" #include "messages/node_info.h" #include "messages/subscribe_info.h" #include "messages/subscribe_node_set.h" @@ -22,7 +24,9 @@ namespace laps::peering { * subscriber objects. * */ - class PeerSession : public quicr::ITransport::TransportDelegate + class PeerSession + : public quicr::Connection::Delegate + , public std::enable_shared_from_this { public: static constexpr std::size_t kControlMessageBufferSize = 4096; @@ -46,12 +50,10 @@ namespace laps::peering { * @brief Constructor to create a new peer session * * @param is_inbound True indicates the peering session is inbound (server accepted) - * @param conn_id Connection ID from the transport for the connection * @param cfg Global config * @param remote Transport remote peer config/parameters */ PeerSession(const bool is_inbound, - const std::uint64_t conn_id, const Config& cfg, const NodeInfo& node_info, const quicr::TransportRemote& remote, @@ -70,28 +72,47 @@ namespace laps::peering { StatusValue Status(); /** - * @brief Set the transport - * @details Setting the transport is not required and should not be used for outbound connections. This - * Server,incoming mode requires the server transport to be used. + * @brief Set the transport and accepted connection + * + * @details Only used for inbound (server accepted) sessions, which share the server transport. Outbound + * sessions create their own transport and connection in Connect(). */ - void SetTransport(std::shared_ptr transport) { transport_ = transport; } + void SetConnection(std::shared_ptr transport, std::shared_ptr connection) + { + transport_ = std::move(transport); + connection_ = std::move(connection); + } /** * @brief Get the peer session ID */ - PeerSessionId GetSessionId() const { return t_conn_id_; } + PeerSessionId GetSessionId() const { return connection_ ? connection_->GetID() : 0; } - uint64_t CreateStream(SubscribeNodeSetId sns_id, uint8_t priority) const; - void CloseStream(SubscribeNodeSetId sns_id, uint64_t stream_id, quicr::StreamClosedFlag flag); + /** + * @brief Open an outbound stream to carry objects for a subscribe node set + * + * @details The handle is held by the caller for as long as it keeps sending on the stream, rather + * than being looked up per object. + * + * @returns The stream, or nullptr if the session has no usable connection + */ + std::shared_ptr CreateStream(uint8_t priority) const; + + void CloseStream(const std::shared_ptr& stream, quicr::StreamClosedFlag flag); void SendNodeInfo(const NodeInfo& node_info, bool withdraw = false) const; void SendSubscribeInfo(SubscribeInfo& subscribe_info, bool withdraw = false) const; void SendAnnounceInfo(const AnnounceInfo& announce_info, bool withdraw = false); void SendSns(const SubscribeNodeSet& sns, bool withdraw = false) const; + + /** + * @brief Send an object to the peer + * + * @param stream Stream to send on, ignored when the flags ask for a datagram + */ void SendData(uint8_t priority, uint32_t ttl, - SubscribeNodeSetId sns_id, - uint64_t stream_id, - const quicr::ITransport::EnqueueFlags& eflags, + const std::shared_ptr& stream, + const quicr::Transport::EnqueueFlags& eflags, std::shared_ptr> data); /** @@ -151,38 +172,60 @@ namespace laps::peering { NodeIdValueType sub_node_id); /* - * Delegate functions mainly for Outgoing but does include incoming + * Connection delegate callbacks */ - void OnNewDataContext(const std::uint64_t&, const std::uint64_t&) override {} - void OnConnectionStatus(const std::uint64_t& conn_id, const quicr::TransportStatus status) override; - void OnNewConnection(const std::uint64_t& conn_id, const quicr::TransportRemote& remote) override; - void OnRecvStream(const std::uint64_t& conn_id, - uint64_t stream_id, - std::optional data_ctx_id, - const bool is_bidir = false) override; - void OnRecvDgram(const std::uint64_t& conn_id, std::optional data_ctx_id) override; + void OnConnectionStatus(quicr::Connection::Status status) override; + + void OnRecvStream(std::uint64_t stream_id, + const std::shared_ptr& rx_ctx, + const std::shared_ptr& stream, + bool is_bidir = false) override; + + void OnRecvDgram() override; void OnConnectionMetricsSampled(const quicr::MetricsTimeStamp sample_time, - const std::uint64_t conn_id, const quicr::QuicConnectionMetrics& quic_connection_metrics) override; - void OnStreamClosed(const std::uint64_t& connection_handle, - std::uint64_t stream_id, + void OnStreamMetricsStampled(const quicr::MetricsTimeStamp, + std::uint64_t, + const quicr::QuicStreamMetrics&, + bool) override + { + } + + void OnStreamClosed(std::uint64_t stream_id, std::shared_ptr rx_context, - std::optional request_id, quicr::StreamClosedFlag flag) override; // --------------------------------------- private: + /// True when the transport and connection are both present, so transport calls are safe to make + bool IsUsable() const { return transport_ != nullptr && connection_ != nullptr; } + void SendConnect(); void SendConnectOk() const; + /** + * @brief Next subscribe node set ID for this session + * + * @details SNS IDs identify a flow of objects on the wire within a peering session. They used to be + * the transport's data context ID for the flow, but the transport no longer has that concept. + */ + SubscribeNodeSetId NextSnsId(); + + /// Send a control message on the session's bidirectional control stream + void SendControl(std::vector msg) const; + void ProcessControlMessage(); - bool ProcessReceivedData(std::optional stream_id, - std::any& ctx, - std::shared_ptr> data) const; + /** + * @brief Forward an object received on a stream to the other peers interested in it + * + * @returns False when the start of the stream is too short to hold its header, so the caller should + * wait for more of it to arrive + */ + bool ProcessReceivedData(std::uint64_t stream_id, std::shared_ptr> data); public: quicr::TransportRemote peer_config_; @@ -216,12 +259,32 @@ namespace laps::peering { /// Key is the ingress peer session ID and SNS ID, value is the SNS egress via this peer std::map, SubscribeNodeSet> peer_sns_; - std::uint64_t t_conn_id_; /// Transport connection context ID (aka peer session id) - std::uint64_t control_data_ctx_id_; /// Control data context ID - uint64_t control_stream_id_{ 0 }; /// control bidir stream + /// Last SNS ID handed out on this session, which is scoped to the session + SubscribeNodeSetId last_sns_id_{ 0 }; + + /** + * @brief Bidirectional stream carrying the peering control messages + * + * @details Opened by the outbound side and learned by the inbound side from the first control + * message it receives, so both sides send on the one stream. + */ + std::shared_ptr control_stream_; + + /// QUIC stream ID of the control stream, learned from the first bidirectional receive callback + std::optional control_stream_id_; + std::vector controL_msg_buffer_; /// Working buffer of control message being processed - std::shared_ptr transport_; /// Transport used for the peering connection + /** + * @brief Header of each inbound object stream, by stream ID + * + * @details Parsed from the start of the stream and needed again for the objects that follow it, + * which do not repeat it. Only touched from the connection's callback thread. + */ + std::map rx_stream_headers_; + + std::shared_ptr connection_; /// Connection (aka peer session) to the peer + std::shared_ptr transport_; /// Transport used for the peering connection }; } // namespace laps \ No newline at end of file diff --git a/src/publish_namespace_handler.h b/src/publish_namespace_handler.h index 6363d9f..fb37f05 100644 --- a/src/publish_namespace_handler.h +++ b/src/publish_namespace_handler.h @@ -3,6 +3,8 @@ #include "quicr/handlers/publish_namespace_handler.h" #include "quicr/track_name.h" +#include + namespace laps { class PublishTrackHandler; diff --git a/src/relay_health_check.cpp b/src/relay_health_check.cpp index bef299a..d0f1e3e 100644 --- a/src/relay_health_check.cpp +++ b/src/relay_health_check.cpp @@ -3,9 +3,10 @@ #include "relay_health_check_internal.h" -#include "quicr/client.h" #include "quicr/handlers/publish_track_handler.h" #include "quicr/handlers/subscribe_track_handler.h" +#include "quicr/session.h" +#include "quicr/session_manager.h" #include @@ -113,7 +114,9 @@ namespace { std::shared_ptr> result_; }; - std::shared_ptr MakeClient(const std::string& endpoint_id, const Options& options) + std::shared_ptr MakeClient(SessionManager& session_manager, + const std::string& endpoint_id, + const Options& options) { ClientConfig config; config.endpoint_id = endpoint_id; @@ -123,11 +126,12 @@ namespace { const std::uint64_t idle_multiplier = options.gateway_enabled ? 4 : 2; config.transport_config.idle_timeout_ms = static_cast(options.timeout.count()) * idle_multiplier; - return Client::Create(config); + + return session_manager.AddTransport(config).lock(); } - bool RunPubSubProbe(Client& subscriber, - Client& publisher, + bool RunPubSubProbe(Session& subscriber, + Session& publisher, const Options& options, const std::string& unique_suffix, std::string& error) @@ -192,7 +196,7 @@ namespace { return true; } - bool RunSubscribeProbe(Client& subscriber, + bool RunSubscribeProbe(Session& subscriber, const FullTrackName& track, const std::string& track_display, const Bytes& expected_payload, @@ -248,37 +252,41 @@ namespace { { const auto unique_suffix = options.name.value_or(DefaultTrackName()); - auto subscriber = MakeClient("relay-health-subscriber-" + unique_suffix, options); - auto publisher = MakeClient("relay-health-publisher-" + unique_suffix, options); + SessionManager session_manager; + + auto subscriber = MakeClient(session_manager, "relay-health-subscriber-" + unique_suffix, options); + auto publisher = MakeClient(session_manager, "relay-health-publisher-" + unique_suffix, options); - subscriber->Start(); - publisher->Start(); + if (subscriber == nullptr || publisher == nullptr) { + error = "[relay] failed to create the publisher and subscriber sessions"; + return false; + } const bool connected = WaitFor( [&subscriber, &publisher]() { - return subscriber->GetStatus() == Client::Status::kReady && - publisher->GetStatus() == Client::Status::kReady; + return subscriber->GetStatus() == Session::Status::kReady && + publisher->GetStatus() == Session::Status::kReady; }, options.timeout); if (!connected) { error = "[relay] publisher and subscriber did not both connect before timeout"; - subscriber->Stop(); - publisher->Stop(); + subscriber->Disconnect(); + publisher->Disconnect(); return false; } const bool relay_ok = RunPubSubProbe(*subscriber, *publisher, options, unique_suffix, error); if (!relay_ok) { - subscriber->Stop(); - publisher->Stop(); + subscriber->Disconnect(); + publisher->Disconnect(); return false; } if (options.gateway_enabled) { if (options.gateway_name.empty()) { error = "[gateway] LIBQUICR_GATEWAY_HEALTH_NAME (or --gateway-name) is empty"; - subscriber->Stop(); - publisher->Stop(); + subscriber->Disconnect(); + publisher->Disconnect(); return false; } @@ -293,14 +301,14 @@ namespace { const bool gateway_ok = RunSubscribeProbe(*subscriber, gateway_track, display, expected_payload, options.timeout, error); if (!gateway_ok) { - subscriber->Stop(); - publisher->Stop(); + subscriber->Disconnect(); + publisher->Disconnect(); return false; } } - subscriber->Stop(); - publisher->Stop(); + subscriber->Disconnect(); + publisher->Disconnect(); return true; } } diff --git a/src/spdlog_logger.h b/src/spdlog_logger.h new file mode 100644 index 0000000..35b5c35 --- /dev/null +++ b/src/spdlog_logger.h @@ -0,0 +1,69 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 Cisco Systems +// SPDX-License-Identifier: BSD-2-Clause + +#pragma once + +#include + +#include + +#include +#include +#include +#include + +namespace laps { + /** + * @brief quicr::Logger backed by an spdlog logger + * + * @details libquicr logs through its own abstract logger interface, so relay logging is bridged to the + * spdlog logger the rest of the relay uses. Levels are not cached here, so changing the level on the + * spdlog logger also changes what libquicr logs. + */ + class SpdlogLogger : public quicr::Logger + { + public: + explicit SpdlogLogger(std::shared_ptr logger) + : logger_(std::move(logger)) + { + } + + void SetLevel(Level max_level) override { logger_->set_level(ToSpdlogLevel(max_level)); } + + bool ShouldLog(Level level) const noexcept override { return logger_->should_log(ToSpdlogLevel(level)); } + + /// @note quicr::Logger has already formatted the message, so this only forwards it + void Log(Level level, std::string_view msg, std::source_location location) override + { + logger_->log( + spdlog::source_loc(location.file_name(), static_cast(location.line()), location.function_name()), + ToSpdlogLevel(level), + msg); + } + + private: + static spdlog::level::level_enum ToSpdlogLevel(Level level) noexcept + { + switch (level) { + case Level::Trace: + return spdlog::level::trace; + case Level::Debug: + return spdlog::level::debug; + case Level::Info: + return spdlog::level::info; + case Level::Warn: + return spdlog::level::warn; + case Level::Error: + return spdlog::level::err; + case Level::Critical: + return spdlog::level::critical; + case Level::Off: + return spdlog::level::off; + } + + return spdlog::level::info; + } + + std::shared_ptr logger_; + }; +} // namespace laps diff --git a/src/subscribe_handler.cc b/src/subscribe_handler.cc index c01320d..d466858 100644 --- a/src/subscribe_handler.cc +++ b/src/subscribe_handler.cc @@ -522,8 +522,10 @@ namespace laps { default: break; } + // The alias is assigned when the handler is bound to its connection, which is after a + // publisher-initiated subscribe is accepted, so it can still be unset here SPDLOG_DEBUG("Track alias: {} subscribe status change reason: {} status: {}", - GetTrackAlias().value(), + GetTrackAlias().value_or(0), reason, static_cast(status)); } diff --git a/src/subscribe_handler.h b/src/subscribe_handler.h index 022374a..3974666 100644 --- a/src/subscribe_handler.h +++ b/src/subscribe_handler.h @@ -41,6 +41,15 @@ namespace laps { void SetFromPeer(); + /** + * @brief Record that forwarding is paused without sending a subscribe update + * + * @details A publisher-initiated subscribe is not bound to its connection until the relay accepts the + * publish, so Pause() has no session to send the update on. The relay instead answers the publish + * with forwarding off, and uses this so that a later Resume() sends the update. + */ + void MarkPaused() { SetStatus(Status::kPaused); } + std::optional GetPendingNewRquestId() { return pending_new_group_request_id_; }; struct PublisherLastUpdateInfo diff --git a/src/transport_helpers.h b/src/transport_helpers.h new file mode 100644 index 0000000..8b52958 --- /dev/null +++ b/src/transport_helpers.h @@ -0,0 +1,63 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 Cisco Systems +// SPDX-License-Identifier: BSD-2-Clause + +#pragma once + +#include +#include + +#include +#include +#include +#include +#include +#include + +namespace laps { + + /** + * @brief Peer IP address and port of a connection + * + * @details libquicr no longer reports the remote endpoint with the new connection callback, so it has to + * be queried from the transport that owns the connection. + * + * @param transport Transport that owns the connection + * @param connection Connection to get the peer address of + * + * @returns Pair of IP address string and port. Empty address and zero port if unavailable. + */ + inline std::pair GetPeerAddress(const std::shared_ptr& transport, + const std::shared_ptr& connection) + { + if (transport == nullptr || connection == nullptr) { + return {}; + } + + sockaddr_storage addr{}; + if (!transport->GetPeerAddrInfo(connection, &addr)) { + return {}; + } + + char ip[INET6_ADDRSTRLEN]{ 0 }; + + switch (addr.ss_family) { + case AF_INET: { + const auto* sa = reinterpret_cast(&addr); + if (inet_ntop(AF_INET, &sa->sin_addr, ip, sizeof(ip)) == nullptr) { + return {}; + } + return { ip, ntohs(sa->sin_port) }; + } + case AF_INET6: { + const auto* sa = reinterpret_cast(&addr); + if (inet_ntop(AF_INET6, &sa->sin6_addr, ip, sizeof(ip)) == nullptr) { + return {}; + } + return { ip, ntohs(sa->sin6_port) }; + } + default: + return {}; + } + } + +} // namespace laps diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt index 6e14dac..d87c356 100644 --- a/test/CMakeLists.txt +++ b/test/CMakeLists.txt @@ -24,7 +24,7 @@ add_executable(laps_test ) target_include_directories(laps_test PRIVATE ${PROJECT_SOURCE_DIR}/src) -target_link_libraries(laps_test PRIVATE quicr doctest::doctest relay_health_check_lib) +target_link_libraries(laps_test PRIVATE quicr spdlog::spdlog doctest::doctest relay_health_check_lib) target_compile_options(laps_test PRIVATE