Readers of a btree use atomic read with acquire memory ordering when getting

btree internal node child ref.

Writer of a btree uses atomic write with release memory ordering when
setting btree internal node child ref in node that is visible to readers
(e.g. when moving child node as part of compacting btree nodes).

1 files changed, 29 insertions, 19 deletions

diff --git a/vespalib/src/vespa/vespalib/btree/btreenode.h b/vespalib/src/vespa/vespalib/btree/btreenode.h
index deeaad40ae6..e457af73654 100644
--- a/vespalib/src/vespa/vespalib/btree/btreenode.h
+++ b/vespalib/src/vespa/vespalib/btree/btreenode.h
@@ -5,7 +5,7 @@
 #include "noaggregated.h"
 #include "minmaxaggregated.h"
 #include "btree_key_data.h"
-#include <vespa/vespalib/datastore/entryref.h>
+#include <vespa/vespalib/datastore/atomic_entry_ref.h>
 #include <vespa/vespalib/datastore/handle.h>
 #include <cassert>
 #include <utility>
@@ -64,6 +64,7 @@ protected:
 
 public:
     using Ref = datastore::EntryRef;
+    using ChildRef = datastore::AtomicEntryRef;
 
     bool isLeaf() const { return _level == 0u; }
     bool getFrozen() const { return _isFrozen; }
@@ -281,10 +282,10 @@ public:
 };
 
 template <typename KeyT, typename AggrT, uint32_t NumSlots = 16>
-class BTreeInternalNode : public BTreeNodeTT<KeyT, BTreeNode::Ref, AggrT, NumSlots>
+class BTreeInternalNode : public BTreeNodeTT<KeyT, BTreeNode::ChildRef, AggrT, NumSlots>
 {
 public:
-    typedef BTreeNodeTT<KeyT, BTreeNode::Ref, AggrT, NumSlots> ParentType;
+    typedef BTreeNodeTT<KeyT, BTreeNode::ChildRef, AggrT, NumSlots> ParentType;
     typedef BTreeInternalNode<KeyT, AggrT, NumSlots> InternalNodeType;
     template <typename, typename, typename, size_t, size_t>
     friend class BTreeNodeAllocator;
@@ -301,12 +302,15 @@ public:
     typedef BTreeNode::Ref Ref;
     typedef datastore::Handle<InternalNodeType> RefPair;
     using ParentType::_keys;
+    using ParentType::_data;
     using ParentType::validSlots;
     using ParentType::_validSlots;
     using ParentType::getFrozen;
     using ParentType::getData;
+    using ParentType::insert;
     using ParentType::setData;
     using ParentType::setLevel;
+    using ParentType::update;
     using ParentType::EMPTY_LEVEL;
     typedef KeyT KeyType;
     typedef Ref DataType;
@@ -335,11 +339,17 @@ private:
     uint32_t countValidLeaves(uint32_t start, uint32_t end, NodeAllocatorType &allocator);
 
 public:
-    BTreeNode::Ref getChild(uint32_t idx) const { return getData(idx); }
-    BTreeNode::Ref get_child_relaxed(uint32_t idx) const { return getData(idx); }
-    void setChild(uint32_t idx, BTreeNode::Ref child) { setData(idx, child); }
-    void set_child_relaxed(uint32_t idx, BTreeNode::Ref child) { setData(idx, child); }
+    BTreeNode::Ref getChild(uint32_t idx) const { return _data[idx].load_acquire(); }
+    BTreeNode::Ref get_child_relaxed(uint32_t idx) const { return _data[idx].load_relaxed(); }
+    void setChild(uint32_t idx, BTreeNode::Ref child) { _data[idx].store_release(child); }
+    void set_child_relaxed(uint32_t idx, BTreeNode::Ref child) { _data[idx].store_relaxed(child); }
     BTreeNode::Ref get_last_child_relaxed() const { return get_child_relaxed(validSlots() - 1); }
+    void update(uint32_t idx, const KeyT & key, BTreeNode::Ref child) {
+        update(idx, key, BTreeNode::ChildRef(child));
+    }
+    void insert(uint32_t idx, const KeyT & key, BTreeNode::Ref child) {
+        insert(idx, key, BTreeNode::ChildRef(child));
+    }
     uint32_t validLeaves() const { return _validLeaves; }
     void setValidLeaves(uint32_t newValidLeaves) { _validLeaves = newValidLeaves; }
     void incValidLeaves(uint32_t delta) { _validLeaves += delta; }
@@ -363,15 +373,15 @@ public:
 
     template <typename NodeStoreType, typename FunctionType>
     void foreach_key(NodeStoreType &store, FunctionType func) const {
-        const BTreeNode::Ref *it = this->_data;
-        const BTreeNode::Ref *ite = it + _validSlots;
+        const BTreeNode::ChildRef *it = this->_data;
+        const BTreeNode::ChildRef *ite = it + _validSlots;
         if (this->getLevel() > 1u) {
             for (; it != ite; ++it) {
-                store.mapInternalRef(*it)->foreach_key(store, func);
+                store.mapInternalRef(it->load_acquire())->foreach_key(store, func);
             }
         } else {
             for (; it != ite; ++it) {
-                store.mapLeafRef(*it)->foreach_key(func);
+                store.mapLeafRef(it->load_acquire())->foreach_key(func);
             }
         }
     }
@@ -382,31 +392,31 @@ public:
      */
     template <typename NodeStoreType, typename FunctionType>
     void foreach_key_range(NodeStoreType &store, uint32_t start_idx, uint32_t end_idx, FunctionType func) const {
-        const BTreeNode::Ref *it = this->_data;
-        const BTreeNode::Ref *ite = it + end_idx;
+        const BTreeNode::ChildRef *it = this->_data;
+        const BTreeNode::ChildRef *ite = it + end_idx;
         it += start_idx;
         if (this->getLevel() > 1u) {
             for (; it != ite; ++it) {
-                store.mapInternalRef(*it)->foreach_key(store, func);
+                store.mapInternalRef(it->load_acquire())->foreach_key(store, func);
             }
         } else {
             for (; it != ite; ++it) {
-                store.mapLeafRef(*it)->foreach_key(func);
+                store.mapLeafRef(it->load_acquire())->foreach_key(func);
             }
         }
     }
 
     template <typename NodeStoreType, typename FunctionType>
     void foreach(NodeStoreType &store, FunctionType func) const {
-        const BTreeNode::Ref *it = this->_data;
-        const BTreeNode::Ref *ite = it + _validSlots;
+        const BTreeNode::ChildRef *it = this->_data;
+        const BTreeNode::ChildRef *ite = it + _validSlots;
         if (this->getLevel() > 1u) {
             for (; it != ite; ++it) {
-                store.mapInternalRef(*it)->foreach(store, func);
+                store.mapInternalRef(it->load_acquire())->foreach(store, func);
             }
         } else {
             for (; it != ite; ++it) {
-                store.mapLeafRef(*it)->foreach(func);
+                store.mapLeafRef(it->load_acquire())->foreach(func);
             }
         }
     }