From abedfb29b17019b0da39ea75d3cf44ae6c414c44 Mon Sep 17 00:00:00 2001 From: Goran Zuzic Date: Tue, 22 Aug 2017 17:13:12 +0200 Subject: [PATCH] Make Distributed Singleton (and remove unused code) Reviewers: sasa.stanko, buda Reviewed By: sasa.stanko Differential Revision: https://phabricator.memgraph.io/D694 --- .../distributed/src/reactors_distributed.cpp | 8 +- .../distributed/src/reactors_distributed.hpp | 42 ++++++----- .../distributed/src/reactors_local.hpp | 2 +- .../distributed/tests/distributed_test.cpp | 73 ++++++++++--------- .../distributed/tests/network_chat.cpp | 20 ++--- .../distributed/tests/network_client.cpp | 2 +- 6 files changed, 75 insertions(+), 72 deletions(-) diff --git a/experimental/distributed/src/reactors_distributed.cpp b/experimental/distributed/src/reactors_distributed.cpp index e09a02ce2..09379fc10 100644 --- a/experimental/distributed/src/reactors_distributed.cpp +++ b/experimental/distributed/src/reactors_distributed.cpp @@ -21,12 +21,12 @@ uint16_t SenderMessage::Port() const { return port_; } std::string SenderMessage::ReactorName() const { return reactor_; } std::string SenderMessage::ChannelName() const { return channel_; } -std::shared_ptr SenderMessage::GetChannelToSender( - Distributed *distributed) const { +std::shared_ptr SenderMessage::GetChannelToSender() const { if (address_ == FLAGS_address && port_ == FLAGS_port) { return System::GetInstance().FindChannel(reactor_, channel_); + } else { + // TODO(zuza): we should probably assert here if services have been already started. + return Distributed::GetInstance().network().Resolve(address_, port_, reactor_, channel_); } - if (distributed) - return distributed->network().Resolve(address_, port_, reactor_, channel_); assert(false); } diff --git a/experimental/distributed/src/reactors_distributed.hpp b/experimental/distributed/src/reactors_distributed.hpp index ccdeb6f4f..53d866e77 100644 --- a/experimental/distributed/src/reactors_distributed.hpp +++ b/experimental/distributed/src/reactors_distributed.hpp @@ -225,8 +225,6 @@ class Network { std::unique_ptr server_{nullptr}; }; -class Distributed; - /** * Message that includes the sender channel used to respond. */ @@ -240,7 +238,7 @@ class SenderMessage : public Message { std::string ReactorName() const; std::string ChannelName() const; - std::shared_ptr GetChannelToSender(Distributed *distributed = nullptr) const; + std::shared_ptr GetChannelToSender() const; template void serialize(Archive &ar) { @@ -274,17 +272,23 @@ class ChannelResolvedMessage : public Message { }; /** - * Placeholder for all functionality related to non-local communication + * Placeholder for all functionality related to non-local communication. + * * E.g. resolve remote channels by memgraph node id, etc. + * Alive through the entire process lifetime. + * Singleton class. Created automatically on first use. */ class Distributed { public: - Distributed() {} - - Distributed(const Distributed &) = delete; - Distributed(Distributed &&) = delete; - Distributed &operator=(const Distributed &) = delete; - Distributed &operator=(Distributed &&) = delete; + /** + * Get the (singleton) instance of Distributed. + * + * More info: https://stackoverflow.com/questions/1008019/c-singleton-design-pattern + */ + static Distributed &GetInstance() { + static Distributed distributed; // guaranteed to be destroyed, initialized on first use + return distributed; + } void StartServices() { network_.StartClient(4); @@ -321,15 +325,13 @@ class Distributed { Network &network() { return network_; } protected: + Distributed() {} + Network network_; -}; - - -class DistributedReactor : public Reactor { - public: - DistributedReactor(std::string name, Distributed &distributed) - : Reactor(name), distributed_(distributed) {} - - protected: - Distributed &distributed_; + + private: + Distributed(const Distributed &) = delete; + Distributed(Distributed &&) = delete; + Distributed &operator=(const Distributed &) = delete; + Distributed &operator=(Distributed &&) = delete; }; diff --git a/experimental/distributed/src/reactors_local.hpp b/experimental/distributed/src/reactors_local.hpp index 4f053e31d..6cbafcb02 100644 --- a/experimental/distributed/src/reactors_local.hpp +++ b/experimental/distributed/src/reactors_local.hpp @@ -465,7 +465,7 @@ class Reactor { * * E.g. holds set of reactors, channels for all reactors. * Alive through the entire process lifetime. - * Singleton class. Created automatically. + * Singleton class. Created automatically on first use. */ class System { public: diff --git a/experimental/distributed/tests/distributed_test.cpp b/experimental/distributed/tests/distributed_test.cpp index 52d37351d..718b8099c 100644 --- a/experimental/distributed/tests/distributed_test.cpp +++ b/experimental/distributed/tests/distributed_test.cpp @@ -13,7 +13,15 @@ class MemgraphDistributed : public Distributed { using Location = std::pair; public: - MemgraphDistributed(System &system) : Distributed() {} + /** + * Get the (singleton) instance of MemgraphDistributed. + * + * More info: https://stackoverflow.com/questions/1008019/c-singleton-design-pattern + */ + static MemgraphDistributed &GetInstance() { + static MemgraphDistributed distributed; // guaranteed to be destroyed, initialized on first use + return distributed; + } /** Register memgraph node id to the given location. */ void RegisterMemgraphNode(int64_t mnid, const std::string &address, uint16_t port) { @@ -29,9 +37,17 @@ class MemgraphDistributed : public Distributed { return Distributed::FindChannel(location.first, location.second, reactor, channel); } + protected: + MemgraphDistributed() {} + private: std::recursive_mutex mutex_; std::unordered_map mnodes_; + + MemgraphDistributed(const MemgraphDistributed &) = delete; + MemgraphDistributed(MemgraphDistributed &&) = delete; + MemgraphDistributed &operator=(const MemgraphDistributed &) = delete; + MemgraphDistributed &operator=(MemgraphDistributed &&) = delete; }; /** @@ -49,8 +65,7 @@ class MemgraphDistributed : public Distributed { * @return Pair (master mnid, list of worker's id). */ std::pair> - ParseConfigAndRegister(const std::string &filename, - MemgraphDistributed &distributed) { + ParseConfigAndRegister(const std::string &filename) { std::ifstream file(filename, std::ifstream::in); assert(file.good()); int64_t master_mnid; @@ -59,6 +74,7 @@ std::pair> std::string address; uint16_t port; file >> master_mnid >> address >> port; + MemgraphDistributed &distributed = MemgraphDistributed::GetInstance(); distributed.RegisterMemgraphNode(master_mnid, address, port); while (file.good()) { file >> mnid >> address >> port; @@ -71,21 +87,6 @@ std::pair> return std::make_pair(master_mnid, worker_mnids); } -/** - * Interface to the Memgraph distributed system. - * - * E.g. get channel to other Memgraph nodes. - */ -class MemgraphReactor : public Reactor { - public: - MemgraphReactor(std::string name, - MemgraphDistributed &distributed) - : Reactor(name), distributed_(distributed) {} - - protected: - MemgraphDistributed &distributed_; -}; - /** * Sends a text message and has a return address. */ @@ -108,16 +109,16 @@ protected: CEREAL_REGISTER_TYPE(TextMessage); -class Master : public MemgraphReactor { +class Master : public Reactor { public: - Master(std::string name, MemgraphDistributed &distributed, - int64_t mnid, std::vector &&worker_mnids) - : MemgraphReactor(name, distributed), mnid_(mnid), + Master(std::string name, int64_t mnid, std::vector &&worker_mnids) + : Reactor(name), mnid_(mnid), worker_mnids_(std::move(worker_mnids)) {} virtual void Run() { - std::cout << "Master (" << mnid_ << ") @ " << distributed_.network().Address() - << ":" << distributed_.network().Port() << std::endl; + MemgraphDistributed &distributed = MemgraphDistributed::GetInstance(); + std::cout << "Master (" << mnid_ << ") @ " << distributed.network().Address() + << ":" << distributed.network().Port() << std::endl; auto stream = main_.first; @@ -138,7 +139,7 @@ class Master : public MemgraphReactor { // send a TextMessage to each worker for (auto wmnid : worker_mnids_) { - auto stream = distributed_.FindChannel(wmnid, "worker", "main"); + auto stream = distributed.FindChannel(wmnid, "worker", "main"); stream->OnEventOnce() .ChainOnce([this, stream](const ChannelResolvedMessage &msg){ msg.channel()->Send("master", "main", "hi from master"); @@ -153,16 +154,16 @@ class Master : public MemgraphReactor { std::vector worker_mnids_; }; -class Worker : public MemgraphReactor { +class Worker : public Reactor { public: - Worker(std::string name, MemgraphDistributed &distributed, - int64_t mnid, int64_t master_mnid) - : MemgraphReactor(name, distributed), mnid_(mnid), + Worker(std::string name, int64_t mnid, int64_t master_mnid) + : Reactor(name), mnid_(mnid), master_mnid_(master_mnid) {} virtual void Run() { - std::cout << "Worker (" << mnid_ << ") @ " << distributed_.network().Address() - << ":" << distributed_.network().Port() << std::endl; + MemgraphDistributed &distributed = MemgraphDistributed::GetInstance(); + std::cout << "Worker (" << mnid_ << ") @ " << distributed.network().Address() + << ":" << distributed.network().Port() << std::endl; auto stream = main_.first; // wait until master sends us a TextMessage, then reply back and close @@ -170,7 +171,7 @@ class Worker : public MemgraphReactor { .ChainOnce([this](const TextMessage &msg) { std::cout << "Message from " << msg.Address() << ":" << msg.Port() << " .. " << msg.text << "\n"; - msg.GetChannelToSender(&distributed_) + msg.GetChannelToSender() ->Send("worker", "main", "hi from worker"); // Sleep for a while so we can read output in the terminal. @@ -190,13 +191,13 @@ int main(int argc, char *argv[]) { gflags::ParseCommandLineFlags(&argc, &argv, true); System &system = System::GetInstance(); - MemgraphDistributed distributed(system); - auto mnids = ParseConfigAndRegister(FLAGS_config_filename, distributed); + MemgraphDistributed& distributed = MemgraphDistributed::GetInstance(); + auto mnids = ParseConfigAndRegister(FLAGS_config_filename); distributed.StartServices(); if (FLAGS_my_mnid == mnids.first) - system.Spawn("master", distributed, FLAGS_my_mnid, std::move(mnids.second)); + system.Spawn("master", FLAGS_my_mnid, std::move(mnids.second)); else - system.Spawn("worker", distributed, FLAGS_my_mnid, mnids.first); + system.Spawn("worker", FLAGS_my_mnid, mnids.first); system.AwaitShutdown(); distributed.StopServices(); diff --git a/experimental/distributed/tests/network_chat.cpp b/experimental/distributed/tests/network_chat.cpp index 2e624e5db..10174b8bb 100644 --- a/experimental/distributed/tests/network_chat.cpp +++ b/experimental/distributed/tests/network_chat.cpp @@ -36,10 +36,10 @@ class ChatACK : public ChatMessage { }; CEREAL_REGISTER_TYPE(ChatACK); -class ChatServer : public DistributedReactor { +class ChatServer : public Reactor { public: - ChatServer(std::string name, Distributed &distributed) - : DistributedReactor(name, distributed) {} + ChatServer(std::string name) + : Reactor(name) {} virtual void Run() { std::cout << "ChatServer is active" << std::endl; @@ -64,10 +64,10 @@ class ChatServer : public DistributedReactor { } }; -class ChatClient : public DistributedReactor { +class ChatClient : public Reactor { public: - ChatClient(std::string name, Distributed &distributed) - : DistributedReactor(name, distributed) {} + ChatClient(std::string name) + : Reactor(name) {} virtual void Run() { std::cout << "ChatClient is active" << std::endl; @@ -79,7 +79,7 @@ class ChatClient : public DistributedReactor { std::cin >> address >> port >> message; auto channel = - distributed_.network().Resolve(address, port, "server", "chat"); + Distributed::GetInstance().network().Resolve(address, port, "server", "chat"); if (channel != nullptr) { channel->Send("server", "chat", message); } else { @@ -92,10 +92,10 @@ class ChatClient : public DistributedReactor { int main(int argc, char *argv[]) { gflags::ParseCommandLineFlags(&argc, &argv, true); System& system = System::GetInstance(); - Distributed distributed; + Distributed &distributed = Distributed::GetInstance(); distributed.StartServices(); - system.Spawn("server", distributed); - system.Spawn("client", distributed); + system.Spawn("server"); + system.Spawn("client"); system.AwaitShutdown(); distributed.StopServices(); return 0; diff --git a/experimental/distributed/tests/network_client.cpp b/experimental/distributed/tests/network_client.cpp index b8ab44236..0d3ebe096 100644 --- a/experimental/distributed/tests/network_client.cpp +++ b/experimental/distributed/tests/network_client.cpp @@ -2,7 +2,7 @@ int main(int argc, char *argv[]) { google::InitGoogleLogging(argv[0]); - Distributed distributed; + Distributed &distributed = Distributed::GetInstance(); distributed.network().StartClient(1); auto channel = distributed.network().Resolve("127.0.0.1", 10000, "master", "main"); std::cout << channel << std::endl;