diff --git a/CMakeLists.txt b/CMakeLists.txt index a32b870..958864a 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -18,7 +18,7 @@ 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.23 + VERSION 0.18.31 DESCRIPTION "Latency Aware Publish/Subscribe" LANGUAGES CXX) @@ -57,10 +57,11 @@ 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) +# quicr/utilities/format.h selects std::format or {fmt} with QUICR_HAS_STD_FORMAT. +# libquicr only adds that definition under its own directory, so relay sources that +# include quicr headers have to repeat the same choice. +if(QUICR_HAS_STD_FORMAT) + add_compile_definitions(QUICR_HAS_STD_FORMAT) endif() # spdlog compiles a bundled copy of fmt by default. libquicr also links fmt::fmt diff --git a/dependencies/libquicr b/dependencies/libquicr index 6fcf5e5..aa73094 160000 --- a/dependencies/libquicr +++ b/dependencies/libquicr @@ -1 +1 @@ -Subproject commit 6fcf5e5287d2e080a7b517a42faccf3672afc114 +Subproject commit aa73094dc3b4d1024e01d5f246d6c41b6c56e716 diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index 20bd703..f2de364 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -5,7 +5,6 @@ if(CLANG_TIDY_EXE) set_target_properties(quicr PROPERTIES CXX_CLANG_TIDY "${CLANG_TIDY_EXE}") endif() - add_executable(lapsRelay main.cc config.cc @@ -30,7 +29,11 @@ add_executable(lapsRelay ) target_link_libraries(lapsRelay PRIVATE quicr spdlog::spdlog) -target_include_directories(lapsRelay PRIVATE ${PROJECT_SOURCE_DIR}/dependencies/oss ) +target_include_directories(lapsRelay PRIVATE + ${PROJECT_SOURCE_DIR}/dependencies/oss + # Peer sessions read quicr::Stream directly. That type is defined for the library, not installed + # with the public headers. + ${PROJECT_SOURCE_DIR}/dependencies/libquicr/src) target_compile_options(lapsRelay PRIVATE diff --git a/src/client_manager.cc b/src/client_manager.cc index 04974d9..b438969 100644 --- a/src/client_manager.cc +++ b/src/client_manager.cc @@ -122,7 +122,7 @@ namespace laps { // Remove all subscribe announces for this connection handle std::vector remove_ns; - for (auto& [ns, conns] : state_.subscribes_namespaces) { + for (auto& [ns, conns] : state_.subscribe_tracks) { auto it = conns.find(connection_handle); if (it != conns.end()) { remove_ns.emplace_back(ns); @@ -133,6 +133,16 @@ namespace laps { UnsubscribeNamespaceReceived(session, ns); } + std::vector remove_sub_ns; + for (const auto& [ns, conns] : state_.subscribe_namespaces) { + if (conns.find(connection_handle) != conns.end()) { + remove_sub_ns.push_back(ns); + } + } + for (const auto& ns : remove_sub_ns) { + DropSubscribeNamespace(connection_handle, ns); + } + // Clean up subscribe states std::vector> unsub_list; for (auto it = state_.subscribe_alias_req_id.lower_bound({ connection_handle, 0 }); @@ -277,17 +287,13 @@ namespace laps { 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, - }); + quicr::SubscribeAttributes update_attrs; + update_attrs.priority = kDefaultPriority; + update_attrs.delivery_timeout = std::chrono::milliseconds(kDefaultObjectTtl); + update_attrs.expires = std::chrono::milliseconds(0); + update_attrs.forward = 1; + update_attrs.new_group_request_id = group_id; + peer_manager_.ClientSubscribeUpdate(track_full_name, update_attrs); // Notify all publishers that there is a new group request for (auto it = state_.pub_subscribes.lower_bound({ th.track_fullname_hash, 0 }); @@ -316,7 +322,7 @@ namespace laps { std::uint64_t exclude_connection) { // TODO: Fix O(prefix namespaces) matching - for (const auto& [prefix, conns] : state_.subscribes_namespaces) { + for (const auto& [prefix, conns] : state_.subscribe_tracks) { if (!prefix.HasSamePrefix(track_namespace)) { continue; } @@ -363,9 +369,94 @@ namespace laps { } } - quicr::Reply ClientManager::PublishNamespaceDoneReceived( - const std::shared_ptr& session, - std::uint64_t request_id) + void ClientManager::NotifySubscribeNamespaces(const quicr::TrackNamespace& track_namespace, + std::uint64_t exclude_connection, + bool done) + { + for (const auto& [prefix, connections] : state_.subscribe_namespaces) { + const auto match = prefix.IsPrefixOf(track_namespace); + if (match == std::partial_ordering::unordered || match == std::partial_ordering::greater) { + continue; + } + + for (const auto connection_handle : connections) { + if (connection_handle == exclude_connection && !config_.allow_self) { + continue; + } + + const auto key = std::make_pair(track_namespace, connection_handle); + if (done) { + if (notified_namespaces_.erase(key) == 0) { + continue; + } + } else if (!notified_namespaces_.emplace(key).second) { + continue; + } + + auto session = GetSession(connection_handle); + if (session == nullptr) { + if (!done) { + notified_namespaces_.erase(key); + } + continue; + } + + if (done) { + session->SendNamespaceDone(track_namespace); + } else { + session->SendNamespace(track_namespace); + } + + SPDLOG_LOGGER_DEBUG(LOGGER, + "{} subscribe namespace connection handle: {} namespace hash: {}", + done ? "NAMESPACE_DONE to" : "NAMESPACE to", + connection_handle, + quicr::TrackHash({ track_namespace, {} }).track_namespace_hash); + } + } + } + + bool ClientManager::NamespaceStillPublished(const quicr::TrackNamespace& track_namespace) const + { + for (const auto& [key, _] : state_.pub_namespace_active) { + if (key.first == track_namespace) { + return true; + } + } + + for (const auto& [_, handler] : state_.pub_subscribes) { + if (handler != nullptr && handler->GetFullTrackName().name_space == track_namespace) { + return true; + } + } + + return false; + } + + void ClientManager::DropSubscribeNamespace(std::uint64_t connection_handle, const quicr::TrackNamespace& prefix) + { + const auto it = state_.subscribe_namespaces.find(prefix); + if (it == state_.subscribe_namespaces.end() || it->second.erase(connection_handle) == 0) { + return; + } + + if (it->second.empty()) { + state_.subscribe_namespaces.erase(it); + } + + for (auto notified = notified_namespaces_.begin(); notified != notified_namespaces_.end();) { + const auto match = prefix.IsPrefixOf(notified->first); + if (notified->second == connection_handle && + (match == std::partial_ordering::less || match == std::partial_ordering::equivalent)) { + notified = notified_namespaces_.erase(notified); + } else { + ++notified; + } + } + } + + void ClientManager::PublishNamespaceDoneReceived(const std::shared_ptr& session, + std::uint64_t request_id) { const auto connection_handle = ConnectionHandle(session); @@ -377,7 +468,7 @@ namespace laps { connection_handle, request_id); - return {}; + return; } SPDLOG_LOGGER_DEBUG(LOGGER, @@ -410,10 +501,12 @@ namespace laps { state_.pub_namespace_active.erase({ track_namespace, connection_handle }); + if (!NamespaceStillPublished(track_namespace)) { + NotifySubscribeNamespaces(track_namespace, connection_handle, true); + } + if (connection_handle) peer_manager_.ClientUnannounce({ track_namespace, {} }); - - return {}; } void ClientManager::PurgePublishState(std::uint64_t connection_handle) @@ -522,7 +615,7 @@ namespace laps { auto sub_track_handler = std::make_shared(sub_ftn, 0, std::nullopt, *this, config_.tick_service_); - // Add subsribers to publisher subscribe handler + // Add subscribers to publisher subscribe handler for (const auto& sub_info : sub_tracks) { sub_track_handler->AddSubscriber(sub_info.connection_handle, sub_info.request_id, @@ -578,6 +671,8 @@ namespace laps { peer_manager_.ClientAnnounce({ track_namespace, {} }, attrs, false, true); } + NotifySubscribeNamespaces(track_namespace, connection_handle, false); + return {}; } @@ -623,7 +718,7 @@ namespace laps { publish_attributes.track_full_name.NameStr()); // PublishTrack within publish namespace handler if matched - for (auto& [tn, conns] : state_.subscribes_namespaces) { + for (auto& [tn, conns] : state_.subscribe_tracks) { const auto prefix_match = tn.IsPrefixOf(publish_attributes.track_full_name.name_space); if (prefix_match == std::partial_ordering::less || prefix_match == std::partial_ordering::equivalent) { @@ -644,7 +739,8 @@ namespace laps { const auto handler = PublishTrackHandler::Create( publish_attributes.track_full_name, quicr::TrackMode::kStream, - publish_attributes.default_publisher_priority, + publish_attributes.publisher_priority, + publish_attributes.publisher_priority, publish_attributes.delivery_timeout.value_or(config_.object_ttl_), publish_attributes.largest_object.value_or(quicr::messages::Location{ 0, 0 }), *this); @@ -664,6 +760,7 @@ namespace laps { *this, config_.tick_service_, true); + sub_track_handler->SetSubscriberPriority(publish_attributes.publisher_priority); if (bypass_resolve) { // Accepting the publish normally binds the handler to its connection, which is what assigns the @@ -692,7 +789,21 @@ namespace laps { } // Check if there are any subscribers - bool has_subs{ peer_manager_.HasSubscribers(th.track_fullname_hash) }; + bool has_peer_subs{ peer_manager_.HasSubscribers(th.track_fullname_hash) }; + bool has_subs{ has_peer_subs }; + + if (has_peer_subs) { + // libquicr copies the publisher's timeout onto the handler only after this callback returns, and + // the largest object is optional on the publish. A peer that is already subscribed hits this + // before either value exists. + sub_track_handler->AddSubscriber( + 0, + 0, + sub_track_handler->GetSubscriberPriority(), + std::chrono::milliseconds(publish_attributes.delivery_timeout.value_or(config_.object_ttl_)), + publish_attributes.largest_object.value_or(quicr::messages::Location{ 0, 0 })); + } + for (auto it = state_.subscribes.lower_bound({ th.track_fullname_hash, 0 }); it != state_.subscribes.end(); ++it) { if (it->first.first == th.track_fullname_hash) { @@ -706,7 +817,7 @@ namespace laps { } quicr::TrackNamespace sub_ns; - for (const auto& [ns_prefix, conns] : state_.subscribes_namespaces) { + for (const auto& [ns_prefix, conns] : state_.subscribe_tracks) { if (ns_prefix.HasSamePrefix(publish_attributes.track_full_name.name_space) && !conns.empty()) { has_subs = true; @@ -749,6 +860,8 @@ namespace laps { state_.conn_state_metrics[connection_handle].published_tracks++; + NotifySubscribeNamespaces(publish_attributes.track_full_name.name_space, connection_handle, false); + return quicr::PublishResponse{ { .subscriber_priority = std::nullopt, .group_order = std::nullopt, .filter = {}, @@ -759,6 +872,96 @@ namespace laps { sub_track_handler }; } + quicr::Reply, quicr::RequestErrorCode> ClientManager::SubscribeNamespaceReceived( + const std::shared_ptr& session, + const quicr::TrackNamespace& prefix_namespace, + [[maybe_unused]] const quicr::SubscribeNamespaceAttributes& attributes) + { + + const auto connection_handle = ConnectionHandle(session); + auto th = quicr::TrackHash({ prefix_namespace, {} }); + + auto [ns_it, _] = state_.subscribe_namespaces.try_emplace(prefix_namespace); + const auto inserted = ns_it->second.insert(connection_handle).second; + + if (inserted) { + SPDLOG_INFO( + "Subscribe namespace received connection handle: {} for namespace_hash: {} prefix: {}, adding to state", + connection_handle, + th.track_namespace_hash, + prefix_namespace.Str()); + } else { + SPDLOG_INFO("Duplicate subscribe namespace connection handle: {} for namespace_hash: {} prefix: {}", + connection_handle, + th.track_namespace_hash, + prefix_namespace.Str()); + return quicr::Unexpected>( + quicr::RequestErrorCode::kDuplicateSubscription, "Duplicate subscribe namespace"); + } + + std::vector matched_ns; + + // TODO: Fix O(prefix namespaces) matching + for (const auto& [key, _] : state_.pub_namespace_active) { + if (!config_.allow_self && key.second == connection_handle) { + continue; + } + + // Add matching announced namespaces to vector without duplicates + const auto prefix_match = prefix_namespace.IsPrefixOf(key.first); + if ((prefix_match == std::partial_ordering::less || prefix_match == std::partial_ordering::equivalent) && + (matched_ns.empty() || matched_ns.back() != key.first)) { + matched_ns.push_back(key.first); + notified_namespaces_.emplace(key.first, connection_handle); + } + } + + // 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) { + if (!handler || (!config_.allow_self && ta_conn.second == connection_handle)) { + continue; + } + + const auto& track_full_name = handler->GetFullTrackName(); + const auto prefix_match = prefix_namespace.IsPrefixOf(track_full_name.name_space); + if (prefix_match == std::partial_ordering::less || prefix_match == std::partial_ordering::equivalent) { + std::optional largest_location = GetLargestAvailable(track_full_name); + + /* + * PublishTrack within the namespace handler determines if the track should be published or not + * based on filters + */ + const auto pub_handler = PublishTrackHandler::Create( + track_full_name, + quicr::TrackMode::kStream, + handler->GetSubscriberPriority(), + handler->GetSubscriberPriority(), + handler->GetDeliveryTimeout().value_or(std::chrono::milliseconds(kDefaultObjectTtl)).count(), + largest_location.value_or(quicr::messages::Location{ 0, 0 }), + *this); + + if (!pub_handler->GetTrackAlias().has_value()) { + auto pub_th = quicr::TrackHash(track_full_name); + pub_handler->SetTrackAlias(pub_th.track_fullname_hash); + } + + if (notified_namespaces_.emplace(track_full_name.name_space, connection_handle).second) { + matched_ns.push_back(track_full_name.name_space); + } + + SPDLOG_LOGGER_DEBUG( + LOGGER, + "Matched PUBLISH track for SUBSCRIBE_NAMESPACE: conn: {} track_alias: {} track_hash: {}", + connection_handle, + ta_conn.first, + quicr::TrackHash(track_full_name).track_fullname_hash); + } + } + + return matched_ns; + } + quicr::Reply, quicr::RequestErrorCode> ClientManager::SubscribeTracksReceived( const std::shared_ptr& session, const quicr::TrackNamespace& prefix_namespace, @@ -767,7 +970,7 @@ namespace laps { const auto connection_handle = ConnectionHandle(session); auto th = quicr::TrackHash({ prefix_namespace, {} }); - auto [it, is_new] = state_.subscribes_namespaces.try_emplace(prefix_namespace); + auto [it, is_new] = state_.subscribe_tracks.try_emplace(prefix_namespace); auto handler = PublishNamespaceHandler::Create(prefix_namespace, GetTickService()); PublishNamespace(connection_handle, handler); @@ -833,7 +1036,8 @@ namespace laps { const auto pub_handler = PublishTrackHandler::Create( track_full_name, quicr::TrackMode::kStream, - handler->GetPriority(), + handler->GetSubscriberPriority(), + handler->GetSubscriberPriority(), handler->GetDeliveryTimeout().value_or(std::chrono::milliseconds(kDefaultObjectTtl)).count(), largest_location.value_or(quicr::messages::Location{ 0, 0 }), *this); @@ -859,14 +1063,15 @@ namespace laps { return matched_ns; } - quicr::Reply ClientManager::UnsubscribeNamespaceReceived( - const std::shared_ptr& session, - const quicr::TrackNamespace& prefix_namespace) + void 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 {}; + DropSubscribeNamespace(connection_handle, prefix_namespace); + + auto it = state_.subscribe_tracks.find(prefix_namespace); + if (it == state_.subscribe_tracks.end()) { + return; } auto th = quicr::TrackHash({ prefix_namespace, {} }); @@ -878,9 +1083,9 @@ namespace laps { auto pub_it = it->second.find(connection_handle); if (pub_it == it->second.end()) { if (it->second.empty()) { - state_.subscribes_namespaces.erase(it); + state_.subscribe_tracks.erase(it); } - return {}; + return; } // Loop through publishes and remove subscribe namespace @@ -902,28 +1107,24 @@ namespace laps { it->second.erase(pub_it); if (it->second.empty()) { - state_.subscribes_namespaces.erase(it); + state_.subscribe_tracks.erase(it); track_rankings_.erase(th.track_namespace_hash); } - - return {}; } - quicr::Reply ClientManager::ClientSetupReceived( + quicr::Expected> ClientManager::SetupReceived( const std::shared_ptr& session, - const quicr::ClientSetupAttributes& client_setup_attributes) + const quicr::SetupAttributes& setup_attributes) { const auto connection_handle = ConnectionHandle(session); - metrics_publisher_.SetConnectionEndpointId(connection_handle, client_setup_attributes.endpoint_id); + metrics_publisher_.SetConnectionEndpointId(connection_handle, setup_attributes.endpoint_id); - SPDLOG_LOGGER_INFO(LOGGER, "Client setup received from endpoint_id: {0}", client_setup_attributes.endpoint_id); + SPDLOG_LOGGER_INFO(LOGGER, "Setup received from endpoint_id: {0}", setup_attributes.endpoint_id); return {}; } - quicr::Reply ClientManager::PublishDoneReceived( - const std::shared_ptr& session, - uint64_t request_id) + void ClientManager::PublishDoneReceived(const std::shared_ptr& session, uint64_t request_id) { const auto connection_handle = ConnectionHandle(session); SPDLOG_LOGGER_INFO( @@ -935,7 +1136,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) @@ -1002,12 +1203,12 @@ namespace laps { } } - const auto& pub_ns = s_it->second->GetFullTrackName().name_space; - const auto& pub_ta = s_it->second->GetTrackAlias().value_or(0); + const auto published_namespace = publisher_handler->GetFullTrackName().name_space; + const auto pub_ta = publisher_handler->GetTrackAlias().value_or(0); // UnPublishTrack within publish namespace handler if matched - for (auto& [tn, conns] : state_.subscribes_namespaces) { - const auto prefix_match = tn.IsPrefixOf(pub_ns); + for (auto& [tn, conns] : state_.subscribe_tracks) { + const auto prefix_match = tn.IsPrefixOf(published_namespace); if (prefix_match == std::partial_ordering::less || prefix_match == std::partial_ordering::equivalent) { for (auto& [_, pub_ns_h] : conns) { @@ -1028,12 +1229,12 @@ namespace laps { publisher_handler->RemoveFromTrackRanking(); } - return {}; + if (!NamespaceStillPublished(published_namespace)) { + NotifySubscribeNamespaces(published_namespace, connection_handle, true); + } } - quicr::Reply ClientManager::UnsubscribeReceived( - const std::shared_ptr& session, - uint64_t request_id) + void 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); @@ -1042,7 +1243,7 @@ namespace laps { 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); @@ -1055,7 +1256,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; @@ -1092,8 +1293,6 @@ namespace laps { state_.subscribes.erase(sub_it); RemoveOrPausePublisherSubscribe(th.track_fullname_hash); - - return {}; } void ClientManager::RemoveOrPausePublisherSubscribe(std::uint64_t track_fullname_hash) @@ -1147,18 +1346,16 @@ namespace laps { } } - quicr::Reply ClientManager::TrackStatusReceived( + 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, - "Track status request connection handle: {} request_id: {} track alias: {}", + "Track status request connection handle: {} track alias: {}", connection_handle, - request_id, th.track_fullname_hash); const auto largest = GetLargestAvailable(track_full_name); @@ -1172,10 +1369,9 @@ namespace laps { } if (it->first.second != connection_handle) { - return quicr::RequestResponse{ - .is_publisher_initiated = it->second->IsPublisherInitiated(), - .largest_location = largest, - }; + quicr::TrackStatusResponse response; + response.largest_location = largest; + return response; } } @@ -1183,7 +1379,7 @@ namespace laps { "Track does not exist"); } - quicr::Reply ClientManager::SubscribeReceived( + quicr::Reply ClientManager::SubscribeReceived( const std::shared_ptr& session, uint64_t request_id, const quicr::FullTrackName& track_full_name, @@ -1226,10 +1422,10 @@ namespace laps { */ ProcessSubscribe(connection_handle, request_id, th, track_full_name, attrs, largest); - return quicr::RequestResponse{ - .is_publisher_initiated = attrs.is_publisher_initiated, - .largest_location = largest, - }; + quicr::SubscribeResponse response; + response.largest_location = largest; + response.is_publisher_initiated = attrs.is_publisher_initiated; + return response; } std::optional ClientManager::GetLargestAvailable(const quicr::FullTrackName& track_name) @@ -1264,7 +1460,6 @@ namespace laps { quicr::messages::FetchEndLocation end) { using Reply = quicr::Reply; - using Result = Reply::ResultType; using Failure = quicr::Unexpected>; const auto th = quicr::TrackHash(track_full_name); @@ -1311,7 +1506,7 @@ namespace laps { * 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 { + return Reply::Defer([=, this](Reply::CompletionType&& complete) { auto track_handler = FetchTrackHandler::Create(pub_fetch_h, track_full_name, priority, @@ -1333,7 +1528,8 @@ namespace laps { if (pub_connection_handle == 0) { UnbindFetchTrack(connection_handle, pub_fetch_h); - return Failure(*cache_miss, "Cannot process fetch"); + complete(Failure(*cache_miss, "Cannot process fetch")); + return; } SPDLOG_LOGGER_DEBUG(LOGGER, @@ -1373,18 +1569,21 @@ namespace laps { break; case quicr::FetchTrackHandler::Status::kError: - return Failure(quicr::FetchErrorCode::kNoObjects, "Cannot process fetch"); + complete(Failure(quicr::FetchErrorCode::kNoObjects, "Cannot process fetch")); + return; default: - return Failure(quicr::FetchErrorCode::kInternalError, "Cannot process fetch"); + complete(Failure(quicr::FetchErrorCode::kInternalError, "Cannot process fetch")); + return; } const auto upstream_largest = track_handler->GetLatestLocation(); if (!upstream_largest.has_value()) { - return Failure(quicr::FetchErrorCode::kNoObjects, "Cannot process fetch"); + complete(Failure(quicr::FetchErrorCode::kNoObjects, "Cannot process fetch")); + return; } - return quicr::FetchResponse{ upstream_largest, resolved_group_order }; + complete(quicr::FetchResponse{ upstream_largest, resolved_group_order }); }); } @@ -1563,6 +1762,7 @@ namespace laps { { auto start_location = attrs.start_location; + bool sub_from_peer{ false }; if (largest.has_value()) { SPDLOG_LOGGER_INFO(LOGGER, "Subscribe largest group: {} object: {}", largest->group, largest->object); @@ -1577,6 +1777,8 @@ namespace laps { attrs.new_group_request_id ? std::to_string(*attrs.new_group_request_id) : std::string("none"), track_full_name.NamespaceStr(), track_full_name.NameStr()); + + sub_from_peer = true; } else { @@ -1616,22 +1818,7 @@ namespace laps { attrs.group_order, {} }); - // TODO: Check draft Section 7.3 re forwarding group order - auto params = - quicr::messages::Parameters{}.Add(quicr::messages::ParameterType::kSubscriberPriority, attrs.priority); - if (attrs.group_order.has_value()) { - params.Add(quicr::messages::ParameterType::kGroupOrder, *attrs.group_order); - } - - auto sub_data = quicr::messages::Message() - .Append(request_id) - .Append(track_full_name.name_space) - .Append(track_full_name.name) - .Append(params); - - // TODO: Current new group is not sent by client in subscribe. It's only in subscribe updates. - - peer_manager_.ClientSubscribe(track_full_name, attrs, sub_data.ToByteSpan()); + peer_manager_.ClientSubscribe(track_full_name, attrs); } // Resume publisher initiated subscribes @@ -1641,14 +1828,11 @@ namespace laps { if (it->first.first != th.track_fullname_hash) { break; } - if (it->second->IsPublisherInitiated()) { - it->second->Resume(); - } it->second->AddSubscriber( connection_handle, request_id, attrs.priority, attrs.delivery_timeout, attrs.start_location); - DampenOrUpdateTrackSubscription(it->second, attrs.new_group_request_id.has_value()); + DampenOrUpdateTrackSubscription(it->second, attrs.new_group_request_id.has_value() || sub_from_peer); } // Subscribe to announcer if announcer is active @@ -1681,10 +1865,6 @@ namespace laps { state_.pub_subscribes_by_req_id[{ sub_track_h->GetRequestId().value(), key.second }] = sub_track_h; state_.pub_subscribes[{ th.track_fullname_hash, key.second }] = sub_track_h; - if (attrs.new_group_request_id) { - sub_track_h->RequestNewGroup(); - } - } else { auto pub_handler_it = state_.pub_subscribes.find({ th.track_fullname_hash, key.second }); if (pub_handler_it != state_.pub_subscribes.end()) { @@ -1731,14 +1911,7 @@ namespace laps { } if (stream_id.has_value()) { - if (is_new_stream) { - quicr::InitialStreamData initial_buffer; - initial_buffer.buffer.Push(*data); - initial_buffer.source_buffers.push_back(std::move(data)); - it->second->StreamDataRecv(*stream_id, std::move(initial_buffer)); - } else { - it->second->StreamDataRecv(*stream_id, std::move(data)); - } + it->second->PeerStreamDataReceived(*stream_id, std::move(data), is_new_stream); } else { it->second->DgramDataRecv(std::move(data)); } @@ -1751,7 +1924,7 @@ namespace laps { return; } - it->second->StreamClosed(stream_id, reset); + it->second->PeerStreamClosedReceived(stream_id, reset); } void ClientManager::MetricsSampled(const std::shared_ptr& session, diff --git a/src/client_manager.h b/src/client_manager.h index bfb22e9..91ed32a 100644 --- a/src/client_manager.h +++ b/src/client_manager.h @@ -164,27 +164,30 @@ namespace laps { void MetricsSampled(const std::shared_ptr& session, const quicr::ConnectionMetrics& metrics) override; - quicr::Reply ClientSetupReceived( + quicr::Expected> SetupReceived( const std::shared_ptr& session, - const quicr::ClientSetupAttributes& client_setup_attributes) override; + const quicr::SetupAttributes& 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( + quicr::Reply, quicr::RequestErrorCode> SubscribeNamespaceReceived( const std::shared_ptr& session, - const quicr::TrackNamespace& prefix_namespace) override; + const quicr::TrackNamespace& prefix_namespace, + const quicr::SubscribeNamespaceAttributes& attributes) override; + + void 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; + void PublishNamespaceDoneReceived(const std::shared_ptr& session, + std::uint64_t request_id) override; quicr::Reply PublishReceived( const std::shared_ptr& session, @@ -192,24 +195,21 @@ namespace laps { 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; + void PublishDoneReceived(const std::shared_ptr& session, std::uint64_t request_id) override; - quicr::Reply SubscribeReceived( + 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; + void 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( + quicr::Reply TrackStatusReceived( const std::shared_ptr& session, - std::uint64_t request_id, const quicr::FullTrackName& track_full_name) override; quicr::Reply StandaloneFetchReceived( @@ -259,6 +259,27 @@ namespace laps { */ void FanoutPublishNamespaceDone(const quicr::TrackNamespace& track_namespace); + /** + * @brief Send NAMESPACE or NAMESPACE_DONE to subscribe-namespace subscribers whose prefix matches + * + * @param track_namespace Fully qualified namespace that became active or ended + * @param exclude_connection Publishing connection. Skipped unless self-forwarding is allowed + * @param done True sends NAMESPACE_DONE, false sends NAMESPACE + */ + void NotifySubscribeNamespaces(const quicr::TrackNamespace& track_namespace, + std::uint64_t exclude_connection, + bool done); + + /** + * @brief True when some publisher still has this namespace announced or a track in it + */ + bool NamespaceStillPublished(const quicr::TrackNamespace& track_namespace) const; + + /** + * @brief Drop one subscribe-namespace subscription and any NAMESPACE already sent for it + */ + void DropSubscribeNamespace(std::uint64_t connection_handle, const quicr::TrackNamespace& prefix); + quicr::Reply PublishReceivedInternal( std::uint64_t connection_handle, uint64_t request_id, @@ -298,6 +319,9 @@ namespace laps { std::map, std::shared_ptr> fanout_namespaces_; + /// Namespaces already reported with NAMESPACE, keyed by namespace and subscriber connection + std::set> notified_namespaces_; + /** * @brief Map of atomic bools to mark if a fetch thread should be interrupted. */ diff --git a/src/fetch_handler.cc b/src/fetch_handler.cc index 815eea3..fd78053 100644 --- a/src/fetch_handler.cc +++ b/src/fetch_handler.cc @@ -18,64 +18,14 @@ namespace laps { { } - void FetchTrackHandler::StreamDataRecv(uint64_t stream_id, quicr::InitialStreamData&& initial_buffer) + void FetchTrackHandler::ObjectReceived( + const quicr::ObjectHeaders& object_headers, + quicr::BytesSpan data, + [[maybe_unused]] std::optional stream_mode) { - subscribe_track_metrics_.bytes_received += initial_buffer.buffer.Size(); - - auto [stream_it, inserted] = - streams_.try_emplace(stream_id, StreamContext{ .buffer = std::move(initial_buffer.buffer) }); - if (!inserted) { - SPDLOG_ERROR("StreamDataRecv got new stream for existing Stream ID {}", stream_id); - return; - } - - stream_it->second.buffer.InitAny(); - TryForwardInitialStreamData(stream_id, stream_it->second); - } - - void FetchTrackHandler::StreamDataRecv(uint64_t stream_id, std::shared_ptr> data) - { - subscribe_track_metrics_.bytes_received += data->size(); - - if (initial_stream_data_forwarded_) { - publish_fetch_handler_->ForwardPublishedData(false, 0, 0, std::move(data)); - return; - } - - const auto stream_it = streams_.find(stream_id); - if (stream_it == streams_.end()) { - SPDLOG_ERROR("StreamDataRecv had no stream for expected Stream ID {}", stream_id); - return; - } - - stream_it->second.buffer.Push(*data); - TryForwardInitialStreamData(stream_id, stream_it->second); - } - - void FetchTrackHandler::TryForwardInitialStreamData(uint64_t stream_id, StreamContext& stream) - { - auto& f_hdr = stream.buffer.GetAny(); - if (!(stream.buffer >> f_hdr)) { - return; - } - - SPDLOG_DEBUG("Fetch header added in rid: {} out rid: {} remaining data size: {}", - f_hdr.request_id, - *publish_fetch_handler_->GetRequestId(), - stream.buffer.Size()); - - f_hdr.request_id = *publish_fetch_handler_->GetRequestId(); - auto bytes = std::make_shared(); - *bytes << f_hdr; - - const auto remaining_data = stream.buffer.Data(); - if (!remaining_data.empty()) { - bytes->insert(bytes->end(), remaining_data.begin(), remaining_data.end()); - } - - publish_fetch_handler_->ForwardPublishedData(true, 0, 0, std::move(bytes)); - initial_stream_data_forwarded_ = true; - streams_.erase(stream_id); + // The session has already parsed the fetch header, so the outbound fetch writes its own, using this + // handler's request id. + publish_fetch_handler_->PublishObject(object_headers, data); } void FetchTrackHandler::StatusChanged(Status status) diff --git a/src/fetch_handler.h b/src/fetch_handler.h index 1955136..6ae90fd 100644 --- a/src/fetch_handler.h +++ b/src/fetch_handler.h @@ -32,13 +32,11 @@ namespace laps { } void StatusChanged(Status status) override; - void StreamDataRecv(uint64_t stream_id, quicr::InitialStreamData&& initial_buffer) override; - void StreamDataRecv(uint64_t stream_id, std::shared_ptr> data) override; + void ObjectReceived(const quicr::ObjectHeaders& object_headers, + quicr::BytesSpan data, + std::optional stream_mode = std::nullopt) override; private: - void TryForwardInitialStreamData(uint64_t stream_id, StreamContext& stream); - - bool initial_stream_data_forwarded_{ false }; std::shared_ptr publish_fetch_handler_; }; } // namespace laps diff --git a/src/peering/info_base.h b/src/peering/info_base.h index 62e7604..2977cf2 100644 --- a/src/peering/info_base.h +++ b/src/peering/info_base.h @@ -7,9 +7,11 @@ #include "peer_session.h" #include "peering/messages/subscribe_info.h" +#include #include #include #include +#include namespace laps::peering { @@ -148,17 +150,45 @@ namespace laps::peering { std::map> subscribes_; + /** + * @brief Identity of one egress stream held by a forwarding entry + * + * @details Client entries use `a` as the group id and `b` as the subgroup id. Peer entries use `a` as + * the ingress QUIC stream id and leave `b` at 0. Those two maps do not share entries. + */ + struct StreamKey + { + uint64_t a{ 0 }; + uint64_t b{ 0 }; + + friend bool operator==(const StreamKey& lhs, const StreamKey& rhs) noexcept + { + return lhs.a == rhs.a && lhs.b == rhs.b; + } + }; + + struct StreamKeyHash + { + std::size_t operator()(const StreamKey& key) const noexcept + { + const auto h1 = std::hash{}(key.a); + const auto h2 = std::hash{}(key.b); + return h1 ^ (h2 + 0x9e3779b97f4a7c15ULL + (h1 << 6) + (h1 >> 2)); + } + }; + struct FibEntry { uint64_t update_ref{ 0 }; ///< Random reference number to detect if entry was updated or not /** - * @brief Egress stream for each ingress stream, keyed by ingress stream ID + * @brief Egress stream for each forwarded flow * * @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. + * stream usable for as long as objects are being forwarded onto it. See StreamKey for how + * each map identifies a flow. */ - std::unordered_map> streams; + std::unordered_map, StreamKeyHash> streams; SubscribeNodeSetId out_sns_id; ///< Egress SNS ID decltype(nodes_best_)::mapped_type peer_session; diff --git a/src/peering/messages/subscribe_info.cc b/src/peering/messages/subscribe_info.cc index ba535b5..4a6dc4c 100644 --- a/src/peering/messages/subscribe_info.cc +++ b/src/peering/messages/subscribe_info.cc @@ -3,7 +3,29 @@ #include "subscribe_info.h" +#include + namespace laps::peering { + namespace { + template + T Read(std::span::iterator& it, const std::span::iterator end) + { + if (static_cast(end - it) < sizeof(T)) { + throw std::out_of_range("Subscribe info truncated"); + } + + const T value = ValueOf({ it, it + static_cast(sizeof(T)) }); + it += sizeof(T); + return value; + } + + template + void Append(std::vector& data, const T& value) + { + const auto bytes = BytesOf(value); + data.insert(data.end(), bytes.rbegin(), bytes.rend()); + } + } SubscribeInfo::SubscribeInfo([[maybe_unused]] std::uint64_t track_fullname_hash, NodeIdValueType source_node_id, @@ -16,64 +38,91 @@ namespace laps::peering { uint32_t SubscribeInfo::SizeBytes() const { return sizeof(seq) + sizeof(source_node_id) + 24 /* namespace, name, and full name hashes */ - + 4 /* size of sub data */ + subscribe_data.size(); + + sizeof(priority) + sizeof(delivery_timeout) + sizeof(expires) + sizeof(forward) + + 1 /* new group present */ + + (new_group_request_id.has_value() ? sizeof(uint64_t) : 0) + 1 /* namespace entry count */ + + sizeof(uint16_t) * static_cast(name_space.GetEntries().size()) + + static_cast(name_space.size()) + sizeof(uint16_t) + static_cast(name.size()); } SubscribeInfo::SubscribeInfo(std::span serialized_data) : track_hash({}) { auto it = serialized_data.begin(); + const auto end = serialized_data.end(); + + seq = Read(it, end); + source_node_id = Read(it, end); + + track_hash.track_namespace_hash = Read(it, end); + track_hash.track_name_hash = Read(it, end); + track_hash.track_fullname_hash = Read(it, end); - seq = ValueOf({ it, it + 2 }); - it += 2; + priority = Read(it, end); + delivery_timeout = Read(it, end); + expires = Read(it, end); + forward = Read(it, end); - source_node_id = ValueOf({ it, it + 8 }); - it += 8; + if (Read(it, end) != 0) { + new_group_request_id = Read(it, end); + } - track_hash.track_namespace_hash = ValueOf({ it, it + 8 }); - it += 8; - track_hash.track_name_hash = ValueOf({ it, it + 8 }); - it += 8; - track_hash.track_fullname_hash = ValueOf({ it, it + 8 }); - it += 8; + const auto num_entries = Read(it, end); + std::vector> entries; + entries.reserve(num_entries); - uint32_t sub_size = ValueOf({ it, it + 4 }); - it += 4; + for (uint8_t i = 0; i < num_entries; ++i) { + const auto len = Read(it, end); + if (static_cast(end - it) < len) { + throw std::out_of_range("Subscribe namespace entry exceeds serialized data"); + } - if (sub_size == 0) { - return; + entries.emplace_back(it, it + len); + it += len; } - if (sub_size > serialized_data.end() - it) { - throw std::out_of_range("Subscribe data size is larger than serialized data size"); + name_space = quicr::TrackNamespace(std::span>{ entries }); + + const auto name_size = Read(it, end); + if (static_cast(end - it) < name_size) { + throw std::out_of_range("Subscribe name exceeds serialized data"); } - subscribe_data.assign(it, it + sub_size); - // it += sub_size; + if (name_size != 0) { + name.assign(it, it + name_size); + } } std::vector& operator<<(std::vector& data, const SubscribeInfo& subscribe_info) { - auto seq_bytes = BytesOf(subscribe_info.seq); - data.insert(data.end(), seq_bytes.rbegin(), seq_bytes.rend()); - - auto src_node_bytes = BytesOf(subscribe_info.source_node_id); - data.insert(data.end(), src_node_bytes.rbegin(), src_node_bytes.rend()); + Append(data, subscribe_info.seq); + Append(data, subscribe_info.source_node_id); - auto namespace_bytes = BytesOf(subscribe_info.track_hash.track_namespace_hash); - data.insert(data.end(), namespace_bytes.rbegin(), namespace_bytes.rend()); + Append(data, subscribe_info.track_hash.track_namespace_hash); + Append(data, subscribe_info.track_hash.track_name_hash); + Append(data, subscribe_info.track_hash.track_fullname_hash); - auto name_bytes = BytesOf(subscribe_info.track_hash.track_name_hash); - data.insert(data.end(), name_bytes.rbegin(), name_bytes.rend()); + Append(data, subscribe_info.priority); + Append(data, subscribe_info.delivery_timeout); + Append(data, subscribe_info.expires); + Append(data, subscribe_info.forward); - auto full_name_bytes = BytesOf(subscribe_info.track_hash.track_fullname_hash); - data.insert(data.end(), full_name_bytes.rbegin(), full_name_bytes.rend()); + Append(data, static_cast(subscribe_info.new_group_request_id.has_value() ? 1 : 0)); + if (subscribe_info.new_group_request_id.has_value()) { + Append(data, *subscribe_info.new_group_request_id); + } - uint32_t sub_size = subscribe_info.subscribe_data.size(); - auto sub_size_bytes = BytesOf(sub_size); - data.insert(data.end(), sub_size_bytes.rbegin(), sub_size_bytes.rend()); + const auto& entries = subscribe_info.name_space.GetEntries(); + data.push_back(static_cast(entries.size())); + for (const auto& entry : entries) { + Append(data, static_cast(entry.size())); + data.insert(data.end(), entry.begin(), entry.end()); + } - data.insert(data.end(), subscribe_info.subscribe_data.begin(), subscribe_info.subscribe_data.end()); + Append(data, static_cast(subscribe_info.name.size())); + if (!subscribe_info.name.empty()) { + data.insert(data.end(), subscribe_info.name.begin(), subscribe_info.name.end()); + } return data; } diff --git a/src/peering/messages/subscribe_info.h b/src/peering/messages/subscribe_info.h index be70e35..8990dc2 100644 --- a/src/peering/messages/subscribe_info.h +++ b/src/peering/messages/subscribe_info.h @@ -4,6 +4,7 @@ #include "node_info.h" #include +#include #include #include @@ -29,12 +30,20 @@ namespace laps::peering { NodeIdValueType source_node_id; ///< Id of the originating source node quicr::TrackHash track_hash; ///< Full name hash - std::vector - subscribe_data; /// Original MoQ subscribe message (wire format) that initiated this subscribe - // Data not in wire message (e.g., serialized) + /// Track identity previously carried inside the MoQ subscribe message. + quicr::TrackNamespace name_space; + std::vector name; - // End not serialized + /** + * Subscribe attributes. Encoded on the wire as fixed-width integers, the same way as track_hash. + * new_group_request_id is preceded by a presence byte and omitted when unset. + */ + uint8_t priority{ 0 }; + uint64_t delivery_timeout{ 0 }; ///< Milliseconds + uint64_t expires{ 0 }; ///< Milliseconds + uint64_t forward{ 0 }; ///< Non-zero forwards data, zero pauses it + std::optional new_group_request_id; /** * @brief Encode node object into bytes that can be written on the wire diff --git a/src/peering/peer_manager.cc b/src/peering/peer_manager.cc index df3e7e7..2db926f 100644 --- a/src/peering/peer_manager.cc +++ b/src/peering/peer_manager.cc @@ -12,6 +12,46 @@ #include namespace laps::peering { + namespace { + void ApplySubscribeAttributes(SubscribeInfo& info, const quicr::SubscribeAttributes& attrs) + { + info.priority = attrs.priority; + info.delivery_timeout = static_cast(attrs.delivery_timeout.count()); + info.expires = static_cast(attrs.expires.count()); + info.forward = attrs.forward; + info.new_group_request_id = attrs.new_group_request_id; + } + + quicr::SubscribeAttributes MakeSubscribeAttributes(const SubscribeInfo& info) + { + quicr::SubscribeAttributes attrs; + attrs.priority = info.priority; + attrs.delivery_timeout = std::chrono::milliseconds(static_cast(info.delivery_timeout)); + attrs.expires = std::chrono::milliseconds(static_cast(info.expires)); + attrs.forward = static_cast(info.forward); + attrs.new_group_request_id = info.new_group_request_id; + return attrs; + } + + InfoBase::StreamKey SubgroupStreamKey(uint64_t group_id, uint64_t subgroup_id) + { + return { .a = group_id, .b = subgroup_id }; + } + + InfoBase::StreamKey IngressStreamKey(uint64_t stream_id) + { + return { .a = stream_id, .b = 0 }; + } + + void AdvanceSeqIfUnchanged(SubscribeInfo& info, uint16_t seq_before) + { + if (info.seq != seq_before) { + return; + } + + info.seq = info.seq < 0xFFFF ? static_cast(info.seq + 1) : static_cast(0); + } + } // ------------------------------------------------------------ // Methods used by peer session to feedback info and actions @@ -102,18 +142,10 @@ namespace laps::peering { if (not withdraw) { uint64_t update_ref = rand(); - auto msg_bytes = quicr::BytesSpan(subscribe_info.subscribe_data); - - // subscribe headers in expected order, must each be parsed - [[maybe_unused]] const auto request_id = quicr::messages::Message::ParseField(msg_bytes); - const auto track_namespace = quicr::messages::Message::ParseField(msg_bytes); - const auto track_name = quicr::messages::Message::ParseField(msg_bytes); - const auto parameters = quicr::messages::Message::ParseField(msg_bytes); - - auto tfn = quicr::FullTrackName{ track_namespace, track_name }; - - auto priority = parameters.Get(quicr::messages::ParameterType::kSubscriberPriority); - auto ngr_id = parameters.GetOptional(quicr::messages::ParameterType::kNewGroupRequest); + const auto& track_namespace = subscribe_info.name_space; + const auto& track_name = subscribe_info.name; + const auto priority = subscribe_info.priority; + const auto s_attrs = MakeSubscribeAttributes(subscribe_info); std::lock_guard _(state_.state_mutex); bool announce_matches{ false }; @@ -145,10 +177,6 @@ namespace laps::peering { subscribe_info.track_hash.track_fullname_hash); if (client_manager_ != nullptr) { - quicr::SubscribeAttributes s_attrs; - s_attrs.priority = 10; - s_attrs.new_group_request_id = ngr_id; - SPDLOG_LOGGER_INFO(LOGGER, "Subscribe to client manager track alias: {} new_group_request: {}", subscribe_info.track_hash.track_fullname_hash, @@ -158,37 +186,36 @@ namespace laps::peering { client_manager_->ProcessSubscribe( 0, 0, subscribe_info.track_hash, { track_namespace, track_name }, s_attrs, std::nullopt); } + } - auto bp_it = info_base_->nodes_best_.find(subscribe_info.source_node_id); - if (bp_it != info_base_->nodes_best_.end()) { - const auto& peer_session = bp_it->second.lock(); - SPDLOG_LOGGER_DEBUG( - LOGGER, - "Best peer session for subscribe fullname: {} source_node: {} is via peer_session_id: {}", - subscribe_info.track_hash.track_fullname_hash, - subscribe_info.source_node_id, - peer_session->GetSessionId()); - - if (auto [sns_id, is_new] = peer_session->AddSubscribeSourceNode( - subscribe_info.track_hash.track_fullname_hash, subscribe_info.source_node_id, priority); + auto bp_it = info_base_->nodes_best_.find(subscribe_info.source_node_id); + if (bp_it != info_base_->nodes_best_.end()) { + const auto& peer_session = bp_it->second.lock(); + SPDLOG_LOGGER_DEBUG( + LOGGER, + "Best peer session for subscribe fullname: {} source_node: {} is via peer_session_id: {}", + subscribe_info.track_hash.track_fullname_hash, + subscribe_info.source_node_id, + peer_session->GetSessionId()); + + if (auto [sns_id, is_new] = peer_session->AddSubscribeSourceNode( + subscribe_info.track_hash.track_fullname_hash, subscribe_info.source_node_id, priority); + is_new) { + SPDLOG_LOGGER_INFO(LOGGER, + "New source added to peer session for subscribe fullname: {} source_node: {} is " + "via peer_session_id: {} sns_id: {}", + subscribe_info.track_hash.track_fullname_hash, + subscribe_info.source_node_id, + peer_session->GetSessionId(), + sns_id); + + if (auto [_, is_new] = info_base_->client_fib_.try_emplace( + { subscribe_info.track_hash.track_fullname_hash, peer_session_id }, + InfoBase::FibEntry{ update_ref, {}, sns_id, bp_it->second }); is_new) { - SPDLOG_LOGGER_INFO( - LOGGER, - "New source added to peer session for subscribe fullname: {} source_node: {} is " - "via peer_session_id: {} sns_id: {}", - subscribe_info.track_hash.track_fullname_hash, - subscribe_info.source_node_id, - peer_session->GetSessionId(), - sns_id); - - if (auto [_, is_new] = info_base_->client_fib_.try_emplace( - { subscribe_info.track_hash.track_fullname_hash, peer_session_id }, - InfoBase::FibEntry{ update_ref, {}, sns_id, bp_it->second }); - is_new) { - SPDLOG_LOGGER_INFO(LOGGER, - "New subscribe fullname: {}, sending subscribe to client manager", - subscribe_info.track_hash.track_fullname_hash); - } + SPDLOG_LOGGER_INFO(LOGGER, + "New subscribe fullname: {}, sending subscribe to client manager", + subscribe_info.track_hash.track_fullname_hash); } } } @@ -290,7 +317,7 @@ namespace laps::peering { .default_publisher_group_order = quicr::messages::GroupOrder::kAscending, .dynamic_groups = true, - .default_publisher_priority = 64, + .publisher_priority = 64, .max_cache_duration = std::nullopt, .delivery_timeout = kDefaultObjectTtl, .track_properties = {} }; @@ -330,6 +357,7 @@ namespace laps::peering { // Remove or find new best peer for subscribe info std::vector remove_sub; std::vector> update_sub; + std::unique_lock ib_lock(info_base_->mutex_); for (const auto& subs : info_base_->subscribes_) { auto sub_it = subs.second.find(remote_node_info.id); @@ -345,7 +373,11 @@ namespace laps::peering { info_base_->client_fib_.find({ si.track_hash.track_fullname_hash, peer_session_id }); if (cfib_it != info_base_->client_fib_.end()) { info_base_->client_fib_.erase(cfib_it); + + ib_lock.unlock(); auto best_peer = info_base_->GetBestPeerSession(si.source_node_id); + ib_lock.lock(); + if (const auto& peer_sess = best_peer.lock()) { // best path found, update entry to use new path update_sub.emplace_back(peer_sess->GetSessionId(), si); @@ -406,8 +438,6 @@ namespace laps::peering { auto it = info_base_->peer_fib_.find({ peer_session_id, data_header.sns_id }); if (it != info_base_->peer_fib_.end()) { // TODO(tievens): Replace data_out with new transport multiple data - std::vector data_out(data->begin(), data->end()); - for (auto& [out_peer_sess_id, entry] : it->second) { if (out_peer_sess_id == peer_session_id) continue; // Skip; don't send back to same peer or if it's self @@ -424,7 +454,6 @@ namespace laps::peering { continue; } - // TODO(tievens): Change to use DataStorage to avoid copies auto client_data = data; if (data_offset != 0) { client_data = std::make_shared>(data->begin() + data_offset, data->end()); @@ -437,21 +466,26 @@ namespace laps::peering { client_manager_->PeerDataReceived( si_it->second.track_hash.track_fullname_hash, false, std::nullopt, client_data); } - continue; } + auto data_out = std::make_shared>(data->begin(), data->end()); + // Update SNS_ID if new stream header included or if datagram (both have sns_id) if (is_new_stream || eflags.use_reliable == false) { auto sns_id_bytes = BytesOf(entry.out_sns_id); - std::copy(sns_id_bytes.rbegin(), sns_id_bytes.rend(), data_out.begin() + 2); + std::copy(sns_id_bytes.rbegin(), sns_id_bytes.rend(), data_out->begin() + 2); } auto out_peer_sess = entry.peer_session.lock(); + if (out_peer_sess == nullptr) { + continue; + } std::shared_ptr out_stream; if (eflags.use_reliable) { - auto sid_it = entry.streams.find(stream_id); + const auto key = IngressStreamKey(stream_id); + auto sid_it = entry.streams.find(key); if (sid_it == entry.streams.end()) { if (!is_new_stream) { @@ -459,16 +493,21 @@ namespace laps::peering { } out_stream = out_peer_sess->CreateStream(data_header.priority); - entry.streams.emplace(stream_id, out_stream); + if (out_stream == nullptr) { + continue; + } + + const auto [inserted_it, inserted] = entry.streams.try_emplace(key, out_stream); + if (!inserted) { + out_peer_sess->CloseStream(out_stream, quicr::StreamClosedFlag::kReset); + out_stream = inserted_it->second; + } } else { 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, out_stream, eflags, std::move(data_out_shared)); + out_peer_sess->SendData(data_header.priority, data_header.ttl, out_stream, eflags, std::move(data_out)); } } else { SPDLOG_LOGGER_DEBUG(config_.logger_, @@ -488,7 +527,7 @@ namespace laps::peering { uint64_t subgroup_id, bool reset) { - uint64_t in_stream_id = group_id << 16 | static_cast(subgroup_id); + const auto key = SubgroupStreamKey(group_id, subgroup_id); for (auto it = info_base_->client_fib_.lower_bound({ track_full_name_hash, 0 }); it != info_base_->client_fib_.end(); @@ -506,10 +545,13 @@ namespace laps::peering { peer_sess->GetSessionId(), fib_entry.out_sns_id); - auto stream_it = fib_entry.streams.find(in_stream_id); + auto stream_it = fib_entry.streams.find(key); if (stream_it != fib_entry.streams.end()) { - peer_sess->CloseStream(stream_it->second, - reset ? quicr::StreamClosedFlag::kReset : quicr::StreamClosedFlag::kFin); + if (reset) { + peer_sess->CloseStream(stream_it->second, quicr::StreamClosedFlag::kReset); + } else { + peer_sess->FinishStream(stream_it->second); + } fib_entry.streams.erase(stream_it); } } @@ -575,32 +617,42 @@ namespace laps::peering { std::shared_ptr out_stream; if (eflags.use_reliable) { - uint64_t in_stream_id = group_id << 16 | static_cast(subgroup_id); + const auto key = SubgroupStreamKey(group_id, subgroup_id); - auto stream_it = fib_entry.streams.find(in_stream_id); + auto stream_it = fib_entry.streams.find(key); if (stream_it == fib_entry.streams.end()) { if (data_header.type != DataType::kNewStream) { - return; + continue; } out_stream = peer_sess->CreateStream(priority); - fib_entry.streams.try_emplace(in_stream_id, out_stream); + if (out_stream == nullptr) { + continue; + } + + const auto [inserted_it, inserted] = fib_entry.streams.try_emplace(key, out_stream); + if (!inserted) { + peer_sess->CloseStream(out_stream, quicr::StreamClosedFlag::kReset); + out_stream = inserted_it->second; + } } else { out_stream = stream_it->second; } - SPDLOG_LOGGER_TRACE( - LOGGER, - "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, - track_full_name_hash, - group_id, - subgroup_id, - fib_entry.streams.size(), - net_data->size()); + if (out_stream == nullptr) { + continue; + } + + SPDLOG_LOGGER_TRACE(LOGGER, + "Data object send, peer_session: {} egress SNS_ID: {} tfn_hash: {} " + "group_id: {} subgroup_id: {} streams: {} data len: {}", + peer_sess->GetSessionId(), + fib_entry.out_sns_id, + track_full_name_hash, + group_id, + subgroup_id, + fib_entry.streams.size(), + net_data->size()); } peer_sess->SendData(priority, ttl, out_stream, eflags, send_data); @@ -634,98 +686,57 @@ namespace laps::peering { void PeerManager::ClientSubscribeUpdate(const quicr::FullTrackName& track_full_name, const quicr::SubscribeAttributes& attrs) { - auto tfn = track_full_name; - auto th = quicr::TrackHash(tfn); + const auto th = quicr::TrackHash(track_full_name); - // Check for existing subscribe and update if (auto si = info_base_->GetSubscribe(th.track_fullname_hash, node_info_.id)) { - try { - // Update subscription params with new group request - - // subscribe headers in expected order, must each be parsed - auto msg_bytes = quicr::BytesSpan(si->subscribe_data); - [[maybe_unused]] const auto request_id = quicr::messages::Message::ParseField(msg_bytes); - [[maybe_unused]] const auto track_namespace = - quicr::messages::Message::ParseField(msg_bytes); - [[maybe_unused]] const auto track_name = quicr::messages::Message::ParseField(msg_bytes); - auto parameters = quicr::messages::Message::ParseField(msg_bytes); - - auto ngr_id = parameters.GetOptional(quicr::messages::ParameterType::kNewGroupRequest); - - if (!ngr_id.has_value()) { - parameters.AddOptional(quicr::messages::ParameterType::kNewGroupRequest, ngr_id); - } - - auto new_group_request = - parameters.GetOptional(quicr::messages::ParameterType::kNewGroupRequest); - if (new_group_request.has_value() && !attrs.new_group_request_id.has_value()) { - // Remove new group request since it's not requested but was found - parameters.Remove(quicr::messages::ParameterType::kNewGroupRequest); + ApplySubscribeAttributes(*si, attrs); - auto sub_data = quicr::messages::Message() - .Append(request_id) - .Append(track_namespace) - .Append(track_name) - .Append(parameters); - - const auto sub_data_bytes = sub_data.ToByteSpan(); - si->subscribe_data.assign(sub_data_bytes.begin(), sub_data_bytes.end()); - } - - if (attrs.new_group_request_id.has_value() && not new_group_request.has_value()) { - parameters.Add(quicr::messages::ParameterType::kNewGroupRequest, *attrs.new_group_request_id); - auto sub_data = quicr::messages::Message() - .Append(request_id) - .Append(track_namespace) - .Append(track_name) - .Append(parameters); - - si->subscribe_data.assign(sub_data.ToByteSpan().begin(), sub_data.ToByteSpan().end()); - } - - } catch (const std::exception& e) { - SPDLOG_LOGGER_ERROR(LOGGER, "Unable to parse subscribe message {}", e.what()); - return; - } - - // Existing subscribe, update it with new data and attributes + const auto seq_before = si->seq; for (const auto& sess : client_peer_sessions_) { SPDLOG_LOGGER_DEBUG(LOGGER, - "Sending subscribe update fullname: {} peer_session_id: {} new_group: {}", + "Sending subscribe update fullname: {} peer_session_id: {} priority: {} " + "delivery_timeout: {} expires: {} forward: {} new_group: {}", th.track_fullname_hash, sess.first, + attrs.priority, + attrs.delivery_timeout.count(), + attrs.expires.count(), + attrs.forward, attrs.new_group_request_id.has_value() ? *attrs.new_group_request_id : -1); sess.second->SendSubscribeInfo(*si, false); } for (const auto& sess : server_peer_sessions_) { SPDLOG_LOGGER_DEBUG(LOGGER, - "Sending subscribe update fullname: {} peer_session_id: {} new_group: {}", + "Sending subscribe update fullname: {} peer_session_id: {} priority: {} " + "delivery_timeout: {} expires: {} forward: {} new_group: {}", th.track_fullname_hash, sess.first, + attrs.priority, + attrs.delivery_timeout.count(), + attrs.expires.count(), + attrs.forward, attrs.new_group_request_id.has_value() ? *attrs.new_group_request_id : -1); sess.second->SendSubscribeInfo(*si, false); } + + AdvanceSeqIfUnchanged(*si, seq_before); + info_base_->AddSubscribe(*si); } } void PeerManager::ClientSubscribe(const quicr::FullTrackName& track_full_name, - const quicr::SubscribeAttributes& attrs, - std::span subscribe_data) + const quicr::SubscribeAttributes& attrs) { - auto tfn = track_full_name; - - if (subscribe_data.empty()) - return; // Empty means it was supposed to be an update which didn't happen - SubscribeInfo si; - si.track_hash = quicr::TrackHash(tfn); - si.subscribe_data.assign(subscribe_data.begin(), subscribe_data.end()); + si.track_hash = quicr::TrackHash(track_full_name); + si.name_space = track_full_name.name_space; + si.name = track_full_name.name; si.source_node_id = node_info_.id; + ApplySubscribeAttributes(si, attrs); - info_base_->AddSubscribe(si); - + const auto seq_before = si.seq; for (const auto& sess : client_peer_sessions_) { SPDLOG_LOGGER_DEBUG(LOGGER, "Sending subscribe fullname: {} peer_session_id: {}", @@ -741,6 +752,9 @@ namespace laps::peering { sess.first); sess.second->SendSubscribeInfo(si, false); } + + AdvanceSeqIfUnchanged(si, seq_before); + info_base_->AddSubscribe(si); } void PeerManager::ClientAnnounce(const quicr::FullTrackName& track_full_name, @@ -804,77 +818,54 @@ namespace laps::peering { continue; const auto& sub_info = si_it.second; - try { - // subscribe headers in expected order, must each be parsed - auto msg_bytes = quicr::BytesSpan(sub_info.subscribe_data); - [[maybe_unused]] const auto request_id = - quicr::messages::Message::ParseField(msg_bytes); - const auto track_namespace = - quicr::messages::Message::ParseField(msg_bytes); - [[maybe_unused]] const auto track_name = - quicr::messages::Message::ParseField(msg_bytes); - auto parameters = quicr::messages::Message::ParseField(msg_bytes); - - auto ngr_id = - parameters.GetOptional(quicr::messages::ParameterType::kNewGroupRequest); - - if (track_full_name.name_space.HasSamePrefix(track_namespace)) { - - if (sub_info.source_node_id == node_info_.id) - continue; - - if (auto cm = client_manager_) { - quicr::SubscribeAttributes s_attrs; - s_attrs.priority = 10; - s_attrs.new_group_request_id = - parameters.GetOptional(quicr::messages::ParameterType::kNewGroupRequest); - - SPDLOG_LOGGER_INFO(LOGGER, - "Subscribe to client manager track alias: {}", - sub_info.track_hash.track_fullname_hash); - - cm->ProcessSubscribe( - 0, 0, sub_info.track_hash, { track_namespace, track_name }, s_attrs, std::nullopt); - } + if (track_full_name.name_space.HasSamePrefix(sub_info.name_space)) { + + if (sub_info.source_node_id == node_info_.id) + continue; + + const auto s_attrs = MakeSubscribeAttributes(sub_info); + + if (auto cm = client_manager_) { + SPDLOG_LOGGER_INFO(LOGGER, + "Subscribe to client manager track alias: {}", + sub_info.track_hash.track_fullname_hash); + + cm->ProcessSubscribe( + 0, 0, sub_info.track_hash, { sub_info.name_space, sub_info.name }, s_attrs, std::nullopt); + } - auto bp_it = info_base_->nodes_best_.find(sub_info.source_node_id); - if (bp_it != info_base_->nodes_best_.end()) { - const auto& peer_session = bp_it->second.lock(); - SPDLOG_LOGGER_DEBUG(LOGGER, - "Best peer session for subscribe fullname: {} source_node: {} is " - "via peer_session_id: {}", - sub_info.track_hash.track_fullname_hash, - sub_info.source_node_id, - peer_session->GetSessionId()); - - if (auto [sns_id, is_new] = peer_session->AddSubscribeSourceNode( - sub_info.track_hash.track_fullname_hash, - sub_info.source_node_id, - parameters.Get(quicr::messages::ParameterType::kSubscriberPriority)); + auto bp_it = info_base_->nodes_best_.find(sub_info.source_node_id); + if (bp_it != info_base_->nodes_best_.end()) { + const auto& peer_session = bp_it->second.lock(); + SPDLOG_LOGGER_DEBUG(LOGGER, + "Best peer session for subscribe fullname: {} source_node: {} is " + "via peer_session_id: {}", + sub_info.track_hash.track_fullname_hash, + sub_info.source_node_id, + peer_session->GetSessionId()); + + if (auto [sns_id, is_new] = peer_session->AddSubscribeSourceNode( + sub_info.track_hash.track_fullname_hash, sub_info.source_node_id, sub_info.priority); + is_new) { + SPDLOG_LOGGER_INFO( + LOGGER, + "New source added to peer session for subscribe fullname: {} source_node: {} is " + "via peer_session_id: {} sns_id: {}", + sub_info.track_hash.track_fullname_hash, + sub_info.source_node_id, + peer_session->GetSessionId(), + sns_id); + + if (auto [_, is_new] = info_base_->client_fib_.try_emplace( + { sub_info.track_hash.track_fullname_hash, peer_session->GetSessionId() }, + InfoBase::FibEntry{ update_ref, {}, sns_id, bp_it->second }); is_new) { - SPDLOG_LOGGER_INFO( - LOGGER, - "New source added to peer session for subscribe fullname: {} source_node: {} is " - "via peer_session_id: {} sns_id: {}", - sub_info.track_hash.track_fullname_hash, - sub_info.source_node_id, - peer_session->GetSessionId(), - sns_id); - - if (auto [_, is_new] = info_base_->client_fib_.try_emplace( - { sub_info.track_hash.track_fullname_hash, peer_session->GetSessionId() }, - InfoBase::FibEntry{ update_ref, {}, sns_id, bp_it->second }); - is_new) { - SPDLOG_LOGGER_INFO(LOGGER, - "New subscribe fullname: {} added to client fib", - sub_info.track_hash.track_fullname_hash); - } + SPDLOG_LOGGER_INFO(LOGGER, + "New subscribe fullname: {} added to client fib", + sub_info.track_hash.track_fullname_hash); } } } - } catch (const std::exception& e) { - SPDLOG_LOGGER_ERROR(LOGGER, "Unable to parse subscribe message {}", e.what()); - continue; } } } @@ -1038,7 +1029,8 @@ namespace laps::peering { NodeId().Value(node_id)); } else { - // Existing entry + // Existing entry. Keep the egress streams already carrying this subgroup. + // Replacing the entry drops that map, and later objects are discarded until a new stream. const auto [o_sns_id, is_new] = peer_sess->AddPeerSnsSourceNode(peer_session.GetSessionId(), sns.id, node_id, sns.priority); if (is_new) { @@ -1049,8 +1041,9 @@ namespace laps::peering { NodeId().Value(node_id)); } - fib_it->second[peer_sess->GetSessionId()] = - InfoBase::FibEntry{ update_ref, {}, o_sns_id, peer_sess_weak }; + it->second.update_ref = update_ref; + it->second.out_sns_id = o_sns_id; + it->second.peer_session = peer_sess_weak; } } } @@ -1318,7 +1311,7 @@ namespace laps::peering { 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); + peer_sess->FinishStream(out_stream); } } @@ -1344,7 +1337,7 @@ namespace laps::peering { continue; } - auto stream_it = entry.streams.find(stream_id); + auto stream_it = entry.streams.find(IngressStreamKey(stream_id)); if (stream_it != entry.streams.end()) { if (auto out_peer_sess = entry.peer_session.lock()) { out_peer_sess->CloseStream(stream_it->second, flag); @@ -1353,21 +1346,6 @@ namespace laps::peering { } } } - - // Notify the client manager of closed stream - for (auto& [key, entry] : info_base_->client_fib_) { - if (key.second != peer_session_id) { - continue; - } - - if (auto out_peer_sess = entry.peer_session.lock()) { - 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); - } - } - } } } // namespace laps diff --git a/src/peering/peer_manager.h b/src/peering/peer_manager.h index fa5e851..8905940 100644 --- a/src/peering/peer_manager.h +++ b/src/peering/peer_manager.h @@ -70,9 +70,7 @@ namespace laps::peering { void ClientSubscribeUpdate(const quicr::FullTrackName& track_full_name, const quicr::SubscribeAttributes&); - void ClientSubscribe(const quicr::FullTrackName& track_full_name, - const quicr::SubscribeAttributes&, - std::span subscribe_data); + void ClientSubscribe(const quicr::FullTrackName& track_full_name, const quicr::SubscribeAttributes&); void ClientUnsubscribe(uint64_t track_fullname_hash); diff --git a/src/peering/peer_session.cc b/src/peering/peer_session.cc index 47d54a3..29107cb 100644 --- a/src/peering/peer_session.cc +++ b/src/peering/peer_session.cc @@ -8,8 +8,11 @@ #include "peering/messages/node_info.h" #include "peering/messages/subscribe_info.h" +#include + #include #include +#include namespace laps::peering { @@ -111,7 +114,7 @@ namespace laps::peering { auto [__, is_new] = sns.nodes.emplace(sub_node_id); - if (is_new) { + if (is_new and node_info_.id != sub_node_id) { SendSns(sns, false); } @@ -225,6 +228,31 @@ namespace laps::peering { transport_->CloseStream(connection_, stream, operation); } + void PeerSession::FinishStream(const std::shared_ptr& stream) + { + if (stream == nullptr) { + return; + } + + // The transport FINs a stream when a queued object asks for it and the send queue has drained. + // Calling CloseStream here instead cuts off bytes still waiting to be written. The TTL has to + // stay within the transport time queue, which is twice the object TTL. + if (status_ == StatusValue::kConnected && IsUsable()) { + quicr::Transport::EnqueueFlags eflags; + eflags.use_reliable = true; + eflags.close_stream = true; + const uint32_t ttl = config_.object_ttl_ > 0 ? config_.object_ttl_ * 2 : config_.object_ttl_; + try { + SendData(0, ttl, stream, eflags, std::make_shared>()); + return; + } catch (const std::invalid_argument&) { + SPDLOG_LOGGER_DEBUG(LOGGER, "Queued stream FIN was rejected, closing stream directly"); + } + } + + CloseStream(stream, quicr::StreamClosedFlag::kFin); + } + void PeerSession::SendData(uint8_t priority, uint32_t ttl, const std::shared_ptr& stream, @@ -286,11 +314,13 @@ namespace laps::peering { if (status_ != StatusValue::kConnected || !IsUsable()) return; SPDLOG_LOGGER_DEBUG(LOGGER, - "Sending subscribe fullname: {} source_node_id: {} withdraw: {} sub_data_size: {}", + "Sending subscribe fullname: {} source_node_id: {} withdraw: {} priority: {} new_group: {}", subscribe_info.track_hash.track_fullname_hash, NodeId().Value(subscribe_info.source_node_id), withdraw, - subscribe_info.subscribe_data.size()); + subscribe_info.priority, + subscribe_info.new_group_request_id.has_value() ? *subscribe_info.new_group_request_id + : -1); SendControl(subscribe_info.Serialize(true, withdraw, node_info_.id == subscribe_info.source_node_id)); } @@ -554,47 +584,60 @@ namespace laps::peering { return true; } - void PeerSession::OnRecvStream(std::uint64_t stream_id, - const std::shared_ptr& rx_ctx, - const std::shared_ptr& stream, - const bool is_bidir) + bool PeerSession::OnRecvStream(const std::shared_ptr& stream) { - if (!IsUsable() || rx_ctx == nullptr) { - return; + if (!IsUsable() || stream == nullptr) { + return false; } + const auto stream_id = stream->GetStreamId(); + const bool is_bidir = stream->IsBidirectional(); + /* * 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) { + if (is_bidir) { control_stream_ = stream; control_stream_id_ = stream_id; } - for (int i = 0; i < kReadLoopMaxPerStream; i++) { - if (rx_ctx->data_queue.Empty()) { - break; - } + quicr::Bytes data; + { + std::lock_guard lock(stream->rx_mutex); - auto data_opt = rx_ctx->data_queue.Pop(); - if (not data_opt.has_value()) { - break; + // A new object stream names its header length in the first byte. Leave a short read where it + // is so the next arrival can complete it. + if (!is_bidir && !rx_stream_headers_.contains(stream_id)) { + const auto available = stream->rx_data.Data(); + if (available.empty() || available.size() < available.front()) { + return false; + } } - const auto& data = data_opt.value(); + data = stream->rx_data.TakeAll(); + } + + if (data.empty()) { + return false; + } - // Get common header - if (is_bidir) { // control - controL_msg_buffer_.insert(controL_msg_buffer_.end(), data->begin(), data->end()); + if (is_bidir) { + controL_msg_buffer_.insert(controL_msg_buffer_.end(), data.begin(), data.end()); + // One read can now hold several messages, where each used to arrive as its own chunk. + for (;;) { + const auto buffered = controL_msg_buffer_.size(); ProcessControlMessage(); - - } else if (!ProcessReceivedData(stream_id, std::move(data))) { - i = 59; - continue; // Try once more + if (controL_msg_buffer_.size() == buffered) { + break; + } } + } else { + ProcessReceivedData(stream_id, std::make_shared(std::move(data))); } + + return false; } void PeerSession::OnRecvDgram() @@ -630,10 +673,13 @@ namespace laps::peering { metrics_.srtt_us = quic_connection_metrics.srtt_us.avg; } - void PeerSession::OnStreamClosed(std::uint64_t stream_id, - [[maybe_unused]] std::shared_ptr rx_context, - quicr::StreamClosedFlag flag) + void PeerSession::OnStreamClosed(const std::shared_ptr& stream, quicr::StreamClosedFlag flag) { + if (stream == nullptr) { + return; + } + + const auto stream_id = stream->GetStreamId(); const auto conn_id = GetSessionId(); if (control_stream_ != nullptr && control_stream_id_.has_value() && *control_stream_id_ == stream_id) { diff --git a/src/peering/peer_session.h b/src/peering/peer_session.h index 8071756..07ed280 100644 --- a/src/peering/peer_session.h +++ b/src/peering/peer_session.h @@ -99,6 +99,14 @@ namespace laps::peering { std::shared_ptr CreateStream(uint8_t priority) const; void CloseStream(const std::shared_ptr& stream, quicr::StreamClosedFlag flag); + + /** + * @brief FIN an egress stream after the bytes already queued on it + * + * @details A direct FIN races the send queue and can drop the tail of the subgroup. This queues the + * FIN behind that data instead. + */ + void FinishStream(const std::shared_ptr& stream); 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); @@ -176,10 +184,7 @@ namespace laps::peering { */ 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; + bool OnRecvStream(const std::shared_ptr& stream) override; void OnRecvDgram() override; @@ -193,9 +198,7 @@ namespace laps::peering { { } - void OnStreamClosed(std::uint64_t stream_id, - std::shared_ptr rx_context, - quicr::StreamClosedFlag flag) override; + void OnStreamClosed(const std::shared_ptr& stream, quicr::StreamClosedFlag flag) override; // --------------------------------------- diff --git a/src/publish_handler.cc b/src/publish_handler.cc index cfa62dd..29a66f4 100644 --- a/src/publish_handler.cc +++ b/src/publish_handler.cc @@ -11,14 +11,16 @@ namespace laps { */ PublishTrackHandler::PublishTrackHandler(const quicr::FullTrackName& full_track_name, quicr::TrackMode track_mode, - uint8_t default_priority, + uint8_t publisher_priority, + uint8_t publish_object_priority, uint32_t default_ttl, quicr::messages::Location start_location, ClientManager& server, std::optional stream_mode) : quicr::PublishTrackHandler(full_track_name, track_mode, - default_priority, + publisher_priority, + publish_object_priority, default_ttl, stream_mode, start_location) @@ -113,4 +115,51 @@ namespace laps { server_.peer_manager_.EndSubgroup(GetTrackAlias().value(), group_id, subgroup_id, !completed); } } + + void PublishTrackHandler::RequestOkReceived(const quicr::messages::Parameters& params) + { + quicr::PublishTrackHandler::RequestOkReceived(params); + + switch (GetStatus()) { + case Status::kNewGroupRequested: + [[fallthrough]]; + case Status::kOk: + [[fallthrough]]; + case Status::kPaused: + [[fallthrough]]; + default: { + auto delivery_timeout = params.Get(quicr::messages::ParameterType::kDeliveryTimeout); + auto priority = params.Get(quicr::messages::ParameterType::kSubscriberPriority); + auto group_order = + params.GetOptional(quicr::messages::ParameterType::kGroupOrder); + const auto publisher_default_group_order = quicr::messages::GroupOrder::kAscending; + auto forward = params.Get(quicr::messages::ParameterType::kForward); + auto new_group_request_id = + params.GetOptional(quicr::messages::ParameterType::kNewGroupRequest); + + quicr::messages::Filter filter; + if (params.Contains(quicr::messages::ParameterType::kSubscriptionFilter)) { + filter = params.GetFilter(quicr::messages::FilterType::kSubscriptionFilter); + } else if (params.Contains(quicr::messages::ParameterType::kTrackFilter)) { + filter = params.GetFilter(quicr::messages::FilterType::kTrackFilter); + } + + quicr::SubscribeAttributes attrs; + attrs.delivery_timeout = std::chrono::milliseconds{ delivery_timeout }; + attrs.expires = attrs.delivery_timeout; + attrs.priority = priority; + attrs.group_order = group_order; + attrs.publisher_default_group_order = publisher_default_group_order; + attrs.forward = forward; + attrs.new_group_request_id = new_group_request_id; + + if (priority) { + publish_object_priority_ = priority; + } + + server_.peer_manager_.ClientSubscribe(GetFullTrackName(), attrs); + break; + } + } + } } diff --git a/src/publish_handler.h b/src/publish_handler.h index 813a055..726b3d1 100644 --- a/src/publish_handler.h +++ b/src/publish_handler.h @@ -12,7 +12,8 @@ namespace laps { public: PublishTrackHandler(const quicr::FullTrackName& full_track_name, quicr::TrackMode track_mode, - uint8_t default_priority, + uint8_t publisher_priority, + uint8_t publish_object_priority, uint32_t default_ttl, quicr::messages::Location start_location, ClientManager& server, @@ -32,15 +33,24 @@ namespace laps { static std::shared_ptr Create(const quicr::FullTrackName& full_track_name, quicr::TrackMode track_mode, - uint8_t default_priority, + uint8_t publisher_priority, + uint8_t publish_object_priority, uint32_t default_ttl, quicr::messages::Location start_location, ClientManager& server) { - return std::make_shared( - full_track_name, track_mode, default_priority, default_ttl, start_location, server); + return std::make_shared(full_track_name, + track_mode, + publisher_priority, + publish_object_priority, + default_ttl, + start_location, + server); } + protected: + void RequestOkReceived(const quicr::messages::Parameters& params) override; + private: ClientManager& server_; diff --git a/src/relay_health_check.cpp b/src/relay_health_check.cpp index d0f1e3e..e3026b4 100644 --- a/src/relay_health_check.cpp +++ b/src/relay_health_check.cpp @@ -146,7 +146,7 @@ namespace { const auto result_promise = std::make_shared>(); auto result_future = result_promise->get_future(); auto sub_handler = VerifyingSubscribeTrackHandler::Create(track, expected_payload, result_promise); - auto pub_handler = PublishTrackHandler::Create(track, TrackMode::kStream, 3, 1000, { 0, 0 }); + auto pub_handler = PublishTrackHandler::Create(track, TrackMode::kStream, 3, 3, 1000, { 0, 0 }); subscriber.SubscribeTrack(sub_handler); publisher.PublishTrack(pub_handler); diff --git a/src/state.h b/src/state.h index 0911157..66c0075 100644 --- a/src/state.h +++ b/src/state.h @@ -64,13 +64,22 @@ namespace laps { std::map, std::shared_ptr> pub_subscribes_by_req_id; /** - * @brief Subscribe Namespace by connection to publish namespace handlers - * @details Subscribe namespaces by connection are added to this map. Each will have an associated + * @brief Subscribe tracks by connection to publish namespace handlers + * @details Subscribe tracks by connection are added to this map. Each will have an associated * publish namespace handler. The publish namespace handler is used to establish publish tracks * to the subscriber of the namespace */ std::map>> - subscribes_namespaces; + subscribe_tracks; + + /** + * @brief Subscribe namespace prefix to the subscriber connection handles + * + * @details Each prefix keeps every connection that subscribed to it. NAMESPACE and NAMESPACE_DONE + * for a matching publish are sent on each of those connections' SUBSCRIBE_NAMESPACE streams. + * Removing the last connection removes the prefix. + */ + std::map> subscribe_namespaces; struct SubscribePublishHandlerInfo { diff --git a/src/subscribe_handler.cc b/src/subscribe_handler.cc index 6530707..241d314 100644 --- a/src/subscribe_handler.cc +++ b/src/subscribe_handler.cc @@ -6,7 +6,7 @@ #include "publish_handler.h" #include "publish_namespace_handler.h" -#include +#include #include namespace laps { @@ -16,12 +16,12 @@ namespace laps { ClientManager& server, std::weak_ptr tick_service, bool is_publisher_initiated) - : quicr::SubscribeTrackHandler(full_track_name, - priority, - group_order, - std::monostate{}, - std::nullopt, - is_publisher_initiated) + : quicr::ForwardingSubscribeTrackHandler(full_track_name, + priority, + group_order, + std::monostate{}, + std::nullopt, + is_publisher_initiated) , server_(server) , tick_service_(std::move(tick_service)) { @@ -82,6 +82,7 @@ namespace laps { std::chrono::milliseconds delivery_timeout, quicr::messages::Location start_location) { + // TODO: Consider allowing add Subscriber to trigger NGR for peering when conn_handle == 0 if (subscribers_.contains(conn_handle)) { // Duplicate return; @@ -91,7 +92,8 @@ namespace laps { auto pub_track_h = std::make_shared( GetFullTrackName(), is_datagram_ ? quicr::TrackMode::kDatagram : quicr::TrackMode::kStream, - priority == 0 ? GetPriority() : priority, + GetSubscriberPriority(), + priority == 0 ? GetSubscriberPriority() : priority, delivery_timeout.count() == 0 ? GetDeliveryTimeout().value_or(std::chrono::milliseconds(server_.config_.object_ttl_)).count() : delivery_timeout.count(), @@ -106,7 +108,7 @@ namespace laps { subscribers_.emplace(conn_handle, nullptr); } - Resume(); + RequestNewGroup(); // Use NGR to keep it clean for peering. NGR sets forward=1 } void SubscribeTrackHandler::RemoveSubscriber(std::uint64_t conn_handle) @@ -235,82 +237,277 @@ namespace laps { } } - void SubscribeTrackHandler::StreamDataRecv(uint64_t stream_id, quicr::InitialStreamData&& initial_buffer) + void SubscribeTrackHandler::SubgroupStarted(std::uint64_t group_id, + std::uint64_t subgroup_id, + std::optional priority, + quicr::messages::StreamHeaderProperties properties) { is_datagram_ = false; - auto [stream_it, inserted] = - streams_.try_emplace(stream_id, StreamContext{ .buffer = std::move(initial_buffer.buffer) }); - if (!inserted) { - SPDLOG_ERROR("StreamDataRecv got new stream for existing Stream ID {}", stream_id); + const auto key = std::pair{ group_id, subgroup_id }; + if (open_subgroups_.contains(key)) { + SPDLOG_ERROR("Subgroup already started for group {} subgroup {}", group_id, subgroup_id); + return; + } + + auto& subgroup = open_subgroups_[key]; + subgroup.properties.emplace(properties); + subgroup.priority = priority; + + if (!GetTrackAlias().has_value()) { + SPDLOG_DEBUG("Subgroup started without a track alias"); return; } - stream_it->second.buffer.InitAny(); - pending_source_buffers_.emplace(stream_id, std::move(initial_buffer.source_buffers)); - TryProcessStreamData(stream_id, stream_it->second); + // The alias in the publisher's header belongs to that session. Subscribers and peers use this one. + quicr::messages::StreamHeaderSubGroup header; + header.properties.emplace(properties); + header.track_alias = *GetTrackAlias(); + header.group_id = group_id; + if (properties.subgroup_id_mode == quicr::messages::SubgroupIdType::kExplicit) { + header.subgroup_id = subgroup_id; + } + if (!properties.default_priority) { + header.publisher_priority = priority; + } + + auto bytes = std::make_shared(); + *bytes << header; + ForwardReceivedData(true, group_id, subgroup_id, std::move(bytes)); + } + + void SubscribeTrackHandler::StreamBytesForwarded(std::uint64_t group_id, + std::uint64_t subgroup_id, + quicr::Bytes&& data) + { + is_datagram_ = false; + + const auto subgroup_it = open_subgroups_.find({ group_id, subgroup_id }); + if (subgroup_it == open_subgroups_.end()) { + SPDLOG_ERROR("Stream bytes for unknown group {} subgroup {}", group_id, subgroup_id); + return; + } + + if (data.empty()) { + return; + } + + auto shared = std::make_shared(std::move(data)); + // Before the first object, subscribers are still waiting on PublishObject, so this reaches peers only. + ForwardReceivedData(false, group_id, subgroup_id, shared); + subgroup_it->second.buffer.Push(*shared); + ConsumeForwardedObjects(group_id, subgroup_id, subgroup_it->second); + } + + void SubscribeTrackHandler::ConsumeForwardedObjects(std::uint64_t group_id, + std::uint64_t subgroup_id, + SubgroupForwardState& subgroup) + { + if (!subgroup.properties.has_value()) { + return; + } + + while (!subgroup.buffer.Empty()) { + if (!subgroup.buffer.AnyHasValueB()) { + subgroup.buffer.InitAnyB(); + } + + auto& obj = subgroup.buffer.GetAnyB(); + obj.properties.emplace(*subgroup.properties); + if (!(subgroup.buffer >> obj)) { + return; + } + + subscribe_track_metrics_.objects_received++; + + if (subgroup.next_object_id.has_value()) { + *subgroup.next_object_id += obj.object_delta; + } else { + subgroup.next_object_id = obj.object_delta; + } + + const bool is_first_object = !subgroup.saw_first_object; + const auto object_id = *subgroup.next_object_id; + + ObjectReceived({ group_id, + object_id, + subgroup_id, + obj.payload.size(), + obj.object_status, + subgroup.priority, + std::nullopt, + quicr::TrackMode::kStream, + obj.extensions, + obj.immutable_extensions }, + obj.payload, + subgroup.properties); + + *subgroup.next_object_id += 1; + subgroup.buffer.ResetAnyB(); + subgroup.saw_first_object = true; + + if (is_first_object) { + const auto remaining_data = subgroup.buffer.Data(); + if (!remaining_data.empty()) { + ForwardReceivedData( + false, + group_id, + subgroup_id, + std::make_shared>(remaining_data.begin(), remaining_data.end()), + false); + } + } + } + } + + void SubscribeTrackHandler::SubgroupEnded(std::uint64_t group_id, std::uint64_t subgroup_id, bool reset) + { + open_subgroups_.erase({ group_id, subgroup_id }); + + for (auto& [conn_handle, pub_handler] : subscribers_) { + if (conn_handle == 0) { + if (GetTrackAlias().has_value()) { + server_.peer_manager_.EndSubgroup(GetTrackAlias().value(), group_id, subgroup_id, reset); + } + } else { + pub_handler->EndSubgroup(group_id, subgroup_id, !reset); + } + } + + for (const auto [_, conn_subs] : sub_namespaces_) { + for (const auto& [_, handler] : conn_subs) { + handler->EndSubgroup(group_id, subgroup_id, !reset); + } + } } - void SubscribeTrackHandler::StreamDataRecv(uint64_t stream_id, std::shared_ptr> data) + void SubscribeTrackHandler::PeerStreamDataReceived(uint64_t stream_id, + std::shared_ptr> data, + bool is_new_stream) { is_datagram_ = false; + auto [stream_it, inserted] = peer_streams_.try_emplace(stream_id); - const auto stream_it = streams_.find(stream_id); - if (stream_it == streams_.end()) { - SPDLOG_ERROR("StreamDataRecv had no stream for expected Stream ID {}", stream_id); + if (is_new_stream) { + if (!inserted) { + SPDLOG_ERROR("Peer stream data got a new stream for existing Stream ID {}", stream_id); + return; + } + + stream_it->second.buffer.Push(*data); + stream_it->second.buffer.InitAny(); + pending_source_buffers_.emplace(stream_id, std::vector{ std::move(data) }); + ProcessStartOfStreamData(stream_id, stream_it->second, true); return; } - stream_it->second.buffer.Push(*data); + if (inserted) { + peer_streams_.erase(stream_it); + SPDLOG_ERROR("Peer stream data had no stream for expected Stream ID {}", stream_id); + return; + } + + auto& stream = stream_it->second; + stream.buffer.Push(*data); const auto pending_it = pending_source_buffers_.find(stream_id); if (pending_it != pending_source_buffers_.end()) { + // Header is still incomplete. Keep the original bytes so the start can be forwarded once it parses. pending_it->second.push_back(std::move(data)); + } else if (stream.header_parsed && !stream.peering_started) { + // Subgroup id comes from the first object. Hold these bytes with the header. + stream.pending_peer_bytes.insert(stream.pending_peer_bytes.end(), data->begin(), data->end()); + } else if (stream.peering_started && stream.next_object_id.has_value()) { + // ForwardPeerData already copied this chunk to other peers. Local subscribers follow the raw + // bytes once PublishObject has opened their stream. + ForwardReceivedData(false, stream.current_group_id, stream.current_subgroup_id, data, false); + } + + ProcessStartOfStreamData(stream_id, stream, true); + } + + void SubscribeTrackHandler::CapturePeerStart(uint64_t stream_id, + PeerStream& stream, + quicr::messages::StreamHeaderSubGroup& header) + { + const auto pending_it = pending_source_buffers_.find(stream_id); + if (pending_it == pending_source_buffers_.end()) { + return; + } + + // Adapt publisher track alias to normalized track alias used by subscribers. + const bool rewrite = + GetReceivedTrackAlias().value_or(0) != GetTrackAlias().value() || pending_it->second.empty(); + + stream.pending_peer_bytes.clear(); + if (rewrite) { + header.track_alias = GetTrackAlias().value(); + + quicr::Bytes encoded; + encoded << header; + stream.pending_peer_bytes.insert(stream.pending_peer_bytes.end(), encoded.begin(), encoded.end()); + + const auto remaining = stream.buffer.Data(); + stream.pending_peer_bytes.insert(stream.pending_peer_bytes.end(), remaining.begin(), remaining.end()); } else { - ForwardReceivedData( - false, stream_it->second.current_group_id, stream_it->second.current_subgroup_id, std::move(data)); + for (const auto& source : pending_it->second) { + stream.pending_peer_bytes.insert(stream.pending_peer_bytes.end(), source->begin(), source->end()); + } } - TryProcessStreamData(stream_id, stream_it->second); + pending_source_buffers_.erase(pending_it); } - void SubscribeTrackHandler::TryProcessStreamData(uint64_t stream_id, StreamContext& stream) + void SubscribeTrackHandler::StartPeeringForward(PeerStream& stream) + { + if (stream.peering_started) { + return; + } + + stream.peering_started = true; + if (stream.pending_peer_bytes.empty()) { + return; + } + + // These bytes already went out on the peer pipeline. Forwarding them again opens a second + // stream that stops after this start. Local subscribers are still served by PublishObject. + ForwardReceivedData(true, + stream.current_group_id, + stream.current_subgroup_id, + std::make_shared(std::move(stream.pending_peer_bytes)), + false); + stream.pending_peer_bytes.clear(); + } + + void SubscribeTrackHandler::ProcessStartOfStreamData(uint64_t stream_id, PeerStream& stream, bool is_from_peer) { auto& s_hdr = stream.buffer.GetAny(); - const auto pending_it = pending_source_buffers_.find(stream_id); - if (pending_it != pending_source_buffers_.end()) { + if (!stream.header_parsed) { + const auto pending_it = pending_source_buffers_.find(stream_id); + if (pending_it == pending_source_buffers_.end()) { + return; + } + if (not(stream.buffer >> s_hdr)) { SPDLOG_DEBUG("Not enough data to process new stream headers yet"); return; } - if (s_hdr.priority.has_value()) { - SetPriority(*s_hdr.priority); + // TODO: Do we want to set the subscriber priority to the publisher priority here? + if (s_hdr.publisher_priority.has_value()) { + SetSubscriberPriority(*s_hdr.publisher_priority); } - // Adapt publisher track alias to normalized track alias used by subscribers. - if (GetReceivedTrackAlias().value_or(0) != GetTrackAlias().value() || pending_it->second.empty()) { - s_hdr.track_alias = GetTrackAlias().value(); - - auto updated_data = std::make_shared>(); - *updated_data << s_hdr; - - const auto remaining_data = stream.buffer.Data(); - updated_data->insert(updated_data->end(), remaining_data.begin(), remaining_data.end()); + stream.header_parsed = true; + stream.current_group_id = s_hdr.group_id; + CapturePeerStart(stream_id, stream, s_hdr); - ForwardReceivedData(true, s_hdr.group_id, s_hdr.subgroup_id.value_or(0), std::move(updated_data)); - } else { - bool is_start = true; - for (auto& source_buffer : pending_it->second) { - ForwardReceivedData( - is_start, s_hdr.group_id, s_hdr.subgroup_id.value_or(0), std::move(source_buffer)); - is_start = false; - } + // kSetFromFirstObject leaves the id out of the header. Opening the peering stream before + // the first object would record it under subgroup 0, and the later close would miss it. + if (s_hdr.subgroup_id.has_value()) { + stream.current_subgroup_id = *s_hdr.subgroup_id; + StartPeeringForward(stream); } - - stream.current_group_id = s_hdr.group_id; - stream.current_subgroup_id = s_hdr.subgroup_id.value_or(0); - pending_source_buffers_.erase(pending_it); } while (not stream.buffer.Empty()) { @@ -347,13 +544,14 @@ namespace laps { stream.current_group_id = s_hdr.group_id; stream.current_subgroup_id = s_hdr.subgroup_id.value(); + StartPeeringForward(stream); ObjectReceived({ s_hdr.group_id, stream.next_object_id.value(), s_hdr.subgroup_id.value(), obj.payload.size(), obj.object_status, - s_hdr.priority, + s_hdr.publisher_priority, std::nullopt, quicr::TrackMode::kStream, obj.extensions, @@ -371,9 +569,9 @@ namespace laps { ForwardReceivedData( false, s_hdr.group_id, - s_hdr.subgroup_id.value_or(0), + s_hdr.subgroup_id.value(), std::make_shared>(remaining_data.begin(), remaining_data.end()), - false); + !is_from_peer); } } } @@ -391,7 +589,7 @@ namespace laps { if (subscribers_.contains(0)) { // Is peering subscribed server_.peer_manager_.ClientDataRecv( msg.track_alias, - GetPriority(), + GetSubscriberPriority(), GetDeliveryTimeout().value_or(std::chrono::milliseconds(kDefaultObjectTtl)).count(), peering::DataType::kDatagram, msg.group_id, @@ -461,7 +659,7 @@ namespace laps { if (forward_to_peers) { server_.peer_manager_.ClientDataRecv( *track_alias, - GetPriority(), + GetSubscriberPriority(), GetDeliveryTimeout().value_or(std::chrono::milliseconds(kDefaultObjectTtl)).count(), d_type, group_id, @@ -534,19 +732,19 @@ namespace laps { } } - void SubscribeTrackHandler::StreamClosed(std::uint64_t stream_id, bool use_reset) + void SubscribeTrackHandler::PeerStreamClosedReceived(std::uint64_t stream_id, bool use_reset) { pending_source_buffers_.erase(stream_id); - auto stream_it = streams_.find(stream_id); - if (stream_it == streams_.end()) { + auto stream_it = peer_streams_.find(stream_id); + if (stream_it == peer_streams_.end()) { return; } for (auto& [conn_handle, pub_handler] : subscribers_) { if (conn_handle == 0) { - // Notify peering manager - if (GetTrackAlias().has_value()) { + // Only a stream that was opened under this group and subgroup has one to close. + if (stream_it->second.peering_started && GetTrackAlias().has_value()) { server_.peer_manager_.EndSubgroup(GetTrackAlias().value(), stream_it->second.current_group_id, stream_it->second.current_subgroup_id, @@ -565,7 +763,7 @@ namespace laps { } } - streams_.erase(stream_it); + peer_streams_.erase(stream_it); } void SubscribeTrackHandler::SetFromPeer() diff --git a/src/subscribe_handler.h b/src/subscribe_handler.h index 753a383..a0d147b 100644 --- a/src/subscribe_handler.h +++ b/src/subscribe_handler.h @@ -3,17 +3,19 @@ #include "client_manager.h" #include "publish_namespace_handler.h" -#include +#include +#include #include #include +#include namespace laps { /** * @brief Subscribe track handler * @details Subscribe track handler used for the subscribe command line option. */ - class SubscribeTrackHandler : public quicr::SubscribeTrackHandler + class SubscribeTrackHandler : public quicr::ForwardingSubscribeTrackHandler { public: static constexpr uint64_t kRefreshRankingIntervalMs = 500; @@ -27,10 +29,19 @@ namespace laps { ~SubscribeTrackHandler(); - void StreamClosed(std::uint64_t stream_id, bool use_reset) override; + void SubgroupStarted(std::uint64_t group_id, + std::uint64_t subgroup_id, + std::optional priority, + quicr::messages::StreamHeaderProperties properties) override; + void StreamBytesForwarded(std::uint64_t group_id, std::uint64_t subgroup_id, quicr::Bytes&& data) override; + void SubgroupEnded(std::uint64_t group_id, std::uint64_t subgroup_id, bool reset) override; + + /// MoQT bytes forwarded by a peer, which still include the subgroup header. + void PeerStreamDataReceived(uint64_t stream_id, + std::shared_ptr> data, + bool is_new_stream); + void PeerStreamClosedReceived(uint64_t stream_id, bool reset); - void StreamDataRecv(uint64_t stream_id, quicr::InitialStreamData&& initial_buffer) override; - void StreamDataRecv(uint64_t stream_id, std::shared_ptr> data) override; void DgramDataRecv(std::shared_ptr> data) override; void ObjectReceived(const quicr::ObjectHeaders& object_headers, quicr::BytesSpan data, @@ -106,7 +117,34 @@ namespace laps { std::size_t SubscriberCount() const { return subscribers_.size(); } private: - void TryProcessStreamData(uint64_t stream_id, StreamContext& stream); + /// Bytes of one subgroup after its header, parsed so the cache still sees each object. + struct SubgroupForwardState + { + std::optional properties; + std::optional priority; + quicr::StreamBuffer buffer; + std::optional next_object_id; + bool saw_first_object{ false }; + }; + + /// Peer-injected stream, which is still framed as a whole MoQT subgroup stream. + struct PeerStream + { + quicr::StreamBuffer buffer; + std::uint64_t current_group_id{ 0 }; + std::uint64_t current_subgroup_id{ 0 }; + std::optional next_object_id; + bool header_parsed{ false }; + /// True once peering has been given this stream under current_group_id/current_subgroup_id. + bool peering_started{ false }; + /// Header and object bytes held until the subgroup id is known, then forwarded as one new stream. + std::vector pending_peer_bytes; + }; + + void ConsumeForwardedObjects(std::uint64_t group_id, std::uint64_t subgroup_id, SubgroupForwardState& subgroup); + void ProcessStartOfStreamData(uint64_t stream_id, PeerStream& stream, bool is_from_peer = false); + void CapturePeerStart(uint64_t stream_id, PeerStream& stream, quicr::messages::StreamHeaderSubGroup& header); + void StartPeeringForward(PeerStream& stream); void ForwardReceivedData(bool is_new_stream, uint64_t group_id, @@ -127,6 +165,9 @@ namespace laps { // copying when the track alias is unchanged. std::map>>> pending_source_buffers_; + std::map, SubgroupForwardState> open_subgroups_; + std::map peer_streams_; + /** * @brief Map of subscribers that have subscribed to this content * diff --git a/test/peering_subscribe.cc b/test/peering_subscribe.cc index a0b2b73..b3dc970 100644 --- a/test/peering_subscribe.cc +++ b/test/peering_subscribe.cc @@ -2,32 +2,79 @@ #include "peering/messages/subscribe_info.h" -#include +#include +#include +#include -TEST_CASE("Serialize Subscribe Info") -{ - using namespace laps::peering; +namespace { + laps::peering::SubscribeInfo MakeSubscribeInfo(bool with_new_group) + { + using namespace laps::peering; + + SubscribeInfo subscribe_info; + subscribe_info.source_node_id = 0xff00aabbcc; - SubscribeInfo subscribe_info; - subscribe_info.source_node_id = 0xff00aabbcc; + quicr::TrackHash track_hash({}); + track_hash.track_name_hash = 0x9000; + track_hash.track_namespace_hash = 0xaabbcc; + track_hash.track_fullname_hash = 0x1234567; + subscribe_info.track_hash = track_hash; - quicr::TrackHash track_hash({}); - track_hash.track_name_hash = 0x9000; - track_hash.track_namespace_hash = 0xaabbcc; - track_hash.track_fullname_hash = 0x1234567; + subscribe_info.name_space = quicr::TrackNamespace(std::string("namespace")); + subscribe_info.name = { 't', 'r', 'a', 'c', 'k' }; + subscribe_info.priority = 10; + subscribe_info.delivery_timeout = 5000; + subscribe_info.expires = 1000; + subscribe_info.forward = 1; + if (with_new_group) { + subscribe_info.new_group_request_id = 42; + } - subscribe_info.track_hash = track_hash; + return subscribe_info; + } - auto net_data = subscribe_info.Serialize(false, false, true); + void CheckRoundTrip(laps::peering::SubscribeInfo& subscribe_info) + { + using namespace laps::peering; - CHECK_EQ(net_data.size(), 38); - CHECK_EQ(subscribe_info.seq, 1); + auto net_data = subscribe_info.Serialize(false, false, true); - SubscribeInfo decoded_si(net_data); + CHECK_EQ(net_data.size(), subscribe_info.SizeBytes()); + CHECK_EQ(subscribe_info.seq, 1); - CHECK_EQ(subscribe_info.source_node_id, decoded_si.source_node_id); + SubscribeInfo decoded_si(net_data); - CHECK_EQ(subscribe_info.track_hash.track_namespace_hash, decoded_si.track_hash.track_namespace_hash); - CHECK_EQ(subscribe_info.track_hash.track_name_hash, decoded_si.track_hash.track_name_hash); - CHECK_EQ(subscribe_info.track_hash.track_fullname_hash, decoded_si.track_hash.track_fullname_hash); + CHECK_EQ(subscribe_info.seq, decoded_si.seq); + CHECK_EQ(subscribe_info.source_node_id, decoded_si.source_node_id); + CHECK_EQ(subscribe_info.track_hash.track_namespace_hash, decoded_si.track_hash.track_namespace_hash); + CHECK_EQ(subscribe_info.track_hash.track_name_hash, decoded_si.track_hash.track_name_hash); + CHECK_EQ(subscribe_info.track_hash.track_fullname_hash, decoded_si.track_hash.track_fullname_hash); + CHECK_EQ(subscribe_info.name_space, decoded_si.name_space); + CHECK_EQ(subscribe_info.name, decoded_si.name); + CHECK_EQ(subscribe_info.priority, decoded_si.priority); + CHECK_EQ(subscribe_info.delivery_timeout, decoded_si.delivery_timeout); + CHECK_EQ(subscribe_info.expires, decoded_si.expires); + CHECK_EQ(subscribe_info.forward, decoded_si.forward); + CHECK_EQ(subscribe_info.new_group_request_id, decoded_si.new_group_request_id); + } +} + +TEST_CASE("Serialize Subscribe Info") +{ + auto subscribe_info = MakeSubscribeInfo(true); + CheckRoundTrip(subscribe_info); +} + +TEST_CASE("Serialize Subscribe Info without new group request") +{ + auto subscribe_info = MakeSubscribeInfo(false); + CheckRoundTrip(subscribe_info); + CHECK_FALSE(subscribe_info.new_group_request_id.has_value()); +} + +TEST_CASE("Subscribe Info truncated input throws") +{ + using namespace laps::peering; + const std::vector truncated{ 0x00, 0x01 }; + CHECK_THROWS_AS(SubscribeInfo(std::span{ truncated }), std::out_of_range); }