MB-39745: Add BgFetchers to EPBucket

Move the BgFetchers to EPBucket in preparation for decoupling them from
KVShards. The behaviour of BgFetchers in this patch should be identical
to their behaviour before.

Change-Id: I40f57029b07b27180390e48d9ab460476faa5fae
Reviewed-on: http://review.couchbase.org/c/kv_engine/+/136377
Tested-by: Build Bot <[email protected]>
Reviewed-by: Dave Rigby <[email protected]>

Author: BenHuddleston

engines/ep/src/ep_bucket.cc

@@ -266,6 +266,9 @@ EPBucket::EPBucket(EventuallyPersistentEngine& theEngine)
             engine.getConfiguration(), stats);
 
     vbMap.enablePersistence(*this);
+    for (size_t i = 0; i < vbMap.getNumShards(); i++) {
+        bgFetchers.emplace_back(std::make_unique<BgFetcher>(*this));
+    }
 
     setFlusherBatchSplitTrigger(config.getFlusherTotalBatchLimit());
     config.addValueChangedListener(
@@ -948,21 +951,16 @@ void EPBucket::wakeUpFlusher() {
 }
 
 bool EPBucket::startBgFetcher() {
-    for (const auto& shard : vbMap.shards) {
-        BgFetcher* bgfetcher = shard->getBgFetcher();
-        if (bgfetcher == NULL) {
-            EP_LOG_WARN("Failed to start bg fetcher for shard {}",
-                        shard->getId());
-            return false;
-        }
-        bgfetcher->start();
+    for (const auto& bgFetcher : bgFetchers) {
+        bgFetcher->start();
     }
     return true;
 }
 
 void EPBucket::stopBgFetcher() {
+    EP_LOG_INFO("Stopping bg fetchers");
+
     for (const auto& shard : vbMap.shards) {
-        BgFetcher* bgfetcher = shard->getBgFetcher();
         for (const auto vbid : shard->getVBuckets()) {
             VBucketPtr vb = shard->getBucket(vbid);
             if (vb && vb->hasPendingBGFetchItems()) {
@@ -974,9 +972,10 @@ void EPBucket::stopBgFetcher() {
                 break;
             }
         }
+    }
 
-        EP_LOG_INFO("Stopping bg fetcher for shard:{}", shard->getId());
-        bgfetcher->stop();
+    for (const auto& bgFetcher : bgFetchers) {
+        bgFetcher->stop();
     }
 }
 
@@ -1353,6 +1352,7 @@ VBucketPtr EPBucket::makeVBucket(
                                     engine.getConfiguration(),
                                     eviction_policy,
                                     std::move(manifest),
+                                    this,
                                     initState,
                                     purgeSeqno,
                                     maxCas,
@@ -2041,3 +2041,10 @@ std::ostream& operator<<(std::ostream& os, const EPBucket::FlushResult& res) {
        << (res.wakeupCkptRemover == EPBucket::WakeCkptRemover::Yes) << "}";
     return os;
 }
+
+BgFetcher& EPBucket::getBgFetcher(Vbid vbid) {
+    // For now we just use modulo, same as we do/would for shards to pick out
+    // the associated BgFetcher
+    auto id = vbid.get() % bgFetchers.size();
+    return *bgFetchers.at(id);
+}

engines/ep/src/ep_bucket.h

@@ -19,6 +19,7 @@
 
 #include "kv_bucket.h"
 
+class BgFetcher;
 namespace Collections::VB {
 class Flush;
 }
@@ -236,6 +237,8 @@ class EPBucket : public KVBucket {
             const void* cookie,
             std::unique_ptr<Collections::Manifest>& newManifest) override;
 
+    BgFetcher& getBgFetcher(Vbid vbid);
+
 protected:
     // During the warmup phase we might want to enable external traffic
     // at a given point in time.. The LoadStorageKvPairCallback will be
@@ -333,6 +336,8 @@ class EPBucket : public KVBucket {
     cb::RelaxedAtomic<bool> retainErroneousTombstones;
 
     std::unique_ptr<Warmup> warmupTask;
+
+    std::vector<std::unique_ptr<BgFetcher>> bgFetchers;
 };
 
 std::ostream& operator<<(std::ostream& os, const EPBucket::FlushResult& res);

engines/ep/src/ep_vb.cc

@@ -25,6 +25,7 @@
 #include "dcp/backfill_by_seqno_disk.h"
 #include "durability/active_durability_monitor.h"
 #include "durability/passive_durability_monitor.h"
+#include "ep_bucket.h"
 #include "ep_engine.h"
 #include "ep_time.h"
 #include "executorpool.h"
@@ -56,6 +57,7 @@ EPVBucket::EPVBucket(Vbid i,
                      Configuration& config,
                      EvictionPolicy evictionPolicy,
                      std::unique_ptr<Collections::VB::Manifest> manifest,
+                     EPBucket* bucket,
                      vbucket_state_t initState,
                      uint64_t purgeSeqno,
                      uint64_t maxCas,
@@ -87,7 +89,8 @@ EPVBucket::EPVBucket(Vbid i,
               mightContainXattrs,
               replicationTopology,
               maxVisibleSeqno),
-      shard(kvshard) {
+      shard(kvshard),
+      epBucket(bucket) {
 }
 
 EPVBucket::~EPVBucket() {
@@ -596,7 +599,7 @@ size_t EPVBucket::getPageableMemUsage() {
 
 size_t EPVBucket::queueBGFetchItem(const DocKey& key,
                                    std::unique_ptr<BGFetchItem> fetch,
-                                   BgFetcher* bgFetcher) {
+                                   BgFetcher& bgFetcher) {
     // While a DiskDocKey supports both the committed and prepared namespaces,
     // ep-engine doesn't support evicting prepared SyncWrites and as such
     // we don't allow bgfetching from Prepared namespace - so just construct
@@ -616,7 +619,7 @@ size_t EPVBucket::queueBGFetchItem(const DocKey& key,
     fetch->value = &bgfetch_itm_ctx.value;
     bgfetch_itm_ctx.bgfetched_list.push_back(std::move(fetch));
 
-    bgFetcher->addPendingVB(getId());
+    bgFetcher.addPendingVB(getId());
     return pendingBGFetches.size();
 }
 
@@ -747,12 +750,14 @@ void EPVBucket::bgFetch(const DocKey& key,
                         const void* cookie,
                         EventuallyPersistentEngine& engine,
                         const bool isMeta) {
+    // @TODO could the BgFetcher ever not be there? It should probably be a
+    // reference if that's the case
     // schedule to the current batch of background fetch of the given
     // vbucket
     size_t bgfetch_size = queueBGFetchItem(
             key,
             std::make_unique<FrontEndBGFetchItem>(cookie, isMeta),
-            getShard()->getBgFetcher());
+            getBgFetcher());
     EP_LOG_DEBUG("Queued a background fetch, now at {}",
                  uint64_t(bgfetch_size));
 }
@@ -780,12 +785,9 @@ void EPVBucket::bgFetchForCompactionExpiry(const DocKey& key,
                                            const Item& item) {
     // schedule to the current batch of background fetch of the given
     // vbucket
-    auto shard = getShard();
-    Expects(shard);
-    auto bgFetchSize =
-            queueBGFetchItem(key,
-                             std::make_unique<CompactionBGFetchItem>(item),
-                             shard->getBgFetcher());
+    auto& bgFetcher = getBgFetcher();
+    auto bgFetchSize = queueBGFetchItem(
+            key, std::make_unique<CompactionBGFetchItem>(item), bgFetcher);
     EP_LOG_DEBUG("Queue a background fetch for compaction expiry, now at {}",
                  bgFetchSize);
 }
@@ -1019,3 +1021,7 @@ void EPVBucket::processImplicitlyCompletedPrepare(
     // used after.
     ht.unlocked_del(v.getHBL(), v.release());
 }
+
+BgFetcher& EPVBucket::getBgFetcher() {
+    return epBucket->getBgFetcher(getId());
+}

engines/ep/src/ep_vb.h

@@ -20,8 +20,9 @@
 #include "vbucket.h"
 #include "vbucket_bgfetch_item.h"
 
-struct vbucket_state;
 class BgFetcher;
+class EPBucket;
+struct vbucket_state;
 
 /**
  * Eventually Peristent VBucket (EPVBucket) is a child class of VBucket.
@@ -46,6 +47,7 @@ class EPVBucket : public VBucket {
               Configuration& config,
               EvictionPolicy evictionPolicy,
               std::unique_ptr<Collections::VB::Manifest> manifest,
+              EPBucket* bucket = nullptr,
               vbucket_state_t initState = vbucket_state_dead,
               uint64_t purgeSeqno = 0,
               uint64_t maxCas = 0,
@@ -128,6 +130,8 @@ class EPVBucket : public VBucket {
         return shard;
     }
 
+    BgFetcher& getBgFetcher();
+
     UniqueDCPBackfillPtr createDCPBackfill(EventuallyPersistentEngine& e,
                                            std::shared_ptr<ActiveStream> stream,
                                            uint64_t startSeqno,
@@ -243,7 +247,7 @@ class EPVBucket : public VBucket {
      */
     size_t queueBGFetchItem(const DocKey& key,
                             std::unique_ptr<BGFetchItem> fetch,
-                            BgFetcher* bgFetcher);
+                            BgFetcher& bgFetcher);
 
 private:
     std::tuple<StoredValue*, MutationStatus, VBNotifyCtx> updateStoredValue(
@@ -343,5 +347,10 @@ class EPVBucket : public VBucket {
      */
     std::atomic<uint64_t> deferredDeletionFileRevision;
 
+    /**
+     * Should replace KVShard
+     */
+    EPBucket* epBucket;
+
     friend class EPVBucketTest;
 };

engines/ep/src/kvshard.cc

@@ -18,7 +18,6 @@
 #include <functional>
 #include <memory>
 
-#include "bgfetcher.h"
 #include "couch-kvstore/couch-kvstore-config.h"
 #include "ep_bucket.h"
 #include "ep_engine.h"
@@ -84,7 +83,6 @@ KVShard::KVShard(EventuallyPersistentEngine& engine, id_type id)
 
 void KVShard::enablePersistence(EPBucket& ep) {
     flusher = std::make_unique<Flusher>(&ep, this);
-    bgFetcher = std::make_unique<BgFetcher>(ep);
 }
 
 // Non-inline destructor so we can destruct
@@ -95,10 +93,6 @@ Flusher *KVShard::getFlusher() {
     return flusher.get();
 }
 
-BgFetcher *KVShard::getBgFetcher() {
-    return bgFetcher.get();
-}
-
 VBucketPtr KVShard::getBucket(Vbid id) const {
     if (id.get() < kvConfig->getMaxVBuckets()) {
         auto element = getElement(id);

engines/ep/src/kvshard.h

@@ -45,7 +45,6 @@
  *   | vbuckets: VBucket[] (partitions)|----> [(VBucket),(VBucket)..]
  *   |                                 |
  *   | flusher: Flusher                |
- *   | BGFetcher: bgFetcher            |
  *   |                                 |
  *   | rwUnderlying: KVStore (write)   |----> (CouchKVStore)
  *   | roUnderlying: KVStore (read)    |----> (CouchKVStore)
@@ -67,7 +66,6 @@
  *     1022            2
  *     1023               3
  */
-class BgFetcher;
 class Configuration;
 class EPBucket;
 class EventuallyPersistentEngine;
@@ -82,7 +80,7 @@ class KVShard {
     KVShard(EventuallyPersistentEngine& engine, KVShard::id_type id);
     ~KVShard();
 
-    /// Enable persistence for this KVShard; setting up flusher and BGFetcher.
+    /// Enable persistence for this KVShard; setting up the flusher.
     void enablePersistence(EPBucket& epBucket);
 
     KVStore* getRWUnderlying() {
@@ -115,7 +113,6 @@ class KVShard {
     }
 
     Flusher *getFlusher();
-    BgFetcher *getBgFetcher();
 
     VBucketPtr getBucket(Vbid id) const;
     void setBucket(VBucketPtr vb);
@@ -251,7 +248,6 @@ class KVShard {
     std::unique_ptr<KVStore> roStore;
 
     std::unique_ptr<Flusher> flusher;
-    std::unique_ptr<BgFetcher> bgFetcher;
 
 public:
     std::atomic<size_t> highPriorityCount;

engines/ep/tests/module_tests/evp_store_with_meta.cc

@@ -1387,8 +1387,7 @@ TEST_P(DelWithMetaTest, setting_zero_deleteTime) {
                                  metadata,
                                  deleted,
                                  datatype));
-    MockGlobalTask mockTask(engine->getTaskable(), TaskId::MultiBGFetcherTask);
-    store->getVBucket(vbid)->getShard()->getBgFetcher()->run(&mockTask);
+    runBGFetcherTask();
     EXPECT_EQ(ENGINE_SUCCESS,
               store->getMetaData({"mykey", DocKeyEncodesCollectionId::No},
                                  vbid,

engines/ep/tests/module_tests/evp_vbucket_test.cc

@@ -37,7 +37,7 @@ void EPVBucketTest::TearDown() {
 size_t EPVBucketTest::public_queueBGFetchItem(
         const DocKey& key,
         std::unique_ptr<BGFetchItem> fetchItem,
-        BgFetcher* bgFetcher) {
+        BgFetcher& bgFetcher) {
     return dynamic_cast<EPVBucket&>(*vbucket).queueBGFetchItem(
             key, std::move(fetchItem), bgFetcher);
 }
@@ -63,7 +63,7 @@ TEST_P(EPVBucketTest, GetBGFetchItemsPerformance) {
                                                       /*isMeta*/ false);
         this->public_queueBGFetchItem(makeStoredDocKey(std::to_string(ii)),
                                       std::move(fetchItem),
-                                      &bgFetcher);
+                                      bgFetcher);
     }
     auto items = this->vbucket->getBGFetchItems();
 }

engines/ep/tests/module_tests/evp_vbucket_test.h

@@ -38,5 +38,5 @@ class EPVBucketTest : public ::testing::WithParamInterface<EvictionPolicy>,
     void TearDown() override;
     size_t public_queueBGFetchItem(const DocKey& key,
                                    std::unique_ptr<BGFetchItem> fetchItem,
-                                   BgFetcher* bgFetcher);
+                                   BgFetcher& bgFetcher);
 };

engines/ep/tests/module_tests/kv_bucket_test.cc

@@ -34,6 +34,7 @@
 #include "ep_bucket.h"
 #include "ep_engine.h"
 #include "ep_time.h"
+#include "ep_vb.h"
 #include "evp_store_single_threaded_test.h"
 #include "failover-table.h"
 #include "fakes/fake_executorpool.h"
@@ -302,7 +303,10 @@ bool KVBucketTest::isItemFreqDecayerTaskSnoozed() const {
 void KVBucketTest::runBGFetcherTask() {
     MockGlobalTask mockTask(engine->getTaskable(),
                             TaskId::MultiBGFetcherTask);
-    store->getVBucket(vbid)->getShard()->getBgFetcher()->run(&mockTask);
+    auto vb = store->getVBucket(vbid);
+    ASSERT_TRUE(vb);
+    auto* epVb = dynamic_cast<EPVBucket*>(vb.get());
+    epVb->getBgFetcher().run(&mockTask);
 }
 
 /**