Merge pull request #16725 from vespa-engine/geirst/avoid-memory-accumulation-in-array-attribute-vectors

Avoid memory accumulation in array attribute vectors

7 files changed, 209 insertions, 57 deletions

diff --git a/searchlib/src/tests/attribute/compaction/attribute_compaction_test.cpp b/searchlib/src/tests/attribute/compaction/attribute_compaction_test.cpp
index 36a697eaa12..c3bb9481f0f 100644
--- a/searchlib/src/tests/attribute/compaction/attribute_compaction_test.cpp
+++ b/searchlib/src/tests/attribute/compaction/attribute_compaction_test.cpp
@@ -1,10 +1,11 @@
 // Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
-#include <vespa/vespalib/testkit/test_kit.h>
+#include <vespa/searchlib/attribute/address_space_usage.h>
 #include <vespa/searchlib/attribute/attribute.h>
 #include <vespa/searchlib/attribute/attributefactory.h>
 #include <vespa/searchlib/attribute/attributeguard.h>
 #include <vespa/searchlib/attribute/integerbase.h>
-#include <vespa/searchlib/attribute/address_space_usage.h>
+#include <vespa/vespalib/testkit/test_kit.h>
+#include <vespa/vespalib/util/stringfmt.h>
 
 #include <vespa/log/log.h>
 LOG_SETUP("attribute_compaction_test");
@@ -126,6 +127,12 @@ Config compactAddressSpaceAttributeConfig(bool enableAddressSpaceCompact)
 
 }
 
+double
+calc_alloc_waste(const AttributeStatus& status)
+{
+    return ((double)(status.getAllocated() - status.getUsed())) / status.getAllocated();
+}
+
 class Fixture {
 public:
     AttributePtr _v;
@@ -141,8 +148,9 @@ public:
     AttributeStatus getStatus() { _v->commit(true); return _v->getStatus(); }
     AttributeStatus getStatus(const vespalib::string &prefix) {
         AttributeStatus status(getStatus());
-        LOG(info, "status %s: used=%" PRIu64 ", dead=%" PRIu64 ", onHold=%" PRIu64,
-            prefix.c_str(), status.getUsed(), status.getDead(), status.getOnHold());
+        LOG(info, "status %s: allocated=%" PRIu64 ", used=%" PRIu64 ", dead=%" PRIu64 ", onHold=%" PRIu64 ", waste=%f",
+            prefix.c_str(), status.getAllocated(), status.getUsed(), status.getDead(), status.getOnHold(),
+            calc_alloc_waste(status));
         return status;
     }
     const Config &getConfig() const { return _v->getConfig(); }
@@ -167,6 +175,23 @@ TEST_F("Test that compaction of integer array attribute reduces memory usage", F
     EXPECT_LESS(afterStatus.getUsed(), beforeStatus.getUsed());
 }
 
+TEST_F("Allocated memory is not accumulated in an array attribute when moving between value classes when compaction is active",
+       Fixture({BasicType::INT64, CollectionType::ARRAY}))
+{
+    DocIdRange range = f.addDocs(1000);
+    for (uint32_t i = 0; i < 50; ++i) {
+        uint32_t values = 10 + i;
+        // When moving all documents from one value class to the next,
+        // all elements in the buffers of the previous value class are marked dead.
+        // Those buffers will eventually be compacted. By taking the dead elements into account when
+        // calculating how large the resulting compacted buffer should be,
+        // we don't accumulate allocated memory as part of that process.
+        f.populate(range, values);
+        auto status = f.getStatus(vespalib::make_string("values=%u", values));
+        EXPECT_LESS(calc_alloc_waste(status), 0.68);
+    }
+}
+
 void
 populate_and_hammer(Fixture& f, bool take_attribute_guard)
 {
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 a1e7140fefe..417d8b80d87 100644
--- a/vespalib/src/tests/datastore/array_store/array_store_test.cpp
+++ b/vespalib/src/tests/datastore/array_store/array_store_test.cpp
@@ -150,13 +150,13 @@ TEST("require that we test with trivial and non-trivial types")
 
 TEST_F("control static sizes", NumberFixture(3)) {
 #ifdef _LIBCPP_VERSION
-    EXPECT_EQUAL(392u, sizeof(f.store));
+    EXPECT_EQUAL(400u, sizeof(f.store));
     EXPECT_EQUAL(296u, sizeof(NumberFixture::ArrayStoreType::DataStoreType));
 #else
-    EXPECT_EQUAL(424u, sizeof(f.store));
+    EXPECT_EQUAL(432u, sizeof(f.store));
     EXPECT_EQUAL(328u, sizeof(NumberFixture::ArrayStoreType::DataStoreType));
 #endif
-    EXPECT_EQUAL(64u, sizeof(NumberFixture::ArrayStoreType::SmallArrayType));
+    EXPECT_EQUAL(72u, sizeof(NumberFixture::ArrayStoreType::SmallArrayType));
     MemoryUsage usage = f.store.getMemoryUsage();
     EXPECT_EQUAL(960u, usage.allocatedBytes());
     EXPECT_EQUAL(32u, usage.usedBytes());
diff --git a/vespalib/src/tests/datastore/buffer_type/buffer_type_test.cpp b/vespalib/src/tests/datastore/buffer_type/buffer_type_test.cpp
index 96d94e9a71b..d647a659eb6 100644
--- a/vespalib/src/tests/datastore/buffer_type/buffer_type_test.cpp
+++ b/vespalib/src/tests/datastore/buffer_type/buffer_type_test.cpp
@@ -14,6 +14,7 @@ struct Setup {
     uint32_t  _minArrays;
     ElemCount _usedElems;
     ElemCount _neededElems;
+    ElemCount _deadElems;
     uint32_t  _bufferId;
     float     _allocGrowFactor;
     bool      _resizing;
@@ -21,6 +22,7 @@ struct Setup {
         : _minArrays(0),
           _usedElems(0),
           _neededElems(0),
+          _deadElems(0),
           _bufferId(1),
           _allocGrowFactor(0.5),
           _resizing(false)
@@ -28,30 +30,46 @@ struct Setup {
     Setup &minArrays(uint32_t value) { _minArrays = value; return *this; }
     Setup &used(size_t value) { _usedElems = value; return *this; }
     Setup &needed(size_t value) { _neededElems = value; return *this; }
+    Setup &dead(size_t value) { _deadElems = value; return *this; }
     Setup &bufferId(uint32_t value) { _bufferId = value; return *this; }
     Setup &resizing(bool value) { _resizing = value; return *this; }
 };
 
 struct Fixture {
-    Setup         setup;
+    std::vector<Setup> setups;
     IntBufferType bufferType;
-    ElemCount     deadElems;
     int buffer[ARRAYS_SIZE];
     Fixture(const Setup &setup_)
-        : setup(setup_),
-          bufferType(ARRAYS_SIZE, setup._minArrays, MAX_ARRAYS, NUM_ARRAYS_FOR_NEW_BUFFER, setup._allocGrowFactor),
-          deadElems(0),
+        : setups(),
+          bufferType(ARRAYS_SIZE, setup_._minArrays, MAX_ARRAYS, NUM_ARRAYS_FOR_NEW_BUFFER, setup_._allocGrowFactor),
           buffer()
-    {}
+    {
+        setups.reserve(4);
+        setups.push_back(setup_);
+    }
     ~Fixture() {
-        bufferType.onHold(&setup._usedElems);
-        bufferType.onFree(setup._usedElems);
+        for (auto& setup : setups) {
+            bufferType.onHold(&setup._usedElems, &setup._deadElems);
+            bufferType.onFree(setup._usedElems);
+        }
+    }
+    Setup& curr_setup() {
+        return setups.back();
+    }
+    void add_setup(const Setup& setup_in) {
+        // The buffer type stores pointers to ElemCount (from Setup) and we must ensure these do not move in memory.
+        assert(setups.size() < setups.capacity());
+        setups.push_back(setup_in);
     }
     void onActive() {
-        bufferType.onActive(setup._bufferId, &setup._usedElems, deadElems, &buffer[0]);
+        bufferType.onActive(curr_setup()._bufferId, &curr_setup()._usedElems, &curr_setup()._deadElems, &buffer[0]);
     }
     size_t arraysToAlloc() {
-        return bufferType.calcArraysToAlloc(setup._bufferId, setup._neededElems, setup._resizing);
+        return bufferType.calcArraysToAlloc(curr_setup()._bufferId, curr_setup()._neededElems, curr_setup()._resizing);
+    }
+    void assertArraysToAlloc(size_t exp) {
+        onActive();
+        EXPECT_EQUAL(exp, arraysToAlloc());
     }
 };
 
@@ -59,8 +77,7 @@ void
 assertArraysToAlloc(size_t exp, const Setup &setup)
 {
     Fixture f(setup);
-    f.onActive();
-    EXPECT_EQUAL(exp, f.arraysToAlloc());
+    f.assertArraysToAlloc(exp);
 }
 
 TEST("require that complete arrays are allocated")
@@ -113,4 +130,40 @@ TEST("require that arrays to alloc is capped to min arrays")
     TEST_DO(assertArraysToAlloc(17, Setup().used(34 * 4).needed(4).minArrays(16)));
 }
 
+TEST("arrays to alloc considers used elements across all active buffers (no resizing)")
+{
+    Fixture f(Setup().used(6 * 4));
+    f.assertArraysToAlloc(6 * 0.5);
+    f.add_setup(Setup().used(8 * 4));
+    f.assertArraysToAlloc((6 + 8) * 0.5);
+    f.add_setup(Setup().used(10 * 4));
+    f.assertArraysToAlloc((6 + 8 + 10) * 0.5);
+}
+
+TEST("arrays to alloc only considers used elements in current buffer when resizing")
+{
+    Fixture f(Setup().used(6 * 4));
+    f.assertArraysToAlloc(6 * 0.5);
+    f.add_setup(Setup().used(8 * 4).resizing(true));
+    f.assertArraysToAlloc(8 + 8 * 0.5);
+}
+
+TEST("arrays to alloc considers (and subtracts) dead elements across all active buffers (no resizing)")
+{
+    Fixture f(Setup().used(6 * 4).dead(2 * 4));
+    f.assertArraysToAlloc((6 - 2) * 0.5);
+    f.add_setup(Setup().used(12 * 4).dead(4 * 4));
+    f.assertArraysToAlloc((6 - 2 + 12 - 4) * 0.5);
+    f.add_setup(Setup().used(20 * 4).dead(6 * 4));
+    f.assertArraysToAlloc((6 - 2 + 12 - 4 + 20 - 6) * 0.5);
+}
+
+TEST("arrays to alloc only considers (and subtracts) dead elements in current buffer when resizing")
+{
+    Fixture f(Setup().used(6 * 4).dead(2 * 4));
+    f.assertArraysToAlloc((6 - 2) * 0.5);
+    f.add_setup(Setup().used(12 * 4).dead(4 * 4).resizing(true));
+    f.assertArraysToAlloc(12 + (12 - 4) * 0.5);
+}
+
 TEST_MAIN() { TEST_RUN_ALL(); }
diff --git a/vespalib/src/tests/datastore/datastore/datastore_test.cpp b/vespalib/src/tests/datastore/datastore/datastore_test.cpp
index f3f43fb575b..11b4f5e6631 100644
--- a/vespalib/src/tests/datastore/datastore/datastore_test.cpp
+++ b/vespalib/src/tests/datastore/datastore/datastore_test.cpp
@@ -635,7 +635,7 @@ TEST(DataStoreTest, can_set_memory_allocator)
 }
 
 TEST(DataStoreTest, control_static_sizes) {
-    EXPECT_EQ(64, sizeof(BufferTypeBase));
+    EXPECT_EQ(72, sizeof(BufferTypeBase));
     EXPECT_EQ(32, sizeof(BufferState::FreeList));
     EXPECT_EQ(1, sizeof(BufferState::State));
     EXPECT_EQ(144, sizeof(BufferState));
diff --git a/vespalib/src/vespa/vespalib/datastore/buffer_type.cpp b/vespalib/src/vespa/vespalib/datastore/buffer_type.cpp
index f25ef2c1e52..22e44fe665d 100644
--- a/vespalib/src/vespa/vespalib/datastore/buffer_type.cpp
+++ b/vespalib/src/vespa/vespalib/datastore/buffer_type.cpp
@@ -33,9 +33,8 @@ BufferTypeBase::BufferTypeBase(uint32_t arraySize,
       _allocGrowFactor(allocGrowFactor),
       _activeBuffers(0),
       _holdBuffers(0),
-      _activeUsedElems(0),
       _holdUsedElems(0),
-      _lastUsedElems(nullptr)
+      _aggr_counts()
 {
 }
 
@@ -50,9 +49,8 @@ BufferTypeBase::~BufferTypeBase()
 {
     assert(_activeBuffers == 0);
     assert(_holdBuffers == 0);
-    assert(_activeUsedElems == 0);
     assert(_holdUsedElems == 0);
-    assert(_lastUsedElems == nullptr);
+    assert(_aggr_counts.empty());
 }
 
 ElemCount
@@ -62,38 +60,24 @@ BufferTypeBase::getReservedElements(uint32_t bufferId) const
 }
 
 void
-BufferTypeBase::flushLastUsed()
+BufferTypeBase::onActive(uint32_t bufferId, ElemCount* usedElems, ElemCount* deadElems, void* buffer)
 {
-    if (_lastUsedElems != nullptr) {
-        _activeUsedElems += *_lastUsedElems;
-        _lastUsedElems = nullptr;
-    }
-}
-
-void
-BufferTypeBase::onActive(uint32_t bufferId, ElemCount *usedElems, ElemCount &deadElems, void *buffer)
-{
-    flushLastUsed();
     ++_activeBuffers;
-    _lastUsedElems = usedElems;
+    _aggr_counts.add_buffer(usedElems, deadElems);
     size_t reservedElems = getReservedElements(bufferId);
     if (reservedElems != 0u) {
         initializeReservedElements(buffer, reservedElems);
         *usedElems = reservedElems;
-        deadElems = reservedElems;
+        *deadElems = reservedElems;
     }
 }
 
 void
-BufferTypeBase::onHold(const ElemCount *usedElems)
+BufferTypeBase::onHold(const ElemCount* usedElems, const ElemCount* deadElems)
 {
-    if (usedElems == _lastUsedElems) {
-        flushLastUsed();
-    }
     --_activeBuffers;
     ++_holdBuffers;
-    assert(_activeUsedElems >= *usedElems);
-    _activeUsedElems -= *usedElems;
+    _aggr_counts.remove_buffer(usedElems, deadElems);
     _holdUsedElems += *usedElems;
 }
 
@@ -123,16 +107,25 @@ size_t
 BufferTypeBase::calcArraysToAlloc(uint32_t bufferId, ElemCount elemsNeeded, bool resizing) const
 {
     size_t reservedElems = getReservedElements(bufferId);
-    size_t usedElems = (resizing ? 0 : _activeUsedElems);
-    if (_lastUsedElems != nullptr) {
-        usedElems += *_lastUsedElems;
+    BufferCounts bc;
+    if (resizing) {
+        if (!_aggr_counts.empty()) {
+            bc = _aggr_counts.last_buffer();
+        }
+    } else {
+        bc = _aggr_counts.all_buffers();
     }
-    assert((usedElems % _arraySize) == 0);
-    size_t usedArrays = usedElems / _arraySize;
-    size_t neededArrays = (elemsNeeded + (resizing ? usedElems : reservedElems) + _arraySize - 1) / _arraySize;
-    size_t growArrays = (usedArrays * _allocGrowFactor);
+    assert((bc.used_elems % _arraySize) == 0);
+    assert((bc.dead_elems % _arraySize) == 0);
+    assert(bc.used_elems >= bc.dead_elems);
+    size_t neededArrays = (elemsNeeded + (resizing ? bc.used_elems : reservedElems) + _arraySize - 1) / _arraySize;
+
+    size_t liveArrays = (bc.used_elems - bc.dead_elems) / _arraySize;
+    size_t growArrays = (liveArrays * _allocGrowFactor);
+    size_t usedArrays = bc.used_elems / _arraySize;
     size_t wantedArrays = std::max((resizing ? usedArrays : 0u) + growArrays,
                                    static_cast<size_t>(_minArrays));
+
     size_t result = wantedArrays;
     if (result < neededArrays) {
         result = neededArrays;
@@ -144,6 +137,53 @@ BufferTypeBase::calcArraysToAlloc(uint32_t bufferId, ElemCount elemsNeeded, bool
     return result;
 }
 
+BufferTypeBase::AggregatedBufferCounts::AggregatedBufferCounts()
+    : _counts()
+{
+}
+
+void
+BufferTypeBase::AggregatedBufferCounts::add_buffer(const ElemCount* used_elems, const ElemCount* dead_elems)
+{
+    for (const auto& elem : _counts) {
+        assert(elem.used_ptr != used_elems);
+        assert(elem.dead_ptr != dead_elems);
+    }
+    _counts.emplace_back(used_elems, dead_elems);
+}
+
+void
+BufferTypeBase::AggregatedBufferCounts::remove_buffer(const ElemCount* used_elems, const ElemCount* dead_elems)
+{
+    auto itr = std::find_if(_counts.begin(), _counts.end(),
+                            [=](const auto& elem){ return elem.used_ptr == used_elems; });
+    assert(itr != _counts.end());
+    assert(itr->dead_ptr == dead_elems);
+    _counts.erase(itr);
+}
+
+BufferTypeBase::BufferCounts
+BufferTypeBase::AggregatedBufferCounts::last_buffer() const
+{
+    BufferCounts result;
+    assert(!_counts.empty());
+    const auto& last = _counts.back();
+    result.used_elems += *last.used_ptr;
+    result.dead_elems += *last.dead_ptr;
+    return result;
+}
+
+BufferTypeBase::BufferCounts
+BufferTypeBase::AggregatedBufferCounts::all_buffers() const
+{
+    BufferCounts result;
+    for (const auto& elem : _counts) {
+        result.used_elems += *elem.used_ptr;
+        result.dead_elems += *elem.dead_ptr;
+    }
+    return result;
+}
+
 template class BufferType<char>;
 template class BufferType<uint8_t>;
 template class BufferType<uint32_t>;
diff --git a/vespalib/src/vespa/vespalib/datastore/buffer_type.h b/vespalib/src/vespa/vespalib/datastore/buffer_type.h
index d52fbfec2a3..c901025e69c 100644
--- a/vespalib/src/vespa/vespalib/datastore/buffer_type.h
+++ b/vespalib/src/vespa/vespalib/datastore/buffer_type.h
@@ -4,6 +4,7 @@
 
 #include "atomic_entry_ref.h"
 #include <string>
+#include <vector>
 
 namespace vespalib::alloc { class MemoryAllocator; }
 
@@ -33,6 +34,7 @@ public:
         {}
         void extraBytesCleaned(size_t value);
     };
+
     BufferTypeBase(const BufferTypeBase &rhs) = delete;
     BufferTypeBase & operator=(const BufferTypeBase &rhs) = delete;
     BufferTypeBase(BufferTypeBase &&rhs) noexcept = default;
@@ -58,9 +60,8 @@ public:
     virtual size_t elementSize() const = 0;
     virtual void cleanHold(void *buffer, size_t offset, ElemCount numElems, CleanContext cleanCtx) = 0;
     size_t getArraySize() const { return _arraySize; }
-    void flushLastUsed();
-    virtual void onActive(uint32_t bufferId, ElemCount *usedElems, ElemCount &deadElems, void *buffer);
-    void onHold(const ElemCount *usedElems);
+    virtual void onActive(uint32_t bufferId, ElemCount* usedElems, ElemCount* deadElems, void* buffer);
+    void onHold(const ElemCount* usedElems, const ElemCount* deadElems);
     virtual void onFree(ElemCount usedElems);
     virtual const alloc::MemoryAllocator* get_memory_allocator() const;
 
@@ -75,6 +76,40 @@ public:
     size_t getMaxArrays() const { return _maxArrays; }
     uint32_t getNumArraysForNewBuffer() const { return _numArraysForNewBuffer; }
 protected:
+
+    struct BufferCounts {
+        ElemCount used_elems;
+        ElemCount dead_elems;
+        BufferCounts() : used_elems(0), dead_elems(0) {}
+        BufferCounts(ElemCount used_elems_in, ElemCount dead_elems_in)
+                : used_elems(used_elems_in), dead_elems(dead_elems_in)
+        {}
+    };
+
+    /**
+     * Tracks aggregated counts for all buffers that are in state ACTIVE.
+     */
+    class AggregatedBufferCounts {
+    private:
+        struct Element {
+            const ElemCount* used_ptr;
+            const ElemCount* dead_ptr;
+            Element() noexcept : used_ptr(nullptr), dead_ptr(nullptr) {}
+            Element(const ElemCount* used_ptr_in, const ElemCount* dead_ptr_in) noexcept
+                    : used_ptr(used_ptr_in), dead_ptr(dead_ptr_in)
+            {}
+        };
+        std::vector<Element> _counts;
+
+    public:
+        AggregatedBufferCounts();
+        void add_buffer(const ElemCount* used_elems, const ElemCount* dead_elems);
+        void remove_buffer(const ElemCount* used_elems, const ElemCount* dead_elems);
+        BufferCounts last_buffer() const;
+        BufferCounts all_buffers() const;
+        bool empty() const { return _counts.empty(); }
+    };
+
     uint32_t _arraySize;  // Number of elements in an allocation unit
     uint32_t _minArrays;  // Minimum number of arrays to allocate in a buffer
     uint32_t _maxArrays;  // Maximum number of arrays to allocate in a buffer
@@ -83,9 +118,8 @@ protected:
     float    _allocGrowFactor;
     uint32_t _activeBuffers;
     uint32_t _holdBuffers;
-    size_t   _activeUsedElems; // Number of used elements in all but the last active buffer for this type.
     size_t   _holdUsedElems;  // Number of used elements in all held buffers for this type.
-    const ElemCount *_lastUsedElems; // Number of used elements in the last active buffer for this type.
+    AggregatedBufferCounts _aggr_counts;
 };
 
 /**
diff --git a/vespalib/src/vespa/vespalib/datastore/bufferstate.cpp b/vespalib/src/vespa/vespalib/datastore/bufferstate.cpp
index d3a5402d7b0..68606c5ea19 100644
--- a/vespalib/src/vespa/vespalib/datastore/bufferstate.cpp
+++ b/vespalib/src/vespa/vespalib/datastore/bufferstate.cpp
@@ -133,7 +133,7 @@ BufferState::onActive(uint32_t bufferId, uint32_t typeId,
     assert(typeId <= std::numeric_limits<uint16_t>::max());
     _typeId = typeId;
     _arraySize = _typeHandler->getArraySize();
-    typeHandler->onActive(bufferId, &_usedElems, _deadElems, buffer);
+    typeHandler->onActive(bufferId, &_usedElems, &_deadElems, buffer);
 }
 
 
@@ -148,7 +148,7 @@ BufferState::onHold()
     assert(_holdElems <= (_usedElems - _deadElems));
     _deadElems = 0;
     _holdElems = _usedElems; // Put everyting on hold
-    _typeHandler->onHold(&_usedElems);
+    _typeHandler->onHold(&_usedElems, &_deadElems);
     if ( ! isFreeListEmpty()) {
         removeFromFreeListList();
         FreeList().swap(_freeList);