tursom/memgraph
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Improve concurrency control for on-disk storage (#1154)

Browse Source 
* Remove locking vertices and serialization checks
---------

Co-authored-by: Aidar Samerkhanov <aidar.samerkhanov@memgraph.io>
This commit is contained in:
Andi 2023-08-25 14:42:52 +02:00 committed by GitHub
parent 4bc5d749b2
commit 030b554ffd
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 12 additions and 127 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

139
src/storage/v2/disk/storage.cpp
View File

@ -86,15 +86,9 @@ bool VertexExistsInCache(const utils::SkipList<Vertex>::Accessor &accessor, Gid
} }
bool VertexHasLabel(const Vertex &vertex, LabelId label, Transaction *transaction, View view) { bool VertexHasLabel(const Vertex &vertex, LabelId label, Transaction *transaction, View view) {
bool deleted = false; bool deleted = vertex.deleted;
bool has_label = false; bool has_label = std::find(vertex.labels.begin(), vertex.labels.end(), label) != vertex.labels.end();
Delta *delta = nullptr; Delta *delta = vertex.delta;
{
std::lock_guard<utils::SpinLock> guard(vertex.lock);
deleted = vertex.deleted;
has_label = std::find(vertex.labels.begin(), vertex.labels.end(), label) != vertex.labels.end();
delta = vertex.delta;
}
ApplyDeltasForRead(transaction, delta, view, [&deleted, &has_label, label](const Delta &delta) { ApplyDeltasForRead(transaction, delta, view, [&deleted, &has_label, label](const Delta &delta) {
switch (delta.action) { switch (delta.action) {
case Delta::Action::REMOVE_LABEL: { case Delta::Action::REMOVE_LABEL: {
@ -129,15 +123,9 @@ bool VertexHasLabel(const Vertex &vertex, LabelId label, Transaction *transactio
} }
PropertyValue GetVertexProperty(const Vertex &vertex, PropertyId property, Transaction *transaction, View view) { PropertyValue GetVertexProperty(const Vertex &vertex, PropertyId property, Transaction *transaction, View view) {
bool deleted = false; bool deleted = vertex.deleted;
PropertyValue value; PropertyValue value = vertex.properties.GetProperty(property);
Delta *delta = nullptr; Delta *delta = vertex.delta;
{
std::lock_guard<utils::SpinLock> guard(vertex.lock);
deleted = vertex.deleted;
value = vertex.properties.GetProperty(property);
delta = vertex.delta;
}
ApplyDeltasForRead(transaction, delta, view, [&deleted, &value, property](const Delta &delta) { ApplyDeltasForRead(transaction, delta, view, [&deleted, &value, property](const Delta &delta) {
switch (delta.action) { switch (delta.action) {
case Delta::Action::SET_PROPERTY: { case Delta::Action::SET_PROPERTY: {
@ -857,15 +845,8 @@ std::optional<VertexAccessor> DiskStorage::DiskAccessor::FindVertex(storage::Gid
} }
Result<std::optional<VertexAccessor>> DiskStorage::DiskAccessor::DeleteVertex(VertexAccessor *vertex) { Result<std::optional<VertexAccessor>> DiskStorage::DiskAccessor::DeleteVertex(VertexAccessor *vertex) {
MG_ASSERT(vertex->transaction_ == &transaction_,
"VertexAccessor must be from the same transaction as the storage "
"accessor when deleting a vertex!");
auto *vertex_ptr = vertex->vertex_; auto *vertex_ptr = vertex->vertex_;
std::lock_guard<utils::SpinLock> guard(vertex_ptr->lock);
if (!PrepareForWrite(&transaction_, vertex_ptr)) return Error::SERIALIZATION_ERROR;
if (vertex_ptr->deleted) { if (vertex_ptr->deleted) {
return std::optional<VertexAccessor>{}; return std::optional<VertexAccessor>{};
} }
@ -884,24 +865,12 @@ Result<std::optional<VertexAccessor>> DiskStorage::DiskAccessor::DeleteVertex(Ve
Result<std::optional<std::pair<VertexAccessor, std::vector<EdgeAccessor>>>> Result<std::optional<std::pair<VertexAccessor, std::vector<EdgeAccessor>>>>
DiskStorage::DiskAccessor::DetachDeleteVertex(VertexAccessor *vertex) { DiskStorage::DiskAccessor::DetachDeleteVertex(VertexAccessor *vertex) {
using ReturnType = std::pair<VertexAccessor, std::vector<EdgeAccessor>>; using ReturnType = std::pair<VertexAccessor, std::vector<EdgeAccessor>>;
MG_ASSERT(vertex->transaction_ == &transaction_,
"VertexAccessor must be from the same transaction as the storage "
"accessor when deleting a vertex!");
auto *vertex_ptr = vertex->vertex_; auto *vertex_ptr = vertex->vertex_;
std::vector<std::tuple<EdgeTypeId, Vertex *, EdgeRef>> in_edges; if (vertex_ptr->deleted) return std::optional<ReturnType>{};
std::vector<std::tuple<EdgeTypeId, Vertex *, EdgeRef>> out_edges;
{ std::vector<std::tuple<EdgeTypeId, Vertex *, EdgeRef>> in_edges{vertex_ptr->in_edges};
std::lock_guard<utils::SpinLock> guard(vertex_ptr->lock); std::vector<std::tuple<EdgeTypeId, Vertex *, EdgeRef>> out_edges{vertex_ptr->out_edges};
if (!PrepareForWrite(&transaction_, vertex_ptr)) return Error::SERIALIZATION_ERROR;
if (vertex_ptr->deleted) return std::optional<ReturnType>{};
in_edges = vertex_ptr->in_edges;
out_edges = vertex_ptr->out_edges;
}
std::vector<EdgeAccessor> deleted_edges; std::vector<EdgeAccessor> deleted_edges;
for (const auto &item : in_edges) { for (const auto &item : in_edges) {
@ -933,14 +902,6 @@ DiskStorage::DiskAccessor::DetachDeleteVertex(VertexAccessor *vertex) {
} }
} }
std::lock_guard<utils::SpinLock> guard(vertex_ptr->lock);
// We need to check again for serialization errors because we unlocked the
// vertex. Some other transaction could have modified the vertex in the
// meantime if we didn't have any edges to delete.
if (!PrepareForWrite(&transaction_, vertex_ptr)) return Error::SERIALIZATION_ERROR;
MG_ASSERT(!vertex_ptr->deleted, "Invalid database state!"); MG_ASSERT(!vertex_ptr->deleted, "Invalid database state!");
CreateAndLinkDelta(&transaction_, vertex_ptr, Delta::RecreateObjectTag()); CreateAndLinkDelta(&transaction_, vertex_ptr, Delta::RecreateObjectTag());
@ -1009,37 +970,10 @@ Result<EdgeAccessor> DiskStorage::DiskAccessor::CreateEdge(const VertexAccessor
const std::string_view properties, const std::string_view properties,
const std::string &old_disk_key) { const std::string &old_disk_key) {
OOMExceptionEnabler oom_exception; OOMExceptionEnabler oom_exception;
MG_ASSERT(from->transaction_ == to->transaction_,
"VertexAccessors must be from the same transaction when creating "
"an edge!");
MG_ASSERT(from->transaction_ == &transaction_,
"VertexAccessors must be from the same transaction in when "
"creating an edge!");
auto *from_vertex = from->vertex_; auto *from_vertex = from->vertex_;
auto *to_vertex = to->vertex_; auto *to_vertex = to->vertex_;
// Obtain the locks by `gid` order to avoid lock cycles. if (from_vertex->deleted || to_vertex->deleted) return Error::DELETED_OBJECT;
std::unique_lock<utils::SpinLock> guard_from(from_vertex->lock, std::defer_lock);
std::unique_lock<utils::SpinLock> guard_to(to_vertex->lock, std::defer_lock);
if (from_vertex->gid < to_vertex->gid) {
guard_from.lock();
guard_to.lock();
} else if (from_vertex->gid > to_vertex->gid) {
guard_to.lock();
guard_from.lock();
} else {
// The vertices are the same vertex, only lock one.
guard_from.lock();
}
if (!PrepareForWrite(&transaction_, from_vertex)) return Error::SERIALIZATION_ERROR;
if (from_vertex->deleted) return Error::DELETED_OBJECT;
if (to_vertex != from_vertex) {
if (!PrepareForWrite(&transaction_, to_vertex)) return Error::SERIALIZATION_ERROR;
if (to_vertex->deleted) return Error::DELETED_OBJECT;
}
auto *disk_storage = static_cast<DiskStorage *>(storage_); auto *disk_storage = static_cast<DiskStorage *>(storage_);
disk_storage->edge_id_.store(std::max(disk_storage->edge_id_.load(std::memory_order_acquire), gid.AsUint() + 1), disk_storage->edge_id_.store(std::max(disk_storage->edge_id_.load(std::memory_order_acquire), gid.AsUint() + 1),
@ -1073,37 +1007,10 @@ Result<EdgeAccessor> DiskStorage::DiskAccessor::CreateEdge(const VertexAccessor
Result<EdgeAccessor> DiskStorage::DiskAccessor::CreateEdge(VertexAccessor *from, VertexAccessor *to, Result<EdgeAccessor> DiskStorage::DiskAccessor::CreateEdge(VertexAccessor *from, VertexAccessor *to,
EdgeTypeId edge_type) { EdgeTypeId edge_type) {
MG_ASSERT(from->transaction_ == &transaction_,
"VertexAccessor must be from the same transaction as the storage "
"accessor when deleting a vertex!");
MG_ASSERT(to->transaction_ == &transaction_,
"VertexAccessor must be from the same transaction as the storage "
"accessor when deleting a vertex!");
auto *from_vertex = from->vertex_; auto *from_vertex = from->vertex_;
auto *to_vertex = to->vertex_; auto *to_vertex = to->vertex_;
// Obtain the locks by `gid` order to avoid lock cycles. if (from_vertex->deleted || to_vertex->deleted) return Error::DELETED_OBJECT;
std::unique_lock<utils::SpinLock> guard_from(from_vertex->lock, std::defer_lock);
std::unique_lock<utils::SpinLock> guard_to(to_vertex->lock, std::defer_lock);
if (from_vertex->gid < to_vertex->gid) {
guard_from.lock();
guard_to.lock();
} else if (from_vertex->gid > to_vertex->gid) {
guard_to.lock();
guard_from.lock();
} else {
// The vertices are the same vertex, only lock one.
guard_from.lock();
}
if (!PrepareForWrite(&transaction_, from_vertex)) return Error::SERIALIZATION_ERROR;
if (from_vertex->deleted) return Error::DELETED_OBJECT;
if (to_vertex != from_vertex) {
if (!PrepareForWrite(&transaction_, to_vertex)) return Error::SERIALIZATION_ERROR;
if (to_vertex->deleted) return Error::DELETED_OBJECT;
}
auto *disk_storage = static_cast<DiskStorage *>(storage_); auto *disk_storage = static_cast<DiskStorage *>(storage_);
auto gid = storage::Gid::FromUint(disk_storage->edge_id_.fetch_add(1, std::memory_order_acq_rel)); auto gid = storage::Gid::FromUint(disk_storage->edge_id_.fetch_add(1, std::memory_order_acq_rel));
@ -1141,38 +1048,16 @@ Result<std::optional<EdgeAccessor>> DiskStorage::DiskAccessor::DeleteEdge(EdgeAc
const auto edge_ref = edge->edge_; const auto edge_ref = edge->edge_;
const auto edge_type = edge->edge_type_; const auto edge_type = edge->edge_type_;
std::unique_lock<utils::SpinLock> guard;
if (config_.properties_on_edges) { if (config_.properties_on_edges) {
const auto *edge_ptr = edge_ref.ptr; if (edge_ref.ptr->deleted) return std::optional<EdgeAccessor>{};
guard = std::unique_lock<utils::SpinLock>(edge_ptr->lock);
if (!PrepareForWrite(&transaction_, edge_ptr)) return Error::SERIALIZATION_ERROR;
if (edge_ptr->deleted) return std::optional<EdgeAccessor>{};
} }
auto *from_vertex = edge->from_vertex_; auto *from_vertex = edge->from_vertex_;
auto *to_vertex = edge->to_vertex_; auto *to_vertex = edge->to_vertex_;
// Obtain the locks by `gid` order to avoid lock cycles.
std::unique_lock<utils::SpinLock> guard_from(from_vertex->lock, std::defer_lock);
std::unique_lock<utils::SpinLock> guard_to(to_vertex->lock, std::defer_lock);
if (from_vertex->gid < to_vertex->gid) {
guard_from.lock();
guard_to.lock();
} else if (from_vertex->gid > to_vertex->gid) {
guard_to.lock();
guard_from.lock();
} else {
// The vertices are the same vertex, only lock one.
guard_from.lock();
}
if (!PrepareForWrite(&transaction_, from_vertex)) return Error::SERIALIZATION_ERROR;
MG_ASSERT(!from_vertex->deleted, "Invalid database state!"); MG_ASSERT(!from_vertex->deleted, "Invalid database state!");
if (to_vertex != from_vertex) { if (to_vertex != from_vertex) {
if (!PrepareForWrite(&transaction_, to_vertex)) return Error::SERIALIZATION_ERROR;
MG_ASSERT(!to_vertex->deleted, "Invalid database state!"); MG_ASSERT(!to_vertex->deleted, "Invalid database state!");
} }