Pass number of needed entries to DataStoreBase member functions that

might switch primary buffer for a buffer type.

6 files changed, 48 insertions, 44 deletions

diff --git a/vespalib/src/tests/datastore/datastore/datastore_test.cpp b/vespalib/src/tests/datastore/datastore/datastore_test.cpp
index 77e07ce8047..7790a135768 100644
--- a/vespalib/src/tests/datastore/datastore/datastore_test.cpp
+++ b/vespalib/src/tests/datastore/datastore/datastore_test.cpp
@@ -35,8 +35,8 @@ public:
     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);
+    void ensure_buffer_capacity(size_t entries_needed) {
+        ParentType::ensure_buffer_capacity(0, entries_needed);
     }
     void enableFreeLists() {
         ParentType::enableFreeLists();
diff --git a/vespalib/src/vespa/vespalib/datastore/allocator.hpp b/vespalib/src/vespa/vespalib/datastore/allocator.hpp
index 12a4c2f8749..83087a3286c 100644
--- a/vespalib/src/vespa/vespalib/datastore/allocator.hpp
+++ b/vespalib/src/vespa/vespalib/datastore/allocator.hpp
@@ -20,7 +20,7 @@ template <typename ... Args>
 typename Allocator<EntryT, RefT>::HandleType
 Allocator<EntryT, RefT>::alloc(Args && ... args)
 {
-    _store.ensureBufferCapacity(_typeId, 1);
+    _store.ensure_buffer_capacity(_typeId, 1);
     uint32_t buffer_id = _store.primary_buffer_id(_typeId);
     BufferState &state = _store.getBufferState(buffer_id);
     assert(state.isActive());
@@ -36,7 +36,7 @@ template <typename EntryT, typename RefT>
 typename Allocator<EntryT, RefT>::HandleType
 Allocator<EntryT, RefT>::allocArray(ConstArrayRef array)
 {
-    _store.ensureBufferCapacity(_typeId, array.size());
+    _store.ensure_buffer_capacity(_typeId, 1);
     uint32_t buffer_id = _store.primary_buffer_id(_typeId);
     BufferState &state = _store.getBufferState(buffer_id);
     assert(state.isActive());
@@ -56,12 +56,12 @@ template <typename EntryT, typename RefT>
 typename Allocator<EntryT, RefT>::HandleType
 Allocator<EntryT, RefT>::allocArray()
 {
-    auto size = _store.getBufferState(_store.primary_buffer_id(_typeId)).getArraySize();
-    _store.ensureBufferCapacity(_typeId, size);
+    _store.ensure_buffer_capacity(_typeId, 1);
     uint32_t buffer_id = _store.primary_buffer_id(_typeId);
     BufferState &state = _store.getBufferState(buffer_id);
     assert(state.isActive());
     size_t oldBufferSize = state.size();
+    auto size = state.getArraySize();
     assert((oldBufferSize % size) == 0);
     RefT ref((oldBufferSize / size), buffer_id);
     EntryT *buf = _store.template getEntryArray<EntryT>(ref, size);
diff --git a/vespalib/src/vespa/vespalib/datastore/datastore.h b/vespalib/src/vespa/vespalib/datastore/datastore.h
index 01b81d0fa58..4fcc154f944 100644
--- a/vespalib/src/vespa/vespalib/datastore/datastore.h
+++ b/vespalib/src/vespa/vespalib/datastore/datastore.h
@@ -75,7 +75,7 @@ class DataStore : public DataStoreT<RefT>
 {
 protected:
     using ParentType = DataStoreT<RefT>;
-    using ParentType::ensureBufferCapacity;
+    using ParentType::ensure_buffer_capacity;
     using ParentType::getEntry;
     using ParentType::dropBuffers;
     using ParentType::init_primary_buffers;
diff --git a/vespalib/src/vespa/vespalib/datastore/datastorebase.cpp b/vespalib/src/vespa/vespalib/datastore/datastorebase.cpp
index a40aa713bca..a7115a1ddcf 100644
--- a/vespalib/src/vespa/vespalib/datastore/datastorebase.cpp
+++ b/vespalib/src/vespa/vespalib/datastore/datastorebase.cpp
@@ -105,19 +105,19 @@ DataStoreBase::~DataStoreBase()
 }
 
 void
-DataStoreBase::switch_primary_buffer(uint32_t typeId, size_t elemsNeeded)
+DataStoreBase::switch_primary_buffer(uint32_t typeId, size_t entries_needed)
 {
     size_t buffer_id = getFirstFreeBufferId();
     if (buffer_id >= getMaxNumBuffers()) {
         LOG_ABORT(vespalib::make_string("switch_primary_buffer(%u, %zu): did not find a free buffer",
-                                        typeId, elemsNeeded).c_str());
+                                        typeId, entries_needed).c_str());
     }
-    onActive(buffer_id, typeId, elemsNeeded);
+    on_active(buffer_id, typeId, entries_needed);
     _primary_buffer_ids[typeId] = buffer_id;
 }
 
 bool
-DataStoreBase::consider_grow_active_buffer(uint32_t type_id, size_t elems_needed)
+DataStoreBase::consider_grow_active_buffer(uint32_t type_id, size_t entries_needed)
 {
     auto type_handler = _typeHandlers[type_id];
     uint32_t buffer_id = primary_buffer_id(type_id);
@@ -147,7 +147,7 @@ DataStoreBase::consider_grow_active_buffer(uint32_t type_id, size_t elems_needed
         return false;
     }
     auto array_size = type_handler->getArraySize();
-    if (elems_needed + min_used > type_handler->getMaxArrays() * array_size) {
+    if (entries_needed * array_size + min_used > type_handler->getMaxArrays() * array_size) {
         return false;
     }
     if (min_buffer_id != buffer_id) {
@@ -181,24 +181,24 @@ DataStoreBase::getBufferState(uint32_t buffer_id) noexcept {
 }
 
 void
-DataStoreBase::switch_or_grow_primary_buffer(uint32_t typeId, size_t elemsNeeded)
+DataStoreBase::switch_or_grow_primary_buffer(uint32_t typeId, size_t entries_needed)
 {
     auto typeHandler = _typeHandlers[typeId];
     uint32_t arraySize = typeHandler->getArraySize();
     size_t numArraysForNewBuffer = typeHandler->get_scaled_num_arrays_for_new_buffer();
-    size_t numEntriesForNewBuffer = numArraysForNewBuffer * arraySize;
+    size_t numElemsForNewBuffer = numArraysForNewBuffer * arraySize;
     uint32_t bufferId = primary_buffer_id(typeId);
-    if (elemsNeeded + getBufferState(bufferId).size() >= numEntriesForNewBuffer) {
-        if (consider_grow_active_buffer(typeId, elemsNeeded)) {
+    if (entries_needed * arraySize + getBufferState(bufferId).size() >= numElemsForNewBuffer) {
+        if (consider_grow_active_buffer(typeId, entries_needed)) {
             bufferId = primary_buffer_id(typeId);
-            if (elemsNeeded > getBufferState(bufferId).remaining()) {
-                fallbackResize(bufferId, elemsNeeded);
+            if (entries_needed * arraySize > getBufferState(bufferId).remaining()) {
+                fallback_resize(bufferId, entries_needed);
             }
         } else {
-            switch_primary_buffer(typeId, elemsNeeded);
+            switch_primary_buffer(typeId, entries_needed);
         }
     } else {
-        fallbackResize(bufferId, elemsNeeded);
+        fallback_resize(bufferId, entries_needed);
     }
 }
 
@@ -209,7 +209,7 @@ DataStoreBase::init_primary_buffers()
     for (uint32_t typeId = 0; typeId < numTypes; ++typeId) {
         size_t buffer_id = getFirstFreeBufferId();
         assert(buffer_id <= get_bufferid_limit_relaxed());
-        onActive(buffer_id, typeId, 0u);
+        on_active(buffer_id, typeId, 0u);
         _primary_buffer_ids[typeId] = buffer_id;
     }
 }
@@ -398,7 +398,7 @@ DataStoreBase::getAddressSpaceUsage() const
 }
 
 void
-DataStoreBase::onActive(uint32_t bufferId, uint32_t typeId, size_t elemsNeeded)
+DataStoreBase::on_active(uint32_t bufferId, uint32_t typeId, size_t entries_needed)
 {
     assert(typeId < _typeHandlers.size());
     assert(bufferId <= _bufferIdLimit);
@@ -418,7 +418,9 @@ DataStoreBase::onActive(uint32_t bufferId, uint32_t typeId, size_t elemsNeeded)
         _bufferIdLimit.store(bufferId + 1, std::memory_order_release);
     }
     assert(state->isFree());
-    state->onActive(bufferId, typeId, _typeHandlers[typeId], elemsNeeded, bufferMeta.get_atomic_buffer());
+    auto type_handler = _typeHandlers[typeId];
+    size_t array_size = type_handler->getArraySize();
+    state->onActive(bufferId, typeId, type_handler, entries_needed * array_size, bufferMeta.get_atomic_buffer());
     bufferMeta.setTypeId(typeId);
     bufferMeta.setArraySize(state->getArraySize());
     if (_freeListsEnabled && state->isActive() && !state->getCompacting()) {
@@ -436,14 +438,15 @@ DataStoreBase::finishCompact(const std::vector<uint32_t> &toHold)
 }
 
 void
-DataStoreBase::fallbackResize(uint32_t bufferId, size_t elemsNeeded)
+DataStoreBase::fallback_resize(uint32_t bufferId, size_t entries_needed)
 {
     BufferState &state = getBufferState(bufferId);
     BufferState::Alloc toHoldBuffer;
     size_t oldUsedElems = state.size();
     size_t oldAllocElems = state.capacity();
     size_t elementSize = state.getTypeHandler()->elementSize();
-    state.fallbackResize(bufferId, elemsNeeded, _buffers[bufferId].get_atomic_buffer(), toHoldBuffer);
+    size_t array_size = state.getTypeHandler()->getArraySize();
+    state.fallbackResize(bufferId, entries_needed * array_size, _buffers[bufferId].get_atomic_buffer(), toHoldBuffer);
     auto hold = std::make_unique<FallbackHold>(oldAllocElems * elementSize,
                                                std::move(toHoldBuffer),
                                                oldUsedElems,
diff --git a/vespalib/src/vespa/vespalib/datastore/datastorebase.h b/vespalib/src/vespa/vespalib/datastore/datastorebase.h
index 9cab7a2e375..a2ae5bb9965 100644
--- a/vespalib/src/vespa/vespalib/datastore/datastorebase.h
+++ b/vespalib/src/vespa/vespalib/datastore/datastorebase.h
@@ -35,15 +35,16 @@ public:
     void init_primary_buffers();
 
     /**
-     * Ensure that the primary buffer for the given type has a given number of elements free at end.
+     * Ensure that the primary buffer for the given type has a given number of entries free at end.
      * Switch to new buffer if current buffer is too full.
      *
-     * @param typeId      Registered data type for buffer.
-     * @param elemsNeeded Number of elements needed to be free.
+     * @param typeId         Registered data type for buffer.
+     * @param entries_needed Number of entries needed to be free.
      */
-    void ensureBufferCapacity(uint32_t typeId, size_t elemsNeeded) {
-        if (elemsNeeded > getBufferState(primary_buffer_id(typeId)).remaining()) [[unlikely]] {
-            switch_or_grow_primary_buffer(typeId, elemsNeeded);
+    void ensure_buffer_capacity(uint32_t typeId, size_t entries_needed) {
+        auto &state = getBufferState(primary_buffer_id(typeId));
+        if (entries_needed * state.getArraySize() > state.remaining()) [[unlikely]] {
+            switch_or_grow_primary_buffer(typeId, entries_needed);
         }
     }
 
@@ -58,10 +59,10 @@ public:
      * Switch to a new primary buffer, typically in preparation for compaction
      * or when the current primary buffer no longer has free space.
      *
-     * @param typeId      Registered data type for buffer.
-     * @param elemsNeeded Number of elements needed to be free.
+     * @param typeId         Registered data type for buffer.
+     * @param entries_needed Number of entries needed to be free.
      */
-    void switch_primary_buffer(uint32_t typeId, size_t elemsNeeded);
+    void switch_primary_buffer(uint32_t typeId, size_t entries_needed);
 
     vespalib::MemoryUsage getMemoryUsage() const;
     vespalib::MemoryUsage getDynamicMemoryUsage() const;
@@ -227,8 +228,8 @@ private:
 
     class BufferHold;
 
-    bool consider_grow_active_buffer(uint32_t type_id, size_t elems_needed);
-    void switch_or_grow_primary_buffer(uint32_t typeId, size_t elemsNeeded);
+    bool consider_grow_active_buffer(uint32_t type_id, size_t entries_needed);
+    void switch_or_grow_primary_buffer(uint32_t typeId, size_t entries_needed);
     void markCompacting(uint32_t bufferId);
     /**
      * Hold of buffer has ended.
@@ -238,14 +239,14 @@ private:
     /**
      * Switch buffer state to active for the given buffer.
      *
-     * @param bufferId    Id of buffer to be active.
-     * @param typeId      Registered data type for buffer.
-     * @param elemsNeeded Number of elements needed to be free.
+     * @param bufferId       Id of buffer to be active.
+     * @param typeId         Registered data type for buffer.
+     * @param entries_needed Number of entries needed to be free.
      */
-    void onActive(uint32_t bufferId, uint32_t typeId, size_t elemsNeeded);
+    void on_active(uint32_t bufferId, uint32_t typeId, size_t entries_needed);
 
     void inc_hold_buffer_count();
-    void fallbackResize(uint32_t bufferId, size_t elementsNeeded);
+    void fallback_resize(uint32_t bufferId, size_t entries_needed);
     uint32_t getFirstFreeBufferId();
 
     template<typename FuncType>
@@ -261,7 +262,7 @@ private:
     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.
+    // The primary buffer is used for allocations of new entries if no available slots are found in free lists.
     std::vector<uint32_t>         _primary_buffer_ids;
 
     Stash                         _stash;
diff --git a/vespalib/src/vespa/vespalib/datastore/raw_allocator.hpp b/vespalib/src/vespa/vespalib/datastore/raw_allocator.hpp
index a8792e3a307..a086423747a 100644
--- a/vespalib/src/vespa/vespalib/datastore/raw_allocator.hpp
+++ b/vespalib/src/vespa/vespalib/datastore/raw_allocator.hpp
@@ -18,13 +18,13 @@ template <typename EntryT, typename RefT>
 typename RawAllocator<EntryT, RefT>::HandleType
 RawAllocator<EntryT, RefT>::alloc(size_t num_entries, size_t extra_entries)
 {
-    size_t arraySize = _store.getBufferState(_store.primary_buffer_id(_typeId)).getArraySize();
-    _store.ensureBufferCapacity(_typeId, (num_entries + extra_entries) * arraySize);
+    _store.ensure_buffer_capacity(_typeId, num_entries + extra_entries);
     uint32_t buffer_id = _store.primary_buffer_id(_typeId);
     BufferState &state = _store.getBufferState(buffer_id);
     assert(state.isActive());
     size_t oldBufferSize = state.size();
     // Must perform scaling ourselves, according to array size
+    size_t arraySize = state.getArraySize();
     RefT ref((oldBufferSize / arraySize), buffer_id);
     EntryT *buffer = _store.getEntryArray<EntryT>(ref, arraySize);
     state.stats().pushed_back(num_entries * arraySize);