Various perf updates and fixes

CMakeLists.txt

@@ -5,6 +5,9 @@ option(LAPS_INSTALL "Install the laps Components" ON)
 
 option(LAPS_BUILD_TESTS "Build Tests laps" OFF)
 
+option(LAPS_USE_BROTLI "Enable picotls Brotli TLS certificate compression (libbrotlidec)" OFF)
+set(QUICR_USE_BROTLI "${LAPS_USE_BROTLI}" CACHE BOOL "Enable picotls Brotli TLS certificate compression" FORCE)
+
 set(CMAKE_CXX_STANDARD 20)
 set(CMAKE_CXX_STANDARD_REQUIRED ON)
 set(CMAKE_CXX_EXTENSIONS OFF)
@@ -15,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.16
+        VERSION 0.18.23
         DESCRIPTION "Latency Aware Publish/Subscribe"
         LANGUAGES CXX)
 

src/client_manager.cc

@@ -271,7 +271,7 @@ namespace laps {
     }
 
     quicr::Reply<void, quicr::ErrorCode> ClientManager::NewGroupRequested(const quicr::FullTrackName& track_full_name,
-                                                             std::uint64_t group_id)
+                                                                          std::uint64_t group_id)
     {
         auto th = quicr::TrackHash(track_full_name);
         SPDLOG_INFO("New group requested received track_alais: {} group_id: {} ", th.track_fullname_hash, group_id);
@@ -921,8 +921,9 @@ namespace laps {
         return {};
     }
 
-    quicr::Reply<void, quicr::ErrorCode> ClientManager::PublishDoneReceived(const std::shared_ptr<quicr::Session>& session,
-                                                               uint64_t request_id)
+    quicr::Reply<void, quicr::ErrorCode> ClientManager::PublishDoneReceived(
+      const std::shared_ptr<quicr::Session>& session,
+      uint64_t request_id)
     {
         const auto connection_handle = ConnectionHandle(session);
         SPDLOG_LOGGER_INFO(
@@ -1001,6 +1002,25 @@ namespace laps {
             }
         }
 
+        const auto& pub_ns = s_it->second->GetFullTrackName().name_space;
+        const auto& pub_ta = s_it->second->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);
+
+            if (prefix_match == std::partial_ordering::less || prefix_match == std::partial_ordering::equivalent) {
+                for (auto& [_, pub_ns_h] : conns) {
+                    if (!config_.allow_self && pub_ns_h->GetConnectionId() == connection_handle) {
+                        // Initially do not mirror
+                        continue;
+                    }
+
+                    pub_ns_h->UnpublishTrack(pub_ta);
+                }
+            }
+        }
+
         state_.pub_subscribes_by_req_id.erase(s_it);
         lock.unlock();
 
@@ -1011,8 +1031,9 @@ namespace laps {
         return {};
     }
 
-    quicr::Reply<void, quicr::ErrorCode> ClientManager::UnsubscribeReceived(const std::shared_ptr<quicr::Session>& session,
-                                                               uint64_t request_id)
+    quicr::Reply<void, quicr::ErrorCode> ClientManager::UnsubscribeReceived(
+      const std::shared_ptr<quicr::Session>& 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);

src/client_manager.h

@@ -28,10 +28,7 @@ namespace laps {
         quicr::Bytes data;
 
         // libc++ now compares set keys with std::less<void>, which requires operator<.
-        bool operator<(const CacheObject& other) const noexcept
-        {
-            return headers.object_id < other.headers.object_id;
-        }
+        bool operator<(const CacheObject& other) const noexcept { return headers.object_id < other.headers.object_id; }
     };
 }
 
@@ -196,7 +193,7 @@ namespace laps {
           std::weak_ptr<quicr::SubscribeNamespaceHandler> sub_ns_handler) override;
 
         quicr::Reply<void, quicr::ErrorCode> PublishDoneReceived(const std::shared_ptr<quicr::Session>& session,
-                                                    std::uint64_t request_id) override;
+                                                                 std::uint64_t request_id) override;
 
         quicr::Reply<quicr::RequestResponse, quicr::RequestErrorCode> SubscribeReceived(
           const std::shared_ptr<quicr::Session>& session,
@@ -205,10 +202,10 @@ namespace laps {
           const quicr::SubscribeAttributes& subscribe_attributes) override;
 
         quicr::Reply<void, quicr::ErrorCode> UnsubscribeReceived(const std::shared_ptr<quicr::Session>& session,
-                                                    std::uint64_t request_id) override;
+                                                                 std::uint64_t request_id) override;
 
         quicr::Reply<void, quicr::ErrorCode> NewGroupRequested(const quicr::FullTrackName& track_full_name,
-                                                  std::uint64_t group_id) override;
+                                                               std::uint64_t group_id) override;
 
         quicr::Reply<quicr::RequestResponse, quicr::RequestErrorCode> TrackStatusReceived(
           const std::shared_ptr<quicr::Session>& session,

src/peering/info_base.h

@@ -8,7 +8,7 @@
 #include "peering/messages/subscribe_info.h"
 
 #include <map>
-#include <quicr/messages/messages.h>
+#include <quicr/messages/message_serialisation.h>
 #include <set>
 
 namespace laps::peering {

src/peering/messages/data_header.h

@@ -5,7 +5,7 @@
 #include "peering/common.h"
 #include "subscribe_info.h"
 
-#include <quicr/messages/messages.h>
+#include <quicr/messages/message_serialisation.h>
 #include <quicr/messages/uintvar.h>
 
 namespace laps::peering {

src/peering/peer_manager.cc

@@ -1364,8 +1364,7 @@ namespace laps::peering {
                 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);
+                    client_manager_->PeerStreamClosed(key.first, stream_id, flag == quicr::StreamClosedFlag::kReset);
                 }
             }
         }

src/publish_namespace_handler.cc

@@ -15,6 +15,12 @@ namespace laps {
     }
 }
 
+void
+laps::PublishNamespaceHandler::UnpublishTrack(std::uint64_t track_alias)
+{
+    published_tracks_.erase(track_alias);
+}
+
 void
 laps::PublishNamespaceHandler::PublishTrack(std::shared_ptr<quicr::PublishTrackHandler> handler)
 {
@@ -152,8 +158,6 @@ laps::PublishNamespaceHandler::UpdateTrackRanking(
 
     if (published_tracks_.size() > active_tracks.size()) {
 
-        std::vector<std::uint64_t> remove_publish_tracks;
-
         // Unpublish tracks that are too old
         for (auto& [ta, track] : published_tracks_) {
             if (active_tracks.contains(ta)) {
@@ -182,21 +186,14 @@ laps::PublishNamespaceHandler::UpdateTrackRanking(
                 }
 
                 if (cur_tick - track.last_updated_tick > delay_publish_done_ms_) {
-                    SPDLOG_INFO("Unpublish track, not in top-n track alias: {} conn_id: {} ticks: {} < {}",
-                                ta,
-                                GetConnectionId(),
-                                track.last_updated_tick,
-                                cur_tick);
-                    remove_publish_tracks.push_back(ta);
+                    SPDLOG_TRACE("Unpublish track, not in top-n track alias: {} conn_id: {} ticks: {} < {}",
+                                 ta,
+                                 GetConnectionId(),
+                                 track.last_updated_tick,
+                                 cur_tick);
                 }
             }
         }
-
-        if (!remove_publish_tracks.empty()) {
-            for (const auto& ta: remove_publish_tracks) {
-                published_tracks_.erase(ta);
-            }
-        }
     }
 }
 

src/publish_namespace_handler.h

@@ -44,6 +44,7 @@ namespace laps {
           std::shared_ptr<const std::vector<uint8_t>> data) override;
 
         void PublishTrack(std::shared_ptr<quicr::PublishTrackHandler> handler) override;
+        void UnpublishTrack(std::uint64_t track_alias);
 
         static auto Create(const quicr::TrackNamespace& prefix, std::weak_ptr<timeq::tick_service> tick_service)
         {

src/subscribe_handler.cc

@@ -140,6 +140,9 @@ namespace laps {
             }
 
             if (value.latest_value != recv_value || cur_tick - value.latest_tick_ms > kRefreshRankingIntervalMs) {
+                SPDLOG_TRACE(
+                  "value changed: from {} to {} ta: {} conn: {}", value.latest_value, recv_value, ta, conn_id);
+
                 value.latest_value = recv_value;
                 value.latest_tick_ms = cur_tick;
 

src/subscribe_handler.h

@@ -16,7 +16,7 @@ namespace laps {
     class SubscribeTrackHandler : public quicr::SubscribeTrackHandler
     {
       public:
-        static constexpr uint64_t kRefreshRankingIntervalMs = 120;
+        static constexpr uint64_t kRefreshRankingIntervalMs = 500;
 
         SubscribeTrackHandler(const quicr::FullTrackName& full_track_name,
                               std::uint8_t priority,

src/track_ranking.h

@@ -132,6 +132,8 @@ namespace laps {
                     flat_track_list_.insert(flat_track_list_.end(), sort_tracks.begin(), sort_tracks.end());
                 }
 
+                NotifyNamespaceHandlers();
+
             } else {
                 // Update latest_tick in flat_track_list_ in-place
                 for (auto& [alias, insert_seq_num, latest_tick, conn_id] : flat_track_list_) {
@@ -142,8 +144,6 @@ namespace laps {
                     }
                 }
             }
-
-            NotifyNamespaceHandlers();
         }
 
         void RemoveTrack(const TrackAlias track_alias)