[Backport 2.4][#8037][DocDB] - Refactor memory management for tablets into a separate class

Summary:
Original commit: D11320 / e67b207

Refactors block cache initialization/management, log cache garbage collection, and memstore limit management into its own class.
In the process, creates log cache garbage collection and memstore management thread for the master tablet.

T8037 Refactor Memstore background task

T8037 Refactor all block cache initialization for ts_tablet_manager

T8037 Memory flush refactor to Memory Manager, passing test

T8037 Refactor sys_catalog to use memory manager

T8037 Remove refactored functions from docdb_rocksdb_util

T8037 adds comments and lint changes

Depends on D11522

Test Plan:
Jenkins: rebase: 2.4
Jenkins: hot
Depends on D11522
ybd --cxx-test tserver_ts_tablet_manager-test
ybd --cxx-test master_master-test
ybd --cxx-test integration-tests_flush-test

Reviewers: sergei, rsami, bogdan

Reviewed By: bogdan

Subscribers: ybase, jenkins-bot

Differential Revision: https://phabricator.dev.yugabyte.com/D11577

Author: jmeehan16

src/yb/docdb/docdb_rocksdb_util.cc

@@ -118,31 +118,6 @@ DEFINE_int32(priority_thread_pool_size, -1,
              "If -1 and max_background_compactions is specified - use max_background_compactions. "
              "If -1 and max_background_compactions is not specified - use sqrt(num_cpus).");
 
-namespace {
-  constexpr int kDbCacheSizeUsePercentage = -1;
-  constexpr int kDbCacheSizeCacheDisabled = -2;
-  constexpr int kDbCacheSizeUseDefault = -3;
-}
-
-DEFINE_bool(enable_block_based_table_cache_gc, false,
-            "Set to true to enable block based table garbage collector.");
-
-DEFINE_int64(db_block_cache_size_bytes, kDbCacheSizeUsePercentage,
-             "Size of RocksDB block cache (in bytes). "
-             "This defaults to -1 for system auto-generated default, which would use "
-             "FLAGS_db_block_cache_size_percentage to select a percentage of the total "
-             "memory as the default size for the shared block cache. Value of -2 disables "
-             "block cache.");
-
-DEFINE_int32(db_block_cache_size_percentage, kDbCacheSizeUseDefault,
-             "Default percentage of total available memory to use as block cache size, if not "
-             "asking for a raw number, through FLAGS_db_block_cache_size_bytes. "
-             "Defaults to -3 (use default percentage as defined by master or tserver).");
-
-DEFINE_int32(db_block_cache_num_shard_bits, 4,
-             "Number of bits to use for sharding the block cache (defaults to 4 bits)");
-TAG_FLAG(db_block_cache_num_shard_bits, advanced);
-
 using std::shared_ptr;
 using std::string;
 using std::unique_ptr;
@@ -743,77 +718,5 @@ Status ForceRocksDBCompact(rocksdb::DB* db) {
   return Status::OK();
 }
 
-namespace {
-
-class LRUCacheGC : public GarbageCollector {
- public:
-  explicit LRUCacheGC(std::shared_ptr<rocksdb::Cache> cache) : cache_(std::move(cache)) {}
-
-  void CollectGarbage(size_t required) {
-    if (!FLAGS_enable_block_based_table_cache_gc) {
-      return;
-    }
-
-    auto evicted = cache_->Evict(required);
-    LOG(INFO) << "Evicted from table cache: " << HumanReadableNumBytes::ToString(evicted)
-              << ", new usage: " << HumanReadableNumBytes::ToString(cache_->GetUsage())
-              << ", required: " << HumanReadableNumBytes::ToString(required);
-  }
-
-  virtual ~LRUCacheGC() = default;
-
- private:
-  std::shared_ptr<rocksdb::Cache> cache_;
-};
-
-int64_t GetTargetBlockCacheSize(const int32_t default_block_cache_size_percentage) {
-  int32_t target_block_cache_size_percentage =
-      (FLAGS_db_block_cache_size_percentage == kDbCacheSizeUseDefault) ?
-      default_block_cache_size_percentage : FLAGS_db_block_cache_size_percentage;
-
-  int64_t target_block_cache_size_bytes = FLAGS_db_block_cache_size_bytes;
-  // Auto-compute size of block cache if asked to.
-  if (target_block_cache_size_bytes == kDbCacheSizeUsePercentage) {
-    // Check some bounds.
-    CHECK(target_block_cache_size_percentage > 0 && target_block_cache_size_percentage <= 100)
-        << Substitute(
-               "Flag tablet_block_cache_size_percentage must be between 0 and 100. Current value: "
-               "$0",
-               target_block_cache_size_percentage);
-
-    const int64_t total_ram_avail = MemTracker::GetRootTracker()->limit();
-    target_block_cache_size_bytes = total_ram_avail * target_block_cache_size_percentage / 100;
-  }
-  return target_block_cache_size_bytes;
-}
-
-}  // namespace
-
-std::shared_ptr<MemTracker> InitBlockCacheMemTracker(
-    const int32_t default_block_cache_size_percentage,
-    const std::shared_ptr<MemTracker>& mem_tracker) {
-  int64_t block_cache_size_bytes = GetTargetBlockCacheSize(default_block_cache_size_percentage);
-
-  return MemTracker::FindOrCreateTracker(block_cache_size_bytes, "BlockBasedTable", mem_tracker);
-}
-
-std::shared_ptr<GarbageCollector> InitBlockCache(
-    const scoped_refptr<MetricEntity>& metrics,
-    const int32_t default_block_cache_size_percentage,
-    MemTracker* block_based_table_mem_tracker,
-    tablet::TabletOptions* options) {
-  std::shared_ptr<GarbageCollector> block_based_table_gc;
-  int64_t block_cache_size_bytes = GetTargetBlockCacheSize(default_block_cache_size_percentage);
-
-  if (block_cache_size_bytes != kDbCacheSizeCacheDisabled) {
-    options->block_cache = rocksdb::NewLRUCache(block_cache_size_bytes,
-                                                FLAGS_db_block_cache_num_shard_bits);
-    options->block_cache->SetMetrics(metrics);
-    block_based_table_gc = std::make_shared<LRUCacheGC>(options->block_cache);
-    block_based_table_mem_tracker->AddGarbageCollector(block_based_table_gc);
-  }
-  return block_based_table_gc;
-}
-
 }  // namespace docdb
 }  // namespace yb

src/yb/docdb/docdb_rocksdb_util.h

@@ -106,16 +106,6 @@ std::unique_ptr<IntentAwareIterator> CreateIntentAwareIterator(
 // Request RocksDB compaction and wait until it completes.
 CHECKED_STATUS ForceRocksDBCompact(rocksdb::DB* db);
 
-std::shared_ptr<MemTracker> InitBlockCacheMemTracker(
-    const int32_t default_block_cache_size_percentage,
-    const std::shared_ptr<MemTracker>& mem_tracker);
-
-std::shared_ptr<GarbageCollector> InitBlockCache(
-    const scoped_refptr<MetricEntity>& metrics,
-    const int32_t default_block_cache_size_percentage,
-    MemTracker* block_based_table_mem_tracker,
-    tablet::TabletOptions* options);
-
 // Initialize the RocksDB 'options'.
 // The 'statistics' object provided by the caller will be used by RocksDB to maintain the stats for
 // the tablet.

src/yb/integration-tests/flush-test.cc

@@ -26,6 +26,7 @@
 
 #include "yb/tserver/mini_tablet_server.h"
 #include "yb/tserver/tablet_server.h"
+#include "yb/tserver/tablet_memory_manager.h"
 #include "yb/tserver/ts_tablet_manager.h"
 
 using namespace std::literals;
@@ -70,7 +71,7 @@ Result<TabletId> GetTabletIdFromSstFilename(const std::string& filename) {
   }
 }
 
-class TabletManagerListener : public tserver::TsTabletManagerListener {
+class TabletManagerListener : public tserver::TabletMemoryManagerListenerIf {
  public:
   void StartedFlush(const TabletId& tablet_id) override {
     std::lock_guard<std::mutex> lock(mutex_);
@@ -126,7 +127,8 @@ class FlushITest : public YBTest {
   }
 
   void SetupCluster() {
-    cluster_->GetTabletManager(0)->TEST_listeners.push_back(tablet_manager_listener_);
+    cluster_->GetTabletManager(0)->tablet_memory_manager()->TEST_listeners.push_back(
+        tablet_manager_listener_);
   }
 
   void SetupWorkload(

src/yb/master/sys_catalog.cc

@@ -54,7 +54,6 @@
 #include "yb/consensus/quorum_util.h"
 
 #include "yb/docdb/doc_rowwise_iterator.h"
-#include "yb/docdb/docdb_rocksdb_util.h"
 
 #include "yb/fs/fs_manager.h"
 #include "yb/gutil/strings/numbers.h"
@@ -69,6 +68,7 @@
 #include "yb/tablet/tablet_options.h"
 #include "yb/tablet/operations/write_operation.h"
 
+#include "yb/tserver/tablet_memory_manager.h"
 #include "yb/tserver/ts_tablet_manager.h"
 
 #include "yb/util/flag_tags.h"
@@ -157,6 +157,9 @@ SysCatalogTable::~SysCatalogTable() {
 }
 
 void SysCatalogTable::StartShutdown() {
+  if (mem_manager_) {
+    mem_manager_->Shutdown();
+  }
   if (tablet_peer()) {
     CHECK(std::atomic_load(&tablet_peer_)->StartShutdown());
   }
@@ -520,21 +523,24 @@ Status SysCatalogTable::OpenTablet(const scoped_refptr<tablet::RaftGroupMetadata
   RETURN_NOT_OK(tablet_peer()->SetBootstrapping());
   tablet::TabletOptions tablet_options;
 
-  block_based_table_mem_tracker_ = docdb::InitBlockCacheMemTracker(
+  // Returns a vector that includes the tablet peer associated with master.
+  const auto get_peers_lambda = [this]() -> std::vector<tablet::TabletPeerPtr> {
+    return { tablet_peer() };
+  };
+
+  mem_manager_ = std::make_shared<tserver::TabletMemoryManager>(
+      &tablet_options,
+      master_->mem_tracker(),
       kDefaultMasterBlockCacheSizePercentage,
-      master_->mem_tracker());
-  block_based_table_gc_ = docdb::InitBlockCache(
       GetMetricEntity(),
-      kDefaultMasterBlockCacheSizePercentage,
-      block_based_table_mem_tracker_.get(),
-      &tablet_options);
+      get_peers_lambda);
 
   tablet::TabletInitData tablet_init_data = {
       .metadata = metadata,
       .client_future = master_->async_client_initializer().get_client_future(),
       .clock = scoped_refptr<server::Clock>(master_->clock()),
       .parent_mem_tracker = master_->mem_tracker(),
-      .block_based_table_mem_tracker = block_based_table_mem_tracker_,
+      .block_based_table_mem_tracker = mem_manager_->block_based_table_mem_tracker(),
       .metric_registry = metric_registry_,
       .log_anchor_registry = tablet_peer()->log_anchor_registry(),
       .tablet_options = tablet_options,
@@ -587,6 +593,8 @@ Status SysCatalogTable::OpenTablet(const scoped_refptr<tablet::RaftGroupMetadata
   if (!tablet->schema()->Equals(schema_)) {
     return STATUS(Corruption, "Unexpected schema", tablet->schema()->ToString());
   }
+  RETURN_NOT_OK(mem_manager_->Init());
+
   return Status::OK();
 }
 

src/yb/master/sys_catalog.h

@@ -44,6 +44,8 @@
 #include "yb/tablet/snapshot_coordinator.h"
 #include "yb/tablet/tablet_peer.h"
 
+#include "yb/tserver/tablet_memory_manager.h"
+
 #include "yb/util/mem_tracker.h"
 #include "yb/util/pb_util.h"
 #include "yb/util/status.h"
@@ -258,9 +260,7 @@ class SysCatalogTable {
 
   std::unordered_map<std::string, scoped_refptr<AtomicGauge<uint64>>> visitor_duration_metrics_;
 
-  std::shared_ptr<MemTracker> block_based_table_mem_tracker_;
-
-  std::shared_ptr<GarbageCollector> block_based_table_gc_;
+  std::shared_ptr<tserver::TabletMemoryManager> mem_manager_;
 
   DISALLOW_COPY_AND_ASSIGN(SysCatalogTable);
 };

src/yb/tserver/CMakeLists.txt

@@ -163,6 +163,7 @@ set(TSERVER_SRCS
   remote_bootstrap_session.cc
   remote_bootstrap_snapshots.cc
   service_util.cc
+  tablet_memory_manager.cc
   tablet_server.cc
   tablet_server_options.cc
   tablet_service.cc