Load Cachestream singleton at DB open, rather than on compaction

Author: jcarin

cachestream/cachestream.cc

@@ -181,4 +181,4 @@ void Cachestream::remove_tgid(int tgid) {
     }
 }
 
-} // namespace ROCKSDB_NAMESPACE
+} // namespace ROCKSDB_NAMESPACE

db/compaction/compaction_job.cc

@@ -60,6 +60,8 @@
 #include "cachestream/cachestream.h"
 #include <sys/syscall.h>
 
+#include <iostream>
+
 namespace ROCKSDB_NAMESPACE {
 
 const char* GetCompactionReasonString(CompactionReason compaction_reason) {
@@ -719,9 +721,6 @@ Status CompactionJob::Run() {
         compact_->compaction->mutable_cf_options()->prefix_extractor;
     std::atomic<size_t> next_file_idx(0);
     auto verify_table = [&](Status& output_status) {
-      auto& cachestream = Cachestream::getInstance();
-      int tid = syscall(SYS_gettid);
-      CachestreamTidGuard tid_guard(cachestream, tid);
 
       while (true) {
         size_t file_idx = next_file_idx.fetch_add(1);
@@ -1082,7 +1081,6 @@ void CompactionJob::ProcessKeyValueCompaction(SubcompactionState* sub_compact) {
   assert(sub_compact);
   assert(sub_compact->compaction);
 
-  // Add current thread's TID to cachestream map using RAII
   auto& cachestream = Cachestream::getInstance();
   int tid = syscall(SYS_gettid);
   CachestreamTidGuard tid_guard(cachestream, tid);

db/db_impl/db_impl_open.cc

@@ -28,6 +28,8 @@
 #include "util/string_util.h"
 #include "util/udt_util.h"
 
+#include "cachestream/cachestream.h"
+
 namespace ROCKSDB_NAMESPACE {
 Options SanitizeOptions(const std::string& dbname, const Options& src,
                         bool read_only, Status* logger_creation_s) {
@@ -1952,6 +1954,9 @@ Status DBImpl::Open(const DBOptions& db_options, const std::string& dbname,
   assert(handles);
   handles->clear();
 
+  // Initialize Cachestream singleton, joins cgroup and loads BPF program.
+  Cachestream::getInstance();
+  
   size_t max_write_buffer_size = 0;
   for (auto cf : column_families) {
     max_write_buffer_size =