Use the generic hold list for entry refs in a datastore.

11 files changed, 152 insertions, 126 deletions

diff --git a/vespalib/src/tests/datastore/datastore/datastore_test.cpp b/vespalib/src/tests/datastore/datastore/datastore_test.cpp
index 2c4e68467da..1cb4f3e2307 100644
--- a/vespalib/src/tests/datastore/datastore/datastore_test.cpp
+++ b/vespalib/src/tests/datastore/datastore/datastore_test.cpp
@@ -31,8 +31,8 @@ public:
     void transferHoldLists(generation_t generation) {
         ParentType::transferHoldLists(generation);
     }
-    void trimElemHoldList(generation_t usedGen) override {
-        ParentType::trimElemHoldList(usedGen);
+    void reclaim_entry_refs(generation_t oldest_used_gen) override {
+        ParentType::reclaim_entry_refs(oldest_used_gen);
     }
     void ensureBufferCapacity(size_t sizeNeeded) {
         ParentType::ensureBufferCapacity(0, sizeNeeded);
@@ -314,11 +314,11 @@ TEST(DataStoreTest, require_that_we_can_hold_and_trim_elements)
     EXPECT_EQ(1, s.getEntry(r1));
     EXPECT_EQ(2, s.getEntry(r2));
     EXPECT_EQ(3, s.getEntry(r3));
-    s.trimElemHoldList(11);
+    s.reclaim_entry_refs(11);
     EXPECT_EQ(0, s.getEntry(r1));
     EXPECT_EQ(2, s.getEntry(r2));
     EXPECT_EQ(3, s.getEntry(r3));
-    s.trimElemHoldList(31);
+    s.reclaim_entry_refs(31);
     EXPECT_EQ(0, s.getEntry(r1));
     EXPECT_EQ(0, s.getEntry(r2));
     EXPECT_EQ(0, s.getEntry(r3));
@@ -363,12 +363,12 @@ TEST(DataStoreTest, require_that_we_can_use_free_lists)
     expect_successive_refs(r1, r2);
     s.holdElem(r2, 1);
     s.transferHoldLists(20);
-    s.trimElemHoldList(11);
+    s.reclaim_entry_refs(11);
     auto r3 = s.addEntry(3); // reuse r1
     EXPECT_EQ(r1, r3);
     auto r4 = s.addEntry(4);
     expect_successive_refs(r2, r4);
-    s.trimElemHoldList(21);
+    s.reclaim_entry_refs(21);
     auto r5 = s.addEntry(5); // reuse r2
     EXPECT_EQ(r2, r5);
     auto r6 = s.addEntry(6);
@@ -394,7 +394,7 @@ TEST(DataStoreTest, require_that_we_can_use_free_lists_with_raw_allocator)
     s.holdElem(h1.ref, 3);
     s.holdElem(h2.ref, 3);
     s.transferHoldLists(10);
-    s.trimElemHoldList(11);
+    s.reclaim_entry_refs(11);
 
     auto h3 = allocator.alloc(3); // reuse h2.ref from free list
     EXPECT_EQ(h2, h3);
diff --git a/vespalib/src/tests/util/generation_hold_list/generation_hold_list_test.cpp b/vespalib/src/tests/util/generation_hold_list/generation_hold_list_test.cpp
index 8305b711d5f..7e56e17990c 100644
--- a/vespalib/src/tests/util/generation_hold_list/generation_hold_list_test.cpp
+++ b/vespalib/src/tests/util/generation_hold_list/generation_hold_list_test.cpp
@@ -3,17 +3,16 @@
 #include <vespa/vespalib/gtest/gtest.h>
 #include <vespa/vespalib/util/generation_hold_list.hpp>
 #include <vespa/vespalib/util/generationholder.h>
-#include <iostream>
+#include <cstdint>
 
-using vespalib::GenerationHeldBase;
-using vespalib::GenerationHoldList;
+using namespace vespalib;
 
 using MyElem = GenerationHeldBase;
-using MyHoldList = GenerationHoldList<MyElem::UP, true, false>;
+using generation_t = GenerationHandler::generation_t;
 
-TEST(GenerationHoldListTest, holding_of_unique_ptr_elements_with_tracking_of_held_bytes)
+TEST(GenerationHolderTest, holding_of_unique_ptr_elements_with_tracking_of_held_bytes)
 {
-    MyHoldList h;
+    GenerationHolder h;
     h.insert(std::make_unique<MyElem>(3));
     h.assign_generation(0);
     h.insert(std::make_unique<MyElem>(5));
@@ -43,4 +42,54 @@ TEST(GenerationHoldListTest, holding_of_unique_ptr_elements_with_tracking_of_hel
     EXPECT_EQ(0, h.get_held_bytes());
 }
 
+TEST(GenerationHolderTest, reclaim_all_clears_everything)
+{
+    GenerationHolder h;
+    h.insert(std::make_unique<MyElem>(3));
+    h.insert(std::make_unique<MyElem>(5));
+    h.assign_generation(1);
+    h.reclaim_all();
+    EXPECT_EQ(0, h.get_held_bytes());
+}
+
+using IntVector = std::vector<int32_t>;
+using IntHoldList = GenerationHoldList<int32_t, false, true>;
+
+struct IntHoldListTest : public testing::Test {
+    IntHoldList h;
+    IntHoldListTest() : h() {}
+    void assert_reclaim(const IntVector& exp, generation_t oldest_used_gen) {
+        IntVector act;
+        h.reclaim(oldest_used_gen, [&](int elem){ act.push_back(elem); });
+        EXPECT_EQ(exp, act);
+    }
+    void assert_reclaim_all(const IntVector& exp) {
+        IntVector act;
+        h.reclaim_all([&](int elem){ act.push_back(elem); });
+        EXPECT_EQ(exp, act);
+    }
+};
+
+TEST_F(IntHoldListTest, reclaim_calls_callback_for_reclaimed_elements)
+{
+    h.insert(3);
+    h.assign_generation(1);
+    h.insert(5);
+    h.insert(7);
+    h.assign_generation(2);
+
+    assert_reclaim({}, 1);
+    assert_reclaim({3}, 2);
+    assert_reclaim({5, 7}, 3);
+}
+
+TEST_F(IntHoldListTest, reclaim_all_calls_callback_for_all_elements)
+{
+    h.insert(3);
+    h.insert(5);
+    h.assign_generation(2);
+    assert_reclaim_all({3, 5});
+    assert_reclaim_all({});
+}
+
 GTEST_MAIN_RUN_ALL_TESTS()
diff --git a/vespalib/src/vespa/vespalib/datastore/datastore.cpp b/vespalib/src/vespa/vespalib/datastore/datastore.cpp
index 15686a9f79d..76d622f3704 100644
--- a/vespalib/src/vespa/vespalib/datastore/datastore.cpp
+++ b/vespalib/src/vespa/vespalib/datastore/datastore.cpp
@@ -1,7 +1,6 @@
 // Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
 
 #include "datastore.hpp"
-#include <vespa/vespalib/util/array.hpp>
 #include <vespa/vespalib/util/rcuvector.hpp>
 
 namespace vespalib::datastore {
@@ -10,7 +9,6 @@ template class DataStoreT<EntryRefT<22> >;
 
 }
 
-template void vespalib::Array<vespalib::datastore::DataStoreBase::ElemHold1ListElem>::increase(size_t);
 template class vespalib::RcuVector<vespalib::datastore::EntryRef>;
 template class vespalib::RcuVectorBase<vespalib::datastore::EntryRef>;
 template class vespalib::RcuVector<vespalib::datastore::AtomicEntryRef>;
diff --git a/vespalib/src/vespa/vespalib/datastore/datastore.h b/vespalib/src/vespa/vespalib/datastore/datastore.h
index 6ff71238a89..25b451deae7 100644
--- a/vespalib/src/vespa/vespalib/datastore/datastore.h
+++ b/vespalib/src/vespa/vespalib/datastore/datastore.h
@@ -50,14 +50,9 @@ public:
     }
     void holdElem(EntryRef ref, size_t numElems, size_t extraBytes);
 
-    /**
-     * Trim elem hold list, freeing elements that no longer needs to be held.
-     *
-     * @param usedGen       lowest generation that is still used.
-     */
-    void trimElemHoldList(generation_t usedGen) override;
+    void reclaim_entry_refs(generation_t oldest_used_gen) override;
 
-    void clearElemHoldList() override;
+    void reclaim_all_entry_refs() override;
 
     bool getCompacting(EntryRef ref) const {
         return getBufferState(RefType(ref).bufferId()).getCompacting();
diff --git a/vespalib/src/vespa/vespalib/datastore/datastore.hpp b/vespalib/src/vespa/vespalib/datastore/datastore.hpp
index 18abe29d04f..190ef994c0b 100644
--- a/vespalib/src/vespa/vespalib/datastore/datastore.hpp
+++ b/vespalib/src/vespa/vespalib/datastore/datastore.hpp
@@ -7,7 +7,7 @@
 #include "free_list_allocator.hpp"
 #include "free_list_raw_allocator.hpp"
 #include "raw_allocator.hpp"
-#include <vespa/vespalib/util/array.hpp>
+#include <vespa/vespalib/util/generation_hold_list.hpp>
 
 namespace vespalib::datastore {
 
@@ -36,39 +36,26 @@ DataStoreT<RefT>::holdElem(EntryRef ref, size_t numElems, size_t extraBytes)
     RefType intRef(ref);
     BufferState &state = getBufferState(intRef.bufferId());
     if (!state.hold_elems(numElems, extraBytes)) {
-        _elemHold1List.push_back(ElemHold1ListElem(ref, numElems));
+        _entry_ref_hold_list.insert({ref, numElems});
     }
 }
 
 template <typename RefT>
 void
-DataStoreT<RefT>::trimElemHoldList(generation_t usedGen)
+DataStoreT<RefT>::reclaim_entry_refs(generation_t oldest_used_gen)
 {
-    auto it = _elemHold2List.begin();
-    auto ite = _elemHold2List.end();
-    uint32_t freed = 0;
-    for (; it != ite; ++it) {
-        if (static_cast<sgeneration_t>(it->_generation - usedGen) >= 0) {
-            break;
-        }
-        free_elem_internal(it->_ref, it->_len, true);
-        ++freed;
-    }
-    if (freed != 0) {
-        _elemHold2List.erase(_elemHold2List.begin(), it);
-    }
+    _entry_ref_hold_list.reclaim(oldest_used_gen, [this](const auto& elem) {
+        free_elem_internal(elem.ref, elem.num_elems, true);
+    });
 }
 
 template <typename RefT>
 void
-DataStoreT<RefT>::clearElemHoldList()
+DataStoreT<RefT>::reclaim_all_entry_refs()
 {
-    auto it = _elemHold2List.begin();
-    auto ite = _elemHold2List.end();
-    for (; it != ite; ++it) {
-        free_elem_internal(it->_ref, it->_len, true);
-    }
-    _elemHold2List.clear();
+    _entry_ref_hold_list.reclaim_all([this](const auto& elem) {
+        free_elem_internal(elem.ref, elem.num_elems, true);
+    });
 }
 
 template <typename RefT>
diff --git a/vespalib/src/vespa/vespalib/datastore/datastorebase.cpp b/vespalib/src/vespa/vespalib/datastore/datastorebase.cpp
index 5962e6cf2f7..4589fbba6fa 100644
--- a/vespalib/src/vespa/vespalib/datastore/datastorebase.cpp
+++ b/vespalib/src/vespa/vespalib/datastore/datastorebase.cpp
@@ -5,9 +5,8 @@
 #include "compacting_buffers.h"
 #include "compaction_spec.h"
 #include "compaction_strategy.h"
-#include <vespa/vespalib/util/array.hpp>
+#include <vespa/vespalib/util/generation_hold_list.hpp>
 #include <vespa/vespalib/util/stringfmt.h>
-#include <algorithm>
 #include <limits>
 #include <cassert>
 
@@ -16,6 +15,10 @@ LOG_SETUP(".vespalib.datastore.datastorebase");
 
 using vespalib::GenerationHeldBase;
 
+namespace vespalib {
+template class GenerationHoldList<datastore::DataStoreBase::EntryRefHoldElem, false, true>;
+}
+
 namespace vespalib::datastore {
 
 namespace {
@@ -88,8 +91,7 @@ DataStoreBase::DataStoreBase(uint32_t numBuffers, uint32_t offset_bits, size_t m
       _free_lists(),
       _freeListsEnabled(false),
       _initializing(false),
-      _elemHold1List(),
-      _elemHold2List(),
+      _entry_ref_hold_list(),
       _numBuffers(numBuffers),
       _offset_bits(offset_bits),
       _hold_buffer_count(0u),
@@ -102,9 +104,6 @@ DataStoreBase::DataStoreBase(uint32_t numBuffers, uint32_t offset_bits, size_t m
 DataStoreBase::~DataStoreBase()
 {
     disableFreeLists();
-
-    assert(_elemHold1List.empty());
-    assert(_elemHold2List.empty());
 }
 
 void
@@ -221,21 +220,10 @@ DataStoreBase::addType(BufferTypeBase *typeHandler)
 }
 
 void
-DataStoreBase::transferElemHoldList(generation_t generation)
-{
-    for (const auto& elemHold1 : _elemHold1List) {
-        _elemHold2List.push_back(ElemHold2ListElem(elemHold1, generation));
-    }
-    _elemHold1List.clear();
-}
-
-void
 DataStoreBase::transferHoldLists(generation_t generation)
 {
     _genHolder.assign_generation(generation);
-    if (hasElemHold1()) {
-        transferElemHoldList(generation);
-    }
+    _entry_ref_hold_list.assign_generation(generation);
 }
 
 void
@@ -249,15 +237,15 @@ DataStoreBase::doneHoldBuffer(uint32_t bufferId)
 void
 DataStoreBase::trimHoldLists(generation_t usedGen)
 {
-    trimElemHoldList(usedGen);  // Trim entries before trimming buffers
+    reclaim_entry_refs(usedGen);  // Trim entries before trimming buffers
     _genHolder.reclaim(usedGen);
 }
 
 void
 DataStoreBase::clearHoldLists()
 {
-    transferElemHoldList(0);
-    clearElemHoldList();
+    _entry_ref_hold_list.assign_generation(0);
+    reclaim_all_entry_refs();
     _genHolder.reclaim_all();
 }
 
diff --git a/vespalib/src/vespa/vespalib/datastore/datastorebase.h b/vespalib/src/vespa/vespalib/datastore/datastorebase.h
index 28c47207a9f..520c13742b5 100644
--- a/vespalib/src/vespa/vespalib/datastore/datastorebase.h
+++ b/vespalib/src/vespa/vespalib/datastore/datastorebase.h
@@ -7,6 +7,7 @@
 #include "memory_stats.h"
 #include <vespa/vespalib/util/address_space.h>
 #include <vespa/vespalib/util/generationholder.h>
+#include <vespa/vespalib/util/generation_hold_list.h>
 #include <vespa/vespalib/util/memoryusage.h>
 #include <atomic>
 #include <deque>
@@ -26,42 +27,18 @@ class CompactionStrategy;
 class DataStoreBase
 {
 protected:
-    /**
-     * Hold list element used in the first phase of holding (before freeze),
-     * before knowing how long elements must be held.
-     */
-    class ElemHold1ListElem
-    {
-    public:
-        EntryRef _ref;
-        size_t   _len;  // Aligned length
-
-        ElemHold1ListElem(EntryRef ref, size_t len)
-            : _ref(ref),
-              _len(len)
-        { }
+    struct EntryRefHoldElem {
+        EntryRef ref;
+        size_t   num_elems;
+
+        EntryRefHoldElem(EntryRef ref_in, size_t num_elems_in)
+            : ref(ref_in),
+              num_elems(num_elems_in)
+        {}
     };
 
+    using EntryRefHoldList = GenerationHoldList<EntryRefHoldElem, false, true>;
     using generation_t = vespalib::GenerationHandler::generation_t;
-    using sgeneration_t = vespalib::GenerationHandler::sgeneration_t;
-
-    /**
-     * Hold list element used in the second phase of holding (at freeze),
-     * when knowing how long elements must be held.
-     */
-    class ElemHold2ListElem : public ElemHold1ListElem
-    {
-    public:
-        generation_t _generation;
-
-        ElemHold2ListElem(const ElemHold1ListElem &hold1, generation_t generation)
-                : ElemHold1ListElem(hold1),
-                  _generation(generation)
-        { }
-    };
-
-    using ElemHold1List = vespalib::Array<ElemHold1ListElem>;
-    using ElemHold2List = std::deque<ElemHold2ListElem>;
 
 private:
     class BufferAndTypeId {
@@ -113,10 +90,7 @@ protected:
     std::vector<FreeList> _free_lists;
     bool _freeListsEnabled;
     bool _initializing;
-
-    ElemHold1List _elemHold1List;
-    ElemHold2List _elemHold2List;
-
+    EntryRefHoldList _entry_ref_hold_list;
     const uint32_t _numBuffers;
     const uint32_t _offset_bits;
     uint32_t       _hold_buffer_count;
@@ -153,11 +127,11 @@ protected:
     /**
      * Trim elem hold list, freeing elements that no longer needs to be held.
      *
-     * @param usedGen       lowest generation that is still used.
+     * @param oldest_used_gen the oldest generation that is still used.
      */
-    virtual void trimElemHoldList(generation_t usedGen) = 0;
+    virtual void reclaim_entry_refs(generation_t oldest_used_gen) = 0;
 
-    virtual void clearElemHoldList() = 0;
+    virtual void reclaim_all_entry_refs() = 0;
 
     void markCompacting(uint32_t bufferId);
 
@@ -213,14 +187,6 @@ public:
     BufferState &getBufferState(uint32_t bufferId) { return _states[bufferId]; }
     uint32_t getNumBuffers() const { return _numBuffers; }
 
-private:
-    bool hasElemHold1() const { return !_elemHold1List.empty(); }
-
-    /**
-     * Transfer element holds from hold1 list to hold2 list.
-     */
-    void transferElemHoldList(generation_t generation);
-
 public:
     /**
      * Transfer holds from hold1 to hold2 lists, assigning generation.
@@ -344,3 +310,7 @@ public:
 };
 
 }
+
+namespace vespalib {
+extern template class GenerationHoldList<datastore::DataStoreBase::EntryRefHoldElem, false, true>;
+}
diff --git a/vespalib/src/vespa/vespalib/util/generation_hold_list.h b/vespalib/src/vespa/vespalib/util/generation_hold_list.h
index b2f90934e84..bdb58afb504 100644
--- a/vespalib/src/vespa/vespalib/util/generation_hold_list.h
+++ b/vespalib/src/vespa/vespalib/util/generation_hold_list.h
@@ -26,9 +26,16 @@ private:
             : elem(std::move(elem_in)),
               gen(gen_in)
         {}
-        size_t byte_size() const { return elem->byte_size(); }
+        size_t byte_size() const {
+            if constexpr (track_bytes_held) {
+                return elem->byte_size();
+            }
+            return 0;
+        }
     };
 
+    struct NoopFunc { void operator()(const T&){} };
+
     using ElemList = std::vector<T>;
     using ElemWithGenList = std::conditional_t<use_deque,
             std::deque<ElemWithGen>,
@@ -43,8 +50,8 @@ private:
      */
     void assign_generation_internal(generation_t current_gen);
 
-    void reclaim_internal(generation_t oldest_used_gen);
-
+    template<typename Func>
+    void reclaim_internal(generation_t oldest_used_gen, Func callback);
 
 public:
     GenerationHoldList();
@@ -67,18 +74,31 @@ public:
 
     /**
      * Reclaim all data elements where the assigned generation < oldest used generation.
+     * The callback function is called for each data element reclaimed.
      **/
-    void reclaim(generation_t oldest_used_gen) {
+    template<typename Func>
+    void reclaim(generation_t oldest_used_gen, Func callback) {
         if (!_phase_2_list.empty() && (_phase_2_list.front().gen < oldest_used_gen)) {
-            reclaim_internal(oldest_used_gen);
+            reclaim_internal(oldest_used_gen, callback);
         }
     }
 
+    void reclaim(generation_t oldest_used_gen) {
+        reclaim(oldest_used_gen, NoopFunc());
+    }
+
     /**
      * Reclaim all data elements from this hold list.
      */
     void reclaim_all();
 
+    /**
+     * Reclaim all data elements from this hold list.
+     * The callback function is called for all data elements reclaimed.
+     */
+    template<typename Func>
+    void reclaim_all(Func callback);
+
     size_t get_held_bytes() const { return _held_bytes.load(std::memory_order_relaxed); }
 
 };
diff --git a/vespalib/src/vespa/vespalib/util/generation_hold_list.hpp b/vespalib/src/vespa/vespalib/util/generation_hold_list.hpp
index ed4a99c4753..4855d1c651d 100644
--- a/vespalib/src/vespa/vespalib/util/generation_hold_list.hpp
+++ b/vespalib/src/vespa/vespalib/util/generation_hold_list.hpp
@@ -18,8 +18,9 @@ GenerationHoldList<T, track_bytes_held, use_deque>::assign_generation_internal(g
 }
 
 template <typename T, bool track_bytes_held, bool use_deque>
+template <typename Func>
 void
-GenerationHoldList<T, track_bytes_held, use_deque>::reclaim_internal(generation_t oldest_used_gen)
+GenerationHoldList<T, track_bytes_held, use_deque>::reclaim_internal(generation_t oldest_used_gen, Func func)
 {
     auto itr = _phase_2_list.begin();
     auto ite = _phase_2_list.end();
@@ -27,7 +28,9 @@ GenerationHoldList<T, track_bytes_held, use_deque>::reclaim_internal(generation_
         if (itr->gen >= oldest_used_gen) {
             break;
         }
-        if (track_bytes_held) {
+        const auto& elem = itr->elem;
+        func(elem);
+        if constexpr (track_bytes_held) {
             _held_bytes.store(get_held_bytes() - itr->byte_size(), std::memory_order_relaxed);
         }
     }
@@ -57,7 +60,7 @@ void
 GenerationHoldList<T, track_bytes_held, use_deque>::insert(T data)
 {
     _phase_1_list.push_back(std::move(data));
-    if (track_bytes_held) {
+    if constexpr (track_bytes_held) {
         _held_bytes.store(get_held_bytes() + _phase_1_list.back()->byte_size(), std::memory_order_relaxed);
     }
 }
@@ -71,4 +74,15 @@ GenerationHoldList<T, track_bytes_held, use_deque>::reclaim_all()
     _held_bytes = 0;
 }
 
+template <typename T, bool track_bytes_held, bool use_deque>
+template <typename Func>
+void
+GenerationHoldList<T, track_bytes_held, use_deque>::reclaim_all(Func func)
+{
+    for (const auto& elem_with_gen : _phase_2_list) {
+        func(elem_with_gen.elem);
+    }
+    reclaim_all();
+}
+
 }
diff --git a/vespalib/src/vespa/vespalib/util/generationholder.cpp b/vespalib/src/vespa/vespalib/util/generationholder.cpp
index 5f31f610c68..07bde82f007 100644
--- a/vespalib/src/vespa/vespalib/util/generationholder.cpp
+++ b/vespalib/src/vespa/vespalib/util/generationholder.cpp
@@ -5,10 +5,15 @@
 
 namespace vespalib {
 
+template class GenerationHoldList<GenerationHeldBase::UP, true, false>;
+
+template void GenerationHolderParent::reclaim_internal
+        <GenerationHolderParent::NoopFunc>(generation_t oldest_used_gen, NoopFunc func);
+
 GenerationHeldBase::~GenerationHeldBase() = default;
 
 GenerationHolder::GenerationHolder()
-    : GenerationHoldList<GenerationHeldBase::UP, true, false>()
+    : GenerationHolderParent()
 {
 }
 
diff --git a/vespalib/src/vespa/vespalib/util/generationholder.h b/vespalib/src/vespa/vespalib/util/generationholder.h
index ec2cded2e84..86d402f9b3b 100644
--- a/vespalib/src/vespa/vespalib/util/generationholder.h
+++ b/vespalib/src/vespa/vespalib/util/generationholder.h
@@ -27,13 +27,13 @@ public:
     size_t byte_size() const { return _byte_size; }
 };
 
-template class GenerationHoldList<GenerationHeldBase::UP, true, false>;
+using GenerationHolderParent = GenerationHoldList<GenerationHeldBase::UP, true, false>;
 
 /*
  * GenerationHolder is meant to hold large elements until readers can
  * no longer access them.
  */
-class GenerationHolder : public GenerationHoldList<GenerationHeldBase::UP, true, false> {
+class GenerationHolder : public GenerationHolderParent {
 public:
     GenerationHolder();
 };