Merge pull request #21409 from vespa-engine/toregge/use-atomic-instructions-to-get-and-set-btree-internal-child-refs

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

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);
             }
         }
     }