Fix optimistic B-tree traversal under weak memory ordering

- `BTree.h:1246` dereferences `next` before the parent lease is validated.
- During concurrent insert/split, the reader can compute `idx` from one
  snapshot and read `children[idx]` from a partially
  visible later snapshot.
- On aarch64, weaker store visibility makes this much more
  plausible: `numElements` can become observable before the
  matching child slot is safely observable to a stale optimistic
  reader.
- `OptimisticReadWriteLock` is also too weak for this
  seqlock-style usage, especially around validation
  and writer publication.

Author: quentin

src/include/souffle/datastructure/BTree.h

@@ -1241,6 +1241,12 @@ class btree {
 
                 // get next pointer
                 auto next = cur->getChild(idx);
+                if (next == nullptr) {
+                    if (cur->lock.validate(cur_lease)) {
+                        assert(false && "B-tree inner node has null child");
+                    }
+                    return insert(k, hints);
+                }
 
                 // get lease on next level
                 auto next_lease = next->lock.start_read();

src/include/souffle/datastructure/BTreeDelete.h

@@ -1294,6 +1294,12 @@ class btree_delete {
 
                 // get next pointer
                 auto next = cur->getChild(idx);
+                if (next == nullptr) {
+                    if (cur->lock.validate(cur_lease)) {
+                        assert(false && "B-tree inner node has null child");
+                    }
+                    return insert(k, hints);
+                }
 
                 // get lease on next level
                 auto next_lease = next->lock.start_read();

src/include/souffle/datastructure/LambdaBTree.h

@@ -198,6 +198,12 @@ class LambdaBTree : public btree<Key, Comparator, Allocator, blockSize, SearchSt
 
                 // get next pointer
                 auto next = cur->getChild(idx);
+                if (next == nullptr) {
+                    if (cur->lock.validate(cur_lease)) {
+                        assert(false && "B-tree inner node has null child");
+                    }
+                    return insert(k, hints, f);
+                }
 
                 // get lease on next level
                 auto next_lease = next->lock.start_read();

src/include/souffle/utility/ParallelUtil.h

@@ -444,7 +444,7 @@ class OptimisticReadWriteLock {
     bool validate(const Lease& lease) {
         // check whether version number has changed in the mean-while
         std::atomic_thread_fence(std::memory_order_acquire);
-        return lease.version == version.load(std::memory_order_relaxed);
+        return lease.version == version.load(std::memory_order_acquire);
     }
 
     /**
@@ -466,14 +466,14 @@ class OptimisticReadWriteLock {
         detail::Waiter wait;
 
         // set last bit => make it odd
-        auto v = version.fetch_or(0x1, std::memory_order_acquire);
+        auto v = version.fetch_or(0x1, std::memory_order_acq_rel);
 
         // check for concurrent writes
         while ((v & 0x1) == 1) {
             // wait for a moment
             wait();
             // get an updated version
-            v = version.fetch_or(0x1, std::memory_order_acquire);
+            v = version.fetch_or(0x1, std::memory_order_acq_rel);
         }
 
         // done
@@ -486,7 +486,7 @@ class OptimisticReadWriteLock {
      * @return true if write permission has been granted, false otherwise.
      */
     bool try_start_write() {
-        auto v = version.fetch_or(0x1, std::memory_order_acquire);
+        auto v = version.fetch_or(0x1, std::memory_order_acq_rel);
         return !(v & 0x1);
     }
 
@@ -499,7 +499,7 @@ class OptimisticReadWriteLock {
      *      be granted, false otherwise.
      */
     bool try_upgrade_to_write(const Lease& lease) {
-        auto v = version.fetch_or(0x1, std::memory_order_acquire);
+        auto v = version.fetch_or(0x1, std::memory_order_acq_rel);
 
         // check whether write privileges have been gained
         if (v & 0x1) return false;  // there is another writer already
@@ -539,7 +539,7 @@ class OptimisticReadWriteLock {
      * @return true if so, false otherwise
      */
     bool is_write_locked() const {
-        return version & 0x1;
+        return version.load(std::memory_order_relaxed) & 0x1;
     }
 };