From 7f1c7a46ccc5b7b6123aa3cd1583cfbe4a3353e2 Mon Sep 17 00:00:00 2001 From: Goran Zuzic Date: Thu, 24 Aug 2017 10:29:21 +0200 Subject: [PATCH] Distributed tests Summary: 1. added distributed tests, currently only one can be run (TODO) 2. usability changes: add Close() to Subscription 3. EventStream::Subscription -> Subscription 4. Add Subscription to ChainOnce 5. Added README.md conventions5 Reviewers: sasa.stanko Reviewed By: sasa.stanko Subscribers: pullbot, buda Differential Revision: https://phabricator.memgraph.io/D701 --- experimental/distributed/README.md | 20 +++-- .../distributed/src/reactors_distributed.hpp | 6 +- .../distributed/src/reactors_local.cpp | 27 +++---- .../distributed/src/reactors_local.hpp | 57 +++++++------- .../distributed/tests/distributed_test.cpp | 8 +- .../distributed/tests/local_memgraph.cpp | 4 +- .../distributed/tests/network_chat.cpp | 4 +- .../tests/reactors_distributed_unit.cpp | 78 +++++++++++++++++++ .../distributed/tests/reactors_local_unit.cpp | 38 ++++----- 9 files changed, 165 insertions(+), 77 deletions(-) create mode 100644 experimental/distributed/tests/reactors_distributed_unit.cpp diff --git a/experimental/distributed/README.md b/experimental/distributed/README.md index 3cb0d5caa..62ebf1dda 100644 --- a/experimental/distributed/README.md +++ b/experimental/distributed/README.md @@ -3,22 +3,32 @@ This subdirectory structure implements distributed infrastructure of Memgraph. -## terminology +## Terminology * Memgraph Node Id (mnid): a machine (processs) that runs a (distributed) Memgraph program. * Node: a computer that performs (distributed) work. * Vertex: an abstract graph concept. * Reactor: a unit of concurrent execution, lives on its own thread. * Channel: a (one-way) communication abstraction between Reactors. The reactors can be on the same machine or on different processes. +* Message: gets sent through channels. Must be serializable if sent via network layer (library: cereal). +* Event: arrival of a (subclass of) Message. You can register callbacks. Register exact callbacks (not for derivated/subclasses). * EventStream: read-end of a channel, is owned by exactly one Reactor/thread. * ChannelWriter: write-end of a channel, can be owned (wrote into) by multiple threads. -## conventions +## Ownership: -1. Locked: A method having a Locked... prefix indicates that you -have to lock the appropriate mutex before calling this function. +* System, Distributed are singletons. They should be always alive. +* ChannelWriter (write-end) should be lightweight and can be copied arbitrarily. +* EventStream (read-end) should never be written by anyone except the owner (the reactor that created it). -## dependencies +## Code Conventions + +* Locked: A method having a Locked... prefix indicates that you + have to lock the appropriate mutex before calling this function. +* ALWAYS close channels. You will memory leak if you don't. + Reactor::CloseChannel or Subscription::Close will do the trick. + +## Dependencies * cereal * \ No newline at end of file diff --git a/experimental/distributed/src/reactors_distributed.hpp b/experimental/distributed/src/reactors_distributed.hpp index 323848a91..39fc038ff 100644 --- a/experimental/distributed/src/reactors_distributed.hpp +++ b/experimental/distributed/src/reactors_distributed.hpp @@ -95,6 +95,8 @@ class Network { /** Start a threadpool that dispatches the messages from the (outgoing) queue to the sockets */ void StartClient(int worker_count) { LOG(INFO) << "Starting " << worker_count << " client workers"; + client_run_ = true; + for (int i = 0; i < worker_count; ++i) { pool_.push_back(std::thread([worker_count, this]() { while (this->client_run_) { @@ -129,6 +131,7 @@ class Network { for (size_t i = 0; i < pool_.size(); ++i) { pool_[i].join(); } + pool_.clear(); } class RemoteChannelWriter : public ChannelWriter { @@ -209,6 +212,7 @@ class Network { if (server_ != nullptr) { server_->Shutdown(); thread_.join(); + server_ = nullptr; } } @@ -217,7 +221,7 @@ class Network { SpinLock mutex_; std::vector pool_; std::queue queue_; - std::atomic client_run_{true}; + std::atomic client_run_; // server variables std::thread thread_; diff --git a/experimental/distributed/src/reactors_local.cpp b/experimental/distributed/src/reactors_local.cpp index 2c3254b19..cc3a60919 100644 --- a/experimental/distributed/src/reactors_local.cpp +++ b/experimental/distributed/src/reactors_local.cpp @@ -1,9 +1,13 @@ #include "reactors_local.hpp" -void EventStream::Subscription::unsubscribe() const { +void EventStream::Subscription::Unsubscribe() const { event_queue_.RemoveCb(*this); } +void EventStream::Subscription::Close() const { + event_queue_.Close(); +} + thread_local Reactor* current_reactor_ = nullptr; std::string Channel::LocalChannelWriter::ReactorName() { @@ -14,7 +18,9 @@ std::string Channel::LocalChannelWriter::Name() { return channel_name_; } -void Channel::LocalEventStream::Close() { +void Channel::Close() { + // TODO(zuza): there will be major problems if a reactor tries to close a stream that isn't theirs + // luckily this should never happen if the framework is used as expected. current_reactor_->CloseChannel(channel_name_); } @@ -59,11 +65,7 @@ void Reactor::CloseChannel(const std::string &s) { auto it = channels_.find(s); assert(it != channels_.end()); channels_.erase(it); -} - -void Reactor::CloseAllChannels() { - std::unique_lock lock(*mutex_); - channels_.clear(); + cvar_->notify_all(); } void Reactor::RunEventLoop() { @@ -76,6 +78,9 @@ void Reactor::RunEventLoop() { std::unique_lock lock(*mutex_); while (true) { + // Not fair because was taken earlier, talk to lion. + msg_and_cb = LockedGetPendingMessages(); + if (msg_and_cb.first != nullptr) break; // Exit the loop if there are no more Channels. if (channels_.empty()) { @@ -83,10 +88,6 @@ void Reactor::RunEventLoop() { break; } - // Not fair because was taken earlier, talk to lion. - msg_and_cb = LockedGetPendingMessages(); - if (msg_and_cb.first != nullptr) break; - cvar_->wait(lock); } @@ -112,13 +113,13 @@ auto Reactor::LockedGetPendingMessages() -> MsgAndCbInfo { if (msg_ptr == nullptr) continue; std::type_index tidx = msg_ptr->GetTypeIndex(); - std::vector > cb_info; + std::vector > cb_info; auto msg_type_cb_iter = event_queue.callbacks_.find(tidx); if (msg_type_cb_iter != event_queue.callbacks_.end()) { // There is a callback for this type. for (auto &tidx_cb_key_value : msg_type_cb_iter->second) { uint64_t uid = tidx_cb_key_value.first; EventStream::Callback cb = tidx_cb_key_value.second; - cb_info.emplace_back(cb, EventStream::Subscription(event_queue, tidx, uid)); + cb_info.emplace_back(cb, Subscription(event_queue, tidx, uid)); } } diff --git a/experimental/distributed/src/reactors_local.hpp b/experimental/distributed/src/reactors_local.hpp index 6c0664aa4..1828e7595 100644 --- a/experimental/distributed/src/reactors_local.hpp +++ b/experimental/distributed/src/reactors_local.hpp @@ -69,8 +69,8 @@ class ChannelWriter { */ class EventStream { public: - class Subscription; class OnEventOnceChainer; + class Subscription; /** * Register a callback that will be called whenever an event arrives. @@ -99,6 +99,7 @@ class EventStream { * Get the name of the channel. */ virtual const std::string &ChannelName() = 0; + /** * Subscription Service. * @@ -109,7 +110,12 @@ class EventStream { /** * Unsubscribe. Call only once. */ - void unsubscribe() const; + void Unsubscribe() const; + + /** + * Close the stream. Convenience method. + */ + void Close() const; private: friend class Reactor; @@ -151,12 +157,12 @@ class EventStream { } template - OnEventOnceChainer &ChainOnce(std::function &&cb) { + OnEventOnceChainer &ChainOnce(std::function &&cb) { std::function wrap = [cb = std::move(cb)](const Message &general_msg, const Subscription &subscription) { const MsgType &correct_msg = dynamic_cast(general_msg); - subscription.unsubscribe(); - cb(correct_msg); // Warning: this can close the Channel, be careful what you put after it! + subscription.Unsubscribe(); + cb(correct_msg, subscription); // Warning: this can close the Channel, be careful what you put after it! }; cbs_.emplace_back(typeid(MsgType), std::move(wrap)); return *this; @@ -196,6 +202,8 @@ private: virtual void OnEventHelper(std::type_index tidx, Callback callback) = 0; }; +using Subscription = EventStream::Subscription; // To write less. + /** * Implementation of a channel. * @@ -218,7 +226,7 @@ class Channel { reactor_name_(params.reactor_name), mutex_(params.mutex), cvar_(params.cvar), - stream_(mutex_, channel_name_, this) {} + stream_(mutex_, this) {} /** * LocalChannelWriter represents the channels to reactors living in the same reactor system (write-end of the channels). @@ -268,20 +276,20 @@ class Channel { public: friend class Channel; - LocalEventStream(std::shared_ptr mutex, std::string channel_name, - Channel *queue) : mutex_(mutex), channel_name_(channel_name), queue_(queue) {} - std::unique_ptr AwaitEvent() { - std::unique_lock lock(*mutex_); - return queue_->LockedAwaitPop(lock); - } + LocalEventStream(std::shared_ptr mutex, Channel *queue) : mutex_(mutex), queue_(queue) {} + void OnEventHelper(std::type_index tidx, Callback callback) { std::unique_lock lock(*mutex_); queue_->LockedOnEventHelper(tidx, callback); } + const std::string &ChannelName() { return queue_->channel_name_; } - void Close(); + + void Close() { + queue_->Close(); + } private: std::shared_ptr mutex_; @@ -289,6 +297,11 @@ class Channel { Channel *queue_; }; + /** + * Close the channel. Must be called from the reactor that owns the channel. + */ + void Close(); + Channel(const Channel &other) = delete; Channel(Channel &&other) = default; Channel &operator=(const Channel &other) = delete; @@ -318,17 +331,6 @@ private: return std::make_shared(mutex_, reactor_name_, channel_name_, self_ptr_); } - std::unique_ptr LockedAwaitPop(std::unique_lock &lock) { - while (true) { - std::unique_ptr m = LockedRawPop(); - if (!m) { - cvar_->wait(lock); - } else { - return m; - } - } - } - std::unique_ptr LockedPop() { return LockedRawPop(); } @@ -395,13 +397,6 @@ class Reactor { */ void CloseChannel(const std::string &s); - /** - * close all channels (typically during shutdown). - * - * Should only be called from the Reactor thread. - */ - void CloseAllChannels(); - /** * Get Reactor name */ diff --git a/experimental/distributed/tests/distributed_test.cpp b/experimental/distributed/tests/distributed_test.cpp index b0dc665d1..f36c6bcd2 100644 --- a/experimental/distributed/tests/distributed_test.cpp +++ b/experimental/distributed/tests/distributed_test.cpp @@ -126,11 +126,11 @@ class Master : public Reactor { // wait until every worker sends a ReturnAddressMsg back, then close stream->OnEvent([this](const TextMessage &msg, - const EventStream::Subscription &subscription) { + const Subscription &subscription) { std::cout << "Message from " << msg.Address() << ":" << msg.Port() << " .. " << msg.text << "\n"; ++workers_seen; if (workers_seen == worker_mnids_.size()) { - subscription.unsubscribe(); + subscription.Unsubscribe(); // Sleep for a while so we can read output in the terminal. // (start_distributed.py runs each process in a new tab which is // closed immediately after process has finished) @@ -143,7 +143,7 @@ class Master : public Reactor { for (auto wmnid : worker_mnids_) { auto stream = memgraph.FindChannel(wmnid, "worker", "main"); stream->OnEventOnce() - .ChainOnce([this, stream](const ChannelResolvedMessage &msg){ + .ChainOnce([this, stream](const ChannelResolvedMessage &msg, const Subscription&){ msg.channelWriter()->Send("master", "main", "hi from master"); stream->Close(); }); @@ -171,7 +171,7 @@ class Worker : public Reactor { auto stream = main_.first; // wait until master sends us a TextMessage, then reply back and close stream->OnEventOnce() - .ChainOnce([this](const TextMessage &msg) { + .ChainOnce([this](const TextMessage &msg, const Subscription&) { std::cout << "Message from " << msg.Address() << ":" << msg.Port() << " .. " << msg.text << "\n"; msg.GetReturnChannelWriter() diff --git a/experimental/distributed/tests/local_memgraph.cpp b/experimental/distributed/tests/local_memgraph.cpp index c7e18cd7f..775e843c5 100644 --- a/experimental/distributed/tests/local_memgraph.cpp +++ b/experimental/distributed/tests/local_memgraph.cpp @@ -151,12 +151,12 @@ // } // } -// stream->OnEvent([this](const Message &msg, const EventStream::Subscription& subscription) { +// stream->OnEvent([this](const Message &msg, const Subscription& subscription) { // std::cout << "Processing Query via Callback" << std::endl; // const Query &query = // dynamic_cast(msg); // exception bad_cast // ProcessQuery(&query); -// subscription.unsubscribe(); +// subscription.Unsubscribe(); // }); // } diff --git a/experimental/distributed/tests/network_chat.cpp b/experimental/distributed/tests/network_chat.cpp index 52447bf25..cf151c40e 100644 --- a/experimental/distributed/tests/network_chat.cpp +++ b/experimental/distributed/tests/network_chat.cpp @@ -46,13 +46,13 @@ class ChatServer : public Reactor { auto chat = Open("chat").first; - chat->OnEvent([](const ChatACK& ack, const EventStream::Subscription&) { + chat->OnEvent([](const ChatACK& ack, const Subscription&) { std::cout << "Received ACK from " << ack.Address() << ":" << ack.Port() << " -> '" << ack.Message() << "'" << std::endl; }); - chat->OnEvent([this](const ChatMessage& msg, const EventStream::Subscription&) { + chat->OnEvent([this](const ChatMessage& msg, const Subscription&) { std::cout << "Received message from " << msg.Address() << ":" << msg.Port() << " -> '" << msg.Message() << "'" << std::endl; diff --git a/experimental/distributed/tests/reactors_distributed_unit.cpp b/experimental/distributed/tests/reactors_distributed_unit.cpp new file mode 100644 index 000000000..d616eded0 --- /dev/null +++ b/experimental/distributed/tests/reactors_distributed_unit.cpp @@ -0,0 +1,78 @@ +/** + * This test file test the Distributed Reactors API on ONLY one process (no real networking). + * In other words, we send a message from one process to itself. + */ + +#include "gtest/gtest.h" +#include "reactors_distributed.hpp" + +#include +#include +#include +#include +#include +#include +#include +#include + +TEST(SimpleTests, StartAndStopServices) { + System &system = System::GetInstance(); + Distributed &distributed = Distributed::GetInstance(); + distributed.StartServices(); + + // do nothing + std::this_thread::sleep_for(std::chrono::milliseconds(500)); + + system.AwaitShutdown(); + distributed.StopServices(); +} + +TEST(SimpleTests, SendEmptyMessage) { + struct Master : public Reactor { + Master(std::string name) : Reactor(name) {} + + virtual void Run() { + Distributed::GetInstance().FindChannel("127.0.0.1", 10000, "worker", "main") + ->OnEventOnce() + .ChainOnce([this](const ChannelResolvedMessage& msg, + const Subscription& subscription) { + msg.channelWriter()->Send(); + subscription.Close(); + }); + + CloseChannel("main"); + } + }; + + struct Worker : public Reactor { + Worker(std::string name) : Reactor(name) {} + + virtual void Run() { + main_.first->OnEventOnce() + .ChainOnce([this](const Message&, const Subscription& subscription) { + // if this message isn't delivered, the main channel will never be closed + subscription.Close(); // close "main" + }); + } + }; + + // emulate flags like it's a multiprocess system, these may be alredy set by default + FLAGS_address = "127.0.0.1"; + FLAGS_port = 10000; + + System &system = System::GetInstance(); + Distributed &distributed = Distributed::GetInstance(); + distributed.StartServices(); + + system.Spawn("master"); + system.Spawn("worker"); + + system.AwaitShutdown(); // this must be called before StopServices + distributed.StopServices(); +} + + +int main(int argc, char **argv) { + ::testing::InitGoogleTest(&argc, argv); + return RUN_ALL_TESTS(); +} diff --git a/experimental/distributed/tests/reactors_local_unit.cpp b/experimental/distributed/tests/reactors_local_unit.cpp index b3fa3e367..fee571a4e 100644 --- a/experimental/distributed/tests/reactors_local_unit.cpp +++ b/experimental/distributed/tests/reactors_local_unit.cpp @@ -92,7 +92,7 @@ TEST(SimpleSendTest, OneCallback) { virtual void Run() { EventStream* stream = main_.first; - stream->OnEvent([this](const MessageInt &msg, const EventStream::Subscription&) { + stream->OnEvent([this](const MessageInt &msg, const Subscription&) { ASSERT_EQ(msg.x, 888); CloseChannel("main"); }); @@ -132,7 +132,7 @@ TEST(SimpleSendTest, IgnoreAfterClose) { virtual void Run() { EventStream* stream = main_.first; - stream->OnEvent([this](const MessageInt& msg, const EventStream::Subscription&) { + stream->OnEvent([this](const MessageInt& msg, const Subscription&) { CloseChannel("main"); ASSERT_EQ(msg.x, 101); }); @@ -156,12 +156,12 @@ TEST(SimpleSendTest, DuringFirstEvent) { virtual void Run() { EventStream* stream = main_.first; - stream->OnEvent([this](const Message &msg, const EventStream::Subscription &subscription) { + stream->OnEvent([this](const Message &msg, const Subscription &subscription) { const MessageInt &msgint = dynamic_cast(msg); if (msgint.x == 101) FindChannel("main")->Send(102); if (msgint.x == 102) { - subscription.unsubscribe(); + subscription.Unsubscribe(); CloseChannel("main"); p_.set_value(777); } @@ -220,19 +220,19 @@ TEST(MultipleSendTest, UnsubscribeService) { virtual void Run() { EventStream* stream = main_.first; - stream->OnEvent([this](const MessageInt &msgint, const EventStream::Subscription &subscription) { + stream->OnEvent([this](const MessageInt &msgint, const Subscription &subscription) { ASSERT_TRUE(msgint.x == 55 || msgint.x == 66); ++num_msgs_received; if (msgint.x == 66) { - subscription.unsubscribe(); // receive only two of them + subscription.Unsubscribe(); // receive only two of them } }); - stream->OnEvent([this](const MessageChar &msgchar, const EventStream::Subscription &subscription) { + stream->OnEvent([this](const MessageChar &msgchar, const Subscription &subscription) { char c = msgchar.x; ++num_msgs_received; ASSERT_TRUE(c == 'a' || c == 'b' || c == 'c'); if (num_msgs_received == 5) { - subscription.unsubscribe(); + subscription.Unsubscribe(); CloseChannel("main"); } }); @@ -281,19 +281,19 @@ TEST(MultipleSendTest, OnEvent) { EventStream* stream = main_.first; correct_vals = 0; - stream->OnEvent([this](const MessageInt &msgint, const EventStream::Subscription&) { + stream->OnEvent([this](const MessageInt &msgint, const Subscription&) { ASSERT_TRUE(msgint.x == 101 || msgint.x == 103); ++correct_vals; main_.second->Send(); }); - stream->OnEvent([this](const MessageChar &msgchar, const EventStream::Subscription&) { + stream->OnEvent([this](const MessageChar &msgchar, const Subscription&) { ASSERT_TRUE(msgchar.x == 'a' || msgchar.x == 'b'); ++correct_vals; main_.second->Send(); }); - stream->OnEvent([this](const EndMessage&, const EventStream::Subscription&) { + stream->OnEvent([this](const EndMessage&, const Subscription&) { ASSERT_LE(correct_vals, 4); if (correct_vals == 4) { CloseChannel("main"); @@ -334,13 +334,13 @@ TEST(MultipleSendTest, Chaining) { EventStream* stream = main_.first; stream->OnEventOnce() - .ChainOnce([this](const MessageInt &msg) { + .ChainOnce([this](const MessageInt &msg, const Subscription&) { ASSERT_EQ(msg.x, 55); }) - .ChainOnce([](const MessageInt &msg) { + .ChainOnce([](const MessageInt &msg, const Subscription&) { ASSERT_EQ(msg.x, 66); }) - .ChainOnce([this](const MessageInt &msg) { + .ChainOnce([this](const MessageInt &msg, const Subscription&) { ASSERT_EQ(msg.x, 77); CloseChannel("main"); }); @@ -386,13 +386,13 @@ TEST(MultipleSendTest, ChainingInRightOrder) { EventStream* stream = main_.first; stream->OnEventOnce() - .ChainOnce([this](const MessageInt &msg) { + .ChainOnce([this](const MessageInt &msg, const Subscription&) { ASSERT_EQ(msg.x, 55); }) - .ChainOnce([](const MessageChar &msg) { + .ChainOnce([](const MessageChar &msg, const Subscription&) { ASSERT_EQ(msg.x, 'b'); }) - .ChainOnce([this](const MessageInt &msg) { + .ChainOnce([this](const MessageInt &msg, const Subscription&) { ASSERT_EQ(msg.x, 77); CloseChannel("main"); }); @@ -440,12 +440,12 @@ TEST(MultipleSendTest, ProcessManyMessages) { EventStream* stream = main_.first; vals = 0; - stream->OnEvent([this](const Message&, const EventStream::Subscription&) { + stream->OnEvent([this](const Message&, const Subscription&) { ++vals; main_.second->Send(); }); - stream->OnEvent([this](const Message&, const EventStream::Subscription&) { + stream->OnEvent([this](const Message&, const Subscription&) { ASSERT_LE(vals, num_tests); if (vals == num_tests) { CloseChannel("main");