Rework delta creation in storage v2

Reviewers: mtomic, teon.banek

Reviewed By: mtomic

Subscribers: pullbot

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

src/storage/v2/mvcc.hpp

@@ -52,20 +52,10 @@ inline void ApplyDeltasForRead(Transaction *transaction, Delta *delta,
 /// transaction) and returns a `bool` value indicating whether the caller can
 /// proceed with a write operation.
 inline bool PrepareForWrite(Transaction *transaction, Vertex *vertex) {
-  if (vertex->delta == nullptr) {
-    if (transaction->modified_vertices.empty() ||
-        transaction->modified_vertices.back() != vertex) {
-      transaction->modified_vertices.push_back(vertex);
-    }
-    return true;
-  }
+  if (vertex->delta == nullptr) return true;
 
   auto ts = vertex->delta->timestamp->load(std::memory_order_acquire);
   if (ts == transaction->transaction_id || ts < transaction->start_timestamp) {
-    if (transaction->modified_vertices.empty() ||
-        transaction->modified_vertices.back() != vertex) {
-      transaction->modified_vertices.push_back(vertex);
-    }
     return true;
   }
 
@@ -73,4 +63,33 @@ inline bool PrepareForWrite(Transaction *transaction, Vertex *vertex) {
   return false;
 }
 
+/// This function creates a delta in the transaction and returns a pointer to
+/// the created delta. It doesn't perform any linking of the delta and is
+/// primarily used to create the first delta for an object.
+inline Delta *CreateDelta(Transaction *transaction, Delta::Action action,
+                          uint64_t value) {
+  return &transaction->deltas.emplace_back(
+      action, value, &transaction->commit_timestamp, transaction->command_id);
+}
+
+/// This function creates a delta in the transaction for the Vertex object and
+/// links the delta into the Vertex's delta list. It also adds the Vertex to the
+/// transaction's modified vertices list.
+inline void CreateAndLinkDelta(Transaction *transaction, Vertex *vertex,
+                               Delta::Action action, uint64_t value) {
+  auto delta = &transaction->deltas.emplace_back(
+      action, value, &transaction->commit_timestamp, transaction->command_id);
+
+  if (vertex->delta) {
+    vertex->delta->prev = delta;
+  }
+  delta->next.store(vertex->delta, std::memory_order_release);
+  vertex->delta = delta;
+
+  if (transaction->modified_vertices.empty() ||
+      transaction->modified_vertices.back() != vertex) {
+    transaction->modified_vertices.push_back(vertex);
+  }
+}
+
 }  // namespace storage

src/storage/v2/storage.hpp

@@ -81,7 +81,7 @@ class Storage final {
     VertexAccessor CreateVertex() {
       auto gid = storage_->vertex_id_.fetch_add(1, std::memory_order_acq_rel);
       auto acc = storage_->vertices_.access();
-      auto delta = transaction_->CreateDelta(Delta::Action::DELETE_OBJECT, 0);
+      auto delta = CreateDelta(transaction_, Delta::Action::DELETE_OBJECT, 0);
       auto [it, inserted] =
           acc.insert(Vertex{storage::Gid::FromUint(gid), delta});
       CHECK(inserted) << "The vertex must be inserted here!";

src/storage/v2/transaction.hpp

@@ -37,10 +37,6 @@ struct Transaction {
 
   ~Transaction() {}
 
-  Delta *CreateDelta(Delta::Action action, uint64_t value) {
-    return &deltas.emplace_back(action, value, &commit_timestamp, command_id);
-  }
-
   uint64_t transaction_id;
   uint64_t start_timestamp;
   std::atomic<uint64_t> commit_timestamp;

src/storage/v2/vertex_accessor.hpp

@@ -53,12 +53,8 @@ class VertexAccessor final {
         vertex_->labels.end())
       return Result<bool>{false};
 
-    auto delta = transaction_->CreateDelta(Delta::Action::REMOVE_LABEL, label);
-    if (vertex_->delta) {
-      vertex_->delta->prev = delta;
-    }
-    delta->next = vertex_->delta;
-    vertex_->delta = delta;
+    CreateAndLinkDelta(transaction_, vertex_, Delta::Action::REMOVE_LABEL,
+                       label);
 
     vertex_->labels.push_back(label);
     return Result<bool>{true};
@@ -73,12 +69,7 @@ class VertexAccessor final {
     auto it = std::find(vertex_->labels.begin(), vertex_->labels.end(), label);
     if (it == vertex_->labels.end()) return Result<bool>{false};
 
-    auto delta = transaction_->CreateDelta(Delta::Action::ADD_LABEL, label);
-    if (vertex_->delta) {
-      vertex_->delta->prev = delta;
-    }
-    delta->next = vertex_->delta;
-    vertex_->delta = delta;
+    CreateAndLinkDelta(transaction_, vertex_, Delta::Action::ADD_LABEL, label);
 
     std::swap(*it, *vertex_->labels.rbegin());
     vertex_->labels.pop_back();