Compare commits

...

1 Commits

Author SHA1 Message Date
gvolfing
23474fd6cc initial changes 2023-05-04 08:31:08 +02:00
2 changed files with 171 additions and 132 deletions

View File

@@ -12,6 +12,7 @@
#include "storage/v2/storage.hpp" #include "storage/v2/storage.hpp"
#include <algorithm> #include <algorithm>
#include <atomic> #include <atomic>
#include <limits>
#include <memory> #include <memory>
#include <mutex> #include <mutex>
#include <variant> #include <variant>
@@ -1439,11 +1440,17 @@ void Storage::CollectGarbage() {
if constexpr (force) { if constexpr (force) {
// We take the unique lock on the main storage lock so we can forcefully clean // We take the unique lock on the main storage lock so we can forcefully clean
// everything we can // everything we can
// TODO(gvolfing) Verify this! This should not just try, it should lock() always.
if (!main_lock_.try_lock()) { if (!main_lock_.try_lock()) {
CollectGarbage<false>(); CollectGarbage<false>();
return; return;
} }
} else { } else {
// DEBUG
// return;
// Because the garbage collector iterates through the indices and constraints // Because the garbage collector iterates through the indices and constraints
// to clean them up, it must take the main lock for reading to make sure that // to clean them up, it must take the main lock for reading to make sure that
// the indices and constraints aren't concurrently being modified. // the indices and constraints aren't concurrently being modified.
@@ -1484,19 +1491,46 @@ void Storage::CollectGarbage() {
deleted_vertices_->swap(current_deleted_vertices); deleted_vertices_->swap(current_deleted_vertices);
deleted_edges_->swap(current_deleted_edges); deleted_edges_->swap(current_deleted_edges);
const bool is_analytical_storage_mode = storage_mode_ == StorageMode::IN_MEMORY_ANALYTICAL;
// Flag that will be used to determine whether the Index GC should be run. It // 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 // 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 // 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. // eliminates high CPU usage when the GC doesn't have to clean up anything.
bool run_index_cleanup = !committed_transactions_->empty() || !garbage_undo_buffers_->empty(); // bool run_index_cleanup = !committed_transactions_->empty() || !garbage_undo_buffers_->empty();
bool run_index_cleanup =
!committed_transactions_->empty() || !garbage_undo_buffers_->empty() || is_analytical_storage_mode;
if (is_analytical_storage_mode) {
oldest_active_start_timestamp = std::numeric_limits<decltype(oldest_active_start_timestamp)>::max();
if constexpr (force) {
auto vertices = vertices_.access();
for (auto &vertex : vertices) {
if (vertex.deleted) {
current_deleted_vertices.push_back(vertex.gid);
}
}
// TODO add edges as well
} else {
// CollectGarbage<true>();
// return;
auto vertices = vertices_.access();
for (auto &vertex : vertices) {
if (vertex.deleted) {
current_deleted_vertices.push_back(vertex.gid);
}
}
}
} else {
while (true) { while (true) {
// We don't want to hold the lock on commited transactions for too long, // We don't want to hold the lock on commited transactions for too long,
// because that prevents other transactions from committing. // because that prevents other transactions from committing.
Transaction *transaction; Transaction *transaction;
{ {
auto committed_transactions_ptr = committed_transactions_.Lock(); auto committed_transactions_ptr = committed_transactions_.Lock();
if (committed_transactions_ptr->empty()) { if (committed_transactions_ptr->empty() || is_analytical_storage_mode) {
break; break;
} }
transaction = &committed_transactions_ptr->front(); transaction = &committed_transactions_ptr->front();
@@ -1619,7 +1653,7 @@ void Storage::CollectGarbage() {
committed_transactions.pop_front(); committed_transactions.pop_front();
}); });
} }
}
// After unlinking deltas from vertices, we refresh the indices. That way // After unlinking deltas from vertices, we refresh the indices. That way
// we're sure that none of the vertices from `current_deleted_vertices` // 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 // appears in an index, and we can safely remove the from the main storage
@@ -1632,8 +1666,11 @@ void Storage::CollectGarbage() {
} }
{ {
uint64_t mark_timestamp = 0;
if (!is_analytical_storage_mode) {
std::unique_lock<utils::SpinLock> guard(engine_lock_); std::unique_lock<utils::SpinLock> guard(engine_lock_);
uint64_t mark_timestamp = timestamp_; // uint64_t mark_timestamp = timestamp_;
mark_timestamp = timestamp_;
// Take garbage_undo_buffers lock while holding the engine lock to make // Take garbage_undo_buffers lock while holding the engine lock to make
// sure that entries are sorted by mark timestamp in the list. // sure that entries are sorted by mark timestamp in the list.
garbage_undo_buffers_.WithLock([&](auto &garbage_undo_buffers) { garbage_undo_buffers_.WithLock([&](auto &garbage_undo_buffers) {
@@ -1648,6 +1685,7 @@ void Storage::CollectGarbage() {
} }
garbage_undo_buffers.splice(garbage_undo_buffers.end(), unlinked_undo_buffers); garbage_undo_buffers.splice(garbage_undo_buffers.end(), unlinked_undo_buffers);
}); });
}
for (auto vertex : current_deleted_vertices) { for (auto vertex : current_deleted_vertices) {
garbage_vertices_.emplace_back(mark_timestamp, vertex); garbage_vertices_.emplace_back(mark_timestamp, vertex);
} }

View File

@@ -67,6 +67,7 @@ class DeltaGenerator final {
auto gid = memgraph::storage::Gid::FromUint(gen_->vertices_count_++); auto gid = memgraph::storage::Gid::FromUint(gen_->vertices_count_++);
auto delta = memgraph::storage::CreateDeleteObjectDelta(&transaction_); auto delta = memgraph::storage::CreateDeleteObjectDelta(&transaction_);
auto &it = gen_->vertices_.emplace_back(gid, delta); auto &it = gen_->vertices_.emplace_back(gid, delta);
// TODO(gvolfing) This is likely problematic when deleting vertices in ANALYTICAL MODE...
if (delta != nullptr) { if (delta != nullptr) {
delta->prev.Set(&it); delta->prev.Set(&it);
} }