Merge pull request #17221 from vespa-engine/toregge/control-document-meta-store-compaction-pace

Control document meta store compaction pace.

5 files changed, 57 insertions, 8 deletions

diff --git a/searchcore/src/tests/proton/documentmetastore/documentmetastore_test.cpp b/searchcore/src/tests/proton/documentmetastore/documentmetastore_test.cpp
index 84dadff0104..5dd6f0ff10f 100644
--- a/searchcore/src/tests/proton/documentmetastore/documentmetastore_test.cpp
+++ b/searchcore/src/tests/proton/documentmetastore/documentmetastore_test.cpp
@@ -2091,6 +2091,16 @@ TEST(DocumentMetaStoreTest, call_to_remove_is_notified)
     EXPECT_EQ(1, listener->remove_cnt);
 }
 
+namespace {
+
+void try_compact_document_meta_store(DocumentMetaStore &dms)
+{
+    dms.removeAllOldGenerations();
+    dms.commit(true);
+}
+
+}
+
 TEST(DocumentMetaStoreTest, gid_to_lid_map_can_be_compacted)
 {
     auto dms = std::make_shared<DocumentMetaStore>(createBucketDB());
@@ -2106,17 +2116,35 @@ TEST(DocumentMetaStoreTest, gid_to_lid_map_can_be_compacted)
     auto status_before = dms->getStatus();
     EXPECT_LT(0, status_before.getOnHold());
     guard = AttributeGuard();
-    dms->removeAllOldGenerations();
-    dms->commit(true);
-    auto status_middle = dms->getStatus();
-    EXPECT_LT(status_before.getDead(), status_middle.getDead());
-    EXPECT_EQ(0, status_middle.getOnHold());
-    for (uint32_t i = 0; i < 15; ++i) {
-        dms->commit(true);
+    try_compact_document_meta_store(*dms);
+    auto status_early = dms->getStatus();
+    EXPECT_LT(status_before.getDead(), status_early.getDead());
+    EXPECT_EQ(0, status_early.getOnHold());
+    bool compaction_done = false;
+    for (uint32_t i = 0; i < 15 && !compaction_done; ++i) {
+        AttributeGuard guard2(dms);
+        auto status_loop_iteration_start = dms->getStatus();
+        try_compact_document_meta_store(*dms);
+        try_compact_document_meta_store(*dms);
+        auto status_second = dms->getStatus();
+        if (i > 0) {
+            EXPECT_GT(status_before.getUsed(), status_second.getUsed());
+        }
+        EXPECT_GT(status_early.getDead(), status_second.getDead());
+        try_compact_document_meta_store(*dms);
+        auto status_third = dms->getStatus();
+        EXPECT_EQ(status_second.getDead(), status_third.getDead());
+        EXPECT_EQ(status_second.getUsed(), status_third.getUsed());
+        EXPECT_EQ(status_second.getOnHold(), status_third.getOnHold());
+        EXPECT_GE(status_loop_iteration_start.getDead(), status_third.getDead());
+        if (status_loop_iteration_start.getDead() == status_third.getDead()) {
+            compaction_done = true;
+        }
     }
+    EXPECT_TRUE(compaction_done);
     auto status_after = dms->getStatus();
     EXPECT_GT(status_before.getUsed(), status_after.getUsed());
-    EXPECT_GT(status_middle.getDead(), status_after.getDead());
+    EXPECT_GT(status_early.getDead(), status_after.getDead());
     EXPECT_EQ(0, status_after.getOnHold());
 }
 
diff --git a/searchcore/src/vespa/searchcore/proton/documentmetastore/documentmetastore.cpp b/searchcore/src/vespa/searchcore/proton/documentmetastore/documentmetastore.cpp
index 25c0e29fbda..d0cada2224f 100644
--- a/searchcore/src/vespa/searchcore/proton/documentmetastore/documentmetastore.cpp
+++ b/searchcore/src/vespa/searchcore/proton/documentmetastore/documentmetastore.cpp
@@ -209,6 +209,9 @@ constexpr size_t DEAD_BYTES_SLACK = 0x10000u;
 bool
 DocumentMetaStore::consider_compact_gid_to_lid_map()
 {
+    if (_gidToLidMap.getAllocator().getNodeStore().has_held_buffers()) {
+        return false;
+    }
     auto &compaction_strategy = getConfig().getCompactionStrategy();
     size_t used_bytes = _cached_gid_to_lid_map_memory_usage.usedBytes();
     size_t dead_bytes = _cached_gid_to_lid_map_memory_usage.deadBytes();
diff --git a/vespalib/src/vespa/vespalib/btree/btreenodestore.h b/vespalib/src/vespa/vespalib/btree/btreenodestore.h
index 0b273b54c3a..81017447580 100644
--- a/vespalib/src/vespa/vespalib/btree/btreenodestore.h
+++ b/vespalib/src/vespa/vespalib/btree/btreenodestore.h
@@ -191,6 +191,10 @@ public:
         return _store.getCompacting(ref);
     }
 
+    bool has_held_buffers() const {
+        return _store.has_held_buffers();
+    }
+    
     template <typename FunctionType>
     void foreach_key(EntryRef ref, FunctionType func) const {
         if (!ref.valid())
diff --git a/vespalib/src/vespa/vespalib/datastore/datastorebase.cpp b/vespalib/src/vespa/vespalib/datastore/datastorebase.cpp
index 2ec8cb3bda3..cce0b0695c2 100644
--- a/vespalib/src/vespa/vespalib/datastore/datastorebase.cpp
+++ b/vespalib/src/vespa/vespalib/datastore/datastorebase.cpp
@@ -66,6 +66,7 @@ public:
           _dsb(dsb),
           _bufferId(bufferId)
     {
+        _dsb.inc_hold_buffer_count();
     }
 
     ~BufferHold() override
@@ -85,6 +86,7 @@ DataStoreBase::DataStoreBase(uint32_t numBuffers, size_t maxArrays)
       _elemHold1List(),
       _elemHold2List(),
       _numBuffers(numBuffers),
+      _hold_buffer_count(0u),
       _maxArrays(maxArrays),
       _compaction_count(0u),
       _genHolder()
@@ -187,6 +189,8 @@ DataStoreBase::transferHoldLists(generation_t generation)
 void
 DataStoreBase::doneHoldBuffer(uint32_t bufferId)
 {
+    assert(_hold_buffer_count > 0);
+    --_hold_buffer_count;
     _states[bufferId].onFree(_buffers[bufferId].getBuffer());
 }
 
@@ -515,5 +519,12 @@ DataStoreBase::startCompactWorstBuffers(bool compactMemory, bool compactAddressS
     return result;
 }
 
+void
+DataStoreBase::inc_hold_buffer_count()
+{
+    assert(_hold_buffer_count < std::numeric_limits<uint32_t>::max());
+    ++_hold_buffer_count;
+}
+
 }
 
diff --git a/vespalib/src/vespa/vespalib/datastore/datastorebase.h b/vespalib/src/vespa/vespalib/datastore/datastorebase.h
index 2617973d239..13c242e85a7 100644
--- a/vespalib/src/vespa/vespalib/datastore/datastorebase.h
+++ b/vespalib/src/vespa/vespalib/datastore/datastorebase.h
@@ -156,6 +156,7 @@ protected:
     ElemHold2List _elemHold2List;
 
     const uint32_t _numBuffers;
+    uint32_t       _hold_buffer_count;
     const size_t   _maxArrays;
     mutable std::atomic<uint64_t> _compaction_count;
 
@@ -350,6 +351,7 @@ private:
      */
     void onActive(uint32_t bufferId, uint32_t typeId, size_t elemsNeeded);
 
+    void inc_hold_buffer_count();
 public:
     uint32_t getTypeId(uint32_t bufferId) const {
         return _buffers[bufferId].getTypeId();
@@ -368,6 +370,7 @@ public:
     std::vector<uint32_t> startCompactWorstBuffers(bool compactMemory, bool compactAddressSpace);
     uint64_t get_compaction_count() const { return _compaction_count.load(std::memory_order_relaxed); }
     void inc_compaction_count() const { ++_compaction_count; }
+    bool has_held_buffers() const noexcept { return _hold_buffer_count != 0u; }
 };
 
 }