- Just start from the beginning to locate a free buffer.

- Assert that the requested buffer id is legal.

2 files changed, 25 insertions, 30 deletions

diff --git a/vespalib/src/vespa/vespalib/datastore/datastorebase.cpp b/vespalib/src/vespa/vespalib/datastore/datastorebase.cpp
index a232bc2ef87..99bdb19576f 100644
--- a/vespalib/src/vespa/vespalib/datastore/datastorebase.cpp
+++ b/vespalib/src/vespa/vespalib/datastore/datastorebase.cpp
@@ -106,15 +106,8 @@ DataStoreBase::~DataStoreBase()
 void
 DataStoreBase::switch_primary_buffer(uint32_t typeId, size_t elemsNeeded)
 {
-    size_t buffer_id = primary_buffer_id(typeId);
-    for (size_t i = 0; i < getNumBuffers(); ++i) {
-        // start using next buffer
-        buffer_id = nextBufferId(buffer_id);
-        if (getBufferState(buffer_id).isFree()) {
-            break;
-        }
-    }
-    if (!getBufferState(buffer_id).isFree()) {
+    size_t buffer_id = getFirstFreeBufferId();
+    if ((buffer_id < _states.size()) && !getBufferState(buffer_id).isFree()) {
         LOG_ABORT(vespalib::make_string("switch_primary_buffer(%u, %zu): did not find a free buffer",
                                         typeId, elemsNeeded).c_str());
     }
@@ -164,6 +157,23 @@ DataStoreBase::consider_grow_active_buffer(uint32_t type_id, size_t elems_needed
     return true;
 }
 
+uint32_t
+DataStoreBase::getFirstFreeBufferId() {
+    for (uint32_t buffer_id = 0; buffer_id < _states.size(); buffer_id++) {
+        if (getBufferState(buffer_id).isFree()) {
+            return buffer_id;
+        }
+    }
+    // Need next(new) buffer
+    return _states.size();
+}
+
+BufferState &
+DataStoreBase::getBufferState(uint32_t buffer_id) noexcept {
+    assert(buffer_id < _states.size());
+    return _states[buffer_id];
+}
+
 void
 DataStoreBase::switch_or_grow_primary_buffer(uint32_t typeId, size_t elemsNeeded)
 {
@@ -191,15 +201,8 @@ DataStoreBase::init_primary_buffers()
 {
     uint32_t numTypes = _primary_buffer_ids.size();
     for (uint32_t typeId = 0; typeId < numTypes; ++typeId) {
-        size_t buffer_id = 0;
-        for (size_t i = 0; i < getNumBuffers(); ++i) {
-            if (getBufferState(buffer_id).isFree()) {
-                 break;
-            }
-            // start using next buffer
-            buffer_id = nextBufferId(buffer_id);
-        }
-        assert(getBufferState(buffer_id).isFree());
+        size_t buffer_id = getFirstFreeBufferId();
+        assert((buffer_id == _states.size()) || getBufferState(buffer_id).isFree());
         onActive(buffer_id, typeId, 0u);
         _primary_buffer_ids[typeId] = buffer_id;
     }
@@ -463,7 +466,7 @@ DataStoreBase::start_compact_worst_buffers(CompactionSpec compaction_spec, const
                                           compaction_strategy.get_active_buffers_ratio(),
                                           compaction_strategy.getMaxDeadAddressSpaceRatio() / 2,
                                           CompactionStrategy::DEAD_ADDRESS_SPACE_SLACK);
-    uint32_t free_buffers = 0;
+    uint32_t free_buffers = _buffers.size() - _states.size();
     for (uint32_t bufferId = 0; bufferId < _numBuffers; ++bufferId) {
         const auto &state = getBufferState(bufferId);
         if (state.isActive()) {
diff --git a/vespalib/src/vespa/vespalib/datastore/datastorebase.h b/vespalib/src/vespa/vespalib/datastore/datastorebase.h
index db91ce8d8cd..8749f2a27e6 100644
--- a/vespalib/src/vespa/vespalib/datastore/datastorebase.h
+++ b/vespalib/src/vespa/vespalib/datastore/datastorebase.h
@@ -73,8 +73,8 @@ public:
      * Get the primary buffer id for the given type id.
      */
     uint32_t primary_buffer_id(uint32_t typeId) const { return _primary_buffer_ids[typeId]; }
-    BufferState &getBufferState(uint32_t bufferId) { return _states[bufferId]; }
-    const BufferAndMeta & getBufferMeta(uint32_t bufferId) const { return _buffers[bufferId]; }
+    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; }
 
     /**
@@ -211,15 +211,6 @@ private:
 
     class BufferHold;
 
-    /**
-     * Get the next buffer id after the given buffer id.
-     */
-    uint32_t nextBufferId(uint32_t bufferId) {
-        uint32_t ret = bufferId + 1;
-        if (ret == _numBuffers)
-            ret = 0;
-        return ret;
-    }
     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);
     void markCompacting(uint32_t bufferId);
@@ -243,6 +234,7 @@ private:
 
     void inc_hold_buffer_count();
     void fallbackResize(uint32_t bufferId, size_t elementsNeeded);
+    uint32_t getFirstFreeBufferId();
 
     virtual void reclaim_all_entry_refs() = 0;