hashDb.hpp

Go to the documentation of this file.

#pragma once
 
#include "numa.hpp"
#include <atomic>
#include <jaffarCommon/concurrent.hpp>
#include <jaffarCommon/hash.hpp>
#include <jaffarCommon/json.hpp>
#include <jaffarCommon/logger.hpp>
#include <jaffarCommon/parallel.hpp>
 
namespace jaffarPlus
{
 
class HashDb final
{
public:
  HashDb(const nlohmann::json& config)
  {
    _maxStoreCount  = jaffarCommon::json::getNumber<size_t>(config, "Max Store Count");
    _maxStoreSizeMb = jaffarCommon::json::getNumber<double>(config, "Max Store Size (Mb)");
 
    // For testing purposes, the maximum size can be overriden by environment variables
    if (auto* value = std::getenv("JAFFAR_ENGINE_OVERRIDE_MAX_HASHDB_SIZE_MB")) _maxStoreSizeMb = std::stoul(value);
  }
 
  __INLINE__ size_t getMaxBudgetBytes() const { return (size_t)(_maxStoreSizeMb * 1024.0 * 1024.0) * _maxStoreCount; }
 
  ~HashDb()
  {
    for (auto* s : _l1) delete s;
    if (_l2 != nullptr) delete _l2;
  }
 
  __INLINE__ void initialize()
  {
    // The configured "Max Store Size" is the ABSOLUTE per-generation budget for the whole hash DB:
    // the global L2 plus all per-domain L1 caches combined never exceed it (so the total footprint is
    // Max Store Size x Max Store Count, independent of NUMA layout -- the tiering does not inflate it).
    // Rolling is driven by each store's actual measured memory (see getStoreSizeBytes).
    const size_t totalBudgetBytes = (size_t)(_maxStoreSizeMb * 1024.0 * 1024.0);
 
    if (_numaCount > 1)
    {
      // Multi-domain: split the budget so L2 + all L1 == the configured budget. The global L2 gets 3/4
      // and the per-domain L1 caches share the remaining 1/4. L2 holds the complete dedup set, which
      // benefits directly from more capacity (fewer rolled-out / re-explored states). The L1s only
      // absorb within-domain repeats, whose working set is strongly biased toward spatiotemporally
      // close states -- so a smaller cache still captures most of the local hit rate. Hence the 3:1
      // weighting toward L2.
      _maxStoreSizeBytes   = (totalBudgetBytes * 3) / 4;
      _l1MaxStoreSizeBytes = (totalBudgetBytes / 4) / (size_t)_numaCount;
      _l1.resize(_numaCount);
      for (int i = 0; i < _numaCount; i++) _l1[i] = makeStore();
    }
    else
    {
      // Single domain: no L1 tier, so the one global store gets the whole budget.
      _maxStoreSizeBytes = totalBudgetBytes;
    }
 
    // The single global authoritative store (holds the complete dedup set)
    _l2 = makeStore();
  }
 
  void printInfo() const
  {
    constexpr double toMb = 1.0 / (1024.0 * 1024.0);
 
    // Sum a store's measured resident memory across its rolling generations
    auto storeSizeMb = [this](const NUMAStore_t* s)
    {
      size_t bytes = 0;
      for (const auto& h : s->_hashStores) bytes += getStoreSizeBytes(h);
      return (double)bytes * toMb;
    };
 
    // Print the per-generation breakdown of a rolling store (newest first). Exposes each store's Id
    // and Age, so window rolling is observable (an Id > 0 means a new store was created), plus its
    // entries, size, and check/collision counts (current store from the live per-thread counters,
    // archived stores from their snapshot taken at roll time).
    auto printStores = [this, toMb, &storeSizeMb](const NUMAStore_t* s, const char* label)
    {
      size_t idx = 0;
      for (auto it = s->_hashStores.rbegin(); it != s->_hashStores.rend(); ++it, ++idx)
      {
        const bool   cur = (it == s->_hashStores.rbegin());
        const size_t q   = cur ? s->getQueryCount() : it->queryCount;
        const size_t c   = cur ? s->getCollisionCount() : it->collisionCount;
        jaffarCommon::logger::log("[J+]      + %sStore %lu/%lu - Id: %lu - Age: %lu, Entries: %.3f M, Size: %.1f Mb, Check Count: %lu, Collision Count: %lu (Rate %.3f%%)\n", label,
                                  idx + 1, _maxStoreCount, it->id, it->age, (double)it->hashSet.size() * toMb, (double)getStoreSizeBytes(*it) * toMb, q, c,
                                  q == 0 ? 0.0 : 100.0 * (double)c / (double)q);
      }
    };
 
    // Single-domain (no L1 tier): report L2 as a single store, with its rolling-window breakdown
    if (_l1.empty())
    {
      jaffarCommon::logger::log("[J+]   Single-Store Dedup (single NUMA domain), budget %.1f Mb:\n", (double)(_maxStoreSizeBytes * _maxStoreCount) * toMb);
      printStores(_l2, "");
      return;
    }
 
    // Two-tier: aggregate L1 across domains + the global L2
    size_t l1q = 0, l1c = 0, l1entries = 0;
    double l1SizeMb = 0.0;
    // Use LIFETIME (since-start) totals for the aggregate ratios so L1 and L2 are compared over the
    // same whole-run window. The per-store getQueryCount()/getCollisionCount() are reset on every roll
    // and the tiers roll at very different rates (L1 budget is a small fraction of L2's, so L1 rolls
    // far more often) -- comparing those current-window counts made l2q/l1q and the dup rate exceed
    // 100%. The lifetime totals make every ratio below a true fraction bounded by 100%.
    for (auto* const g : _l1)
    {
      l1q += g->getTotalQueryCount();
      l1c += g->getTotalCollisionCount();
      for (const auto& s : g->_hashStores) l1entries += s.hashSet.size();
      l1SizeMb += storeSizeMb(g);
    }
    const size_t l2q       = _l2->getTotalQueryCount();
    const size_t l2c       = _l2->getTotalCollisionCount();
    size_t       l2entries = 0;
    for (const auto& s : _l2->_hashStores) l2entries += s.hashSet.size();
 
    // Budgets: each L1 rolls at _l1MaxStoreSizeBytes and keeps up to _maxStoreCount generations
    // (times numaCount domains); L2 rolls at _maxStoreSizeBytes. The two budgets sum to the
    // configured Max Store Size x Max Store Count (3/4 to L2, 1/4 shared across the L1 caches).
    const double l1BudgetMb = (double)(_l1MaxStoreSizeBytes * _maxStoreCount * (size_t)_numaCount) * toMb;
    const double l2BudgetMb = (double)(_maxStoreSizeBytes * _maxStoreCount) * toMb;
 
    jaffarCommon::logger::log("[J+]   Two-Tier Dedup (%d NUMA-local L1 + global L2):\n", _numaCount);
    jaffarCommon::logger::log("[J+]      + L1 (local):                           Checks: %lu, Local Hits: %lu (%.3f%%), Entries: %.3f M, Size: %.1f / %.1f Mb\n", l1q, l1c,
                              l1q == 0 ? 0.0 : 100.0 * (double)l1c / (double)l1q, (double)l1entries * toMb, l1SizeMb, l1BudgetMb);
    jaffarCommon::logger::log(
        "[J+]      + L2 (global):                          Checks (L1 misses): %lu (%.3f%% of all), Cross-Domain Hits: %lu, Entries: %.3f M, Size: %.1f / %.1f Mb\n", l2q,
        l1q == 0 ? 0.0 : 100.0 * (double)l2q / (double)l1q, l2c, (double)l2entries * toMb, storeSizeMb(_l2), l2BudgetMb);
    jaffarCommon::logger::log("[J+]      + Served locally (no L2 access):        %.3f%% | Total dup rate: %.3f%%\n", l1q == 0 ? 0.0 : 100.0 * (double)l1c / (double)l1q,
                              l1q == 0 ? 0.0 : 100.0 * (double)(l1c + l2c) / (double)l1q);
    // L2 rolling-window breakdown (Id/Age make window rolling observable)
    printStores(_l2, "L2 ");
  }
 
  __INLINE__ bool checkHashExists(const jaffarCommon::hash::hash_t hash)
  {
    const auto threadId = jaffarCommon::parallel::getThreadId();
 
    // Single-domain: just the one local store
    if (_l1.empty())
    {
      _l2->_threadQueryCount[threadId].value++;
      const bool collision = probeStore(_l2, hash);
      if (collision) _l2->_threadCollisionCount[threadId].value++;
      return collision;
    }
 
    // L1: this domain's local store. By construction every hash in any L1 was also inserted into L2,
    // so an L1 hit is a genuine duplicate -- answered locally, with no L2 (possibly remote) access.
    // probeStore() also inserts the hash into L1's current store (caching it locally).
    auto* const l1 = _l1[_preferredNumaDomain];
    l1->_threadQueryCount[threadId].value++;
    if (probeStore(l1, hash))
    {
      l1->_threadCollisionCount[threadId].value++;
      return true;
    }
 
    // L1 miss -> consult the authoritative global L2. probeStore() inserts the hash into L2's current
    // store, making it globally visible (so other domains detect it as a duplicate). This is what
    // preserves complete, global dedup: every globally-new hash is registered in L2 exactly once, by
    // whichever domain first encounters it.
    _l2->_threadQueryCount[threadId].value++;
    const bool collision = probeStore(_l2, hash);
    if (collision) _l2->_threadCollisionCount[threadId].value++;
    return collision;
  }
 
  __INLINE__ void insertHash(const jaffarCommon::hash::hash_t hash)
  {
    if (_l1.empty() == false) _l1[_preferredNumaDomain]->_hashStores.rbegin()->hashSet.insert(hash);
    _l2->_hashStores.rbegin()->hashSet.insert(hash);
  }
 
  __INLINE__ void advanceStep()
  {
    for (auto* const g : _l1) rollStore(g, _l1MaxStoreSizeBytes);
    rollStore(_l2, _maxStoreSizeBytes);
  }
 
private:
  struct hashStore_t
  {
    const size_t id; 
 
    const size_t age; 
 
    size_t queryCount; 
 
    size_t collisionCount; 
 
    jaffarCommon::concurrent::HashSet_t<jaffarCommon::hash::hash_t> hashSet = {}; 
  };
 
  struct alignas(64) paddedCounter_t
  {
    size_t value = 0; 
  };
 
  struct NUMAStore_t
  {
    size_t _currentHashStoreId = 0; 
 
    size_t _currentAge = 0;
 
    std::deque<hashStore_t> _hashStores;
 
    std::vector<paddedCounter_t> _threadQueryCount; 
 
    std::vector<paddedCounter_t> _threadCollisionCount; 
 
    size_t _lifetimeQueryCount     = 0;
    size_t _lifetimeCollisionCount = 0; 
 
    // Aggregating helpers (called only at reporting / store-rollover time, never in the hot path)
 
    __INLINE__ size_t getQueryCount() const
    {
      size_t total = 0;
      for (const auto& c : _threadQueryCount) total += c.value;
      return total;
    }
    __INLINE__ size_t getCollisionCount() const
    {
      size_t total = 0;
      for (const auto& c : _threadCollisionCount) total += c.value;
      return total;
    }
    __INLINE__ void resetCounts()
    {
      for (auto& c : _threadQueryCount) c.value = 0;
      for (auto& c : _threadCollisionCount) c.value = 0;
    }
    __INLINE__ size_t getTotalQueryCount() const { return _lifetimeQueryCount + getQueryCount(); }
    __INLINE__ size_t getTotalCollisionCount() const { return _lifetimeCollisionCount + getCollisionCount(); }
  };
 
  __INLINE__ NUMAStore_t* makeStore()
  {
    auto* s = new NUMAStore_t;
    s->_hashStores.push_back(hashStore_t({.id = s->_currentHashStoreId++, .age = s->_currentAge, .queryCount = 0, .collisionCount = 0}));
    s->_threadQueryCount.resize(_threadCount);
    s->_threadCollisionCount.resize(_threadCount);
    return s;
  }
 
  __INLINE__ bool probeStore(NUMAStore_t* const store, const jaffarCommon::hash::hash_t hash)
  {
    auto   itr             = store->_hashStores.rbegin();
    size_t curHashStoreIdx = 0;
    while (itr != store->_hashStores.rend())
    {
      bool collisionFound = false;
      if (curHashStoreIdx == 0)
        collisionFound = itr->hashSet.insert(hash).second == false;
      else
        collisionFound = itr->hashSet.contains(hash);
      if (collisionFound == true) return true;
      itr++;
      curHashStoreIdx++;
    }
    return false;
  }
 
  __INLINE__ void rollStore(NUMAStore_t* const g, const size_t maxBytes)
  {
    auto& currentHashStore = *g->_hashStores.rbegin();
 
    if (getStoreSizeBytes(currentHashStore) >= maxBytes)
    {
      // Snapshot the (summed per-thread) counters into the current store before any structural
      // change, then reset. This must precede pop_front(): when maxStoreCount == 1 the front store
      // IS the current (back) store, so popping it first would leave currentHashStore dangling.
      currentHashStore.queryCount     = g->getQueryCount();
      currentHashStore.collisionCount = g->getCollisionCount();
      // Fold this window's counts into the lifetime accumulators BEFORE the reset, so the since-start
      // totals survive both the reset and the eventual discard (pop_front) of this generation.
      g->_lifetimeQueryCount += currentHashStore.queryCount;
      g->_lifetimeCollisionCount += currentHashStore.collisionCount;
      g->resetCounts();
 
      // If we already reached the maximum hash stores, discard the oldest one
      if (g->_hashStores.size() == _maxStoreCount) g->_hashStores.pop_front();
 
      // Now create the new one, by pushing it from the back
      g->_hashStores.push_back(hashStore_t({.id = g->_currentHashStoreId++, .age = g->_currentAge, .queryCount = 0, .collisionCount = 0}));
    }
 
    // Increasing age
    g->_currentAge++;
  }
 
  __INLINE__ size_t getStoreSizeBytes(const hashStore_t& store) const { return store.hashSet.capacity() * _bytesPerSlot + _hashStoreFixedOverhead; }
 
  static constexpr size_t _bytesPerSlot = sizeof(jaffarCommon::hash::hash_t) + sizeof(uint8_t);
 
  static constexpr size_t _hashStoreFixedOverhead = sizeof(jaffarCommon::concurrent::HashSet_t<jaffarCommon::hash::hash_t>);
 
  size_t _maxStoreCount; 
 
  double _maxStoreSizeMb; 
 
  size_t _maxStoreSizeBytes; 
 
  size_t _l1MaxStoreSizeBytes = 0; 
 
  std::vector<NUMAStore_t*> _l1; 
 
  NUMAStore_t* _l2 = nullptr; 
};
 
} // namespace jaffarPlus