Use atomic counters in vespalib::datastore::BufferState.

6 files changed, 124 insertions, 109 deletions

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 a31dfffc25b..6988e41add1 100644
--- a/vespalib/src/tests/datastore/buffer_type/buffer_type_test.cpp
+++ b/vespalib/src/tests/datastore/buffer_type/buffer_type_test.cpp
@@ -12,9 +12,9 @@ constexpr uint32_t NUM_ARRAYS_FOR_NEW_BUFFER(0);
 
 struct Setup {
     uint32_t  _minArrays;
-    ElemCount _usedElems;
+    std::atomic<ElemCount> _usedElems;
     ElemCount _neededElems;
-    ElemCount _deadElems;
+    std::atomic<ElemCount> _deadElems;
     uint32_t  _bufferId;
     float     _allocGrowFactor;
     bool      _resizing;
@@ -27,6 +27,7 @@ struct Setup {
           _allocGrowFactor(0.5),
           _resizing(false)
     {}
+    Setup(const Setup& rhs);
     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; }
@@ -35,6 +36,17 @@ struct Setup {
     Setup &resizing(bool value) { _resizing = value; return *this; }
 };
 
+Setup::Setup(const Setup& rhs)
+    : _minArrays(rhs._minArrays),
+      _usedElems(rhs._usedElems.load(std::memory_order_relaxed)),
+      _neededElems(rhs._neededElems),
+      _deadElems(rhs._deadElems.load(std::memory_order_relaxed)),
+      _bufferId(rhs._bufferId),
+      _allocGrowFactor(rhs._allocGrowFactor),
+      _resizing(rhs._resizing)
+{
+}
+
 struct Fixture {
     std::vector<Setup> setups;
     IntBufferType bufferType;
diff --git a/vespalib/src/vespa/vespalib/datastore/buffer_type.cpp b/vespalib/src/vespa/vespalib/datastore/buffer_type.cpp
index ca908d48210..dec24d6fe46 100644
--- a/vespalib/src/vespa/vespalib/datastore/buffer_type.cpp
+++ b/vespalib/src/vespa/vespalib/datastore/buffer_type.cpp
@@ -17,10 +17,12 @@ constexpr float DEFAULT_ALLOC_GROW_FACTOR = 0.2;
 void
 BufferTypeBase::CleanContext::extraBytesCleaned(size_t value)
 {
-    assert(_extraUsedBytes >= value);
-    assert(_extraHoldBytes >= value);
-    _extraUsedBytes -= value;
-    _extraHoldBytes -= value;
+    size_t extra_used_bytes = _extraUsedBytes.load(std::memory_order_relaxed);
+    size_t extra_hold_bytes = _extraHoldBytes.load(std::memory_order_relaxed);
+    assert(extra_used_bytes >= value);
+    assert(extra_hold_bytes >= value);
+    _extraUsedBytes.store(extra_used_bytes - value, std::memory_order_relaxed);
+    _extraHoldBytes.store(extra_hold_bytes - value, std::memory_order_relaxed);
 }
 
 BufferTypeBase::BufferTypeBase(uint32_t arraySize,
@@ -62,7 +64,7 @@ BufferTypeBase::getReservedElements(uint32_t bufferId) const
 }
 
 void
-BufferTypeBase::onActive(uint32_t bufferId, ElemCount* usedElems, ElemCount* deadElems, void* buffer)
+BufferTypeBase::onActive(uint32_t bufferId, std::atomic<ElemCount>* usedElems, std::atomic<ElemCount>* deadElems, void* buffer)
 {
     _aggr_counts.add_buffer(usedElems, deadElems);
     assert(std::find(_active_buffers.begin(), _active_buffers.end(), bufferId) == _active_buffers.end());
@@ -76,7 +78,7 @@ BufferTypeBase::onActive(uint32_t bufferId, ElemCount* usedElems, ElemCount* dea
 }
 
 void
-BufferTypeBase::onHold(uint32_t buffer_id, const ElemCount* usedElems, const ElemCount* deadElems)
+BufferTypeBase::onHold(uint32_t buffer_id, const std::atomic<ElemCount>* usedElems, const std::atomic<ElemCount>* deadElems)
 {
     ++_holdBuffers;
     auto itr = std::find(_active_buffers.begin(), _active_buffers.end(), buffer_id);
@@ -95,7 +97,7 @@ BufferTypeBase::onFree(ElemCount usedElems)
 }
 
 void
-BufferTypeBase::resume_primary_buffer(uint32_t buffer_id, ElemCount* used_elems, ElemCount* dead_elems)
+BufferTypeBase::resume_primary_buffer(uint32_t buffer_id, std::atomic<ElemCount>* used_elems, std::atomic<ElemCount>* dead_elems)
 {
     auto itr = std::find(_active_buffers.begin(), _active_buffers.end(), buffer_id);
     assert(itr != _active_buffers.end());
@@ -174,7 +176,7 @@ BufferTypeBase::AggregatedBufferCounts::AggregatedBufferCounts()
 }
 
 void
-BufferTypeBase::AggregatedBufferCounts::add_buffer(const ElemCount* used_elems, const ElemCount* dead_elems)
+BufferTypeBase::AggregatedBufferCounts::add_buffer(const std::atomic<ElemCount>* used_elems, const std::atomic<ElemCount>* dead_elems)
 {
     for (const auto& elem : _counts) {
         assert(elem.used_ptr != used_elems);
@@ -184,7 +186,7 @@ BufferTypeBase::AggregatedBufferCounts::add_buffer(const ElemCount* used_elems,
 }
 
 void
-BufferTypeBase::AggregatedBufferCounts::remove_buffer(const ElemCount* used_elems, const ElemCount* dead_elems)
+BufferTypeBase::AggregatedBufferCounts::remove_buffer(const std::atomic<ElemCount>* used_elems, const std::atomic<ElemCount>* dead_elems)
 {
     auto itr = std::find_if(_counts.begin(), _counts.end(),
                             [=](const auto& elem){ return elem.used_ptr == used_elems; });
@@ -199,8 +201,8 @@ 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;
+    result.used_elems += last.used_ptr->load(std::memory_order_relaxed);
+    result.dead_elems += last.dead_ptr->load(std::memory_order_relaxed);
     return result;
 }
 
@@ -209,8 +211,8 @@ BufferTypeBase::AggregatedBufferCounts::all_buffers() const
 {
     BufferCounts result;
     for (const auto& elem : _counts) {
-        result.used_elems += *elem.used_ptr;
-        result.dead_elems += *elem.dead_ptr;
+        result.used_elems += elem.used_ptr->load(std::memory_order_relaxed);
+        result.dead_elems += elem.dead_ptr->load(std::memory_order_relaxed);
     }
     return result;
 }
diff --git a/vespalib/src/vespa/vespalib/datastore/buffer_type.h b/vespalib/src/vespa/vespalib/datastore/buffer_type.h
index 3394af18b04..53436df432f 100644
--- a/vespalib/src/vespa/vespalib/datastore/buffer_type.h
+++ b/vespalib/src/vespa/vespalib/datastore/buffer_type.h
@@ -25,10 +25,10 @@ public:
     using ElemCount = vespalib::datastore::ElemCount;
     class CleanContext {
     private:
-        size_t &_extraUsedBytes;
-        size_t &_extraHoldBytes;
+        std::atomic<size_t> &_extraUsedBytes;
+        std::atomic<size_t> &_extraHoldBytes;
     public:
-        CleanContext(size_t &extraUsedBytes, size_t &extraHoldBytes)
+        CleanContext(std::atomic<size_t>& extraUsedBytes, std::atomic<size_t>& extraHoldBytes)
             : _extraUsedBytes(extraUsedBytes),
               _extraHoldBytes(extraHoldBytes)
         {}
@@ -60,10 +60,10 @@ 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; }
-    virtual void onActive(uint32_t bufferId, ElemCount* usedElems, ElemCount* deadElems, void* buffer);
-    void onHold(uint32_t buffer_id, const ElemCount* usedElems, const ElemCount* deadElems);
+    virtual void onActive(uint32_t bufferId, std::atomic<ElemCount>* usedElems, std::atomic<ElemCount>* deadElems, void* buffer);
+    void onHold(uint32_t buffer_id, const std::atomic<ElemCount>* usedElems, const std::atomic<ElemCount>* deadElems);
     virtual void onFree(ElemCount usedElems);
-    void resume_primary_buffer(uint32_t buffer_id, ElemCount* used_elems, ElemCount* dead_elems);
+    void resume_primary_buffer(uint32_t buffer_id, std::atomic<ElemCount>* used_elems, std::atomic<ElemCount>* dead_elems);
     virtual const alloc::MemoryAllocator* get_memory_allocator() const;
 
     /**
@@ -95,10 +95,10 @@ protected:
     class AggregatedBufferCounts {
     private:
         struct Element {
-            const ElemCount* used_ptr;
-            const ElemCount* dead_ptr;
+            const std::atomic<ElemCount>* used_ptr;
+            const std::atomic<ElemCount>* dead_ptr;
             Element() noexcept : used_ptr(nullptr), dead_ptr(nullptr) {}
-            Element(const ElemCount* used_ptr_in, const ElemCount* dead_ptr_in) noexcept
+            Element(const std::atomic<ElemCount>* used_ptr_in, const std::atomic<ElemCount>* dead_ptr_in) noexcept
                     : used_ptr(used_ptr_in), dead_ptr(dead_ptr_in)
             {}
         };
@@ -106,8 +106,8 @@ protected:
 
     public:
         AggregatedBufferCounts();
-        void add_buffer(const ElemCount* used_elems, const ElemCount* dead_elems);
-        void remove_buffer(const ElemCount* used_elems, const ElemCount* dead_elems);
+        void add_buffer(const std::atomic<ElemCount>* used_elems, const std::atomic<ElemCount>* dead_elems);
+        void remove_buffer(const std::atomic<ElemCount>* used_elems, const std::atomic<ElemCount>* dead_elems);
         BufferCounts last_buffer() const;
         BufferCounts all_buffers() const;
         bool empty() const { return _counts.empty(); }
diff --git a/vespalib/src/vespa/vespalib/datastore/bufferstate.cpp b/vespalib/src/vespa/vespalib/datastore/bufferstate.cpp
index e0104c8bb71..cde891b477d 100644
--- a/vespalib/src/vespa/vespalib/datastore/bufferstate.cpp
+++ b/vespalib/src/vespa/vespalib/datastore/bufferstate.cpp
@@ -30,7 +30,7 @@ BufferState::BufferState()
       _buffer(Alloc::alloc(0, MemoryAllocator::HUGEPAGE_SIZE)),
       _arraySize(0),
       _typeId(0),
-      _state(FREE),
+      _state(State::FREE),
       _disableElemHoldList(false),
       _compacting(false)
 {
@@ -38,7 +38,7 @@ BufferState::BufferState()
 
 BufferState::~BufferState()
 {
-    assert(_state == FREE);
+    assert(getState() == State::FREE);
     assert(_freeListList == nullptr);
     assert(_nextHasFree == nullptr);
     assert(_prevHasFree == nullptr);
@@ -48,8 +48,9 @@ BufferState::~BufferState()
 
 void
 BufferState::decHoldElems(size_t value) {
-    assert(_holdElems >= value);
-    _holdElems -= value;
+    ElemCount hold_elems = getHoldElems();
+    assert(hold_elems >= value);
+    _holdElems.store(hold_elems - value, std::memory_order_relaxed);
 }
 
 namespace {
@@ -105,14 +106,14 @@ BufferState::onActive(uint32_t bufferId, uint32_t typeId,
 {
     assert(buffer.load(std::memory_order_relaxed) == nullptr);
     assert(_buffer.get() == nullptr);
-    assert(_state == FREE);
+    assert(getState() == State::FREE);
     assert(_typeHandler == nullptr);
-    assert(_allocElems == 0);
-    assert(_usedElems == 0);
-    assert(_deadElems == 0u);
-    assert(_holdElems == 0);
-    assert(_extraUsedBytes == 0);
-    assert(_extraHoldBytes == 0);
+    assert(capacity() == 0);
+    assert(size() == 0);
+    assert(getDeadElems() == 0u);
+    assert(getHoldElems() == 0);
+    assert(getExtraUsedBytes() == 0);
+    assert(getExtraHoldBytes() == 0);
     assert(isFreeListEmpty());
     assert(_nextHasFree == nullptr);
     assert(_prevHasFree == nullptr);
@@ -127,12 +128,12 @@ BufferState::onActive(uint32_t bufferId, uint32_t typeId,
     _buffer.create(alloc.bytes).swap(_buffer);
     assert(_buffer.get() != nullptr || alloc.elements == 0u);
     buffer.store(_buffer.get(), std::memory_order_release);
-    _allocElems = alloc.elements;
-    _state = ACTIVE;
-    _typeHandler = typeHandler;
+    _allocElems.store(alloc.elements, std::memory_order_relaxed);
+    _typeHandler.store(typeHandler, std::memory_order_release);
     assert(typeId <= std::numeric_limits<uint16_t>::max());
     _typeId = typeId;
-    _arraySize = _typeHandler->getArraySize();
+    _arraySize = typeHandler->getArraySize();
+    _state.store(State::ACTIVE, std::memory_order_release);
     typeHandler->onActive(bufferId, &_usedElems, &_deadElems, buffer.load(std::memory_order::relaxed));
 }
 
@@ -140,15 +141,15 @@ BufferState::onActive(uint32_t bufferId, uint32_t typeId,
 void
 BufferState::onHold(uint32_t buffer_id)
 {
-    assert(_state == ACTIVE);
-    assert(_typeHandler != nullptr);
-    _state = HOLD;
+    assert(getState() == State::ACTIVE);
+    assert(getTypeHandler() != nullptr);
+    _state.store(State::HOLD, std::memory_order_release);
     _compacting = false;
-    assert(_deadElems <= _usedElems);
-    assert(_holdElems <= (_usedElems - _deadElems));
-    _deadElems = 0;
-    _holdElems = _usedElems; // Put everyting on hold
-    _typeHandler->onHold(buffer_id, &_usedElems, &_deadElems);
+    assert(getDeadElems() <= size());
+    assert(getHoldElems() <= (size() - getDeadElems()));
+    _deadElems.store(0, std::memory_order_relaxed);
+    _holdElems.store(size(), std::memory_order_relaxed); // Put everyting on hold
+    getTypeHandler()->onHold(buffer_id, &_usedElems, &_deadElems);
     if ( ! isFreeListEmpty()) {
         removeFromFreeListList();
         FreeList().swap(_freeList);
@@ -164,21 +165,21 @@ void
 BufferState::onFree(std::atomic<void*>& buffer)
 {
     assert(buffer.load(std::memory_order_relaxed) == _buffer.get());
-    assert(_state == HOLD);
+    assert(getState() == State::HOLD);
     assert(_typeHandler != nullptr);
-    assert(_deadElems <= _usedElems);
-    assert(_holdElems == _usedElems - _deadElems);
-    _typeHandler->destroyElements(buffer, _usedElems);
+    assert(getDeadElems() <= size());
+    assert(getHoldElems() == size() - getDeadElems());
+    getTypeHandler()->destroyElements(buffer, size());
     Alloc::alloc().swap(_buffer);
-    _typeHandler->onFree(_usedElems);
+    getTypeHandler()->onFree(size());
     buffer.store(nullptr, std::memory_order_release);
-    _usedElems = 0;
-    _allocElems = 0;
-    _deadElems = 0u;
-    _holdElems = 0u;
-    _extraUsedBytes = 0;
-    _extraHoldBytes = 0;
-    _state = FREE;
+    _usedElems.store(0, std::memory_order_relaxed);
+    _allocElems.store(0, std::memory_order_relaxed);
+    _deadElems.store(0, std::memory_order_relaxed);
+    _holdElems.store(0, std::memory_order_relaxed);
+    _extraUsedBytes.store(0, std::memory_order_relaxed);
+    _extraHoldBytes.store(0, std::memory_order_relaxed);
+    _state.store(State::FREE, std::memory_order_release);
     _typeHandler = nullptr;
     _arraySize = 0;
     assert(isFreeListEmpty());
@@ -193,18 +194,18 @@ BufferState::onFree(std::atomic<void*>& buffer)
 void
 BufferState::dropBuffer(uint32_t buffer_id, std::atomic<void*>& buffer)
 {
-    if (_state == FREE) {
+    if (getState() == State::FREE) {
         assert(buffer.load(std::memory_order_relaxed) == nullptr);
         return;
     }
-    assert(buffer.load(std::memory_order_relaxed) != nullptr || _allocElems == 0);
-    if (_state == ACTIVE) {
+    assert(buffer.load(std::memory_order_relaxed) != nullptr || capacity() == 0);
+    if (getState() == State::ACTIVE) {
         onHold(buffer_id);
     }
-    if (_state == HOLD) {
+    if (getState() == State::HOLD) {
         onFree(buffer);
     }
-    assert(_state == FREE);
+    assert(getState() == State::FREE);
     assert(buffer.load(std::memory_order_relaxed) == nullptr);
 }
 
@@ -212,7 +213,7 @@ BufferState::dropBuffer(uint32_t buffer_id, std::atomic<void*>& buffer)
 void
 BufferState::setFreeListList(FreeListList *freeListList)
 {
-    if (_state == FREE && freeListList != nullptr) {
+    if (getState() == State::FREE && freeListList != nullptr) {
         return;
     }
     if (freeListList == _freeListList) {
@@ -285,25 +286,25 @@ BufferState::fallbackResize(uint32_t bufferId,
                             std::atomic<void*>& buffer,
                             Alloc &holdBuffer)
 {
-    assert(_state == ACTIVE);
+    assert(getState() == State::ACTIVE);
     assert(_typeHandler != nullptr);
     assert(holdBuffer.get() == nullptr);
     AllocResult alloc = calcAllocation(bufferId, *_typeHandler, elementsNeeded, true);
-    assert(alloc.elements >= _usedElems + elementsNeeded);
-    assert(alloc.elements > _allocElems);
+    assert(alloc.elements >= size() + elementsNeeded);
+    assert(alloc.elements > capacity());
     Alloc newBuffer = _buffer.create(alloc.bytes);
-    _typeHandler->fallbackCopy(newBuffer.get(), buffer.load(std::memory_order_relaxed), _usedElems);
+    getTypeHandler()->fallbackCopy(newBuffer.get(), buffer.load(std::memory_order_relaxed), size());
     holdBuffer.swap(_buffer);
     std::atomic_thread_fence(std::memory_order_release);
     _buffer = std::move(newBuffer);
     buffer.store(_buffer.get(), std::memory_order_release);
-    _allocElems = alloc.elements;
+    _allocElems.store(alloc.elements, std::memory_order_relaxed);
 }
 
 void
 BufferState::resume_primary_buffer(uint32_t buffer_id)
 {
-    _typeHandler->resume_primary_buffer(buffer_id, &_usedElems, &_deadElems);
+    getTypeHandler()->resume_primary_buffer(buffer_id, &_usedElems, &_deadElems);
 }
 
 }
diff --git a/vespalib/src/vespa/vespalib/datastore/bufferstate.h b/vespalib/src/vespa/vespalib/datastore/bufferstate.h
index 7862f58cfe1..639f4f9dfc6 100644
--- a/vespalib/src/vespa/vespalib/datastore/bufferstate.h
+++ b/vespalib/src/vespa/vespalib/datastore/bufferstate.h
@@ -41,23 +41,23 @@ public:
 
     using FreeList = vespalib::Array<EntryRef>;
 
-    enum State : uint8_t {
+    enum class State : uint8_t {
         FREE,
         ACTIVE,
         HOLD
     };
 
 private:
-    ElemCount     _usedElems;
-    ElemCount     _allocElems;
-    ElemCount     _deadElems;
-    ElemCount     _holdElems;
+    std::atomic<ElemCount>     _usedElems;
+    std::atomic<ElemCount>     _allocElems;
+    std::atomic<ElemCount>     _deadElems;
+    std::atomic<ElemCount>     _holdElems;
     // Number of bytes that are heap allocated by elements that are stored in this buffer.
     // For simple types this is 0.
-    size_t        _extraUsedBytes;
+    std::atomic<size_t>        _extraUsedBytes;
     // Number of bytes that are heap allocated by elements that are stored in this buffer and is now on hold.
     // For simple types this is 0.
-    size_t        _extraHoldBytes;
+    std::atomic<size_t>        _extraHoldBytes;
     FreeList      _freeList;
     FreeListList *_freeListList;    // non-nullptr if free lists are enabled
 
@@ -65,11 +65,11 @@ private:
     BufferState    *_nextHasFree;
     BufferState    *_prevHasFree;
 
-    BufferTypeBase *_typeHandler;
+    std::atomic<BufferTypeBase*> _typeHandler;
     Alloc           _buffer;
     uint32_t        _arraySize;
     uint16_t        _typeId;
-    State           _state : 8;
+    std::atomic<State> _state;
     bool            _disableElemHoldList : 1;
     bool            _compacting : 1;
 
@@ -140,51 +140,51 @@ public:
         if (isFreeListEmpty()) {
             removeFromFreeListList();
         }
-        _deadElems -= _arraySize;
+        _deadElems.store(getDeadElems() - _arraySize, std::memory_order_relaxed);
         return ret;
     }
 
-    size_t size() const { return _usedElems; }
-    size_t capacity() const { return _allocElems; }
-    size_t remaining() const { return _allocElems - _usedElems; }
+    size_t size() const { return _usedElems.load(std::memory_order_relaxed); }
+    size_t capacity() const { return _allocElems.load(std::memory_order_relaxed); }
+    size_t remaining() const { return capacity() - size(); }
     void pushed_back(size_t numElems) {
         pushed_back(numElems, 0);
     }
     void pushed_back(size_t numElems, size_t extraBytes) {
-        _usedElems += numElems;
-        _extraUsedBytes += extraBytes;
+        _usedElems.store(size() + numElems, std::memory_order_relaxed);
+        _extraUsedBytes.store(getExtraUsedBytes() + extraBytes, std::memory_order_relaxed);
     }
     void cleanHold(void *buffer, size_t offset, ElemCount numElems) {
-        _typeHandler->cleanHold(buffer, offset, numElems, BufferTypeBase::CleanContext(_extraUsedBytes, _extraHoldBytes));
+        getTypeHandler()->cleanHold(buffer, offset, numElems, BufferTypeBase::CleanContext(_extraUsedBytes, _extraHoldBytes));
     }
     void dropBuffer(uint32_t buffer_id, std::atomic<void*>& buffer);
     uint32_t getTypeId() const { return _typeId; }
     uint32_t getArraySize() const { return _arraySize; }
-    size_t getDeadElems() const { return _deadElems; }
-    size_t getHoldElems() const { return _holdElems; }
-    size_t getExtraUsedBytes() const { return _extraUsedBytes; }
-    size_t getExtraHoldBytes() const { return _extraHoldBytes; }
+    size_t getDeadElems() const { return _deadElems.load(std::memory_order_relaxed); }
+    size_t getHoldElems() const { return _holdElems.load(std::memory_order_relaxed); }
+    size_t getExtraUsedBytes() const { return _extraUsedBytes.load(std::memory_order_relaxed); }
+    size_t getExtraHoldBytes() const { return _extraHoldBytes.load(std::memory_order_relaxed); }
     bool getCompacting() const { return _compacting; }
     void setCompacting() { _compacting = true; }
-    uint32_t get_used_arrays() const noexcept { return _usedElems / _arraySize; }
+    uint32_t get_used_arrays() const noexcept { return size() / _arraySize; }
     void fallbackResize(uint32_t bufferId, size_t elementsNeeded, std::atomic<void*>& buffer, Alloc &holdBuffer);
 
     bool isActive(uint32_t typeId) const {
-        return ((_state == ACTIVE) && (_typeId == typeId));
+        return (isActive() && (_typeId == typeId));
     }
-    bool isActive() const { return (_state == ACTIVE); }
-    bool isOnHold() const { return (_state == HOLD); }
-    bool isFree() const { return (_state == FREE); }
-    State getState() const { return _state; }
-    const BufferTypeBase *getTypeHandler() const { return _typeHandler; }
-    BufferTypeBase *getTypeHandler() { return _typeHandler; }
-
-    void incDeadElems(size_t value) { _deadElems += value; }
-    void incHoldElems(size_t value) { _holdElems += value; }
+    bool isActive() const { return (getState() == State::ACTIVE); }
+    bool isOnHold() const { return (getState() == State::HOLD); }
+    bool isFree() const { return (getState() == State::FREE); }
+    State getState() const { return _state.load(std::memory_order_relaxed); }
+    const BufferTypeBase *getTypeHandler() const { return _typeHandler.load(std::memory_order_relaxed); }
+    BufferTypeBase *getTypeHandler() { return _typeHandler.load(std::memory_order_relaxed); }
+
+    void incDeadElems(size_t value) { _deadElems.store(getDeadElems() + value, std::memory_order_relaxed); }
+    void incHoldElems(size_t value) { _holdElems.store(getHoldElems() + value, std::memory_order_relaxed); }
     void decHoldElems(size_t value);
-    void incExtraUsedBytes(size_t value) { _extraUsedBytes += value; }
+    void incExtraUsedBytes(size_t value) { _extraUsedBytes.store(getExtraUsedBytes() + value, std::memory_order_relaxed); }
     void incExtraHoldBytes(size_t value) {
-        _extraHoldBytes += value;
+        _extraHoldBytes.store(getExtraHoldBytes() + value, std::memory_order_relaxed);
     }
 
     bool hasDisabledElemHoldList() const { return _disableElemHoldList; }
diff --git a/vespalib/src/vespa/vespalib/datastore/datastorebase.cpp b/vespalib/src/vespa/vespalib/datastore/datastorebase.cpp
index 43e47296029..cf5406bf343 100644
--- a/vespalib/src/vespa/vespalib/datastore/datastorebase.cpp
+++ b/vespalib/src/vespa/vespalib/datastore/datastorebase.cpp
@@ -365,13 +365,13 @@ DataStoreBase::getMemStats() const
     for (const BufferState & bState: _states) {
         auto typeHandler = bState.getTypeHandler();
         BufferState::State state = bState.getState();
-        if ((state == BufferState::FREE) || (typeHandler == nullptr)) {
+        if ((state == BufferState::State::FREE) || (typeHandler == nullptr)) {
             ++stats._freeBuffers;
-        } else if (state == BufferState::ACTIVE) {
+        } else if (state == BufferState::State::ACTIVE) {
             size_t elementSize = typeHandler->elementSize();
             ++stats._activeBuffers;
             add_buffer_state_to_mem_stats(bState, elementSize, stats);
-        } else if (state == BufferState::HOLD) {
+        } else if (state == BufferState::State::HOLD) {
             size_t elementSize = typeHandler->elementSize();
             ++stats._holdBuffers;
             add_buffer_state_to_mem_stats(bState, elementSize, stats);