Hide more details inside BufferState and reduce external API on BufferStats.

Using incDead() directly is no longer supported as marking elements as
dead right before they are put on hold is unnecessary.

12 files changed, 122 insertions, 85 deletions

diff --git a/vespalib/CMakeLists.txt b/vespalib/CMakeLists.txt
index 7aafb7c364e..b84a90465ea 100644
--- a/vespalib/CMakeLists.txt
+++ b/vespalib/CMakeLists.txt
@@ -54,6 +54,7 @@ vespa_define_module(
     src/tests/data/smart_buffer
     src/tests/datastore/array_store
     src/tests/datastore/array_store_config
+    src/tests/datastore/buffer_stats
     src/tests/datastore/buffer_type
     src/tests/datastore/compact_buffer_candidates
     src/tests/datastore/datastore
diff --git a/vespalib/src/tests/datastore/buffer_stats/CMakeLists.txt b/vespalib/src/tests/datastore/buffer_stats/CMakeLists.txt
new file mode 100644
index 00000000000..2463f584133
--- /dev/null
+++ b/vespalib/src/tests/datastore/buffer_stats/CMakeLists.txt
@@ -0,0 +1,9 @@
+# Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+vespa_add_executable(vespalib_datastore_buffer_stats_test_app TEST
+    SOURCES
+    buffer_stats_test.cpp
+    DEPENDS
+    vespalib
+    GTest::GTest
+)
+vespa_add_test(NAME vespalib_datastore_buffer_stats_test_app COMMAND vespalib_datastore_buffer_stats_test_app)
diff --git a/vespalib/src/tests/datastore/buffer_stats/buffer_stats_test.cpp b/vespalib/src/tests/datastore/buffer_stats/buffer_stats_test.cpp
new file mode 100644
index 00000000000..09b2590a5f3
--- /dev/null
+++ b/vespalib/src/tests/datastore/buffer_stats/buffer_stats_test.cpp
@@ -0,0 +1,34 @@
+// Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#include <vespa/vespalib/datastore/buffer_stats.h>
+#include <vespa/vespalib/datastore/memory_stats.h>
+#include <vespa/vespalib/gtest/gtest.h>
+
+using namespace vespalib::datastore;
+
+TEST(BufferStatsTest, buffer_stats_to_memory_stats)
+{
+    InternalBufferStats buf;
+    buf.set_alloc_elems(17);
+    buf.pushed_back(7);
+    buf.set_dead_elems(5);
+    buf.set_hold_elems(3);
+    buf.inc_extra_used_bytes(13);
+    buf.inc_extra_hold_bytes(11);
+
+    MemoryStats mem;
+    constexpr size_t es = 8;
+    buf.add_to_mem_stats(es, mem);
+
+    EXPECT_EQ(17, mem._allocElems);
+    EXPECT_EQ(7, mem._usedElems);
+    EXPECT_EQ(5, mem._deadElems);
+    EXPECT_EQ(3, mem._holdElems);
+    EXPECT_EQ(17 * es + 13, mem._allocBytes);
+    EXPECT_EQ(7 * es + 13, mem._usedBytes);
+    EXPECT_EQ(5 * es, mem._deadBytes);
+    EXPECT_EQ(3 * es + 11, mem._holdBytes);
+}
+
+GTEST_MAIN_RUN_ALL_TESTS()
+
diff --git a/vespalib/src/tests/datastore/datastore/datastore_test.cpp b/vespalib/src/tests/datastore/datastore/datastore_test.cpp
index bf389e5e78e..484da33aeb2 100644
--- a/vespalib/src/tests/datastore/datastore/datastore_test.cpp
+++ b/vespalib/src/tests/datastore/datastore/datastore_test.cpp
@@ -35,9 +35,6 @@ public:
     void trimElemHoldList(generation_t usedGen) override {
         ParentType::trimElemHoldList(usedGen);
     }
-    void incDead(EntryRef ref, uint64_t dead) {
-        ParentType::incDead(ref, dead);
-    }
     void ensureBufferCapacity(size_t sizeNeeded) {
         ParentType::ensureBufferCapacity(0, sizeNeeded);
     }
@@ -429,11 +426,6 @@ TEST(DataStoreTest, require_that_memory_stats_are_calculated)
     m._usedElems++;
     assertMemStats(m, s.getMemStats());
 
-    // inc dead
-    s.incDead(r, 1);
-    m._deadElems++;
-    assertMemStats(m, s.getMemStats());
-
     // hold buffer
     s.addEntry(20);
     s.addEntry(30);
@@ -474,7 +466,7 @@ TEST(DataStoreTest, require_that_memory_stats_are_calculated)
 
     { // increase extra hold bytes
         auto prev_stats = s.getMemStats();
-        s.get_active_buffer_state().stats().inc_extra_hold_bytes(30);
+        s.get_active_buffer_state().hold_elems(0, 30);
         auto curr_stats = s.getMemStats();
         EXPECT_EQ(prev_stats._holdBytes + 30, curr_stats._holdBytes);
     }
@@ -487,7 +479,6 @@ TEST(DataStoreTest, require_that_memory_usage_is_calculated)
     s.addEntry(20);
     s.addEntry(30);
     s.addEntry(40);
-    s.incDead(r, 1);
     s.holdBuffer(r.bufferId());
     s.transferHoldLists(100);
     vespalib::MemoryUsage m = s.getMemoryUsage();
@@ -698,9 +689,9 @@ void test_free_element_to_held_buffer(bool direct, bool before_hold_buffer)
     s.holdBuffer(0); // hold last buffer
     if (!before_hold_buffer) {
         if (direct) {
-            ASSERT_DEATH({ s.freeElem(ref, 1); }, "state.isOnHold\\(\\) && was_held");
+            ASSERT_DEATH({ s.freeElem(ref, 1); }, "isOnHold\\(\\) && was_held");
         } else {
-            ASSERT_DEATH({ s.holdElem(ref, 1); }, "state.isActive\\(\\)");
+            ASSERT_DEATH({ s.holdElem(ref, 1); }, "isActive\\(\\)");
         }
     }
     s.transferHoldLists(100);
diff --git a/vespalib/src/vespa/vespalib/datastore/buffer_stats.cpp b/vespalib/src/vespa/vespalib/datastore/buffer_stats.cpp
index 0d367cf9835..8d97414626e 100644
--- a/vespalib/src/vespa/vespalib/datastore/buffer_stats.cpp
+++ b/vespalib/src/vespa/vespalib/datastore/buffer_stats.cpp
@@ -16,14 +16,6 @@ BufferStats::BufferStats()
 }
 
 void
-BufferStats::dec_hold_elems(size_t value)
-{
-    ElemCount elems = hold_elems();
-    assert(elems >= value);
-    _hold_elems.store(elems - value, std::memory_order_relaxed);
-}
-
-void
 BufferStats::add_to_mem_stats(size_t element_size, MemoryStats& stats) const
 {
     size_t extra_used = extra_used_bytes();
@@ -37,13 +29,13 @@ BufferStats::add_to_mem_stats(size_t element_size, MemoryStats& stats) const
     stats._holdBytes += (hold_elems() * element_size) + extra_hold_bytes();
 }
 
-MutableBufferStats::MutableBufferStats()
+InternalBufferStats::InternalBufferStats()
     : BufferStats()
 {
 }
 
 void
-MutableBufferStats::clear()
+InternalBufferStats::clear()
 {
     _alloc_elems.store(0, std::memory_order_relaxed);
     _used_elems.store(0, std::memory_order_relaxed);
@@ -53,5 +45,13 @@ MutableBufferStats::clear()
     _extra_hold_bytes.store(0, std::memory_order_relaxed);
 }
 
+void
+InternalBufferStats::dec_hold_elems(size_t value)
+{
+    ElemCount elems = hold_elems();
+    assert(elems >= value);
+    _hold_elems.store(elems - value, std::memory_order_relaxed);
+}
+
 }
 
diff --git a/vespalib/src/vespa/vespalib/datastore/buffer_stats.h b/vespalib/src/vespa/vespalib/datastore/buffer_stats.h
index 0df74c0a79d..66f8b532c41 100644
--- a/vespalib/src/vespa/vespalib/datastore/buffer_stats.h
+++ b/vespalib/src/vespa/vespalib/datastore/buffer_stats.h
@@ -47,11 +47,7 @@ public:
     size_t extra_used_bytes() const { return _extra_used_bytes.load(std::memory_order_relaxed); }
     size_t extra_hold_bytes() const { return _extra_hold_bytes.load(std::memory_order_relaxed); }
 
-    void inc_dead_elems(size_t value) { _dead_elems.store(dead_elems() + value, std::memory_order_relaxed); }
-    void inc_hold_elems(size_t value) { _hold_elems.store(hold_elems() + value, std::memory_order_relaxed); }
-    void dec_hold_elems(size_t value);
     void inc_extra_used_bytes(size_t value) { _extra_used_bytes.store(extra_used_bytes() + value, std::memory_order_relaxed); }
-    void inc_extra_hold_bytes(size_t value) { _extra_hold_bytes.store(extra_hold_bytes() + value, std::memory_order_relaxed); }
 
     void add_to_mem_stats(size_t element_size, MemoryStats& stats) const;
 };
@@ -59,13 +55,17 @@ public:
 /**
  * Provides low-level access to buffer stats for integration in BufferState.
  */
-class MutableBufferStats : public BufferStats {
+class InternalBufferStats : public BufferStats {
 public:
-    MutableBufferStats();
+    InternalBufferStats();
     void clear();
     void set_alloc_elems(size_t value) { _alloc_elems.store(value, std::memory_order_relaxed); }
     void set_dead_elems(size_t value) { _dead_elems.store(value, std::memory_order_relaxed); }
     void set_hold_elems(size_t value) { _hold_elems.store(value, std::memory_order_relaxed); }
+    void inc_dead_elems(size_t value) { _dead_elems.store(dead_elems() + value, std::memory_order_relaxed); }
+    void inc_hold_elems(size_t value) { _hold_elems.store(hold_elems() + value, std::memory_order_relaxed); }
+    void dec_hold_elems(size_t value);
+    void inc_extra_hold_bytes(size_t value) { _extra_hold_bytes.store(extra_hold_bytes() + value, std::memory_order_relaxed); }
     std::atomic<ElemCount>& used_elems_ref() { return _used_elems; }
     std::atomic<ElemCount>& dead_elems_ref() { return _dead_elems; }
     std::atomic<size_t>& extra_used_bytes_ref() { return _extra_used_bytes; }
diff --git a/vespalib/src/vespa/vespalib/datastore/bufferstate.cpp b/vespalib/src/vespa/vespalib/datastore/bufferstate.cpp
index 35455a193d2..e9ff243fb08 100644
--- a/vespalib/src/vespa/vespalib/datastore/bufferstate.cpp
+++ b/vespalib/src/vespa/vespalib/datastore/bufferstate.cpp
@@ -176,6 +176,39 @@ BufferState::disableElemHoldList()
     _disableElemHoldList = true;
 }
 
+bool
+BufferState::hold_elems(size_t num_elems, size_t extra_bytes)
+{
+    assert(isActive());
+    if (_disableElemHoldList) {
+        // The elements are directly marked as dead as they are not put on hold.
+        _stats.inc_dead_elems(num_elems);
+        return true;
+    }
+    _stats.inc_hold_elems(num_elems);
+    _stats.inc_extra_hold_bytes(extra_bytes);
+    return false;
+}
+
+void
+BufferState::free_elems(EntryRef ref, size_t num_elems, bool was_held, size_t ref_offset)
+{
+    if (isActive()) {
+        if (_free_list.enabled() && (num_elems == getArraySize())) {
+            _free_list.push_entry(ref);
+        }
+    } else {
+        assert(isOnHold() && was_held);
+    }
+    _stats.inc_dead_elems(num_elems);
+    if (was_held) {
+        _stats.dec_hold_elems(num_elems);
+    }
+    getTypeHandler()->cleanHold(_buffer.get(), (ref_offset * _arraySize), num_elems,
+                                BufferTypeBase::CleanContext(_stats.extra_used_bytes_ref(),
+                                                             _stats.extra_hold_bytes_ref()));
+}
+
 void
 BufferState::fallbackResize(uint32_t bufferId,
                             size_t elementsNeeded,
diff --git a/vespalib/src/vespa/vespalib/datastore/bufferstate.h b/vespalib/src/vespa/vespalib/datastore/bufferstate.h
index c35a51b0c99..0acad7d3ab1 100644
--- a/vespalib/src/vespa/vespalib/datastore/bufferstate.h
+++ b/vespalib/src/vespa/vespalib/datastore/bufferstate.h
@@ -39,7 +39,7 @@ public:
     };
 
 private:
-    MutableBufferStats _stats;
+    InternalBufferStats _stats;
     BufferFreeList _free_list;
     std::atomic<BufferTypeBase*> _typeHandler;
     Alloc           _buffer;
@@ -83,24 +83,34 @@ public:
     void onFree(std::atomic<void*>& buffer);
 
     /**
-     * Disable hold of elements, just mark then as dead without cleanup.
+     * Disable hold of elements, just mark elements as dead without cleanup.
      * Typically used when tearing down data structure in a controlled manner.
      */
     void disableElemHoldList();
 
+    /**
+     * Update stats to reflect that the given elements are put on hold.
+     * Returns true if element hold list is disabled for this buffer.
+     */
+    bool hold_elems(size_t num_elems, size_t extra_bytes);
+
+    /**
+     * Free the given elements and update stats accordingly.
+     *
+     * The given entry ref is put on the free list (if enabled).
+     * Hold cleaning of elements is executed on the buffer type.
+     */
+    void free_elems(EntryRef ref, size_t num_elems, bool was_held, size_t ref_offset);
+
     BufferStats& stats() { return _stats; }
     const BufferStats& stats() const { return _stats; }
-    BufferFreeList& free_list() { return _free_list; }
-    const BufferFreeList& free_list() const { return _free_list; }
+
+    void enable_free_list(FreeList& type_free_list) { _free_list.enable(type_free_list); }
+    void disable_free_list() { _free_list.disable(); }
 
     size_t size() const { return _stats.size(); }
     size_t capacity() const { return _stats.capacity(); }
     size_t remaining() const { return _stats.remaining(); }
-    void cleanHold(void *buffer, size_t offset, ElemCount numElems) {
-        getTypeHandler()->cleanHold(buffer, offset, numElems,
-                                    BufferTypeBase::CleanContext(_stats.extra_used_bytes_ref(),
-                                                                 _stats.extra_hold_bytes_ref()));
-    }
     void dropBuffer(uint32_t buffer_id, std::atomic<void*>& buffer);
     uint32_t getTypeId() const { return _typeId; }
     uint32_t getArraySize() const { return _arraySize; }
@@ -119,7 +129,6 @@ public:
     const BufferTypeBase *getTypeHandler() const { return _typeHandler.load(std::memory_order_relaxed); }
     BufferTypeBase *getTypeHandler() { return _typeHandler.load(std::memory_order_relaxed); }
 
-    bool hasDisabledElemHoldList() const { return _disableElemHoldList; }
     void resume_primary_buffer(uint32_t buffer_id);
 
 };
diff --git a/vespalib/src/vespa/vespalib/datastore/datastore.h b/vespalib/src/vespa/vespalib/datastore/datastore.h
index 86c39b547d3..75c77204837 100644
--- a/vespalib/src/vespa/vespalib/datastore/datastore.h
+++ b/vespalib/src/vespa/vespalib/datastore/datastore.h
@@ -38,17 +38,6 @@ public:
     ~DataStoreT() override;
 
     /**
-     * Increase number of dead elements in buffer.
-     *
-     * @param ref       Reference to dead stored features
-     * @param dead      Number of newly dead elements
-     */
-    void incDead(EntryRef ref, size_t deadElems) {
-        RefType intRef(ref);
-        DataStoreBase::incDead(intRef.bufferId(), deadElems);
-    }
-
-    /**
      * Free element(s).
      */
     void freeElem(EntryRef ref, size_t numElems) { free_elem_internal(ref, numElems, false); }
diff --git a/vespalib/src/vespa/vespalib/datastore/datastore.hpp b/vespalib/src/vespa/vespalib/datastore/datastore.hpp
index 90f7507f80f..e14b248e569 100644
--- a/vespalib/src/vespa/vespalib/datastore/datastore.hpp
+++ b/vespalib/src/vespa/vespalib/datastore/datastore.hpp
@@ -26,19 +26,7 @@ DataStoreT<RefT>::free_elem_internal(EntryRef ref, size_t numElems, bool was_hel
 {
     RefType intRef(ref);
     BufferState &state = getBufferState(intRef.bufferId());
-    if (state.isActive()) {
-        if (state.free_list().enabled() && (numElems == state.getArraySize())) {
-            state.free_list().push_entry(ref);
-        }
-    } else {
-        assert(state.isOnHold() && was_held);
-    }
-    state.stats().inc_dead_elems(numElems);
-    if (was_held) {
-        state.stats().dec_hold_elems(numElems);
-    }
-    state.cleanHold(getBuffer(intRef.bufferId()),
-                    intRef.offset() * state.getArraySize(), numElems);
+    state.free_elems(ref, numElems, was_held, intRef.offset());
 }
 
 template <typename RefT>
@@ -47,14 +35,9 @@ DataStoreT<RefT>::holdElem(EntryRef ref, size_t numElems, size_t extraBytes)
 {
     RefType intRef(ref);
     BufferState &state = getBufferState(intRef.bufferId());
-    assert(state.isActive());
-    if (state.hasDisabledElemHoldList()) {
-        state.stats().inc_dead_elems(numElems);
-        return;
+    if (!state.hold_elems(numElems, extraBytes)) {
+        _elemHold1List.push_back(ElemHold1ListElem(ref, numElems));
     }
-    _elemHold1List.push_back(ElemHold1ListElem(ref, numElems));
-    state.stats().inc_hold_elems(numElems);
-    state.stats().inc_extra_hold_bytes(extraBytes);
 }
 
 template <typename RefT>
diff --git a/vespalib/src/vespa/vespalib/datastore/datastorebase.cpp b/vespalib/src/vespa/vespalib/datastore/datastorebase.cpp
index 302a1b49219..a5ed2c05f22 100644
--- a/vespalib/src/vespa/vespalib/datastore/datastorebase.cpp
+++ b/vespalib/src/vespa/vespalib/datastore/datastorebase.cpp
@@ -300,7 +300,7 @@ DataStoreBase::enableFreeLists()
         if (!bState.isActive() || bState.getCompacting()) {
             continue;
         }
-        bState.free_list().enable(_free_lists[bState.getTypeId()]);
+        bState.enable_free_list(_free_lists[bState.getTypeId()]);
     }
     _freeListsEnabled = true;
 }
@@ -309,7 +309,7 @@ void
 DataStoreBase::disableFreeLists()
 {
     for (BufferState & bState : _states) {
-        bState.free_list().disable();
+        bState.disable_free_list();
     }
     _freeListsEnabled = false;
 }
@@ -321,17 +321,11 @@ DataStoreBase::enableFreeList(uint32_t bufferId)
     if (_freeListsEnabled &&
         state.isActive() &&
         !state.getCompacting()) {
-        state.free_list().enable(_free_lists[state.getTypeId()]);
+        state.enable_free_list(_free_lists[state.getTypeId()]);
     }
 }
 
 void
-DataStoreBase::disableFreeList(uint32_t bufferId)
-{
-    _states[bufferId].free_list().disable();
-}
-
-void
 DataStoreBase::disableElemHoldList()
 {
     for (auto &state : _states) {
@@ -452,7 +446,7 @@ DataStoreBase::markCompacting(uint32_t bufferId)
     assert(!state.getCompacting());
     state.setCompacting();
     state.disableElemHoldList();
-    state.free_list().disable();
+    state.disable_free_list();
     inc_compaction_count();
 }
 
diff --git a/vespalib/src/vespa/vespalib/datastore/datastorebase.h b/vespalib/src/vespa/vespalib/datastore/datastorebase.h
index 4038e12efee..712a171f51c 100644
--- a/vespalib/src/vespa/vespalib/datastore/datastorebase.h
+++ b/vespalib/src/vespa/vespalib/datastore/datastorebase.h
@@ -253,12 +253,6 @@ public:
 
     void dropBuffers();
 
-
-    void incDead(uint32_t bufferId, size_t deadElems) {
-        BufferState &state = _states[bufferId];
-        state.stats().inc_dead_elems(deadElems);
-    }
-
     /**
      * Enable free list management.
      * This only works for fixed size elements.