core: Replace sbf with hyperloglog (#5784)

* core: Replace sbf with hyperloglog

Signed-off-by: Abhijat Malviya <[email protected]>

Author: abhijat

src/core/page_usage_stats.cc

@@ -12,8 +12,6 @@
 
 #include <string>
 
-#include "core/bloom.h"
-
 extern "C" {
 #include <unistd.h>
 
@@ -30,23 +28,17 @@ using absl::StripTrailingAsciiWhitespace;
 
 namespace {
 constexpr auto kUsageHistPoints = std::array{50, 90, 99};
-constexpr auto kInitialSBFCap = 1000;
-constexpr auto kFProb = 0.001;
-constexpr auto kGrowthFactor = 2;
 constexpr auto kHistSignificantFigures = 3;
 
-}  // namespace
-
-FilterWithSize::FilterWithSize()
-    : sbf{SBF{kInitialSBFCap, kFProb, kGrowthFactor, PMR_NS::get_default_resource()}}, size{0} {
+HllBufferPtr InitHllPtr() {
+  HllBufferPtr p;
+  p.size = getDenseHllSize();
+  p.hll = new uint8_t[p.size];
+  CHECK_EQ(0, createDenseHll(p));
+  return p;
 }
 
-void FilterWithSize::Add(uintptr_t address) {
-  const auto s = std::to_string(address);
-  if (sbf.Add(s)) {
-    size += 1;
-  }
-}
+}  // namespace
 
 void CollectedPageStats::Merge(CollectedPageStats&& other, uint16_t shard_id) {
   this->pages_scanned += other.pages_scanned;
@@ -98,34 +90,47 @@ std::string CollectedPageStats::ToString() const {
   return response;
 }
 
-PageUsage::UniquePages::UniquePages() {
+PageUsage::UniquePages::UniquePages()
+    : pages_scanned{InitHllPtr()},
+      pages_marked_for_realloc{InitHllPtr()},
+      pages_full{InitHllPtr()},
+      pages_reserved_for_malloc{InitHllPtr()},
+      pages_with_heap_mismatch{InitHllPtr()},
+      pages_above_threshold{InitHllPtr()} {
   hdr_histogram* h = nullptr;
   const auto init_result = hdr_init(1, 100, kHistSignificantFigures, &h);
   CHECK_EQ(0, init_result) << "failed to initialize histogram";
   page_usage_hist = h;
 }
 
 PageUsage::UniquePages::~UniquePages() {
+  delete[] pages_scanned.hll;
+  delete[] pages_marked_for_realloc.hll;
+  delete[] pages_full.hll;
+  delete[] pages_reserved_for_malloc.hll;
+  delete[] pages_with_heap_mismatch.hll;
+  delete[] pages_above_threshold.hll;
   hdr_close(page_usage_hist);
 }
 
 void PageUsage::UniquePages::AddStat(mi_page_usage_stats_t stat) {
-  const auto address = stat.page_address;
-  pages_scanned.Add(address);
+  const auto data = reinterpret_cast<const unsigned char*>(&stat.page_address);
+  constexpr size_t size = sizeof(stat.page_address);
+  pfadd_dense(pages_scanned, data, size);
   if (stat.flags == MI_DFLY_PAGE_BELOW_THRESHOLD) {
-    pages_marked_for_realloc.Add(address);
+    pfadd_dense(pages_marked_for_realloc, data, size);
   } else {
     if (stat.flags & MI_DFLY_PAGE_FULL) {
-      pages_full.Add(address);
+      pfadd_dense(pages_full, data, size);
     } else if (stat.flags & MI_DFLY_HEAP_MISMATCH) {
-      pages_with_heap_mismatch.Add(address);
+      pfadd_dense(pages_with_heap_mismatch, data, size);
     } else if (stat.flags & MI_DFLY_PAGE_USED_FOR_MALLOC) {
-      pages_reserved_for_malloc.Add(address);
+      pfadd_dense(pages_reserved_for_malloc, data, size);
     } else {
       // We record usage only for pages which have usage above the given threshold but which are not
       // full. This allows tuning the threshold for future commands. This also excludes full pages,
       // so the only pages here have: threshold < usage% < 100
-      pages_above_threshold.Add(address);
+      pfadd_dense(pages_above_threshold, data, size);
       const double perc = static_cast<double>(stat.used) / static_cast<double>(stat.capacity);
       hdr_record_value(page_usage_hist, perc * 100);
     }
@@ -137,16 +142,18 @@ CollectedPageStats PageUsage::UniquePages::CollectedStats() const {
   for (const auto p : kUsageHistPoints) {
     usage[p] = hdr_value_at_percentile(page_usage_hist, p);
   }
-  return CollectedPageStats{.pages_scanned = pages_scanned.size,
-                            .pages_marked_for_realloc = pages_marked_for_realloc.size,
-                            .pages_full = pages_full.size,
-                            .pages_reserved_for_malloc = pages_reserved_for_malloc.size,
-                            .pages_with_heap_mismatch = pages_with_heap_mismatch.size,
-                            .pages_above_threshold = pages_above_threshold.size,
-                            .objects_skipped_not_required = objects_skipped_not_required,
-                            .objects_skipped_not_supported = objects_skipped_not_supported,
-                            .page_usage_hist = std::move(usage),
-                            .shard_wide_summary = {}};
+
+  return CollectedPageStats{
+      .pages_scanned = static_cast<uint64_t>(pfcountSingle(pages_scanned)),
+      .pages_marked_for_realloc = static_cast<uint64_t>(pfcountSingle(pages_marked_for_realloc)),
+      .pages_full = static_cast<uint64_t>(pfcountSingle(pages_full)),
+      .pages_reserved_for_malloc = static_cast<uint64_t>(pfcountSingle(pages_reserved_for_malloc)),
+      .pages_with_heap_mismatch = static_cast<uint64_t>(pfcountSingle(pages_with_heap_mismatch)),
+      .pages_above_threshold = static_cast<uint64_t>(pfcountSingle(pages_above_threshold)),
+      .objects_skipped_not_required = objects_skipped_not_required,
+      .objects_skipped_not_supported = objects_skipped_not_supported,
+      .page_usage_hist = std::move(usage),
+      .shard_wide_summary = {}};
 }
 
 PageUsage::PageUsage(CollectPageStats collect_stats, float threshold)

src/core/page_usage_stats.h

@@ -9,21 +9,16 @@
 #define MI_BUILD_RELEASE 1
 #include <mimalloc/types.h>
 
-#include "core/bloom.h"
+extern "C" {
+#include "redis/hyperloglog.h"
+}
 
 struct hdr_histogram;
 
 namespace dfly {
 
 enum class CollectPageStats : uint8_t { YES, NO };
 
-struct FilterWithSize {
-  FilterWithSize();
-  SBF sbf;
-  size_t size;
-  void Add(uintptr_t);
-};
-
 struct CollectedPageStats {
   double threshold{0.0};
   uint64_t pages_scanned{0};
@@ -72,12 +67,12 @@ class PageUsage {
   float threshold_;
 
   struct UniquePages {
-    FilterWithSize pages_scanned{};
-    FilterWithSize pages_marked_for_realloc{};
-    FilterWithSize pages_full{};
-    FilterWithSize pages_reserved_for_malloc{};
-    FilterWithSize pages_with_heap_mismatch{};
-    FilterWithSize pages_above_threshold{};
+    HllBufferPtr pages_scanned;
+    HllBufferPtr pages_marked_for_realloc;
+    HllBufferPtr pages_full;
+    HllBufferPtr pages_reserved_for_malloc;
+    HllBufferPtr pages_with_heap_mismatch;
+    HllBufferPtr pages_above_threshold;
     hdr_histogram* page_usage_hist{};
 
     uint64_t objects_skipped_not_required{0};

src/core/page_usage_stats_test.cc

@@ -135,8 +135,6 @@ TEST_F(PageUsageStatsTest, StatCollection) {
   }
 
   constexpr auto page_count_per_flag = 150;
-  // allow for collisions in filter
-  constexpr auto expected_min_count = 140;
 
   auto start = 0;
   for (const uint8_t flag : {MI_DFLY_PAGE_FULL, MI_DFLY_PAGE_USED_FOR_MALLOC, MI_DFLY_HEAP_MISMATCH,
@@ -156,9 +154,11 @@ TEST_F(PageUsageStatsTest, StatCollection) {
   st.Merge(p.CollectedStats(), 1);
 
   EXPECT_GT(st.pages_scanned, 12000);
-  EXPECT_GT(st.pages_full, expected_min_count);
-  EXPECT_GT(st.pages_reserved_for_malloc, expected_min_count);
-  EXPECT_GT(st.pages_marked_for_realloc, expected_min_count);
+
+  // Expect a small error margin due to HLL
+  EXPECT_NEAR(st.pages_full, page_count_per_flag, 5);
+  EXPECT_NEAR(st.pages_reserved_for_malloc, page_count_per_flag, 5);
+  EXPECT_NEAR(st.pages_marked_for_realloc, page_count_per_flag, 5);
 
   const auto usage = st.shard_wide_summary;