[improve] modify the negativeACK structure to reduce memory overhead

Author: syslucas

CMakeLists.txt

@@ -124,13 +124,17 @@ if (INTEGRATE_VCPKG)
     find_package(protobuf CONFIG REQUIRED)
     find_package(zstd CONFIG REQUIRED)
     find_package(Snappy CONFIG REQUIRED)
+    find_package(roaring CONFIG REQUIRED)
     set(COMMON_LIBS CURL::libcurl
         ZLIB::ZLIB
         OpenSSL::SSL
         OpenSSL::Crypto
         protobuf::libprotobuf
         $<IF:$<TARGET_EXISTS:zstd::libzstd_shared>,zstd::libzstd_shared,zstd::libzstd_static>
         Snappy::snappy
+        roaring::roaring
+        roaring::roaring-headers
+        roaring::roaring-headers-cpp
         )
     if (USE_ASIO)
         find_package(asio CONFIG REQUIRED)

lib/NegativeAcksTracker.cc

@@ -19,14 +19,20 @@
 
 #include "NegativeAcksTracker.h"
 
+#include <bits/stdint-uintn.h>
+
 #include <functional>
 #include <set>
+#include <utility>
 
 #include "ClientImpl.h"
 #include "ConsumerImpl.h"
 #include "ExecutorService.h"
 #include "LogUtils.h"
 #include "MessageIdUtil.h"
+#include "pulsar/MessageBuilder.h"
+#include "pulsar/MessageId.h"
+#include "pulsar/MessageIdBuilder.h"
 DECLARE_LOG_OBJECT()
 
 namespace pulsar {
@@ -75,13 +81,22 @@ void NegativeAcksTracker::handleTimer(const ASIO_ERROR &ec) {
 
     auto now = Clock::now();
 
+    // The map is sorted by time, so we can exit immediately when we traverse to a time that does not match
     for (auto it = nackedMessages_.begin(); it != nackedMessages_.end();) {
-        if (it->second < now) {
-            messagesToRedeliver.insert(it->first);
-            it = nackedMessages_.erase(it);
-        } else {
-            ++it;
+        if (it->first > now) {
+            // We are done with all the messages that need to be redelivered
+            break;
+        }
+
+        auto ledgerMap = it->second;
+        for (auto ledgerIt = ledgerMap.begin(); ledgerIt != ledgerMap.end(); ++ledgerIt) {
+            auto entrySet = ledgerIt->second;
+            for (auto setIt = entrySet.begin(); setIt != entrySet.end(); ++setIt) {
+                messagesToRedeliver.insert(
+                    MessageIdBuilder().ledgerId(ledgerIt->first).entryId(*setIt).build());
+            }
         }
+        it = nackedMessages_.erase(it);
     }
     lock.unlock();
 
@@ -99,7 +114,7 @@ void NegativeAcksTracker::add(const MessageId &m) {
     {
         std::lock_guard<std::mutex> lock{mutex_};
         // Erase batch id to group all nacks from same batch
-        nackedMessages_[msgId] = now + nackDelay_;
+        nackedMessages_[now][msgId.ledgerId()].add((uint64_t)msgId.entryId());
     }
 
     scheduleTimer();

lib/NegativeAcksTracker.h

@@ -27,6 +27,8 @@
 #include <map>
 #include <memory>
 #include <mutex>
+#include <roaring/roaring64map.hh>
+#include <unordered_map>
 
 #include "AsioDefines.h"
 #include "AsioTimer.h"
@@ -39,6 +41,7 @@ class ClientImpl;
 using ClientImplPtr = std::shared_ptr<ClientImpl>;
 class ExecutorService;
 using ExecutorServicePtr = std::shared_ptr<ExecutorService>;
+using LedgerId = int64_t;
 
 class NegativeAcksTracker : public std::enable_shared_from_this<NegativeAcksTracker> {
    public:
@@ -65,7 +68,7 @@ class NegativeAcksTracker : public std::enable_shared_from_this<NegativeAcksTrac
     std::chrono::milliseconds nackDelay_;
     std::chrono::milliseconds timerInterval_;
     typedef typename std::chrono::steady_clock Clock;
-    std::map<MessageId, Clock::time_point> nackedMessages_;
+    std::map<Clock::time_point, std::unordered_map<LedgerId, roaring::Roaring64Map>> nackedMessages_;
 
     const DeadlineTimerPtr timer_;
     std::atomic_bool closed_{false};

vcpkg.json

@@ -54,6 +54,9 @@
     {
       "name": "zstd",
       "version>=": "1.5.5"
+    },
+    {
+      "name" : "roaring"
     }
   ],
   "features": {