MB-56917: Don't queue BGItems if above mutation watermark

Prevent queuing items into the BGFetchQueue if the size of the
BGFetchItem will increase above the mutation watermark. If the required
memory exceeds the mutation watermark eviction will be invoked in order
to decrease memory usage. This is achieved through calling
memoryCondition().

Change-Id: If5dfd3432345575ee96fee2f110e47d85ab8ffb4
Reviewed-on: https://review.couchbase.org/c/kv_engine/+/201256
Reviewed-by: Jim Walker <[email protected]>
Tested-by: Build Bot <[email protected]>

Author: mohammadzaeem

engines/ep/src/ep_engine.cc

@@ -5439,6 +5439,14 @@ cb::engine_errc EventuallyPersistentEngine::checkPrivilege(
     return cb::engine_errc::failed;
 }
 
+cb::engine_errc EventuallyPersistentEngine::checkMemoryForBGFetch(
+        size_t pendingBytes) {
+    if (memoryTracker->isBelowMutationMemoryQuota(pendingBytes)) {
+        return cb::engine_errc::success;
+    }
+    return memoryCondition();
+}
+
 cb::engine_errc EventuallyPersistentEngine::testPrivilege(
         CookieIface& cookie,
         cb::rbac::Privilege priv,

engines/ep/src/ep_engine.h

@@ -969,6 +969,17 @@ class EventuallyPersistentEngine : public EngineIface, public DcpIface {
                                    cb::rbac::Privilege priv,
                                    CollectionID) const;
 
+    /**
+     * Checks if adding the specified memory size will keep memory usage below
+     * the mutation watermark. If so, returns success; otherwise, invoke
+     * eviction and return no_memory/temp_failure.
+     *
+     * @param pendingBytes The size of memory to be added.
+     * @return cb::engine_errc::success if addition memory size is available,
+     * otherwise no_memory / temp_failure.
+     */
+    cb::engine_errc checkMemoryForBGFetch(size_t pendingBytes);
+
     /**
      * Test the access for the given privilege for the bucket.scope.collection
      * This differs from checkPrivilege in that the error case has no side

engines/ep/src/ep_vb.cc

@@ -829,6 +829,14 @@ cb::engine_errc EPVBucket::bgFetch(HashTable::HashBucketLock&& hbl,
     // reference if that's the case
     // schedule to the current batch of background fetch of the given
     // vbucket
+
+    size_t requiredMemory =
+            StoredValue::getRequiredStorage(key) + sizeof(FrontEndBGFetchItem);
+    auto status = engine.checkMemoryForBGFetch(requiredMemory);
+    if (status != cb::engine_errc::success) {
+        return status;
+    }
+
     const auto filter =
             isMeta ? ValueFilter::KEYS_ONLY
                    : dynamic_cast<EPBucket&>(*bucket)

engines/ep/tests/module_tests/evp_store_test.cc

@@ -773,6 +773,51 @@ TEST_P(EPBucketFullEvictionTest, BgfetchSucceedsUntilMutationWatermark) {
     destroy_mock_cookie(cookie2);
 }
 
+TEST_P(EPBucketFullEvictionTest, DontQueueBGFetchItemAboveMutationWatermark) {
+    // This test relies on precise memory tracking
+    const auto& stats = engine->getEpStats();
+    if (!stats.isMemoryTrackingEnabled()) {
+        GTEST_SKIP();
+    }
+
+    ASSERT_EQ(cb::engine_errc::success,
+              store->setVBucketState(vbid, vbucket_state_active, {}));
+    EXPECT_LT(stats.getPreciseTotalMemoryUsed(), stats.getMaxDataSize());
+
+    int valueSize = 1 * 1024 * 1024;
+
+    // Load document of size 1MiB
+    const std::string value(valueSize, 'x');
+    auto key0 = makeStoredDocKey("key_0");
+
+    // store item
+    store_item(vbid, key0, value);
+    flush_vbucket_to_disk(vbid);
+
+    // evict item
+    evict_key(vbid, key0);
+
+    // try to bgFetch item
+    auto options = static_cast<get_options_t>(
+            QUEUE_BG_FETCH | HONOR_STATES | TRACK_REFERENCE | DELETE_TEMP |
+            HIDE_LOCKED_CAS | TRACK_STATISTICS | GET_DELETED_VALUE);
+    auto cookie = create_mock_cookie();
+
+    // set bucket quota using current memory usage as 93%
+    double mutationWat = engine->getConfiguration().getMutationMemRatio();
+    engine->setMaxDataSize(stats.getPreciseTotalMemoryUsed() / mutationWat);
+
+    auto gv = store->get(key0, vbid, cookie, options);
+    EXPECT_NE(cb::engine_errc::would_block, gv.getStatus());
+
+    // BGFetch queue should remain empty
+    auto vb = store->getVBucket(vbid);
+    EXPECT_FALSE(vb->hasPendingBGFetchItems());
+    EXPECT_LE(stats.getPreciseTotalMemoryUsed(), stats.getMaxDataSize());
+
+    destroy_mock_cookie(cookie);
+}
+
 TEST_P(EPBucketFullEvictionTest, ExpiryFindNonResidentItem) {
     EXPECT_EQ(cb::engine_errc::success,
               store->setVBucketState(vbid, vbucket_state_active, {}));