From 6df79724cacf081f100d26ec48b482e2413ce702 Mon Sep 17 00:00:00 2001 From: Andi Skrgat Date: Wed, 17 May 2023 11:44:00 +0200 Subject: [PATCH] (un)lock path, get info and transaction id deduplication --- src/storage/v2/disk/storage.cpp | 21 +---------- src/storage/v2/disk/storage.hpp | 10 ------ src/storage/v2/inmemory/storage.cpp | 43 +---------------------- src/storage/v2/inmemory/storage.hpp | 10 ------ src/storage/v2/storage.cpp | 41 ++++++++++++++++++++++ src/storage/v2/storage.hpp | 54 ++++++++++++++++------------- 6 files changed, 73 insertions(+), 106 deletions(-) diff --git a/src/storage/v2/disk/storage.cpp b/src/storage/v2/disk/storage.cpp index f33789bc7..cf7f33436 100644 --- a/src/storage/v2/disk/storage.cpp +++ b/src/storage/v2/disk/storage.cpp @@ -88,8 +88,7 @@ DiskStorage::DiskStorage(Config config) : Storage(config), indices_(&constraints_, config.items), isolation_level_(IsolationLevel::SNAPSHOT_ISOLATION), - storage_mode_(StorageMode::IN_MEMORY_TRANSACTIONAL), - config_(config) { + storage_mode_(StorageMode::IN_MEMORY_TRANSACTIONAL) { if (config_.durability.snapshot_wal_mode == Config::Durability::SnapshotWalMode::DISABLED /// TODO(andi): When replication support will be added, uncomment this. // && replication_role_ == ReplicationRole::MAIN) { @@ -876,9 +875,6 @@ Result> DiskStorage::DiskAccessor::DeleteEdge(EdgeAc &disk_storage->indices_, &disk_storage->constraints_, config_, true); } -// this should be handled on an above level of abstraction -void DiskStorage::DiskAccessor::AdvanceCommand() { ++transaction_.command_id; } - void DiskStorage::DiskAccessor::FlushCache() { /// Flush vertex cache. auto vertex_acc = storage_->vertices_.access(); @@ -1235,14 +1231,6 @@ void DiskStorage::DiskAccessor::FinalizeTransaction() { } } -// this should be handled on an above level of abstraction -std::optional DiskStorage::DiskAccessor::GetTransactionId() const { - if (is_transaction_active_) { - return transaction_.transaction_id.load(std::memory_order_acquire); - } - return {}; -} - utils::BasicResult DiskStorage::CreateIndex( LabelId label, const std::optional desired_commit_timestamp) { /// TODO: (andi): Here we will probably use some lock to protect reading from and writing to the RocksDB at the @@ -1336,9 +1324,6 @@ DiskStorage::DropUniqueConstraint(LabelId label, const std::set &pro // this should be handled on an above level of abstraction ConstraintsInfo DiskStorage::ListAllConstraints() const { throw utils::NotYetImplemented("ListAllConstraints"); } -// this should be handled on an above level of abstraction -StorageInfo DiskStorage::GetInfo() const { throw utils::NotYetImplemented("GetInfo"); } - Transaction DiskStorage::CreateTransaction(IsolationLevel isolation_level, StorageMode storage_mode) { /// We acquire the transaction engine lock here because we access (and /// modify) the transaction engine variables (`transaction_id` and @@ -1668,10 +1653,6 @@ utils::BasicResult DiskStorage::CreateSnapshot throw utils::NotYetImplemented("CreateSnapshot"); } -bool DiskStorage::LockPath() { throw utils::NotYetImplemented("LockPath"); } - -bool DiskStorage::UnlockPath() { throw utils::NotYetImplemented("UnlockPath"); } - void DiskStorage::FreeMemory() { throw utils::NotYetImplemented("FreeMemory"); } uint64_t DiskStorage::CommitTimestamp(const std::optional desired_commit_timestamp) { diff --git a/src/storage/v2/disk/storage.hpp b/src/storage/v2/disk/storage.hpp index ae3d3ab85..9874f7272 100644 --- a/src/storage/v2/disk/storage.hpp +++ b/src/storage/v2/disk/storage.hpp @@ -182,8 +182,6 @@ class DiskStorage final : public Storage { throw utils::NotYetImplemented("ListAllConstraints() is not implemented for DiskStorage."); } - void AdvanceCommand() override; - utils::BasicResult Commit( std::optional desired_commit_timestamp = {}) override; @@ -194,8 +192,6 @@ class DiskStorage final : public Storage { /// Currently, it does everything the same as in-memory version. void FinalizeTransaction() override; - std::optional GetTransactionId() const override; - /// Deserializes vertex from the string key and stores it into the vertices_ and lru_vertices_. /// Properties are deserialized from the value. /// The method should be called only when the vertex is not in the cache. @@ -321,11 +317,6 @@ class DiskStorage final : public Storage { ConstraintsInfo ListAllConstraints() const override; - StorageInfo GetInfo() const override; - - bool LockPath() override; - bool UnlockPath() override; - bool SetReplicaRole(io::network::Endpoint endpoint, const replication::ReplicationServerConfig &config) override; bool SetMainReplicationRole() override; @@ -390,7 +381,6 @@ class DiskStorage final : public Storage { Indices indices_; IsolationLevel isolation_level_; StorageMode storage_mode_; - Config config_; // TODO: This isn't really a commit log, it doesn't even care if a // transaction commited or aborted. We could probably combine this with diff --git a/src/storage/v2/inmemory/storage.cpp b/src/storage/v2/inmemory/storage.cpp index 6de8d9bd3..d7d72b690 100644 --- a/src/storage/v2/inmemory/storage.cpp +++ b/src/storage/v2/inmemory/storage.cpp @@ -71,10 +71,9 @@ std::string RegisterReplicaErrorToString(InMemoryStorage::RegisterReplicaError e InMemoryStorage::InMemoryStorage(Config config) : Storage(config), - indices_(&constraints_, config.items), isolation_level_(config.transaction.isolation_level), storage_mode_(StorageMode::IN_MEMORY_TRANSACTIONAL), - config_(config) { + indices_(&constraints_, config.items) { if (config_.durability.snapshot_wal_mode == Config::Durability::SnapshotWalMode::DISABLED && replication_role_ == ReplicationRole::MAIN) { spdlog::warn( @@ -603,8 +602,6 @@ Result> InMemoryStorage::InMemoryAccessor::DeleteEdg &mem_storage->indices_, &mem_storage->constraints_, config_, true); } -void InMemoryStorage::InMemoryAccessor::AdvanceCommand() { ++transaction_.command_id; } - utils::BasicResult InMemoryStorage::InMemoryAccessor::Commit( const std::optional desired_commit_timestamp) { MG_ASSERT(is_transaction_active_, "The transaction is already terminated!"); @@ -909,13 +906,6 @@ void InMemoryStorage::InMemoryAccessor::FinalizeTransaction() { } } -std::optional InMemoryStorage::InMemoryAccessor::GetTransactionId() const { - if (is_transaction_active_) { - return transaction_.transaction_id.load(std::memory_order_acquire); - } - return {}; -} - utils::BasicResult InMemoryStorage::CreateIndex( LabelId label, const std::optional desired_commit_timestamp) { std::unique_lock storage_guard(main_lock_); @@ -1092,18 +1082,6 @@ ConstraintsInfo InMemoryStorage::ListAllConstraints() const { return {ListExistenceConstraints(constraints_), constraints_.unique_constraints.ListConstraints()}; } -StorageInfo InMemoryStorage::GetInfo() const { - auto vertex_count = vertices_.size(); - auto edge_count = edge_count_.load(std::memory_order_acquire); - double average_degree = 0.0; - if (vertex_count) { - // NOLINTNEXTLINE(bugprone-narrowing-conversions, cppcoreguidelines-narrowing-conversions) - average_degree = 2.0 * static_cast(edge_count) / vertex_count; - } - return {vertex_count, edge_count, average_degree, utils::GetMemoryUsage(), - utils::GetDirDiskUsage(config_.durability.storage_directory)}; -} - VerticesIterable InMemoryStorage::InMemoryAccessor::Vertices(LabelId label, View view) { return VerticesIterable( static_cast(storage_)->indices_.label_index.Vertices(label, view, &transaction_)); @@ -1699,25 +1677,6 @@ utils::BasicResult InMemoryStorage::CreateSnapshot return CreateSnapshotError::ReachedMaxNumTries; } -bool InMemoryStorage::LockPath() { - auto locker_accessor = global_locker_.Access(); - return locker_accessor.AddPath(config_.durability.storage_directory); -} - -bool InMemoryStorage::UnlockPath() { - { - auto locker_accessor = global_locker_.Access(); - if (!locker_accessor.RemovePath(config_.durability.storage_directory)) { - return false; - } - } - - // We use locker accessor in seperate scope so we don't produce deadlock - // after we call clean queue. - file_retainer_.CleanQueue(); - return true; -} - void InMemoryStorage::FreeMemory() { CollectGarbage(); diff --git a/src/storage/v2/inmemory/storage.hpp b/src/storage/v2/inmemory/storage.hpp index ea7b0dbb9..6299f4eeb 100644 --- a/src/storage/v2/inmemory/storage.hpp +++ b/src/storage/v2/inmemory/storage.hpp @@ -209,8 +209,6 @@ class InMemoryStorage final : public Storage { mem_storage->constraints_.unique_constraints.ListConstraints()}; } - void AdvanceCommand() override; - /// Returns void if the transaction has been committed. /// Returns `StorageDataManipulationError` if an error occures. Error can be: /// * `ReplicationError`: there is at least one SYNC replica that has not confirmed receiving the transaction. @@ -225,8 +223,6 @@ class InMemoryStorage final : public Storage { void FinalizeTransaction() override; - std::optional GetTransactionId() const override; - private: /// @throw std::bad_alloc VertexAccessor CreateVertex(storage::Gid gid); @@ -322,11 +318,6 @@ class InMemoryStorage final : public Storage { ConstraintsInfo ListAllConstraints() const override; - StorageInfo GetInfo() const override; - - bool LockPath() override; - bool UnlockPath() override; - bool SetReplicaRole(io::network::Endpoint endpoint, const replication::ReplicationServerConfig &config) override; bool SetMainReplicationRole() override; @@ -391,7 +382,6 @@ class InMemoryStorage final : public Storage { StorageMode storage_mode_; Constraints constraints_; Indices indices_; - Config config_; // TODO: This isn't really a commit log, it doesn't even care if a // transaction commited or aborted. We could probably combine this with diff --git a/src/storage/v2/storage.cpp b/src/storage/v2/storage.cpp index 9cc8fe09d..720e3ca1f 100644 --- a/src/storage/v2/storage.cpp +++ b/src/storage/v2/storage.cpp @@ -10,6 +10,7 @@ // licenses/APL.txt. #include "storage/v2/storage.hpp" +#include "utils/stat.hpp" namespace memgraph::storage { @@ -279,6 +280,37 @@ Storage::Accessor::Accessor(Accessor &&other) noexcept other.commit_timestamp_.reset(); } +bool Storage::LockPath() { + auto locker_accessor = global_locker_.Access(); + return locker_accessor.AddPath(config_.durability.storage_directory); +} + +bool Storage::UnlockPath() { + { + auto locker_accessor = global_locker_.Access(); + if (!locker_accessor.RemovePath(config_.durability.storage_directory)) { + return false; + } + } + + // We use locker accessor in seperate scope so we don't produce deadlock + // after we call clean queue. + file_retainer_.CleanQueue(); + return true; +} + +StorageInfo Storage::GetInfo() const { + auto vertex_count = vertices_.size(); + auto edge_count = edge_count_.load(std::memory_order_acquire); + double average_degree = 0.0; + if (vertex_count) { + // NOLINTNEXTLINE(bugprone-narrowing-conversions, cppcoreguidelines-narrowing-conversions) + average_degree = 2.0 * static_cast(edge_count) / vertex_count; + } + return {vertex_count, edge_count, average_degree, utils::GetMemoryUsage(), + utils::GetDirDiskUsage(config_.durability.storage_directory)}; +} + // this should be handled on an above level of abstraction const std::string &Storage::LabelToName(LabelId label) const { return name_id_mapper_.IdToName(label.AsUint()); } @@ -321,4 +353,13 @@ PropertyId Storage::Accessor::NameToProperty(const std::string_view name) { retu EdgeTypeId Storage::Accessor::NameToEdgeType(const std::string_view name) { return storage_->NameToEdgeType(name); } +std::optional Storage::Accessor::GetTransactionId() const { + if (is_transaction_active_) { + return transaction_.transaction_id.load(std::memory_order_acquire); + } + return {}; +} + +void Storage::Accessor::AdvanceCommand() { ++transaction_.command_id; } + } // namespace memgraph::storage diff --git a/src/storage/v2/storage.hpp b/src/storage/v2/storage.hpp index c8756deb0..7a914b237 100644 --- a/src/storage/v2/storage.hpp +++ b/src/storage/v2/storage.hpp @@ -179,7 +179,8 @@ struct StorageInfo { class Storage { public: Storage(Config config) - : snapshot_directory_(config.durability.storage_directory / durability::kSnapshotDirectory), + : config_(config), + snapshot_directory_(config.durability.storage_directory / durability::kSnapshotDirectory), wal_directory_(config.durability.storage_directory / durability::kWalDirectory), lock_file_path_(config.durability.storage_directory / durability::kLockFile), uuid_(utils::GenerateUUID()), @@ -272,19 +273,6 @@ class Storage { /// @throw std::bad_alloc virtual Result> DeleteEdge(EdgeAccessor *edge) = 0; - const std::string &LabelToName(LabelId label) const; - const std::string &PropertyToName(PropertyId property) const; - const std::string &EdgeTypeToName(EdgeTypeId edge_type) const; - - /// @throw std::bad_alloc if unable to insert a new mapping - LabelId NameToLabel(std::string_view name); - - /// @throw std::bad_alloc if unable to insert a new mapping - PropertyId NameToProperty(std::string_view name); - - /// @throw std::bad_alloc if unable to insert a new mapping - EdgeTypeId NameToEdgeType(std::string_view name); - virtual bool LabelIndexExists(LabelId label) const = 0; virtual bool LabelPropertyIndexExists(LabelId label, PropertyId property) const = 0; @@ -293,8 +281,6 @@ class Storage { virtual ConstraintsInfo ListAllConstraints() const = 0; - virtual void AdvanceCommand() = 0; - /// Returns void if the transaction has been committed. /// Returns `StorageDataManipulationError` if an error occures. Error can be: /// * `ReplicationError`: there is at least one SYNC replica that has not confirmed receiving the transaction. @@ -309,7 +295,24 @@ class Storage { virtual void FinalizeTransaction() = 0; - virtual std::optional GetTransactionId() const = 0; + std::optional GetTransactionId() const; + + void AdvanceCommand(); + + const std::string &LabelToName(LabelId label) const; + + const std::string &PropertyToName(PropertyId property) const; + + const std::string &EdgeTypeToName(EdgeTypeId edge_type) const; + + /// @throw std::bad_alloc if unable to insert a new mapping + LabelId NameToLabel(std::string_view name); + + /// @throw std::bad_alloc if unable to insert a new mapping + PropertyId NameToProperty(std::string_view name); + + /// @throw std::bad_alloc if unable to insert a new mapping + EdgeTypeId NameToEdgeType(std::string_view name); protected: Storage *storage_; @@ -319,11 +322,15 @@ class Storage { bool is_transaction_active_; }; - virtual std::unique_ptr Access(std::optional override_isolation_level) = 0; - std::unique_ptr Access() { return Access(std::optional{}); } + bool LockPath(); + bool UnlockPath(); + + StorageInfo GetInfo() const; const std::string &LabelToName(LabelId label) const; + const std::string &PropertyToName(PropertyId property) const; + const std::string &EdgeTypeToName(EdgeTypeId edge_type) const; /// @throw std::bad_alloc if unable to insert a new mapping @@ -335,6 +342,9 @@ class Storage { /// @throw std::bad_alloc if unable to insert a new mapping EdgeTypeId NameToEdgeType(std::string_view name); + virtual std::unique_ptr Access(std::optional override_isolation_level) = 0; + std::unique_ptr Access() { return Access(std::optional{}); } + /// Create an index. /// Returns void if the index has been created. /// Returns `StorageIndexDefinitionError` if an error occures. Error can be: @@ -453,11 +463,6 @@ class Storage { virtual ConstraintsInfo ListAllConstraints() const = 0; - virtual StorageInfo GetInfo() const = 0; - - virtual bool LockPath() = 0; - virtual bool UnlockPath() = 0; - virtual bool SetReplicaRole(io::network::Endpoint endpoint, const replication::ReplicationServerConfig &config) = 0; bool SetReplicaRole(io::network::Endpoint endpoint) { @@ -546,6 +551,7 @@ class Storage { std::atomic edge_count_{0}; NameIdMapper name_id_mapper_; + Config config_; // Transaction engine utils::SpinLock engine_lock_;