Merge branch 'master' of github.com:memgraph/memgraph into gc-proactive-deletion

CMakeLists.txt

@@ -292,7 +292,7 @@ if (MG_ENTERPRISE)
   add_definitions(-DMG_ENTERPRISE)
 endif()
 
-set(ENABLE_JEMALLOC ON)
+option(ENABLE_JEMALLOC "Use jemalloc" ON)
 
 if (ASAN)
   message(WARNING "Disabling jemalloc as it doesn't work well with ASAN")

src/memory/CMakeLists.txt

@@ -3,14 +3,14 @@ set(memory_src_files
     global_memory_control.cpp
     query_memory_control.cpp)
 
-
-
-find_package(jemalloc REQUIRED)
-
 add_library(mg-memory STATIC ${memory_src_files})
 target_link_libraries(mg-memory mg-utils fmt)
 
+message(STATUS "ENABLE_JEMALLOC: ${ENABLE_JEMALLOC}")
 if (ENABLE_JEMALLOC)
+  find_package(jemalloc REQUIRED)
   target_link_libraries(mg-memory Jemalloc::Jemalloc ${CMAKE_DL_LIBS})
   target_compile_definitions(mg-memory PRIVATE USE_JEMALLOC=1)
+else()
+  target_compile_definitions(mg-memory PRIVATE USE_JEMALLOC=0)
 endif()

src/storage/v2/inmemory/storage.cpp

@@ -986,10 +986,14 @@ void InMemoryStorage::InMemoryAccessor::FinalizeTransaction() {
     if (!transaction_.deltas.use().empty()) {
       // Only hand over delta to be GC'ed if there was any deltas
       mem_storage->committed_transactions_.WithLock([&](auto &committed_transactions) {
-        // using mark of 0 as GC will assign a mark_timestamp after unlinking
+        // using commit timestamp as mark, set here reduces atomic contention in GC,
-        committed_transactions.emplace_back(0, std::move(transaction_.deltas),
+        // after unlinking, an actual mark_timestamp will be assigned
+        committed_transactions.emplace_back(*commit_timestamp_, std::move(transaction_.deltas),
                                             std::move(transaction_.commit_timestamp));
       });
+    } else {
+      // proactively clear commit_timestamp, no deltas refer to it
+      transaction_.commit_timestamp.reset();
     }
     commit_timestamp_.reset();
   }
@@ -1210,7 +1214,7 @@ void InMemoryStorage::CollectGarbage(std::unique_lock<utils::ResourceLock> main_
       main_lock_.lock_shared();
     }
   } else {
-    MG_ASSERT(main_guard.mutex() == std::addressof(main_lock_), "main_guard should be only for the main_lock_");
+    DMG_ASSERT(main_guard.mutex() == std::addressof(main_lock_), "main_guard should be only for the main_lock_");
   }
 
   utils::OnScopeExit lock_releaser{[&] {
@@ -1237,19 +1241,20 @@ void InMemoryStorage::CollectGarbage(std::unique_lock<utils::ResourceLock> main_
     return;
   }
 
-  uint64_t oldest_active_start_timestamp = commit_log_->OldestActive();
-
   // Deltas from previous GC runs or from aborts can be cleaned up here
   garbage_undo_buffers_.WithLock([&](auto &garbage_undo_buffers) {
-    if constexpr (force) {
+    if (!garbage_undo_buffers.empty()) {
-      // if force is set to true we can simply delete all the leftover undos because
+      if constexpr (force) {
-      // no transaction is active
+        // if force is set to true we can simply delete all the leftover undos because
-      garbage_undo_buffers.clear();
+        // no transaction is active
-    } else {
+        garbage_undo_buffers.clear();
-      // garbage_undo_buffers is ordered, pop until we can't
+      } else {
-      while (!garbage_undo_buffers.empty() &&
+        uint64_t oldest_active_start_timestamp = commit_log_->OldestActive();
-             garbage_undo_buffers.front().mark_timestamp_ <= oldest_active_start_timestamp) {
+        // garbage_undo_buffers is ordered, pop until we can't
-        garbage_undo_buffers.pop_front();
+        while (!garbage_undo_buffers.empty() &&
+               garbage_undo_buffers.front().mark_timestamp_ <= oldest_active_start_timestamp) {
+          garbage_undo_buffers.pop_front();
+        }
       }
     }
   });
@@ -1259,6 +1264,10 @@ void InMemoryStorage::CollectGarbage(std::unique_lock<utils::ResourceLock> main_
   // garbage_undo_buffers lock.
   std::list<GCDeltas> unlinked_undo_buffers{};
 
+  // same as unlinked_undo_buffers, but it is known none of these deltas are accessible
+  // (because no unlinking was required)
+  std::list<GCDeltas> unlinked_undo_buffers_delete_ASAP{};
+
   // We will only free vertices deleted up until now in this GC cycle, and we
   // will do it after cleaning-up the indices. That way we are sure that all
   // vertices that appear in an index also exist in main storage.
@@ -1275,20 +1284,13 @@ void InMemoryStorage::CollectGarbage(std::unique_lock<utils::ResourceLock> main_
   committed_transactions_.WithLock(
       [&](auto &committed_transactions) { committed_transactions.swap(linked_undo_buffers); });
 
-  // Flag that will be used to determine whether the Index GC should be run. It
-  // should be run when there were any items that were cleaned up (there were
-  // updates between this run of the GC and the previous run of the GC). This
-  // eliminates high CPU usage when the GC doesn't have to clean up anything.
-  bool run_index_cleanup = !linked_undo_buffers.empty() || !garbage_undo_buffers_->empty() || need_full_scan_vertices ||
-                           need_full_scan_edges;
-
   auto const end_linked_undo_buffers = linked_undo_buffers.end();
+  uint64_t oldest_active_start_timestamp = commit_log_->OldestActive();
   for (auto linked_entry = linked_undo_buffers.begin(); linked_entry != end_linked_undo_buffers;) {
     auto const *const commit_timestamp_ptr = linked_entry->commit_timestamp_.get();
-    auto const commit_timestamp = commit_timestamp_ptr->load(std::memory_order_acquire);
 
-    // only process those that are no longer active
+    // only process those that are no longer active (here mark_timestamp_ == *commit_timestamp_)
-    if (commit_timestamp >= oldest_active_start_timestamp) {
+    if (linked_entry->mark_timestamp_ >= oldest_active_start_timestamp) {
       ++linked_entry;  // can not process, skip
       continue;        // must continue to next transaction, because committed_transactions_ was not ordered
     }
@@ -1324,6 +1326,7 @@ void InMemoryStorage::CollectGarbage(std::unique_lock<utils::ResourceLock> main_
     // chain in a broken state.
     // The chain can be only read without taking any locks.
 
+    bool any_unlinks = false;
     for (Delta &delta : linked_entry->deltas_.use()) {
       while (true) {
         auto prev = delta.prev.Get();
@@ -1337,6 +1340,7 @@ void InMemoryStorage::CollectGarbage(std::unique_lock<utils::ResourceLock> main_
               continue;
             }
             vertex->delta = nullptr;
+            any_unlinks = true;
             if (vertex->deleted) {
               DMG_ASSERT(delta.action == memgraph::storage::Delta::Action::RECREATE_OBJECT);
               current_deleted_vertices.push_back(vertex->gid);
@@ -1352,6 +1356,7 @@ void InMemoryStorage::CollectGarbage(std::unique_lock<utils::ResourceLock> main_
               continue;
             }
             edge->delta = nullptr;
+            any_unlinks = true;
             if (edge->deleted) {
               DMG_ASSERT(delta.action == memgraph::storage::Delta::Action::RECREATE_OBJECT);
               current_deleted_edges.push_back(edge->gid);
@@ -1408,8 +1413,8 @@ void InMemoryStorage::CollectGarbage(std::unique_lock<utils::ResourceLock> main_
               // chain.
               continue;
             }
-            Delta *prev_delta = prev.delta;
+            prev.delta->next.store(nullptr, std::memory_order_release);
-            prev_delta->next.store(nullptr, std::memory_order_release);
+            any_unlinks = true;
             break;
           }
           case PreviousPtr::Type::NULLPTR: {
@@ -1423,9 +1428,20 @@ void InMemoryStorage::CollectGarbage(std::unique_lock<utils::ResourceLock> main_
     // Now unlinked, move to unlinked_undo_buffers
     auto const to_move = linked_entry;
     ++linked_entry;  // advanced to next before we move the list node
-    unlinked_undo_buffers.splice(unlinked_undo_buffers.end(), linked_undo_buffers, to_move);
+    if (any_unlinks) {
+      // At least one delta in this collection maybe accessible (move to unlinked_undo_buffers)
+      unlinked_undo_buffers.splice(unlinked_undo_buffers.end(), linked_undo_buffers, to_move);
+    } else {
+      // None of the deltas in this collection are accessible (move to unlinked_undo_buffers_delete_ASAP)
+      // Can't delete here because maybe another linked_entry unlinking way try to read these deltas
+      unlinked_undo_buffers_delete_ASAP.splice(unlinked_undo_buffers_delete_ASAP.end(), linked_undo_buffers, to_move);
+    }
   }
 
+  // Now we have unlinked, we can clear the inaccessible deltas
+  // they are not accessible to any active transaction or any future unlinking
+  unlinked_undo_buffers_delete_ASAP.clear();
+
   if (!linked_undo_buffers.empty()) {
     // some were not able to be collected, add them back to committed_transactions_ for the next GC run
     committed_transactions_.WithLock([&linked_undo_buffers](auto &committed_transactions) {
@@ -1433,6 +1449,13 @@ void InMemoryStorage::CollectGarbage(std::unique_lock<utils::ResourceLock> main_
     });
   }
 
+  // Flag that will be used to determine whether the Index GC should be run. It
+  // should be run when there were any items that were cleaned up (there were
+  // updates between this run of the GC and the previous run of the GC). This
+  // eliminates high CPU usage when the GC doesn't have to clean up anything.
+  bool run_index_cleanup = !current_deleted_vertices.empty() || !current_deleted_edges.empty() ||
+                           need_full_scan_vertices || need_full_scan_edges;
+
   // After unlinking deltas from vertices, we refresh the indices. That way
   // we're sure that none of the vertices from `current_deleted_vertices`
   // appears in an index, and we can safely remove the from the main storage
@@ -1445,7 +1468,7 @@ void InMemoryStorage::CollectGarbage(std::unique_lock<utils::ResourceLock> main_
     mem_unique_constraints->RemoveObsoleteEntries(oldest_active_start_timestamp);
   }
 
-  {
+  if (!unlinked_undo_buffers.empty()) {
     std::unique_lock<utils::SpinLock> guard(engine_lock_);
     uint64_t mark_timestamp = timestamp_;  // a timestamp no active transaction can currently have
 

src/storage/v2/inmemory/storage.hpp

@@ -428,8 +428,9 @@ class InMemoryStorage final : public Storage {
     GCDeltas(GCDeltas &&) = default;
     GCDeltas &operator=(GCDeltas &&) = default;
 
-    uint64_t mark_timestamp_{};                                  //!< a timestamp no active transaction currently has
+    uint64_t mark_timestamp_{};  //!< linked: non-atomic copy of commit_timestamp_, unlinked: a timestamp no active
-    BondPmrLd deltas_;                                           //!< the deltas that need cleaning
+                                 //!< transaction currently had
+    BondPmrLd deltas_;           //!< the deltas that need cleaning
     std::unique_ptr<std::atomic<uint64_t>> commit_timestamp_{};  //!< the timestamp the deltas are pointing at
   };