Add free-list support to ArrayStore

* Add support for enabling freelists via `ArrayStoreConfig`. Currently defaults to false.
* Small array allocations from freelist reuse both entry ref and array buffer
* Large array allocations from freelist reuse entry ref only

6 files changed, 77 insertions, 21 deletions

diff --git a/vespalib/src/tests/datastore/array_store/array_store_test.cpp b/vespalib/src/tests/datastore/array_store/array_store_test.cpp
index d4e294dfd8d..e74a277d964 100644
--- a/vespalib/src/tests/datastore/array_store/array_store_test.cpp
+++ b/vespalib/src/tests/datastore/array_store/array_store_test.cpp
@@ -1,5 +1,6 @@
 // Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
 
+#include <vespa/vespalib/test/datastore/buffer_stats.h>
 #include <vespa/vespalib/test/datastore/memstats.h>
 #include <vespa/vespalib/datastore/array_store.hpp>
 #include <vespa/vespalib/testkit/testapp.h>
@@ -12,6 +13,7 @@ using vespalib::MemoryUsage;
 using vespalib::ArrayRef;
 using generation_t = vespalib::GenerationHandler::generation_t;
 using MemStats = search::datastore::test::MemStats;
+using BufferStats = search::datastore::test::BufferStats;
 
 constexpr float ALLOC_GROW_FACTOR = 0.2;
 
@@ -29,10 +31,10 @@ struct Fixture
     ArrayStoreType store;
     ReferenceStore refStore;
     generation_t generation;
-    Fixture(uint32_t maxSmallArraySize)
+    Fixture(uint32_t maxSmallArraySize, bool enable_free_lists = true)
         : store(ArrayStoreConfig(maxSmallArraySize,
                                  ArrayStoreConfig::AllocSpec(16, RefT::offsetSize(), 8 * 1024,
-                                                             ALLOC_GROW_FACTOR))),
+                                                             ALLOC_GROW_FACTOR)).enable_free_lists(enable_free_lists)),
           refStore(),
           generation(1)
     {}
@@ -66,11 +68,19 @@ struct Fixture
     uint32_t getBufferId(EntryRef ref) const {
         return EntryRefType(ref).bufferId();
     }
-    void assertBufferState(EntryRef ref, const MemStats expStats) const {
+    void assertBufferState(EntryRef ref, const MemStats& expStats) const {
         EXPECT_EQUAL(expStats._used, store.bufferState(ref).size());
         EXPECT_EQUAL(expStats._hold, store.bufferState(ref).getHoldElems());
         EXPECT_EQUAL(expStats._dead, store.bufferState(ref).getDeadElems());
     }
+    void assert_buffer_stats(EntryRef ref, const BufferStats& exp_stats) const {
+        auto& state = store.bufferState(ref);
+        EXPECT_EQUAL(exp_stats._used, state.size());
+        EXPECT_EQUAL(exp_stats._hold, state.getHoldElems());
+        EXPECT_EQUAL(exp_stats._dead, state.getDeadElems());
+        EXPECT_EQUAL(exp_stats._extra_used, state.getExtraUsedBytes());
+        EXPECT_EQUAL(exp_stats._extra_hold, state.getExtraHoldBytes());
+    }
     void assertMemoryUsage(const MemStats expStats) const {
         MemoryUsage act = store.getMemoryUsage();
         EXPECT_EQUAL(expStats._used, act.usedBytes());
@@ -82,6 +92,14 @@ struct Fixture
             TEST_DO(assertGet(elem.first, elem.second));
         }
     }
+    void assert_ref_reused(const EntryVector& first, const EntryVector& second, bool should_reuse) {
+        EntryRef ref1 = add(first);
+        remove(ref1);
+        trimHoldLists();
+        EntryRef ref2 = add(second);
+        EXPECT_EQUAL(should_reuse, (ref2 == ref1));
+        assertGet(ref2, second);
+    }
     EntryRef getEntryRef(const EntryVector &input) {
         for (auto itr = refStore.begin(); itr != refStore.end(); ++itr) {
             if (itr->second == input) {
@@ -166,12 +184,39 @@ TEST_F("require that elements are put on hold when a small array is removed", Nu
 TEST_F("require that elements are put on hold when a large array is removed", NumberFixture(3))
 {
     EntryRef ref = f.add({1,2,3,4});
-    // Note: The first buffer have the first element reserved -> we expect 2 elements used here.
+    // Note: The first buffer has the first element reserved -> we expect 2 elements used here.
     TEST_DO(f.assertBufferState(ref, MemStats().used(2).hold(0).dead(1)));
     f.store.remove(ref);
     TEST_DO(f.assertBufferState(ref, MemStats().used(2).hold(1).dead(1)));
 }
 
+TEST_F("small arrays are allocated from free-lists when enabled", NumberFixture(3, true)) {
+    f.assert_ref_reused({1,2,3}, {4,5,6}, true);
+}
+
+TEST_F("small arrays are NOT allocated from free-lists when disabled", NumberFixture(3, false)) {
+    f.assert_ref_reused({1,2,3}, {4,5,6}, false);
+}
+
+TEST_F("large arrays are allocated from free-lists when enabled", NumberFixture(3, true)) {
+    f.assert_ref_reused({1,2,3,4}, {5,6,7,8}, true);
+}
+
+TEST_F("large arrays are NOT allocated from free-lists when disabled", NumberFixture(3, false)) {
+    f.assert_ref_reused({1,2,3,4}, {5,6,7,8}, false);
+}
+
+TEST_F("track size of large array allocations with free-lists enabled", NumberFixture(3, true)) {
+    EntryRef ref = f.add({1,2,3,4});
+    TEST_DO(f.assert_buffer_stats(ref, BufferStats().used(2).hold(0).dead(1).extra_used(16)));
+    f.remove({1,2,3,4});
+    TEST_DO(f.assert_buffer_stats(ref, BufferStats().used(2).hold(1).dead(1).extra_hold(16).extra_used(16)));
+    f.trimHoldLists();
+    TEST_DO(f.assert_buffer_stats(ref, BufferStats().used(2).hold(0).dead(2).extra_used(0)));
+    f.add({5,6,7,8,9});
+    TEST_DO(f.assert_buffer_stats(ref, BufferStats().used(2).hold(0).dead(1).extra_used(20)));
+}
+
 TEST_F("require that new underlying buffer is allocated when current is full", SmallOffsetNumberFixture(3))
 {
     uint32_t firstBufferId = f.getBufferId(f.add({1,1}));
diff --git a/vespalib/src/vespa/vespalib/datastore/array_store.hpp b/vespalib/src/vespa/vespalib/datastore/array_store.hpp
index 524013652c5..56aa2d5e2c3 100644
--- a/vespalib/src/vespa/vespalib/datastore/array_store.hpp
+++ b/vespalib/src/vespa/vespalib/datastore/array_store.hpp
@@ -53,6 +53,9 @@ ArrayStore<EntryT, RefT>::ArrayStore(const ArrayStoreConfig &cfg)
 {
     initArrayTypes(cfg);
     _store.initActiveBuffers();
+    if (cfg.enable_free_lists()) {
+        _store.enableFreeLists();
+    }
 }
 
 template <typename EntryT, typename RefT>
@@ -81,23 +84,20 @@ EntryRef
 ArrayStore<EntryT, RefT>::addSmallArray(const ConstArrayRef &array)
 {
     uint32_t typeId = getTypeId(array.size());
-    return _store.template allocator<EntryT>(typeId).allocArray(array).ref;
+    using NoOpReclaimer = btree::DefaultReclaimer<EntryT>;
+    return _store.template freeListAllocator<EntryT, NoOpReclaimer>(typeId).allocArray(array).ref;
 }
 
 template <typename EntryT, typename RefT>
 EntryRef
 ArrayStore<EntryT, RefT>::addLargeArray(const ConstArrayRef &array)
 {
-    _store.ensureBufferCapacity(_largeArrayTypeId, 1);
-    uint32_t activeBufferId = _store.getActiveBufferId(_largeArrayTypeId);
-    BufferState &state = _store.getBufferState(activeBufferId);
-    assert(state.isActive());
-    size_t oldBufferSize = state.size();
-    RefT ref(oldBufferSize, activeBufferId);
-    LargeArray *buf = _store.template getEntry<LargeArray>(ref);
-    new (static_cast<void *>(buf)) LargeArray(array.cbegin(), array.cend());
-    state.pushed_back(1, sizeof(EntryT) * array.size());
-    return ref;
+    using NoOpReclaimer = btree::DefaultReclaimer<LargeArray>;
+    auto handle = _store.template freeListAllocator<LargeArray, NoOpReclaimer>(_largeArrayTypeId)
+            .alloc(array.cbegin(), array.cend());
+    auto& state = _store.getBufferState(RefT(handle.ref).bufferId());
+    state.incExtraUsedBytes(sizeof(EntryT) * array.size());
+    return handle.ref;
 }
 
 template <typename EntryT, typename RefT>
diff --git a/vespalib/src/vespa/vespalib/datastore/array_store_config.cpp b/vespalib/src/vespa/vespalib/datastore/array_store_config.cpp
index 0581183f675..ef3d40b98b2 100644
--- a/vespalib/src/vespa/vespalib/datastore/array_store_config.cpp
+++ b/vespalib/src/vespa/vespalib/datastore/array_store_config.cpp
@@ -6,7 +6,8 @@
 namespace search::datastore {
 
 ArrayStoreConfig::ArrayStoreConfig(size_t maxSmallArraySize, const AllocSpec &defaultSpec)
-    : _allocSpecs()
+    : _allocSpecs(),
+      _enable_free_lists(false)
 {
     for (size_t i = 0; i < (maxSmallArraySize + 1); ++i) {
         _allocSpecs.push_back(defaultSpec);
@@ -14,7 +15,8 @@ ArrayStoreConfig::ArrayStoreConfig(size_t maxSmallArraySize, const AllocSpec &de
 }
 
 ArrayStoreConfig::ArrayStoreConfig(const AllocSpecVector &allocSpecs)
-    : _allocSpecs(allocSpecs)
+    : _allocSpecs(allocSpecs),
+      _enable_free_lists(false)
 {
 }
 
diff --git a/vespalib/src/vespa/vespalib/datastore/array_store_config.h b/vespalib/src/vespa/vespalib/datastore/array_store_config.h
index a39c4454308..fba60f76b6c 100644
--- a/vespalib/src/vespa/vespalib/datastore/array_store_config.h
+++ b/vespalib/src/vespa/vespalib/datastore/array_store_config.h
@@ -39,6 +39,7 @@ public:
 
 private:
     AllocSpecVector _allocSpecs;
+    bool _enable_free_lists;
 
     /**
      * Setup an array store with arrays of size [1-(allocSpecs.size()-1)] allocated in buffers and
@@ -56,6 +57,15 @@ public:
 
     size_t maxSmallArraySize() const { return _allocSpecs.size() - 1; }
     const AllocSpec &specForSize(size_t arraySize) const;
+    ArrayStoreConfig& enable_free_lists(bool enable) & noexcept {
+        _enable_free_lists = enable;
+        return *this;
+    }
+    ArrayStoreConfig&& enable_free_lists(bool enable) && noexcept {
+        _enable_free_lists = enable;
+        return std::move(*this);
+    }
+    [[nodiscard]] bool enable_free_lists() const noexcept { return _enable_free_lists; }
 
     /**
      * Generate a config that is optimized for the given memory huge page size.
diff --git a/vespalib/src/vespa/vespalib/datastore/datastore.hpp b/vespalib/src/vespa/vespalib/datastore/datastore.hpp
index 797cd75cb08..40a00c68774 100644
--- a/vespalib/src/vespa/vespalib/datastore/datastore.hpp
+++ b/vespalib/src/vespa/vespalib/datastore/datastore.hpp
@@ -19,9 +19,7 @@ DataStoreT<RefT>::DataStoreT()
 }
 
 template <typename RefT>
-DataStoreT<RefT>::~DataStoreT()
-{
-}
+DataStoreT<RefT>::~DataStoreT() = default;
 
 template <typename RefT>
 void
@@ -30,7 +28,7 @@ DataStoreT<RefT>::freeElem(EntryRef ref, size_t numElems)
     RefType intRef(ref);
     BufferState &state = getBufferState(intRef.bufferId());
     if (state.isActive()) {
-        if (state.freeListList() != NULL && numElems == state.getArraySize()) {
+        if (state.freeListList() != nullptr && numElems == state.getArraySize()) {
             if (state.freeList().empty()) {
                 state.addToFreeListList();
             }
diff --git a/vespalib/src/vespa/vespalib/test/datastore/buffer_stats.h b/vespalib/src/vespa/vespalib/test/datastore/buffer_stats.h
index d39d8e43dcb..73d22ea63a4 100644
--- a/vespalib/src/vespa/vespalib/test/datastore/buffer_stats.h
+++ b/vespalib/src/vespa/vespalib/test/datastore/buffer_stats.h
@@ -3,6 +3,7 @@
 #pragma once
 
 #include <cassert>
+#include <cstddef>
 
 namespace search::datastore::test {