MB-41782: Create one task for scheduleCompaction

Currently multiple calls to scheduleCompaction result
in new tasks being created and scheduled, they will all
eventually run and compact the vbucket many times.

This change aims to improve this by trying to maintain just one
CompactTask per vbucket. The aim is that if multiple
schedculeCompaction calls occur

* The first call creates a task with the given CompactionConfig
* Subsequent calls update the task with the most recent
  CompactionConfig

The task itself is changed so that there is a window/scope inside
the run function where it copies the configuration and starts
compacting. Once this has happened the compaction will complete,
but any schedule calls happening whilst compaction is running will
update the tasks config and set a flag so that when the task
completes it reschedules ready to run again with the new config.

The main purpose of this change is to improve what happens when a
scope is dropped. We have seen from logs that a scope drop may
trigger many compactions as each collection that is dropped
calls scheduleCompaction. As part of this improvement the
scheduleCompaction function now takes a delay parameter which
when 0 means no-delay (schedule now) otherwise schedule but run
in 'delay' seconds. The collections use of scheduleCompaction sets
a short delay and once all collection drops have been flushed
one compaction will run a short time later.

Change-Id: I998b5fe6833b2b7e1681cac64933d58da7b9560f
Reviewed-on: http://review.couchbase.org/c/kv_engine/+/140364
Tested-by: Build Bot <[email protected]>
Reviewed-by: James Harrison <[email protected]>

Author: jimwwalker

engines/ep/configuration.json

@@ -197,6 +197,12 @@
             "dynamic": false,
             "type": "size_t"
         },
+        "collections_drop_compaction_delay" : {
+            "default": "5000",
+            "descr": "How many milliseconds before compaction runs following the drop of a collection",
+            "dynamic": false,
+            "type": "size_t"
+        },
         "collections_enabled" : {
             "default": "true",
             "descr": "Enable the collections functionality, enabling the storage of collection metadata",

engines/ep/src/collections/flush.cc

@@ -23,6 +23,7 @@
 #include "collections/vbucket_manifest.h"
 #include "collections/vbucket_manifest_handles.h"
 #include "ep_bucket.h"
+#include "ep_engine.h"
 #include "item.h"
 
 namespace Collections::VB {
@@ -242,8 +243,16 @@ void Flush::checkAndTriggerPurge(Vbid vbid, EPBucket& bucket) const {
 }
 
 void Flush::triggerPurge(Vbid vbid, EPBucket& bucket) {
-    CompactionConfig config;
-    bucket.scheduleCompaction(vbid, config, nullptr);
+    // There's no requirement for compaction to run 'now', schedule with a delay
+    // which allows for any other drop events in the queue to all end up
+    // 'coalesced' into one run of compaction.
+    bucket.scheduleCompaction(
+            vbid,
+            nullptr,
+            std::chrono::milliseconds(
+                    bucket.getEPEngine()
+                            .getConfiguration()
+                            .getCollectionsDropCompactionDelay()));
 }
 
 static std::pair<bool, std::optional<CollectionID>> getCollectionID(

engines/ep/src/ep_bucket.cc

@@ -1000,9 +1000,11 @@ void EPBucket::stopBgFetcher() {
     }
 }
 
-ENGINE_ERROR_CODE EPBucket::scheduleCompaction(Vbid vbid,
-                                               const CompactionConfig& c,
-                                               const void* cookie) {
+ENGINE_ERROR_CODE EPBucket::scheduleCompaction(
+        Vbid vbid,
+        std::optional<CompactionConfig> config,
+        const void* cookie,
+        std::chrono::milliseconds delay) {
     ENGINE_ERROR_CODE errCode = checkForDBExistence(vbid);
     if (errCode != ENGINE_SUCCESS) {
         return errCode;
@@ -1014,41 +1016,66 @@ ENGINE_ERROR_CODE EPBucket::scheduleCompaction(Vbid vbid,
         return ENGINE_NOT_MY_VBUCKET;
     }
 
-    LockHolder lh(compactionLock);
-    ExTask task = std::make_shared<CompactTask>(*this, vbid, c, cookie);
-    compactionTasks.emplace_back(std::make_pair(vbid, task));
-    bool snoozed = false;
-    if (compactionTasks.size() > 1) {
-        if ((stats.diskQueueSize > compactionWriteQueueCap &&
-             compactionTasks.size() > (vbMap.getNumShards() / 2)) ||
-            engine.getWorkLoadPolicy().getWorkLoadPattern() == READ_HEAVY) {
-            // Snooze a new compaction task.
-            // We will wake it up when one of the existing compaction tasks is
-            // done.
-            task->snooze(60);
-            snoozed = true;
+    auto handle = compactionTasks.wlock();
+
+    // Convert delay to ExecutorPool 'double' e.g. 1500ms = 1.5 secs
+    std::chrono::duration<double> execDelay = delay;
+
+    // try to emplace an empty shared_ptr
+    auto [itr, emplaced] = handle->try_emplace(vbid, nullptr);
+    auto& task = itr->second;
+
+    if (!emplaced) {
+        // The existing task must be poked - it needs to either reschedule if
+        // it is currently running or run with the given config.
+        task->runCompactionWithConfig(config);
+        if (execDelay.count() > 0.0) {
+            ExecutorPool::get()->snooze(task->getId(), execDelay.count());
+        } else {
+            ExecutorPool::get()->wake(task->getId());
+        }
+    } else {
+        // Nothing in the map for this vbid now construct the task
+        itr->second =
+                std::make_shared<CompactTask>(*this, vbid, config, cookie);
+        if (handle->size() > 1) {
+            if ((stats.diskQueueSize > compactionWriteQueueCap &&
+                 handle->size() > (vbMap.getNumShards() / 2)) ||
+                engine.getWorkLoadPolicy().getWorkLoadPattern() == READ_HEAVY) {
+                // Snooze a new compaction task.
+                // We will wake it up when one of the existing compaction tasks
+                // is done.
+                execDelay = std::chrono::seconds(60);
+            }
+        }
+
+        if (execDelay.count() > 0.0) {
+            task->snooze(execDelay.count());
         }
+        ExecutorPool::get()->schedule(task);
     }
 
-    ExecutorPool::get()->schedule(task);
+    return ENGINE_EWOULDBLOCK;
+}
 
-    EP_LOG_INFO(
-            "Compaction of {}, task:{}, purge_before_ts:{}, "
-            "purge_before_seq:{}, "
-            "drop_deletes:{}, snoozed:{}, scheduled (awaiting completion).",
-            vbid,
-            uint64_t(task->getId()),
-            c.purge_before_ts,
-            c.purge_before_seq,
-            c.drop_deletes,
-            snoozed);
+ENGINE_ERROR_CODE EPBucket::scheduleCompaction(
+        Vbid vbid,
+        const CompactionConfig& config,
+        const void* cookie,
+        std::chrono::milliseconds delay) {
+    return scheduleCompaction(
+            vbid, std::optional<CompactionConfig>{config}, cookie, delay);
+}
 
-    return ENGINE_EWOULDBLOCK;
+ENGINE_ERROR_CODE EPBucket::scheduleCompaction(
+        Vbid vbid, const void* cookie, std::chrono::milliseconds delay) {
+    return scheduleCompaction(
+            vbid, std::optional<CompactionConfig>{}, cookie, delay);
 }
 
 ENGINE_ERROR_CODE EPBucket::cancelCompaction(Vbid vbid) {
-    LockHolder lh(compactionLock);
-    for (const auto& task : compactionTasks) {
+    auto handle = compactionTasks.wlock();
+    for (const auto& task : *handle) {
         task.second->cancel();
     }
     return ENGINE_SUCCESS;
@@ -1231,35 +1258,50 @@ bool EPBucket::doCompact(Vbid vbid,
         engine.decrementSessionCtr();
     }
 
-    updateCompactionTasks(vbid);
-
     if (cookie) {
         engine.notifyIOComplete(cookie, err);
     }
-    --stats.pendingCompactions;
+    --stats.pendingCompactions; // just size of map??
     return false;
 }
 
-void EPBucket::updateCompactionTasks(Vbid db_file_id) {
-    LockHolder lh(compactionLock);
-    bool erased = false, woke = false;
-    auto it = compactionTasks.begin();
-    while (it != compactionTasks.end()) {
-        if ((*it).first == db_file_id) {
-            it = compactionTasks.erase(it);
-            erased = true;
-        } else {
-            ExTask& task = (*it).second;
-            if (task->getState() == TASK_SNOOZED) {
-                ExecutorPool::get()->wake(task->getId());
-                woke = true;
+bool EPBucket::updateCompactionTasks(Vbid vbid, bool canErase) {
+    auto handle = compactionTasks.wlock();
+    // Copy the size before we may erase a task
+    auto size = handle->size();
+    bool reschedule = false;
+
+    // process the caller and then find a second task to wake
+    if (canErase) {
+        if (auto itr = handle->find(vbid); itr != handle->end()) {
+            const auto& task = (*itr).second;
+            if (task->isRescheduleRequired()) {
+                // Nope can't erase!
+                reschedule = true;
+            } else {
+                // Done, can now erase from the compactionTasks map
+                handle->erase(itr);
             }
-            ++it;
+        } else {
+            throw std::logic_error(
+                    "EPBucket::updateCompactionTasks no task for vbid:" +
+                    vbid.to_string());
         }
-        if (erased && woke) {
-            break;
+    }
+
+    // If another task does exist, find it and wake it
+    if (size > 1) {
+        for (const auto& [key, task] : *handle) {
+            if (key != vbid) {
+                if (task->getState() == TASK_SNOOZED) {
+                    ExecutorPool::get()->wake(task->getId());
+                    // wake one other task
+                    break;
+                }
+            }
         }
     }
+    return reschedule;
 }
 
 std::pair<uint64_t, bool> EPBucket::getLastPersistedCheckpointId(Vbid vb) {

engines/ep/src/ep_bucket.h

@@ -28,6 +28,7 @@ class Commit;
 }
 enum class ValueFilter;
 class BucketStatCollector;
+class CompactTask;
 struct CompactionContext;
 
 /**
@@ -113,9 +114,24 @@ class EPBucket : public KVBucket {
     /// Stops the background fetcher for each shard.
     void stopBgFetcher();
 
+    /**
+     * Schedule compaction with a config -override of KVBucket method
+     */
+    ENGINE_ERROR_CODE scheduleCompaction(
+            Vbid vbid,
+            const CompactionConfig& c,
+            const void* ck,
+            std::chrono::milliseconds delay) override;
+
+    /**
+     * Schedule compaction with no config. If a CompactTask is already
+     * scheduled then the task will still run, but with whatever config it
+     * already has. If a task is already scheduled, the given delay parameter
+     * takes effect.
+     */
     ENGINE_ERROR_CODE scheduleCompaction(Vbid vbid,
-                                         const CompactionConfig& c,
-                                         const void* ck) override;
+                                         const void* cookie,
+                                         std::chrono::milliseconds delay);
 
     ENGINE_ERROR_CODE cancelCompaction(Vbid vbid) override;
 
@@ -133,6 +149,15 @@ class EPBucket : public KVBucket {
                    CompactionConfig& config,
                    const void* cookie);
 
+    /**
+     * After compaction completes the task can be removed if no further
+     * compaction is required. If other compaction tasks exist, one of them
+     * will be 'poked' to run. This method is called from CompactTask
+     *
+     * @param vbid id of vbucket that has completed compaction
+     */
+    bool updateCompactionTasks(Vbid vbid, bool canErase);
+
     std::pair<uint64_t, bool> getLastPersistedCheckpointId(Vbid vb) override;
 
     ENGINE_ERROR_CODE getFileStats(
@@ -266,13 +291,6 @@ class EPBucket : public KVBucket {
      */
     void compactionCompletionCallback(CompactionContext& ctx);
 
-    /**
-     * Remove completed compaction tasks or wake snoozed tasks
-     *
-     * @param db_file_id vbucket id for couchstore
-     */
-    void updateCompactionTasks(Vbid db_file_id);
-
     void stopWarmup();
 
     /// function which is passed down to compactor for dropping keys
@@ -321,6 +339,11 @@ class EPBucket : public KVBucket {
      */
     void initializeShards();
 
+    ENGINE_ERROR_CODE scheduleCompaction(Vbid vbid,
+                                         std::optional<CompactionConfig> config,
+                                         const void* cookie,
+                                         std::chrono::milliseconds delay);
+
     /**
      * Max number of backill items in a single flusher batch before we split
      * into multiple batches. Actual batch size may be larger as we will not
@@ -340,6 +363,9 @@ class EPBucket : public KVBucket {
     std::unique_ptr<Warmup> warmupTask;
 
     std::vector<std::unique_ptr<BgFetcher>> bgFetchers;
+
+    folly::Synchronized<std::unordered_map<Vbid, std::shared_ptr<CompactTask>>>
+            compactionTasks;
 };
 
 std::ostream& operator<<(std::ostream& os, const EPBucket::FlushResult& res);

engines/ep/src/ep_engine.cc

@@ -6368,7 +6368,8 @@ ENGINE_ERROR_CODE EventuallyPersistentEngine::deleteVBucket(
 
 ENGINE_ERROR_CODE EventuallyPersistentEngine::compactDB(
         Vbid vbid, const CompactionConfig& c, const void* cookie) {
-    return kvBucket->scheduleCompaction(vbid, c, cookie);
+    return kvBucket->scheduleCompaction(
+            vbid, c, cookie, std::chrono::seconds(0));
 }
 
 ENGINE_ERROR_CODE EventuallyPersistentEngine::getAllVBucketSequenceNumbers(

engines/ep/src/ephemeral_bucket.cc

@@ -168,9 +168,11 @@ void EphemeralBucket::attemptToFreeMemory() {
     }
 }
 
-ENGINE_ERROR_CODE EphemeralBucket::scheduleCompaction(Vbid vbid,
-                                                      const CompactionConfig& c,
-                                                      const void* ck) {
+ENGINE_ERROR_CODE EphemeralBucket::scheduleCompaction(
+        Vbid vbid,
+        const CompactionConfig& c,
+        const void* ck,
+        std::chrono::milliseconds delay) {
     return ENGINE_ENOTSUP;
 }
 

engines/ep/src/ephemeral_bucket.h

@@ -36,9 +36,11 @@ class EphemeralBucket : public KVBucket {
 
     bool initialize() override;
 
-    ENGINE_ERROR_CODE scheduleCompaction(Vbid vbid,
-                                         const CompactionConfig& c,
-                                         const void* ck) override;
+    ENGINE_ERROR_CODE scheduleCompaction(
+            Vbid vbid,
+            const CompactionConfig& c,
+            const void* ck,
+            std::chrono::milliseconds delay) override;
 
     ENGINE_ERROR_CODE cancelCompaction(Vbid vbid) override;
 

engines/ep/src/kv_bucket.h

@@ -105,8 +105,6 @@ class VBCBAdaptor : public GlobalTask {
 const uint16_t EP_PRIMARY_SHARD = 0;
 class KVShard;
 
-using CompTaskEntry = std::pair<Vbid, ExTask>;
-
 /**
  * KVBucket is the base class for concrete Key/Value bucket implementations
  * which use the concept of VBuckets to support replication, persistence and
@@ -848,9 +846,6 @@ class KVBucket : public KVBucketIface {
     std::atomic<size_t> lastTransTimePerItem;
     EvictionPolicy eviction_policy;
 
-    std::mutex compactionLock;
-    std::list<CompTaskEntry> compactionTasks;
-
     std::unique_ptr<Collections::Manager> collectionsManager;
 
     /**

engines/ep/src/kv_bucket_iface.h

@@ -448,10 +448,13 @@ class KVBucketIface {
      * @param vbid The vbucket to compact
      * @param c The context for compaction of a DB file
      * @param ck cookie used to notify connection of operation completion
+     * @param delay millisecond delay for the task to execute, 0 is run 'now'
      */
-    virtual ENGINE_ERROR_CODE scheduleCompaction(Vbid vbid,
-                                                 const CompactionConfig& c,
-                                                 const void* ck) = 0;
+    virtual ENGINE_ERROR_CODE scheduleCompaction(
+            Vbid vbid,
+            const CompactionConfig& c,
+            const void* ck,
+            std::chrono::milliseconds delay) = 0;
 
     /**
      * Cancels compaction of a database file

engines/ep/src/kvstore.cc

@@ -601,3 +601,10 @@ IORequest::~IORequest() = default;
 bool IORequest::isDelete() const {
     return item->isDeleted() && !item->isPending();
 }
+
+bool CompactionConfig::operator==(const CompactionConfig& other) const {
+    return purge_before_ts == other.purge_before_ts &&
+           purge_before_seq == other.purge_before_seq &&
+           drop_deletes == other.drop_deletes &&
+           retain_erroneous_tombstones == other.retain_erroneous_tombstones;
+}