diff --git a/customers/otto/parallel_connected_components.cpp b/customers/otto/parallel_connected_components.cpp index c1f22de66..b01f6f1f4 100644 --- a/customers/otto/parallel_connected_components.cpp +++ b/customers/otto/parallel_connected_components.cpp @@ -25,9 +25,9 @@ DECLARE_int32(gc_cycle_sec); static const std::string kLabel{"kLabel"}; static const std::string kProperty{"kProperty"}; -void GenerateGraph(GraphDb &db) { +void GenerateGraph(database::GraphDb &db) { { - GraphDbAccessor dba{db}; + database::GraphDbAccessor dba{db}; dba.BuildIndex(dba.Label(kLabel), dba.Property(kProperty)); dba.Commit(); } @@ -56,7 +56,7 @@ void GenerateGraph(GraphDb &db) { SpinLock vertices_lock; for (int i = 0; i < FLAGS_thread_count; ++i) { threads.emplace_back([&db, &vertex_ids, &vertices, &vertices_lock, i]() { - GraphDbAccessor dba{db}; + database::GraphDbAccessor dba{db}; auto label = dba.Label(kLabel); auto property = dba.Property(kProperty); auto batch_size = FLAGS_vertex_count / FLAGS_thread_count; @@ -76,7 +76,7 @@ void GenerateGraph(GraphDb &db) { << timer.Elapsed().count() << " seconds."; } { - GraphDbAccessor dba{db}; + database::GraphDbAccessor dba{db}; for (int i = 0; i < FLAGS_vertex_count; ++i) vertices[i] = *dba.Transfer(vertices[i]); @@ -96,32 +96,32 @@ void GenerateGraph(GraphDb &db) { } } -auto EdgeIteration(GraphDb &db) { - GraphDbAccessor dba{db}; +auto EdgeIteration(database::GraphDb &db) { + database::GraphDbAccessor dba{db}; int64_t sum{0}; for (auto edge : dba.Edges(false)) sum += edge.from().gid() + edge.to().gid(); return sum; } -auto VertexIteration(GraphDb &db) { - GraphDbAccessor dba{db}; +auto VertexIteration(database::GraphDb &db) { + database::GraphDbAccessor dba{db}; int64_t sum{0}; for (auto v : dba.Vertices(false)) for (auto e : v.out()) sum += e.gid() + e.to().gid(); return sum; } -auto ConnectedComponentsEdges(GraphDb &db) { +auto ConnectedComponentsEdges(database::GraphDb &db) { UnionFind connectivity{FLAGS_vertex_count}; - GraphDbAccessor dba{db}; + database::GraphDbAccessor dba{db}; for (auto edge : dba.Edges(false)) connectivity.Connect(edge.from().gid(), edge.to().gid()); return connectivity.Size(); } -auto ConnectedComponentsVertices(GraphDb &db) { +auto ConnectedComponentsVertices(database::GraphDb &db) { UnionFind connectivity{FLAGS_vertex_count}; - GraphDbAccessor dba{db}; + database::GraphDbAccessor dba{db}; for (auto from : dba.Vertices(false)) { for (auto out_edge : from.out()) connectivity.Connect(from.gid(), out_edge.to().gid()); @@ -129,7 +129,7 @@ auto ConnectedComponentsVertices(GraphDb &db) { return connectivity.Size(); } -auto ConnectedComponentsVerticesParallel(GraphDb &db) { +auto ConnectedComponentsVerticesParallel(database::GraphDb &db) { UnionFind connectivity{FLAGS_vertex_count}; SpinLock connectivity_lock; @@ -143,7 +143,7 @@ auto ConnectedComponentsVerticesParallel(GraphDb &db) { for (int i = 0; i < FLAGS_thread_count; ++i) { threads.emplace_back( [&connectivity, &connectivity_lock, &bounds, &db, i]() { - GraphDbAccessor dba{db}; + database::GraphDbAccessor dba{db}; for (auto from : dba.Vertices(dba.Label(kLabel), dba.Property(kProperty), utils::MakeBoundInclusive(bounds[i]), @@ -159,11 +159,11 @@ auto ConnectedComponentsVerticesParallel(GraphDb &db) { return connectivity.Size(); } -auto Expansion(GraphDb &db) { +auto Expansion(database::GraphDb &db) { std::vector component_ids(FLAGS_vertex_count, -1); int next_component_id{0}; std::stack expansion_stack; - GraphDbAccessor dba{db}; + database::GraphDbAccessor dba{db}; for (auto v : dba.Vertices(false)) { if (component_ids[v.gid()] != -1) continue; auto component_id = next_component_id++; @@ -186,7 +186,7 @@ int main(int argc, char **argv) { google::InitGoogleLogging(argv[0]); FLAGS_gc_cycle_sec = -1; - GraphDb db; + database::SingleNode db; GenerateGraph(db); auto timed_call = [&db](auto callable, const std::string &descr) { LOG(INFO) << "Running " << descr << "..."; diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index 682ce1da6..aa53a46bc 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -9,10 +9,10 @@ set(memgraph_src_files communication/messaging/protocol.cpp communication/rpc/rpc.cpp data_structures/concurrent/skiplist_gc.cpp + database/config.cpp database/counters.cpp database/graph_db.cpp database/graph_db_accessor.cpp - database/graph_db_config.cpp database/state_delta.cpp distributed/coordination_master.cpp distributed/coordination_worker.cpp diff --git a/src/communication/bolt/v1/session.hpp b/src/communication/bolt/v1/session.hpp index a5268c5cb..6df216828 100644 --- a/src/communication/bolt/v1/session.hpp +++ b/src/communication/bolt/v1/session.hpp @@ -26,12 +26,7 @@ namespace communication::bolt { /** Encapsulates Dbms and Interpreter that are passed through the network server * and worker to the session. */ struct SessionData { - /** Constructs a SessionData object. - * @param args - Arguments forwarded to the GraphDb constructor. */ - template - SessionData(TArgs &&... args) : db(std::forward(args)...) {} - - GraphDb db; + database::MasterBase &db; query::Interpreter interpreter; }; @@ -202,7 +197,7 @@ class Session { // TODO: Rethink if there is a way to hide some members. At the momement all // of them are public. TSocket socket_; - GraphDb &db_; + database::MasterBase &db_; query::Interpreter &interpreter_; TimeoutSocket timeout_socket_{*this}; @@ -218,7 +213,7 @@ class Session { State state_{State::Handshake}; // GraphDbAccessor of active transaction in the session, can be null if // there is no associated transaction. - std::unique_ptr db_accessor_; + std::unique_ptr db_accessor_; // Time of the last event. std::chrono::time_point last_event_time_ = std::chrono::steady_clock::now(); diff --git a/src/communication/bolt/v1/states/executing.hpp b/src/communication/bolt/v1/states/executing.hpp index 686ccf342..306d324cc 100644 --- a/src/communication/bolt/v1/states/executing.hpp +++ b/src/communication/bolt/v1/states/executing.hpp @@ -75,13 +75,14 @@ State HandleRun(TSession &session, State state, Marker marker) { // TODO: Possible (but very unlikely) race condition, where we have alive // session during shutdown, but is_accepting_transactions_ isn't yet false. // We should probably create transactions under some locking mechanism. - if (!session.db_.is_accepting_transactions_) { + if (!session.db_.is_accepting_transactions()) { // Db is shutting down and doesn't accept new transactions so we should // close this session. return State::Close; } // Create new transaction. - session.db_accessor_ = std::make_unique(session.db_); + session.db_accessor_ = + std::make_unique(session.db_); } // If there was not explicitly started transaction before maybe we are diff --git a/src/communication/messaging/distributed.hpp b/src/communication/messaging/distributed.hpp index 49dd80b50..294d2f9a2 100644 --- a/src/communication/messaging/distributed.hpp +++ b/src/communication/messaging/distributed.hpp @@ -58,7 +58,7 @@ class System { friend class Writer; System(const std::string &address, uint16_t port); - System(const Endpoint &endpoint); + explicit System(const Endpoint &endpoint); System(const System &) = delete; System(System &&) = delete; System &operator=(const System &) = delete; diff --git a/src/database/graph_db_config.cpp b/src/database/config.cpp similarity index 51% rename from src/database/graph_db_config.cpp rename to src/database/config.cpp index 2ece07751..b41a17ceb 100644 --- a/src/database/graph_db_config.cpp +++ b/src/database/config.cpp @@ -1,19 +1,8 @@ -#include -#include #include -#include - #include "database/graph_db.hpp" #include "utils/flag_validation.hpp" -namespace fs = std::experimental::filesystem; - -// TODO review: tech docs say the default here is 'true', which it is in the -// community config. Should we set the default here to true? On some other -// points the tech docs are consistent with community config, and not with these -// defaults. - // Durability flags. DEFINE_bool(durability_enabled, false, "If durability (database persistence) should be enabled"); @@ -29,15 +18,37 @@ DEFINE_int32(snapshot_max_retained, -1, "Number of retained snapshots, -1 means without limit."); DEFINE_bool(snapshot_on_exit, false, "Snapshot on exiting the database."); -// Misc flags. -DEFINE_int32(gc_cycle_sec, 30, - "Amount of time between starts of two cleaning cycles in seconds. " - "-1 to turn off."); +// Misc flags DEFINE_int32(query_execution_time_sec, 180, "Maximum allowed query execution time. Queries exceeding this " "limit will be aborted. Value of -1 means no limit."); +DEFINE_int32(gc_cycle_sec, 30, + "Amount of time between starts of two cleaning cycles in seconds. " + "-1 to turn off."); -GraphDb::Config::Config() +// Distributed master/worker flags. +DEFINE_HIDDEN_int32(worker_id, 0, + "ID of a worker in a distributed system. Igored in " + "single-node and distributed-master."); +DEFINE_HIDDEN_string(master_host, "0.0.0.0", + "For master node indicates the host served on. For worker " + "node indicates the master location."); +DEFINE_VALIDATED_HIDDEN_int32( + master_port, 0, + "For master node the port on which to serve. For " + "worker node indicates the master's port.", + FLAG_IN_RANGE(0, std::numeric_limits::max())); +DEFINE_HIDDEN_string(worker_host, "0.0.0.0", + "For worker node indicates the host served on. For master " + "node this flag is not used."); +DEFINE_VALIDATED_HIDDEN_int32( + worker_port, 0, + "For master node it's unused. For worker node " + "indicates the port on which to serve. If zero (default value), a port is " + "chosen at random. Sent to the master when registring worker node.", + FLAG_IN_RANGE(0, std::numeric_limits::max())); + +database::Config::Config() // Durability flags. : durability_enabled{FLAGS_durability_enabled}, durability_directory{FLAGS_durability_directory}, @@ -47,4 +58,10 @@ GraphDb::Config::Config() snapshot_on_exit{FLAGS_snapshot_on_exit}, // Misc flags. gc_cycle_sec{FLAGS_gc_cycle_sec}, - query_execution_time_sec{FLAGS_query_execution_time_sec} {} + query_execution_time_sec{FLAGS_query_execution_time_sec}, + // Distributed flags. + worker_id{FLAGS_worker_id}, + master_endpoint{FLAGS_master_host, + static_cast(FLAGS_master_port)}, + worker_endpoint{FLAGS_worker_host, + static_cast(FLAGS_worker_port)} {} diff --git a/src/database/graph_db.cpp b/src/database/graph_db.cpp index 631ef10d0..05d457c2d 100644 --- a/src/database/graph_db.cpp +++ b/src/database/graph_db.cpp @@ -1,95 +1,191 @@ -#include -#include - -#include - #include "database/graph_db.hpp" -#include "database/graph_db_accessor.hpp" + +#include "communication/messaging/distributed.hpp" +#include "distributed/coordination_master.hpp" +#include "distributed/coordination_worker.hpp" #include "durability/paths.hpp" #include "durability/recovery.hpp" #include "durability/snapshooter.hpp" #include "storage/concurrent_id_mapper_master.hpp" +#include "storage/concurrent_id_mapper_single_node.hpp" #include "storage/concurrent_id_mapper_worker.hpp" #include "transactions/engine_master.hpp" #include "transactions/engine_single_node.hpp" #include "transactions/engine_worker.hpp" -#include "utils/timer.hpp" +#include "utils/flag_validation.hpp" -namespace fs = std::experimental::filesystem; +namespace database { +namespace impl { -#define INIT_MAPPERS(type, ...) \ - labels_ = std::make_unique>(__VA_ARGS__); \ - edge_types_ = std::make_unique>(__VA_ARGS__); \ - properties_ = std::make_unique>(__VA_ARGS__); +class Base { + public: + explicit Base(const Config &config) : config_(config) {} + virtual ~Base() {} -GraphDb::GraphDb(Config config) : GraphDb(config, 0) { - tx_engine_ = std::make_unique(&wal_); - counters_ = std::make_unique(); - INIT_MAPPERS(storage::SingleNodeConcurrentIdMapper); - Start(); -} + const Config config_; -GraphDb::GraphDb(communication::messaging::System &system, - distributed::MasterCoordination &master, Config config) - : GraphDb(config, 0) { - tx_engine_ = std::make_unique(system, &wal_); - auto counters = std::make_unique(system); - counters_ = std::move(counters); - INIT_MAPPERS(storage::MasterConcurrentIdMapper, system); - get_endpoint_ = [&master](int worker_id) { - return master.GetEndpoint(worker_id); - }; - Start(); -} + virtual Storage &storage() = 0; + virtual StorageGc &storage_gc() = 0; + virtual durability::WriteAheadLog &wal() = 0; + virtual tx::Engine &tx_engine() = 0; + virtual storage::ConcurrentIdMapper