diff --git a/experimental/distributed/main-client.cpp b/experimental/distributed/main-client.cpp index b9030de5b..77c87e016 100644 --- a/experimental/distributed/main-client.cpp +++ b/experimental/distributed/main-client.cpp @@ -1,12 +1,31 @@ -#include "reactors_distributed.hpp" +#include +#include +#include +#include "reactors_distributed.hpp" #include "memgraph_config.hpp" #include "memgraph_distributed.hpp" #include "memgraph_transactions.hpp" -#include -#include -#include +/** + * List of queries that should be executed. + */ +std::vector queries = {{ + "create vertex", + "create vertex", + "create vertex", + "create vertex", + "create vertex", + "create vertex", + "create vertex", + "create vertex", + "create vertex", + "create vertex", + "vertex count", + "create vertex", + "create vertex", + "vertex count" +}}; /** * This is the client that issues some hard-coded queries. @@ -17,40 +36,31 @@ class Client : public Reactor { } void IssueQueries(std::shared_ptr channel_to_leader) { - const int NUM_VERTS = 10; // (concurrently) create a couple of vertices - for (int num_vert = 0; num_vert < NUM_VERTS; ++num_vert) { + for (int query_idx = 0; query_idx < queries.size(); ++query_idx) { // register callback - - std::string channel_name = "create-node-" + std::to_string(num_vert); - // TODO(zuza): this is actually pretty bad because if SuccessQueryCreateVertex arrives, then - // FailureQueryCreateVertex never gets unsubscribed. This could cause memory leaks - // in the future (not currently since all callbacks get destroyed when channel is closed). - // The best thing to do is to implement a ThenOnce and Either. Perhaps even a ThenClose. + std::string channel_name = "query-" + std::to_string(query_idx); auto stream = Open(channel_name).first; stream ->OnEventOnce() - .ChainOnce([this, num_vert](const SuccessQueryCreateVertex&, const Subscription& sub) { - LOG(INFO) << "successfully created vertex " << num_vert+1 << std::endl; - sub.CloseChannel(); - }); - - stream - ->OnEventOnce() - .ChainOnce([this, num_vert](const FailureQueryCreateVertex&, const Subscription& sub) { - LOG(INFO) << "failed on creating vertex " << num_vert+1 << std::endl; - sub.CloseChannel(); - }); + .ChainOnce([this, query_idx](const ResultMsg &msg, + const Subscription &sub){ + std::cout << "Result of query " << query_idx << " (" + << queries[query_idx] << "):" << std::endl + << " " << msg.result() << std::endl; + sub.CloseChannel(); + }); // then issue the query (to avoid race conditions) - LOG(INFO) << "Issuing command to create vertex " << num_vert+1; - channel_to_leader->Send(channel_name); + std::cout << "Issuing command " << query_idx << " (" + << queries[query_idx] << ")" << std::endl; + channel_to_leader->Send(channel_name, queries[query_idx]); } } virtual void Run() { MemgraphDistributed& memgraph = MemgraphDistributed::GetInstance(); - int mnid = memgraph.LeaderMnid(); + auto mnid = memgraph.LeaderMnid(); memgraph.FindChannel(mnid, "master", "client-queries") ->OnEventOnce() diff --git a/experimental/distributed/main.cpp b/experimental/distributed/main.cpp index 5fed57528..bb8d91640 100644 --- a/experimental/distributed/main.cpp +++ b/experimental/distributed/main.cpp @@ -1,45 +1,23 @@ -#include "memgraph_config.hpp" -#include "memgraph_distributed.hpp" -#include "memgraph_transactions.hpp" - -#include "reactors_distributed.hpp" - #include #include #include #include +#include "memgraph_config.hpp" +#include "memgraph_distributed.hpp" +#include "memgraph_transactions.hpp" +#include "reactors_distributed.hpp" +#include "storage.hpp" + DEFINE_uint64(my_mnid, -1, "Memgraph node id"); // TODO(zuza): this should be assigned by the leader once in the future -/** - * Sends a text message and has a return address. - */ -class TextMessage : public ReturnAddressMsg { - public: - TextMessage(std::string reactor, std::string channel, std::string s) - : ReturnAddressMsg(reactor, channel), text(s) {} - - template - void serialize(Archive &archive) { - archive(cereal::virtual_base_class(this), text); - } - - std::string text; - - protected: - friend class cereal::access; - TextMessage() {} // Cereal needs access to a default constructor. -}; -CEREAL_REGISTER_TYPE(TextMessage); - class Master : public Reactor { public: - Master(std::string name, MnidT mnid) - : Reactor(name), mnid_(mnid) { + Master(std::string name, MnidT mnid) : Reactor(name), mnid_(mnid) { MemgraphDistributed& memgraph = MemgraphDistributed::GetInstance(); worker_mnids_ = memgraph.GetAllMnids(); - // remove the leader (itself), because its not a worker + // remove the leader (itself), because it is not a worker auto leader_it = std::find(worker_mnids_.begin(), worker_mnids_.end(), memgraph.LeaderMnid()); worker_mnids_.erase(leader_it); } @@ -47,47 +25,170 @@ class Master : public Reactor { virtual void Run() { Distributed &distributed = Distributed::GetInstance(); - LOG(INFO) << "Master (" << mnid_ << ") @ " << distributed.network().Address() + std::cout << "Master (" << mnid_ << ") @ " << distributed.network().Address() << ":" << distributed.network().Port() << std::endl; // TODO(zuza): check if all workers are up + // start listening on queries arriving from the client auto stream = Open("client-queries").first; - stream->OnEvent([this](const QueryCreateVertex& msg, const Subscription&) { - std::random_device rd; // slow random number generator + stream->OnEvent([this](const QueryMsg &msg, const Subscription &){ + // process query message + if (msg.query() == "create vertex") { + InstallMakeVertex(msg.GetReturnChannelWriter()); + } else if (msg.query() == "vertex count") { + InstallVertexCount(msg.GetReturnChannelWriter()); + } else { + std::cerr << "unknown query" << std::endl; + } + }); + } - // succeed and fail with 50-50 - if (rd() % 2 == 0) { - msg.GetReturnChannelWriter() - ->Send(); - } else { - msg.GetReturnChannelWriter() - ->Send(); + private: + /** + * Organizes communication with all workers and performs VertexCount. + */ + void InstallVertexCount(std::shared_ptr return_channel) { + // open channel through which answers will arrive + auto channel_name = "response" + std::to_string(xid++); + auto result = Open(channel_name).first; + + // create struct to keep track of responses + struct VertexCountResponse { + VertexCountResponse(int64_t count, int64_t remaining) + : count_(count), remaining_(remaining) {} + + int64_t count_; + int64_t remaining_; + }; + + // allocate it dynamically so it lives outside the scope of this function + // it will be deallocated once all responses arrive and channel is closed + auto response = std::make_shared(0, worker_mnids_.size()); + + // register callbacks + result->OnEvent( + [this, response, return_channel](const ResultQueryVertexCount &msg, + const Subscription &sub){ + response->count_ += msg.count(); + --response->remaining_; + if (response->remaining_ == 0) { + sub.CloseChannel(); + return_channel->Send(std::to_string(response->count_)); } }); + // instruct workers to count vertices + for (auto wmnid : worker_mnids_) + VertexCount(wmnid, channel_name); + } + + /** + * Asynchronously counts vertices on the given node. + * + * @param mnid Id of the node whose vertices should be counted. + * @param channel_name Name of the channel on which response will arrive. + */ + void VertexCount(MnidT mnid, std::string channel_name) { + MemgraphDistributed::GetInstance().FindChannel(mnid, "worker", "main") + ->OnEventOnceThenClose( + [this, channel_name](const ChannelResolvedMessage &msg){ + msg.channelWriter()->Send(channel_name); + }); + } + + /** + * Organizes communication with a random worker and performs MakeVertex. + */ + void InstallMakeVertex(std::shared_ptr return_channel) { + // choose worker on random and instruct it to make vertex + auto wmnid = worker_mnids_[rand() % worker_mnids_.size()]; + + // open channel through which answer will arrive + auto channel_name = "response" + std::to_string(xid++); + auto result = Open(channel_name).first; + + // register callbacks for the answer + // TODO(zuza): this is actually pretty bad because if SuccessQueryCreateVertex arrives, then + // FailureQueryCreateVertex never gets unsubscribed. This could cause memory leaks + // in the future (not currently since all callbacks get destroyed when channel is closed). + // The best thing to do is to implement a ThenOnce and Either. Perhaps even a ThenClose. + // An Either in conjunction with a failure detector event stream should eventually fail + // the transaction and close the channel. + result->OnEventOnceThenClose( + [this, return_channel](const SuccessQueryCreateVertex &) { + return_channel->Send("success"); + }); + result->OnEventOnceThenClose( + [this, return_channel](const FailureQueryCreateVertex &) { + return_channel->Send("failure"); + }); + + // instruct worker to make vertex + MakeVertex(wmnid, channel_name); + } + + /** + * Asynchronously creates vertex on the give node. + * + * @param mnid Id of the node on which vertex should be created. + * @param channel_name Name of the channel on which response will arrive. + */ + void MakeVertex(MnidT mnid, std::string channel_name) { + MemgraphDistributed::GetInstance().FindChannel(mnid, "worker", "main") + ->OnEventOnceThenClose( + [this, channel_name](const ChannelResolvedMessage &msg){ + msg.channelWriter()->Send(channel_name); + }); } protected: - MnidT workers_seen = 0; + // node id const MnidT mnid_; + + // transaction id + int64_t xid{0}; + + // list of ids of nodes that act as worker std::vector worker_mnids_; }; class Worker : public Reactor { public: Worker(std::string name, MnidT mnid) - : Reactor(name), mnid_(mnid) {} + : Reactor(name), mnid_(mnid), storage_(mnid) {} virtual void Run() { Distributed &distributed = Distributed::GetInstance(); - LOG(INFO) << "Worker (" << mnid_ << ") @ " << distributed.network().Address() + std::cout << "Worker (" << mnid_ << ") @ " << distributed.network().Address() << ":" << distributed.network().Port() << std::endl; + + main_.first->OnEvent([this](const QueryCreateVertex& msg, + const Subscription &) { + std::random_device rd; // slow random number generator + + // succeed and fail with 50-50 (just for testing) + // TODO: remove random failure + if (rd() % 2 == 0) { + storage_.MakeVertex(); + std::cout << "Vertex created" << std::endl; + msg.GetReturnChannelWriter()->Send(); + } else { + msg.GetReturnChannelWriter()->Send(); + } + }); + + main_.first->OnEvent([this](const QueryVertexCount &msg, + const Subscription &){ + auto count = storage_.VertexCount(); + msg.GetReturnChannelWriter()->Send(count); + }); } protected: const MnidT mnid_; + ShardedStorage storage_; }; int main(int argc, char *argv[]) { diff --git a/experimental/distributed/src/memgraph_transactions.hpp b/experimental/distributed/src/memgraph_transactions.hpp index ac41c8147..a7fae29ab 100644 --- a/experimental/distributed/src/memgraph_transactions.hpp +++ b/experimental/distributed/src/memgraph_transactions.hpp @@ -1,11 +1,66 @@ #pragma once +#include + #include "reactors_local.hpp" #include "reactors_distributed.hpp" +/** + * Message which encapsulates query. + * It is create on Client and sent to Master which will process it. + */ +class QueryMsg : public ReturnAddressMsg { + public: + QueryMsg(std::string return_channel, std::string query) + : ReturnAddressMsg(return_channel), query_(query) {} + + const std::string &query() const { return query_; } + + template + void serialize(Archive &archive) { + archive(cereal::virtual_base_class(this), query_); + } + +protected: + // Cereal needs access to default constructor. + friend class cereal::access; + QueryMsg() = default; + + std::string query_; +}; +CEREAL_REGISTER_TYPE(QueryMsg); + +/** + * Message which encapuslates result of a query. + * Currently, result is string. + */ +class ResultMsg : public Message { + public: + ResultMsg(std::string result) : result_(result) {} + + const std::string &result() const { return result_; } + + template + void serialize(Archive &archive) { + archive(cereal::virtual_base_class(this), result_); + } + + protected: + friend class cereal::access; + ResultMsg() = default; + + std::string result_; +}; +CEREAL_REGISTER_TYPE(ResultMsg); + +/** + * Below are message that are exchanged between Master and Workers. + */ + class QueryCreateVertex : public ReturnAddressMsg { public: - QueryCreateVertex(std::string return_channel) : ReturnAddressMsg(return_channel) {} + QueryCreateVertex(std::string return_channel) + : ReturnAddressMsg(return_channel) {} template void serialize(Archive &archive) { @@ -13,8 +68,9 @@ public: } protected: + // Cereal needs access to default constructor. friend class cereal::access; - QueryCreateVertex() {} // Cereal needs access to a default constructor. + QueryCreateVertex() {} }; CEREAL_REGISTER_TYPE(QueryCreateVertex); @@ -29,7 +85,6 @@ public: }; CEREAL_REGISTER_TYPE(SuccessQueryCreateVertex); - class FailureQueryCreateVertex : public Message { public: FailureQueryCreateVertex() {} @@ -40,3 +95,41 @@ public: } }; CEREAL_REGISTER_TYPE(FailureQueryCreateVertex); + + +class QueryVertexCount : public ReturnAddressMsg { + public: + QueryVertexCount(std::string return_channel) + : ReturnAddressMsg(return_channel) {} + + template + void serialize(Archive &archive) { + archive(cereal::virtual_base_class(this)); + } + + protected: + // Cereal needs access to default constructor. + friend class cereal::access; + QueryVertexCount() {} +}; +CEREAL_REGISTER_TYPE(QueryVertexCount); + +class ResultQueryVertexCount : public Message { + public: + ResultQueryVertexCount(int64_t count) : count_(count) {} + + int64_t count() const { return count_; } + + template + void serialize(Archive &archive) { + archive(cereal::virtual_base_class(this), count_); + } + + protected: + // Cereal needs access to default constructor. + friend class cereal::access; + ResultQueryVertexCount() {} + + int64_t count_; +}; +CEREAL_REGISTER_TYPE(ResultQueryVertexCount); \ No newline at end of file diff --git a/experimental/distributed/src/reactors_local.hpp b/experimental/distributed/src/reactors_local.hpp index 396a8e7b0..828e02230 100644 --- a/experimental/distributed/src/reactors_local.hpp +++ b/experimental/distributed/src/reactors_local.hpp @@ -77,13 +77,29 @@ class EventStream { */ template void OnEvent(std::function &&cb) { - OnEventHelper(typeid(MsgType), [cb = move(cb)](const Message &general_msg, - const Subscription &subscription) { + OnEventHelper(typeid(MsgType), + [cb = std::move(cb)](const Message &general_msg, + const Subscription &subscription) { const MsgType &correct_msg = dynamic_cast(general_msg); cb(correct_msg, subscription); }); } + /** + * Register a callback that will be called only once. + * Once event is received, channel of this EventStream is closed. + */ + template + void OnEventOnceThenClose(std::function &&cb) { + OnEventHelper(typeid(MsgType), + [cb = std::move(cb)](const Message &general_msg, + const Subscription &subscription) { + const MsgType &correct_msg = dynamic_cast(general_msg); + subscription.CloseChannel(); + cb(correct_msg); + }); + } + /** * Starts a chain to register a callback that fires off only once. * diff --git a/experimental/distributed/src/storage.hpp b/experimental/distributed/src/storage.hpp new file mode 100644 index 000000000..7bdfa30f2 --- /dev/null +++ b/experimental/distributed/src/storage.hpp @@ -0,0 +1,45 @@ +#pragma once + +#include +#include + +#include "uid.hpp" + +/** + * Mock-up storage to test basic hardcoded queries. + * + * Current code is taken from graph.hpp. It will grow as needed to support + * new queries, but not more. Once all functionality from graph.hpp is + * transfered here, this file will be included in graph.hpp. + */ + +class Vertex {}; + +/** + * Storage that is split over multiple nodes. + */ +class ShardedStorage { + public: + ShardedStorage(int64_t mnid) : mnid_(mnid) {} + + /** Returns number of vertices on this node. */ + int64_t VertexCount() const { return vertices_.size(); } + + /** Creates a new vertex on this node. Returns its global id */ + const UniqueVid &MakeVertex() { + UniqueVid new_id(mnid_, next_vertex_sequence_++); + auto new_vertex = vertices_.emplace(std::make_pair(new_id, Vertex())); + return new_vertex.first->first; + }; + + private: + // unique Memgraph node ID + // uniqueness is ensured by the (distributed) system + const int64_t mnid_; + + // counter of vertices created on this node. + int64_t next_vertex_sequence_{0}; + + // vertex storage of this node + std::unordered_map vertices_; +}; \ No newline at end of file diff --git a/experimental/distributed/start_main.py b/experimental/distributed/start_main.py index 6e71ed608..44aee6073 100644 --- a/experimental/distributed/start_main.py +++ b/experimental/distributed/start_main.py @@ -8,7 +8,7 @@ terminal_flags = ' --geometry=200x50 ' # columns x rows config_filename = 'config' log_dir = "logs" -glog_flags = '--alsologtostderr --logbufsecs=0 --minloglevel=0 --log_dir="{}" '.format(log_dir) +glog_flags = '--alsologtostderr --logbufsecs=0 --minloglevel=2 --log_dir="{}" '.format(log_dir) def GetMainCall(my_mnid, address, port): ret = "./main {} --my_mnid {} --address {} --port {} --config_filename={}".format( @@ -31,6 +31,7 @@ def NamedGnomeTab(name, command): if __name__ == "__main__": command = "{} {}".format(terminal_command, terminal_flags) + command += NamedGnomeTab("client", GetClientCall()) f = open(config_filename, 'r') for line in f: @@ -40,7 +41,6 @@ if __name__ == "__main__": port = data[2] command += NamedGnomeTab("mnid={}".format(my_mnid), GetMainCall(my_mnid, address, port)) - command += NamedGnomeTab("client", GetClientCall()) print(command) os.system('mkdir -p {}'.format(log_dir)) os.system(command)