Make previous pointer atomic in storage v2

Reviewers: mtomic, teon.banek

Reviewed By: mtomic

Subscribers: pullbot

Differential Revision: https://phabricator.memgraph.io/D2191

src/storage/v2/delta.hpp

@@ -36,54 +36,56 @@ class PreviousPtr {
     EDGE,
   };
 
-  void Set(Delta *delta) {
-    uintptr_t value = reinterpret_cast<uintptr_t>(delta);
-    CHECK((value & kMask) == 0) << "Invalid pointer!";
-    storage_ = value | kDelta;
-  }
+  struct Pointer {
+    explicit Pointer(Delta *delta) : type(Type::DELTA), delta(delta) {}
+    explicit Pointer(Vertex *vertex) : type(Type::VERTEX), vertex(vertex) {}
+    explicit Pointer(Edge *edge) : type(Type::EDGE), edge(edge) {}
 
-  void Set(Vertex *vertex) {
-    uintptr_t value = reinterpret_cast<uintptr_t>(vertex);
-    CHECK((value & kMask) == 0) << "Invalid pointer!";
-    storage_ = value | kVertex;
-  }
+    Type type;
+    Delta *delta{nullptr};
+    Vertex *vertex{nullptr};
+    Edge *edge{nullptr};
+  };
 
-  void Set(Edge *edge) {
-    uintptr_t value = reinterpret_cast<uintptr_t>(edge);
-    CHECK((value & kMask) == 0) << "Invalid pointer!";
-    storage_ = value | kEdge;
-  }
+  PreviousPtr() : storage_(0) {}
 
-  Type GetType() const {
-    uintptr_t type = storage_ & kMask;
+  PreviousPtr(const PreviousPtr &other) noexcept
+      : storage_(other.storage_.load(std::memory_order_acquire)) {}
+
+  Pointer Get() const {
+    uintptr_t value = storage_.load(std::memory_order_acquire);
+    uintptr_t type = value & kMask;
     if (type == kDelta) {
-      return Type::DELTA;
+      return Pointer{reinterpret_cast<Delta *>(value & ~kMask)};
     } else if (type == kVertex) {
-      return Type::VERTEX;
+      return Pointer{reinterpret_cast<Vertex *>(value & ~kMask)};
     } else if (type == kEdge) {
-      return Type::EDGE;
+      return Pointer{reinterpret_cast<Edge *>(value & ~kMask)};
     } else {
       LOG(FATAL) << "Invalid pointer type!";
     }
   }
 
-  Delta *GetDelta() const {
-    CHECK((storage_ & kMask) == kDelta) << "Can't convert pointer to delta!";
-    return reinterpret_cast<Delta *>(storage_ & ~kMask);
+  void Set(Delta *delta) {
+    uintptr_t value = reinterpret_cast<uintptr_t>(delta);
+    CHECK((value & kMask) == 0) << "Invalid pointer!";
+    storage_.store(value | kDelta, std::memory_order_release);
   }
 
-  Vertex *GetVertex() const {
-    CHECK((storage_ & kMask) == kVertex) << "Can't convert pointer to vertex!";
-    return reinterpret_cast<Vertex *>(storage_ & ~kMask);
+  void Set(Vertex *vertex) {
+    uintptr_t value = reinterpret_cast<uintptr_t>(vertex);
+    CHECK((value & kMask) == 0) << "Invalid pointer!";
+    storage_.store(value | kVertex, std::memory_order_release);
   }
 
-  Edge *GetEdge() const {
-    CHECK((storage_ & kMask) == kEdge) << "Can't convert pointer to edge!";
-    return reinterpret_cast<Edge *>(storage_ & ~kMask);
+  void Set(Edge *edge) {
+    uintptr_t value = reinterpret_cast<uintptr_t>(edge);
+    CHECK((value & kMask) == 0) << "Invalid pointer!";
+    storage_.store(value | kEdge, std::memory_order_release);
   }
 
  private:
-  uintptr_t storage_{0};
+  std::atomic<uintptr_t> storage_;
 };
 
 struct Delta {

src/storage/v2/mvcc.hpp

@@ -85,7 +85,9 @@ inline void CreateAndLinkDelta(Transaction *transaction, TObj *object,
                                                  transaction->command_id);
 
   // The operations are written in such order so that both `next` and `prev`
-  // chains are valid at all times.
+  // chains are valid at all times. The chains must be valid at all times
+  // because garbage collection (which traverses the chains) is done
+  // concurrently (as well as other execution threads).
   delta->prev.Set(object);
   if (object->delta) {
     object->delta->prev.Set(delta);

src/storage/v2/storage.cpp

@@ -307,9 +307,10 @@ void Storage::Accessor::Commit() {
 void Storage::Accessor::Abort() {
   CHECK(transaction_->is_active) << "The transaction is already terminated!";
   for (const auto &delta : transaction_->deltas) {
-    switch (delta.prev.GetType()) {
+    auto prev = delta.prev.Get();
+    switch (prev.type) {
       case PreviousPtr::Type::VERTEX: {
-        auto vertex = delta.prev.GetVertex();
+        auto vertex = prev.vertex;
         std::lock_guard<utils::SpinLock> guard(vertex->lock);
         Delta *current = vertex->delta;
         while (current != nullptr &&
@@ -409,7 +410,7 @@ void Storage::Accessor::Abort() {
         break;
       }
       case PreviousPtr::Type::EDGE: {
-        auto edge = delta.prev.GetEdge();
+        auto edge = prev.edge;
         std::lock_guard<utils::SpinLock> guard(edge->lock);
         Delta *current = edge->delta;
         while (current != nullptr &&