Make Distributed Singleton (and remove unused code)

Reviewers: sasa.stanko, buda

Reviewed By: sasa.stanko

Differential Revision: https://phabricator.memgraph.io/D694

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<Channel> SenderMessage::GetChannelToSender(
-    Distributed *distributed) const {
+std::shared_ptr<Channel> 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);
 }

experimental/distributed/src/reactors_distributed.hpp

@@ -225,8 +225,6 @@ class Network {
   std::unique_ptr<NetworkServer> 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<Channel> GetChannelToSender(Distributed *distributed = nullptr) const;
+  std::shared_ptr<Channel> GetChannelToSender() const;
 
   template <class Archive>
   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;
 };

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:

experimental/distributed/tests/distributed_test.cpp

@@ -13,7 +13,15 @@ class MemgraphDistributed : public Distributed {
   using Location = std::pair<std::string, uint16_t>;
 
  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<int64_t, Location> 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<int64_t, std::vector<int64_t>>
-  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<int64_t, std::vector<int64_t>>
   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<int64_t, std::vector<int64_t>>
   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<int64_t> &&worker_mnids)
-    : MemgraphReactor(name, distributed), mnid_(mnid),
+  Master(std::string name, int64_t mnid, std::vector<int64_t> &&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<ChannelResolvedMessage>([this, stream](const ChannelResolvedMessage &msg){
           msg.channel()->Send<TextMessage>("master", "main", "hi from master");
@@ -153,16 +154,16 @@ class Master : public MemgraphReactor {
   std::vector<int64_t> 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<TextMessage>([this](const TextMessage &msg) {
       std::cout << "Message from " << msg.Address() << ":" << msg.Port() << " .. " << msg.text << "\n";
 
-      msg.GetChannelToSender(&distributed_)
+      msg.GetChannelToSender()
         ->Send<TextMessage>("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>("master", distributed, FLAGS_my_mnid, std::move(mnids.second));
+    system.Spawn<Master>("master", FLAGS_my_mnid, std::move(mnids.second));
   else
-    system.Spawn<Worker>("worker", distributed, FLAGS_my_mnid, mnids.first);
+    system.Spawn<Worker>("worker", FLAGS_my_mnid, mnids.first);
   system.AwaitShutdown();
   distributed.StopServices();
 

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<ChatMessage>("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<ChatServer>("server", distributed);
-  system.Spawn<ChatClient>("client", distributed);
+  system.Spawn<ChatServer>("server");
+  system.Spawn<ChatClient>("client");
   system.AwaitShutdown();
   distributed.StopServices();
   return 0;

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;