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

(un)lock path, get info and transaction id deduplication

Browse Source
This commit is contained in:
Andi Skrgat
2023-05-17 11:44:00 +02:00
parent 6f652fff42
commit 6df79724ca
6 changed files with 73 additions and 106 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<std::optional<EdgeAccessor>> 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<uint64_t> DiskStorage::DiskAccessor::GetTransactionId() const {
if (is_transaction_active_) {
return transaction_.transaction_id.load(std::memory_order_acquire);
}
return {};
}
utils::BasicResult<StorageIndexDefinitionError, void> DiskStorage::CreateIndex(
LabelId label, const std::optional<uint64_t> 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<PropertyId> &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::CreateSnapshotError> 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<uint64_t> desired_commit_timestamp) {

10
src/storage/v2/disk/storage.hpp
View File

@@ -182,8 +182,6 @@ class DiskStorage final : public Storage {
throw utils::NotYetImplemented("ListAllConstraints() is not implemented for DiskStorage.");
}
void AdvanceCommand() override;
utils::BasicResult<StorageDataManipulationError, void> Commit(
std::optional<uint64_t> 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<uint64_t> 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

43
src/storage/v2/inmemory/storage.cpp
View File

@@ -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<std::optional<EdgeAccessor>> InMemoryStorage::InMemoryAccessor::DeleteEdg
&mem_storage->indices_, &mem_storage->constraints_, config_, true);
}
void InMemoryStorage::InMemoryAccessor::AdvanceCommand() { ++transaction_.command_id; }
utils::BasicResult<StorageDataManipulationError, void> InMemoryStorage::InMemoryAccessor::Commit(
const std::optional<uint64_t> desired_commit_timestamp) {
MG_ASSERT(is_transaction_active_, "The transaction is already terminated!");
@@ -909,13 +906,6 @@ void InMemoryStorage::InMemoryAccessor::FinalizeTransaction() {
}
}
std::optional<uint64_t> InMemoryStorage::InMemoryAccessor::GetTransactionId() const {
if (is_transaction_active_) {
return transaction_.transaction_id.load(std::memory_order_acquire);
}
return {};
}
utils::BasicResult<StorageIndexDefinitionError, void> InMemoryStorage::CreateIndex(
LabelId label, const std::optional<uint64_t> desired_commit_timestamp) {
std::unique_lock<utils::RWLock> 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<double>(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<InMemoryStorage *>(storage_)->indices_.label_index.Vertices(label, view, &transaction_));
@@ -1699,25 +1677,6 @@ utils::BasicResult<Storage::CreateSnapshotError> 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<true>();

10
src/storage/v2/inmemory/storage.hpp
View File

@@ -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<uint64_t> 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

41
src/storage/v2/storage.cpp
View File

@@ -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<double>(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<uint64_t> 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

54
src/storage/v2/storage.hpp
View File

@@ -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<std::optional<EdgeAccessor>> 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<uint64_t> GetTransactionId() const = 0;
std::optional<uint64_t> 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<Accessor> Access(std::optional<IsolationLevel> override_isolation_level) = 0;
std::unique_ptr<Accessor> Access() { return Access(std::optional<IsolationLevel>{}); }
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<Accessor> Access(std::optional<IsolationLevel> override_isolation_level) = 0;
std::unique_ptr<Accessor> Access() { return Access(std::optional<IsolationLevel>{}); }
/// 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<uint64_t> edge_count_{0};
NameIdMapper name_id_mapper_;
Config config_;
// Transaction engine
utils::SpinLock engine_lock_;