Allocate BufferState in stash and add pointer to BufferAndMeta

12 files changed, 189 insertions, 140 deletions

diff --git a/vespalib/src/tests/btree/btree_test.cpp b/vespalib/src/tests/btree/btree_test.cpp
index ef64549e16a..b8da9ea6042 100644
--- a/vespalib/src/tests/btree/btree_test.cpp
+++ b/vespalib/src/tests/btree/btree_test.cpp
@@ -1064,7 +1064,8 @@ adjustAllocatedBytes(size_t nodeCount, size_t nodeSize)
 
 TEST_F(BTreeTest, require_that_memory_usage_is_calculated)
 {
-    constexpr size_t BASE = 163912;
+    constexpr size_t BASE_ALLOCATED = 28744u;
+    constexpr size_t BASE_USED = 24984;
     typedef BTreeNodeAllocator<int32_t, int8_t,
         btree::NoAggregated,
         MyTraits::INTERNAL_SLOTS, MyTraits::LEAF_SLOTS> NodeAllocator;
@@ -1083,8 +1084,8 @@ TEST_F(BTreeTest, require_that_memory_usage_is_calculated)
     const uint32_t initialLeafNodes = 128u;
     mu.incAllocatedBytes(adjustAllocatedBytes(initialInternalNodes, sizeof(INode)));
     mu.incAllocatedBytes(adjustAllocatedBytes(initialLeafNodes, sizeof(LNode)));
-    mu.incAllocatedBytes(BASE);
-    mu.incUsedBytes(BASE);
+    mu.incAllocatedBytes(BASE_ALLOCATED);
+    mu.incUsedBytes(BASE_USED);
     mu.incUsedBytes(sizeof(INode));
     mu.incDeadBytes(sizeof(INode));
     EXPECT_TRUE(assertMemoryUsage(mu, tm.getMemoryUsage()));
@@ -1115,8 +1116,8 @@ TEST_F(BTreeTest, require_that_memory_usage_is_calculated)
     mu = vespalib::MemoryUsage();
     mu.incAllocatedBytes(adjustAllocatedBytes(initialInternalNodes, sizeof(INode)));
     mu.incAllocatedBytes(adjustAllocatedBytes(initialLeafNodes, sizeof(LNode)));
-    mu.incAllocatedBytes(BASE);
-    mu.incUsedBytes(BASE);
+    mu.incAllocatedBytes(BASE_ALLOCATED);
+    mu.incUsedBytes(BASE_USED);
     mu.incUsedBytes(sizeof(INode) * 2);
     mu.incDeadBytes(sizeof(INode) * 2);
     mu.incUsedBytes(sizeof(LNode));
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 97e1ddb985d..c32554232ef 100644
--- a/vespalib/src/tests/datastore/array_store/array_store_test.cpp
+++ b/vespalib/src/tests/datastore/array_store/array_store_test.cpp
@@ -210,16 +210,16 @@ INSTANTIATE_TEST_SUITE_P(NumberStoreFreeListsDisabledMultiTest,
 
 TEST_P(NumberStoreTest, control_static_sizes) {
 #ifdef _LIBCPP_VERSION
-    EXPECT_EQ(472u, sizeof(store));
-    EXPECT_EQ(296u, sizeof(NumberStoreTest::ArrayStoreType::DataStoreType));
+    EXPECT_EQ(464u, sizeof(store));
+    EXPECT_EQ(280u, sizeof(NumberStoreTest::ArrayStoreType::DataStoreType));
 #else
-    EXPECT_EQ(504u, sizeof(store));
-    EXPECT_EQ(328u, sizeof(NumberStoreTest::ArrayStoreType::DataStoreType));
+    EXPECT_EQ(496u, sizeof(store));
+    EXPECT_EQ(320u, sizeof(NumberStoreTest::ArrayStoreType::DataStoreType));
 #endif
     EXPECT_EQ(112u, sizeof(NumberStoreTest::ArrayStoreType::SmallBufferType));
     MemoryUsage usage = store.getMemoryUsage();
-    EXPECT_EQ(1312160u, usage.allocatedBytes());
-    EXPECT_EQ(1311232u, usage.usedBytes());
+    EXPECT_EQ(202144u, usage.allocatedBytes());
+    EXPECT_EQ(197776u, usage.usedBytes());
 }
 
 TEST_P(NumberStoreTest, add_and_get_small_arrays_of_trivial_type)
diff --git a/vespalib/src/tests/datastore/datastore/datastore_test.cpp b/vespalib/src/tests/datastore/datastore/datastore_test.cpp
index 794be39ae9b..7121bf5ce11 100644
--- a/vespalib/src/tests/datastore/datastore/datastore_test.cpp
+++ b/vespalib/src/tests/datastore/datastore/datastore_test.cpp
@@ -474,7 +474,8 @@ TEST(DataStoreTest, require_that_memory_stats_are_calculated)
 
 TEST(DataStoreTest, require_that_memory_usage_is_calculated)
 {
-    constexpr size_t BASE = 676;
+    constexpr size_t BASE_ALLOCATED = 4228;
+    constexpr size_t BASE_USED = 308;
     MyStore s;
     MyRef r = s.addEntry(10);
     s.addEntry(20);
@@ -483,8 +484,8 @@ TEST(DataStoreTest, require_that_memory_usage_is_calculated)
     s.holdBuffer(r.bufferId());
     s.assign_generation(100);
     vespalib::MemoryUsage m = s.getMemoryUsage();
-    EXPECT_EQ(MyRef::offsetSize() * sizeof(int) + BASE, m.allocatedBytes());
-    EXPECT_EQ(5 * sizeof(int) + BASE, m.usedBytes());
+    EXPECT_EQ(MyRef::offsetSize() * sizeof(int) + BASE_ALLOCATED, m.allocatedBytes());
+    EXPECT_EQ(5 * sizeof(int) + BASE_USED, m.usedBytes());
     EXPECT_EQ(0 * sizeof(int), m.deadBytes());
     EXPECT_EQ(5 * sizeof(int), m.allocatedBytesOnHold());
     s.reclaim_memory(101);
@@ -492,28 +493,29 @@ TEST(DataStoreTest, require_that_memory_usage_is_calculated)
 
 TEST(DataStoreTest, require_that_we_can_disable_elemement_hold_list)
 {
-    constexpr size_t BASE = 676;
+    constexpr size_t BASE_ALLOCATED = 4228;
+    constexpr size_t BASE_USED = 308;
     MyStore s;
     MyRef r1 = s.addEntry(10);
     MyRef r2 = s.addEntry(20);
     MyRef r3 = s.addEntry(30);
     (void) r3;
     vespalib::MemoryUsage m = s.getMemoryUsage();
-    EXPECT_EQ(MyRef::offsetSize() * sizeof(int) + BASE, m.allocatedBytes());
-    EXPECT_EQ(4 * sizeof(int) + BASE, m.usedBytes());
+    EXPECT_EQ(MyRef::offsetSize() * sizeof(int) + BASE_ALLOCATED, m.allocatedBytes());
+    EXPECT_EQ(4 * sizeof(int) + BASE_USED, m.usedBytes());
     EXPECT_EQ(1 * sizeof(int), m.deadBytes());
     EXPECT_EQ(0 * sizeof(int), m.allocatedBytesOnHold());
     s.holdElem(r1, 1);
     m = s.getMemoryUsage();
-    EXPECT_EQ(MyRef::offsetSize() * sizeof(int) + BASE, m.allocatedBytes());
-    EXPECT_EQ(4 * sizeof(int) + BASE, m.usedBytes());
+    EXPECT_EQ(MyRef::offsetSize() * sizeof(int) + BASE_ALLOCATED, m.allocatedBytes());
+    EXPECT_EQ(4 * sizeof(int) + BASE_USED, m.usedBytes());
     EXPECT_EQ(1 * sizeof(int), m.deadBytes());
     EXPECT_EQ(1 * sizeof(int), m.allocatedBytesOnHold());
     s.disableElemHoldList();
     s.holdElem(r2, 1);
     m = s.getMemoryUsage();
-    EXPECT_EQ(MyRef::offsetSize() * sizeof(int) + BASE, m.allocatedBytes());
-    EXPECT_EQ(4 * sizeof(int) + BASE, m.usedBytes());
+    EXPECT_EQ(MyRef::offsetSize() * sizeof(int) + BASE_ALLOCATED, m.allocatedBytes());
+    EXPECT_EQ(4 * sizeof(int) + BASE_USED, m.usedBytes());
     EXPECT_EQ(2 * sizeof(int), m.deadBytes());
     EXPECT_EQ(1 * sizeof(int), m.allocatedBytesOnHold());
     s.assign_generation(100);
@@ -538,7 +540,7 @@ void assertGrowStats(GrowthStats expSizes,
 
 TEST(DataStoreTest, require_that_buffer_growth_works)
 {
-    constexpr size_t BASE = 41032u;
+    constexpr size_t BASE = 10312;
     // Always switch to new buffer, min size 4
     assertGrowStats({ 4, 4, 4, 4, 8, 16, 16, 32, 64, 64 },
                     { 4 }, 20 + BASE, 4, 0);
diff --git a/vespalib/src/tests/datastore/unique_store/unique_store_test.cpp b/vespalib/src/tests/datastore/unique_store/unique_store_test.cpp
index cf62d238d53..1781a680bac 100644
--- a/vespalib/src/tests/datastore/unique_store/unique_store_test.cpp
+++ b/vespalib/src/tests/datastore/unique_store/unique_store_test.cpp
@@ -470,14 +470,14 @@ TEST_F(DoubleTest, nan_is_handled)
 }
 
 TEST_F(DoubleTest, control_memory_usage) {
-    EXPECT_EQ(464, sizeof(store));
+    EXPECT_EQ(456u, sizeof(store));
     EXPECT_EQ(144u, sizeof(BufferState));
-    EXPECT_EQ(163908u, store.get_values_memory_usage().allocatedBytes());
-    EXPECT_EQ(163892u, store.get_values_memory_usage().usedBytes());
-    EXPECT_EQ(262120u, store.get_dictionary_memory_usage().allocatedBytes());
-    EXPECT_EQ(164176u, store.get_dictionary_memory_usage().usedBytes());
-    EXPECT_EQ(426028u, store.getMemoryUsage().allocatedBytes());
-    EXPECT_EQ(328068u, store.getMemoryUsage().usedBytes());
+    EXPECT_EQ(28740u, store.get_values_memory_usage().allocatedBytes());
+    EXPECT_EQ(24804u, store.get_values_memory_usage().usedBytes());
+    EXPECT_EQ(126952u, store.get_dictionary_memory_usage().allocatedBytes());
+    EXPECT_EQ(25248u, store.get_dictionary_memory_usage().usedBytes());
+    EXPECT_EQ(155692u, store.getMemoryUsage().allocatedBytes());
+    EXPECT_EQ(50052, store.getMemoryUsage().usedBytes());
 }
                 
 GTEST_MAIN_RUN_ALL_TESTS()
diff --git a/vespalib/src/vespa/vespalib/datastore/array_store.h b/vespalib/src/vespa/vespalib/datastore/array_store.h
index dd786e5f2e2..c2b65d72f03 100644
--- a/vespalib/src/vespa/vespalib/datastore/array_store.h
+++ b/vespalib/src/vespa/vespalib/datastore/array_store.h
@@ -76,10 +76,9 @@ public:
             return ConstArrayRef();
         }
         RefT internalRef(ref);
-        uint32_t typeId = _store.getTypeId(internalRef.bufferId());
-        if (typeId != _largeArrayTypeId) [[likely]] {
-            size_t arraySize = _mapper.get_array_size(typeId);
-            return getSmallArray(internalRef, arraySize);
+        const BufferAndMeta & bufferAndMeta = _store.getBufferMeta(internalRef.bufferId());
+        if (bufferAndMeta.getTypeId() != _largeArrayTypeId) [[likely]] {
+            return getSmallArray(internalRef, bufferAndMeta.getArraySize());
         } else {
             return getLargeArray(internalRef);
         }
diff --git a/vespalib/src/vespa/vespalib/datastore/buffer_free_list.cpp b/vespalib/src/vespa/vespalib/datastore/buffer_free_list.cpp
index 0a440e3e867..224ed4b0c8f 100644
--- a/vespalib/src/vespa/vespalib/datastore/buffer_free_list.cpp
+++ b/vespalib/src/vespa/vespalib/datastore/buffer_free_list.cpp
@@ -52,5 +52,23 @@ BufferFreeList::disable()
     _free_list = nullptr;
 }
 
+void
+BufferFreeList::push_entry(EntryRef ref) {
+    if (empty()) {
+        attach();
+    }
+    _free_refs.push_back(ref);
+}
+EntryRef
+BufferFreeList::pop_entry() {
+    EntryRef ret = _free_refs.back();
+    _free_refs.pop_back();
+    if (empty()) {
+        detach();
+    }
+    _dead_elems.store(_dead_elems.load(std::memory_order_relaxed) - _array_size, std::memory_order_relaxed);
+    return ret;
+}
+
 }
 
diff --git a/vespalib/src/vespa/vespalib/datastore/buffer_free_list.h b/vespalib/src/vespa/vespalib/datastore/buffer_free_list.h
index 74fd47056de..148ddd8db88 100644
--- a/vespalib/src/vespa/vespalib/datastore/buffer_free_list.h
+++ b/vespalib/src/vespa/vespalib/datastore/buffer_free_list.h
@@ -41,21 +41,8 @@ public:
     bool enabled() const { return _free_list != nullptr; }
     bool empty() const { return _free_refs.empty(); }
     uint32_t array_size() const { return _array_size; }
-    void push_entry(EntryRef ref) {
-        if (empty()) {
-            attach();
-        }
-        _free_refs.push_back(ref);
-    }
-    EntryRef pop_entry() {
-        EntryRef ret = _free_refs.back();
-        _free_refs.pop_back();
-        if (empty()) {
-            detach();
-        }
-        _dead_elems.store(_dead_elems.load(std::memory_order_relaxed) - _array_size, std::memory_order_relaxed);
-        return ret;
-    }
+    void push_entry(EntryRef ref);
+    EntryRef pop_entry();
 };
 
 }
diff --git a/vespalib/src/vespa/vespalib/datastore/bufferstate.h b/vespalib/src/vespa/vespalib/datastore/bufferstate.h
index aa7f6dfdfa4..a0a8638e65f 100644
--- a/vespalib/src/vespa/vespalib/datastore/bufferstate.h
+++ b/vespalib/src/vespa/vespalib/datastore/bufferstate.h
@@ -134,23 +134,28 @@ public:
 
 class BufferAndMeta {
 public:
-    BufferAndMeta() : BufferAndMeta(nullptr, 0, 0) { }
-    BufferAndMeta(void* buffer, uint32_t typeId, uint32_t arraySize)
-        : _buffer(buffer),
-          _typeId(typeId),
-          _arraySize(arraySize)
-    { }
+    BufferAndMeta() : BufferAndMeta(nullptr, nullptr, 0, 0) { }
     std::atomic<void*>& get_atomic_buffer() noexcept { return _buffer; }
     void* get_buffer_relaxed() noexcept { return _buffer.load(std::memory_order_relaxed); }
     const void* get_buffer_acquire() const noexcept { return _buffer.load(std::memory_order_acquire); }
     uint32_t getTypeId() const { return _typeId; }
     uint32_t getArraySize() const { return _arraySize; }
+    BufferState * get_state_relaxed() const { return _state.load(std::memory_order_relaxed); }
+    BufferState * get_state_acquire() const { return _state.load(std::memory_order_acquire); }
     void setTypeId(uint32_t typeId) { _typeId = typeId; }
     void setArraySize(uint32_t arraySize) { _arraySize = arraySize; }
+    void set_state(BufferState * state) { _state.store(state, std::memory_order_release); }
 private:
-    std::atomic<void*> _buffer;
-    uint32_t   _typeId;
-    uint32_t   _arraySize;
+    BufferAndMeta(void* buffer, BufferState * state, uint32_t typeId, uint32_t arraySize)
+        : _buffer(buffer),
+          _state(state),
+          _typeId(typeId),
+          _arraySize(arraySize)
+    { }
+    std::atomic<void*>        _buffer;
+    std::atomic<BufferState*> _state;
+    uint32_t                  _typeId;
+    uint32_t                  _arraySize;
 };
 
 }
diff --git a/vespalib/src/vespa/vespalib/datastore/datastorebase.cpp b/vespalib/src/vespa/vespalib/datastore/datastorebase.cpp
index 99bdb19576f..a0c5e282f9a 100644
--- a/vespalib/src/vespa/vespalib/datastore/datastorebase.cpp
+++ b/vespalib/src/vespa/vespalib/datastore/datastorebase.cpp
@@ -84,15 +84,15 @@ DataStoreBase::DataStoreBase(uint32_t numBuffers, uint32_t offset_bits, size_t m
     : _entry_ref_hold_list(),
       _buffers(numBuffers),
       _primary_buffer_ids(),
-      _states(numBuffers),
+      _stash(),
       _typeHandlers(),
       _free_lists(),
       _compaction_count(0u),
       _genHolder(),
       _maxArrays(maxArrays),
-      _numBuffers(numBuffers),
-      _offset_bits(offset_bits),
+      _bufferIdLimit(0u),
       _hold_buffer_count(0u),
+      _offset_bits(offset_bits),
       _freeListsEnabled(false),
       _initializing(false)
 {
@@ -107,7 +107,7 @@ void
 DataStoreBase::switch_primary_buffer(uint32_t typeId, size_t elemsNeeded)
 {
     size_t buffer_id = getFirstFreeBufferId();
-    if ((buffer_id < _states.size()) && !getBufferState(buffer_id).isFree()) {
+    if (buffer_id > getMaxNumBuffers()) {
         LOG_ABORT(vespalib::make_string("switch_primary_buffer(%u, %zu): did not find a free buffer",
                                         typeId, elemsNeeded).c_str());
     }
@@ -159,19 +159,22 @@ DataStoreBase::consider_grow_active_buffer(uint32_t type_id, size_t elems_needed
 
 uint32_t
 DataStoreBase::getFirstFreeBufferId() {
-    for (uint32_t buffer_id = 0; buffer_id < _states.size(); buffer_id++) {
-        if (getBufferState(buffer_id).isFree()) {
+    uint32_t buffer_id = 0;
+    for (const auto & buffer : _buffers) {
+        if (buffer.get_state_relaxed() == nullptr || buffer.get_state_relaxed()->isFree()) {
             return buffer_id;
         }
+        buffer_id++;
     }
-    // Need next(new) buffer
-    return _states.size();
+    // No free buffer, return out of bounds
+    return buffer_id;
 }
 
 BufferState &
 DataStoreBase::getBufferState(uint32_t buffer_id) noexcept {
-    assert(buffer_id < _states.size());
-    return _states[buffer_id];
+    assert(buffer_id < getBufferIdLimit());
+    assert(_buffers[buffer_id].get_state_relaxed() != nullptr);
+    return *_buffers[buffer_id].get_state_relaxed();
 }
 
 void
@@ -202,7 +205,7 @@ DataStoreBase::init_primary_buffers()
     uint32_t numTypes = _primary_buffer_ids.size();
     for (uint32_t typeId = 0; typeId < numTypes; ++typeId) {
         size_t buffer_id = getFirstFreeBufferId();
-        assert((buffer_id == _states.size()) || getBufferState(buffer_id).isFree());
+        assert(buffer_id <= getBufferIdLimit());
         onActive(buffer_id, typeId, 0u);
         _primary_buffer_ids[typeId] = buffer_id;
     }
@@ -253,9 +256,13 @@ DataStoreBase::reclaim_all_memory()
 void
 DataStoreBase::dropBuffers()
 {
-    uint32_t numBuffers = _buffers.size();
-    for (uint32_t bufferId = 0; bufferId < numBuffers; ++bufferId) {
-        getBufferState(bufferId).dropBuffer(bufferId, _buffers[bufferId].get_atomic_buffer());
+    uint32_t buffer_id_limit = getBufferIdLimit();
+    for (uint32_t bufferId = 0; bufferId < buffer_id_limit; ++bufferId) {
+        BufferAndMeta & buffer = _buffers[bufferId];
+        BufferState * state = buffer.get_state_relaxed();
+        if (state) {
+            state->dropBuffer(bufferId, buffer.get_atomic_buffer());
+        }
     }
     _genHolder.reclaim_all();
 }
@@ -278,18 +285,17 @@ DataStoreBase::getMemoryUsage() const {
     size_t extra_allocated = 0;
     extra_allocated += _buffers.capacity() * sizeof(BufferAndMeta);
     extra_allocated += _primary_buffer_ids.capacity() * sizeof(uint32_t);
-    extra_allocated += _states.capacity() * sizeof(BufferState);
     extra_allocated += _typeHandlers.capacity() * sizeof(BufferTypeBase *);
     extra_allocated += _free_lists.capacity() * sizeof(FreeList);
 
     size_t extra_used = 0;
     extra_used += _buffers.size() * sizeof(BufferAndMeta);
     extra_used += _primary_buffer_ids.size() * sizeof(uint32_t);
-    extra_used += _states.size() * sizeof(BufferState);
     extra_used += _typeHandlers.size() * sizeof(BufferTypeBase *);
     extra_used += _free_lists.size() * sizeof(FreeList);
     usage.incAllocatedBytes(extra_allocated);
     usage.incUsedBytes(extra_used);
+    usage.merge(_stash.get_memory_usage());
     return usage;
 }
 
@@ -305,11 +311,12 @@ DataStoreBase::holdBuffer(uint32_t bufferId)
 void
 DataStoreBase::enableFreeLists()
 {
-    for (auto& bState : _states) {
-        if (!bState.isActive() || bState.getCompacting()) {
+    for (const auto& buffer : _buffers) {
+        BufferState * bState = buffer.get_state_relaxed();
+        if (!bState || !bState->isActive() || bState->getCompacting()) {
             continue;
         }
-        bState.enable_free_list(_free_lists[bState.getTypeId()]);
+        bState->enable_free_list(_free_lists[bState->getTypeId()]);
     }
     _freeListsEnabled = true;
 }
@@ -317,8 +324,11 @@ DataStoreBase::enableFreeLists()
 void
 DataStoreBase::disableFreeLists()
 {
-    for (auto& bState : _states) {
-        bState.disable_free_list();
+    for (const auto& buffer : _buffers) {
+        BufferState * bState = buffer.get_state_relaxed();
+        if (bState) {
+            bState->disable_free_list();
+        }
     }
     _freeListsEnabled = false;
 }
@@ -335,9 +345,10 @@ DataStoreBase::enableFreeList(uint32_t bufferId)
 void
 DataStoreBase::disableElemHoldList()
 {
-    for (auto &state : _states) {
-        if (!state.isFree()) {
-            state.disableElemHoldList();
+    for (const auto& buffer : _buffers) {
+        BufferState * state = buffer.get_state_relaxed();
+        if (state && !state->isFree()) {
+            state->disableElemHoldList();
         }
     }
 }
@@ -347,21 +358,28 @@ DataStoreBase::getMemStats() const
 {
     MemoryStats stats;
 
-    for (const auto& bState: _states) {
-        auto typeHandler = bState.getTypeHandler();
-        auto state = bState.getState();
-        if ((state == BufferState::State::FREE) || (typeHandler == nullptr)) {
-            ++stats._freeBuffers;
-        } else if (state == BufferState::State::ACTIVE) {
-            size_t elementSize = typeHandler->elementSize();
-            ++stats._activeBuffers;
-            bState.stats().add_to_mem_stats(elementSize, stats);
-        } else if (state == BufferState::State::HOLD) {
-            size_t elementSize = typeHandler->elementSize();
-            ++stats._holdBuffers;
-            bState.stats().add_to_mem_stats(elementSize, stats);
+    uint32_t buffer_id_limit = getBufferIdLimit();
+    stats._freeBuffers = (getMaxNumBuffers() - buffer_id_limit);
+    for (uint32_t bufferId = 0; bufferId < buffer_id_limit; ++bufferId) {
+        BufferState * bState = _buffers[bufferId].get_state_acquire();
+        if (bState != nullptr) {
+            auto typeHandler = bState->getTypeHandler();
+            auto state = bState->getState();
+            if ((state == BufferState::State::FREE) || (typeHandler == nullptr)) {
+                ++stats._freeBuffers;
+            } else if (state == BufferState::State::ACTIVE) {
+                size_t elementSize = typeHandler->elementSize();
+                ++stats._activeBuffers;
+                bState->stats().add_to_mem_stats(elementSize, stats);
+            } else if (state == BufferState::State::HOLD) {
+                size_t elementSize = typeHandler->elementSize();
+                ++stats._holdBuffers;
+                bState->stats().add_to_mem_stats(elementSize, stats);
+            } else {
+                LOG_ABORT("should not be reached");
+            }
         } else {
-            LOG_ABORT("should not be reached");
+            ++stats._freeBuffers;
         }
     }
     size_t genHolderHeldBytes = _genHolder.get_held_bytes();
@@ -374,21 +392,23 @@ DataStoreBase::getMemStats() const
 vespalib::AddressSpace
 DataStoreBase::getAddressSpaceUsage() const
 {
+    uint32_t buffer_id_limit = getBufferIdLimit();
     size_t usedArrays = 0;
     size_t deadArrays = 0;
-    size_t limitArrays = 0;
-    for (const auto& bState: _states) {
-        if (bState.isActive()) {
-            uint32_t arraySize = bState.getArraySize();
-            usedArrays += bState.size() / arraySize;
-            deadArrays += bState.stats().dead_elems() / arraySize;
-            limitArrays += bState.capacity() / arraySize;
-        } else if (bState.isOnHold()) {
-            uint32_t arraySize = bState.getArraySize();
-            usedArrays += bState.size() / arraySize;
-            limitArrays += bState.capacity() / arraySize;
-        } else if (bState.isFree()) {
+    size_t limitArrays = size_t(_maxArrays) * (getMaxNumBuffers() - buffer_id_limit);
+    for (uint32_t bufferId = 0; bufferId < buffer_id_limit; ++bufferId) {
+        BufferState * bState = _buffers[bufferId].get_state_acquire();
+        if (bState == nullptr || bState->isFree()) {
             limitArrays += _maxArrays;
+        } else if (bState->isActive()) {
+            uint32_t arraySize = bState->getArraySize();
+            usedArrays += bState->size() / arraySize;
+            deadArrays += bState->stats().dead_elems() / arraySize;
+            limitArrays += bState->capacity() / arraySize;
+        } else if (bState->isOnHold()) {
+            uint32_t arraySize = bState->getArraySize();
+            usedArrays += bState->size() / arraySize;
+            limitArrays += bState->capacity() / arraySize;
         } else {
             LOG_ABORT("should not be reached");
         }
@@ -400,12 +420,20 @@ void
 DataStoreBase::onActive(uint32_t bufferId, uint32_t typeId, size_t elemsNeeded)
 {
     assert(typeId < _typeHandlers.size());
-    assert(bufferId < _numBuffers);
-    BufferState &state = getBufferState(bufferId);
+    assert(bufferId <= _bufferIdLimit);
+
     BufferAndMeta & bufferMeta = _buffers[bufferId];
-    state.onActive(bufferId, typeId, _typeHandlers[typeId], elemsNeeded, bufferMeta.get_atomic_buffer());
+    BufferState *state = bufferMeta.get_state_relaxed();
+    if (state == nullptr) {
+        BufferState & newState = _stash.create<BufferState>();
+        state = & newState;
+        bufferMeta.set_state(state);
+        _bufferIdLimit.store(bufferId + 1);
+    }
+    assert(state->isFree());
+    state->onActive(bufferId, typeId, _typeHandlers[typeId], elemsNeeded, bufferMeta.get_atomic_buffer());
     bufferMeta.setTypeId(typeId);
-    bufferMeta.setArraySize(state.getArraySize());
+    bufferMeta.setArraySize(state->getArraySize());
     enableFreeList(bufferId);
 }
 
@@ -456,39 +484,40 @@ DataStoreBase::markCompacting(uint32_t bufferId)
 std::unique_ptr<CompactingBuffers>
 DataStoreBase::start_compact_worst_buffers(CompactionSpec compaction_spec, const CompactionStrategy& compaction_strategy)
 {
+    uint32_t buffer_id_limit = getBufferIdLimit();
     // compact memory usage
-    CompactBufferCandidates elem_buffers(_numBuffers, compaction_strategy.get_max_buffers(),
+    CompactBufferCandidates elem_buffers(buffer_id_limit, compaction_strategy.get_max_buffers(),
                                          compaction_strategy.get_active_buffers_ratio(),
                                          compaction_strategy.getMaxDeadBytesRatio() / 2,
                                          CompactionStrategy::DEAD_BYTES_SLACK);
     // compact address space
-    CompactBufferCandidates array_buffers(_numBuffers, compaction_strategy.get_max_buffers(),
+    CompactBufferCandidates array_buffers(buffer_id_limit, compaction_strategy.get_max_buffers(),
                                           compaction_strategy.get_active_buffers_ratio(),
                                           compaction_strategy.getMaxDeadAddressSpaceRatio() / 2,
                                           CompactionStrategy::DEAD_ADDRESS_SPACE_SLACK);
-    uint32_t free_buffers = _buffers.size() - _states.size();
-    for (uint32_t bufferId = 0; bufferId < _numBuffers; ++bufferId) {
-        const auto &state = getBufferState(bufferId);
-        if (state.isActive()) {
-            auto typeHandler = state.getTypeHandler();
+    uint32_t free_buffers = getMaxNumBuffers() - buffer_id_limit;
+    for (uint32_t bufferId = 0; bufferId < buffer_id_limit; ++bufferId) {
+        BufferState * state = _buffers[bufferId].get_state_relaxed();
+        if (state == nullptr || state->isFree()) {
+            free_buffers++;
+        } else if (state->isActive()) {
+            auto typeHandler = state->getTypeHandler();
             uint32_t arraySize = typeHandler->getArraySize();
             uint32_t reservedElements = typeHandler->getReservedElements(bufferId);
-            size_t used_elems = state.size();
-            size_t deadElems = state.stats().dead_elems() - reservedElements;
+            size_t used_elems = state->size();
+            size_t deadElems = state->stats().dead_elems() - reservedElements;
             if (compaction_spec.compact_memory()) {
                 elem_buffers.add(bufferId, used_elems, deadElems);
             }
             if (compaction_spec.compact_address_space()) {
-                array_buffers.add(bufferId,  used_elems / arraySize, deadElems / arraySize);
+                array_buffers.add(bufferId, used_elems / arraySize, deadElems / arraySize);
             }
-        } else if (state.isFree()) {
-            ++free_buffers;
         }
     }
     elem_buffers.set_free_buffers(free_buffers);
     array_buffers.set_free_buffers(free_buffers);
     std::vector<uint32_t> result;
-    result.reserve(std::min(_numBuffers, 2 * compaction_strategy.get_max_buffers()));
+    result.reserve(std::min(buffer_id_limit, 2 * compaction_strategy.get_max_buffers()));
     elem_buffers.select(result);
     array_buffers.select(result);
     std::sort(result.begin(), result.end());
@@ -497,7 +526,7 @@ DataStoreBase::start_compact_worst_buffers(CompactionSpec compaction_spec, const
     for (auto buffer_id : result) {
         markCompacting(buffer_id);
     }
-    return std::make_unique<CompactingBuffers>(*this, _numBuffers, _offset_bits, std::move(result));
+    return std::make_unique<CompactingBuffers>(*this, buffer_id_limit, _offset_bits, std::move(result));
 }
 
 void
diff --git a/vespalib/src/vespa/vespalib/datastore/datastorebase.h b/vespalib/src/vespa/vespalib/datastore/datastorebase.h
index 8749f2a27e6..fd5b97503c9 100644
--- a/vespalib/src/vespa/vespalib/datastore/datastorebase.h
+++ b/vespalib/src/vespa/vespalib/datastore/datastorebase.h
@@ -8,9 +8,8 @@
 #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 <vespa/vespalib/util/stash.h>
 #include <atomic>
-#include <deque>
 #include <vector>
 
 namespace vespalib::datastore {
@@ -75,7 +74,19 @@ public:
     uint32_t primary_buffer_id(uint32_t typeId) const { return _primary_buffer_ids[typeId]; }
     BufferState &getBufferState(uint32_t buffer_id) noexcept;
     const BufferAndMeta & getBufferMeta(uint32_t buffer_id) const { return _buffers[buffer_id]; }
-    uint32_t getNumBuffers() const { return _numBuffers; }
+    uint32_t getMaxNumBuffers() const noexcept { return _buffers.size(); }
+    uint32_t getBufferIdLimit() const noexcept { return _bufferIdLimit.load(std::memory_order_relaxed); }
+
+    template<typename FuncType>
+    void for_each_active_buffer(FuncType func) {
+        uint32_t buffer_id_limit = getBufferIdLimit();
+        for (uint32_t i = 0; i < buffer_id_limit; i++) {
+            const BufferState * state = _buffers[i].get_state_relaxed();
+            if (state && state->isActive()) {
+                func(i, *state);
+            }
+        }
+    }
 
     /**
      * Assign generation on data elements on hold lists added since the last time this function was called.
@@ -238,21 +249,21 @@ private:
 
     virtual void reclaim_all_entry_refs() = 0;
 
-    std::vector<BufferAndMeta>  _buffers; // For fast mapping with known types
+    std::vector<BufferAndMeta>    _buffers; // For fast mapping with known types
 
     // Provides a mapping from typeId -> primary buffer for that type.
     // The primary buffer is used for allocations of new element(s) if no available slots are found in free lists.
     std::vector<uint32_t>         _primary_buffer_ids;
 
-    std::vector<BufferState>      _states;
+    Stash                         _stash;
     std::vector<BufferTypeBase *> _typeHandlers; // TypeId -> handler
     std::vector<FreeList>         _free_lists;
     mutable std::atomic<uint64_t> _compaction_count;
     vespalib::GenerationHolder    _genHolder;
     const uint32_t                _maxArrays;
-    const uint32_t                _numBuffers;
-    const uint32_t                _offset_bits;
+    std::atomic<uint32_t>         _bufferIdLimit;
     uint32_t                      _hold_buffer_count;
+    const uint8_t                 _offset_bits;
     bool                          _freeListsEnabled;
     bool                          _initializing;
 };
diff --git a/vespalib/src/vespa/vespalib/datastore/unique_store.hpp b/vespalib/src/vespa/vespalib/datastore/unique_store.hpp
index aef6ea07290..602b953e4e6 100644
--- a/vespalib/src/vespa/vespalib/datastore/unique_store.hpp
+++ b/vespalib/src/vespa/vespalib/datastore/unique_store.hpp
@@ -102,8 +102,8 @@ private:
     std::unique_ptr<CompactingBuffers> _compacting_buffers;
 
     void allocMapping() {
-        _mapping.resize(RefT::numBuffers());
         auto& data_store = _compacting_buffers->get_store();
+        _mapping.resize(data_store.getBufferIdLimit());
         for (const auto bufferId : _compacting_buffers->get_buffer_ids()) {
             BufferState &state = data_store.getBufferState(bufferId);
             _mapping[bufferId].resize(state.get_used_arrays());
diff --git a/vespalib/src/vespa/vespalib/datastore/unique_store_enumerator.hpp b/vespalib/src/vespa/vespalib/datastore/unique_store_enumerator.hpp
index 6d08a027bf1..8c7a646b287 100644
--- a/vespalib/src/vespa/vespalib/datastore/unique_store_enumerator.hpp
+++ b/vespalib/src/vespa/vespalib/datastore/unique_store_enumerator.hpp
@@ -42,13 +42,10 @@ template <typename RefT>
 void
 UniqueStoreEnumerator<RefT>::allocate_enum_values(DataStoreBase & store)
 {
-    _enumValues.resize(RefType::numBuffers());
-    for (uint32_t bufferId = 0; bufferId < RefType::numBuffers(); ++bufferId) {
-        const BufferState &state = store.getBufferState(bufferId);
-        if (state.isActive()) {
-            _enumValues[bufferId].resize(state.get_used_arrays());
-        }
-    }
+    _enumValues.resize(store.getBufferIdLimit());
+    store.for_each_active_buffer([this](uint32_t buffer_id, const BufferState & state) {
+        _enumValues[buffer_id].resize(state.get_used_arrays());
+    });
 }
 
 template <typename RefT>