Commit 226f1170 authored by Dominik Charousset's avatar Dominik Charousset Committed by GitHub

Merge pull request #51

Implement proper shutdown of multiplexer
parents b7d2db7c 1b230f37
/******************************************************************************_opt /******************************************************************************
* ____ _ _____ * * ____ _ _____ *
* / ___| / \ | ___| C++ * * / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor * * | | / _ \ | |_ Actor *
...@@ -47,6 +47,8 @@ public: ...@@ -47,6 +47,8 @@ public:
error init() override; error init() override;
void stop() override;
endpoint_manager_ptr peer(const node_id& id) override; endpoint_manager_ptr peer(const node_id& id) override;
void resolve(const uri& locator, const actor& listener) override; void resolve(const uri& locator, const actor& listener) override;
......
...@@ -20,6 +20,7 @@ ...@@ -20,6 +20,7 @@
#include <string> #include <string>
#include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/net/fwd.hpp" #include "caf/net/fwd.hpp"
#include "caf/proxy_registry.hpp" #include "caf/proxy_registry.hpp"
...@@ -29,7 +30,7 @@ namespace caf::net { ...@@ -29,7 +30,7 @@ namespace caf::net {
/// Technology-specific backend for connecting to and managing peer /// Technology-specific backend for connecting to and managing peer
/// connections. /// connections.
/// @relates middleman /// @relates middleman
class middleman_backend : public proxy_registry::backend { class CAF_NET_EXPORT middleman_backend : public proxy_registry::backend {
public: public:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
...@@ -48,6 +49,8 @@ public: ...@@ -48,6 +49,8 @@ public:
/// Resolves a path to a remote actor. /// Resolves a path to a remote actor.
virtual void resolve(const uri& locator, const actor& listener) = 0; virtual void resolve(const uri& locator, const actor& listener) = 0;
virtual void stop() = 0;
// -- properties ------------------------------------------------------------- // -- properties -------------------------------------------------------------
const std::string& id() const noexcept { const std::string& id() const noexcept {
......
...@@ -88,6 +88,10 @@ public: ...@@ -88,6 +88,10 @@ public:
/// Polls until no socket event handler remains. /// Polls until no socket event handler remains.
void run(); void run();
/// Signals the multiplexer to initiate shutdown.
/// @thread-safe
void shutdown();
protected: protected:
// -- utility functions ------------------------------------------------------ // -- utility functions ------------------------------------------------------
...@@ -97,6 +101,9 @@ protected: ...@@ -97,6 +101,9 @@ protected:
/// Adds a new socket manager to the pollset. /// Adds a new socket manager to the pollset.
void add(socket_manager_ptr mgr); void add(socket_manager_ptr mgr);
/// Deletes a known socket manager from the pollset.
void del(ptrdiff_t index);
/// Writes `opcode` and pointer to `mgr` the the pipe for handling an event /// Writes `opcode` and pointer to `mgr` the the pipe for handling an event
/// later via the pollset updater. /// later via the pollset updater.
void write_to_pipe(uint8_t opcode, const socket_manager_ptr& mgr); void write_to_pipe(uint8_t opcode, const socket_manager_ptr& mgr);
...@@ -119,6 +126,9 @@ protected: ...@@ -119,6 +126,9 @@ protected:
/// Guards `write_handle_`. /// Guards `write_handle_`.
std::mutex write_lock_; std::mutex write_lock_;
/// Signals shutdown has been requested.
bool shutting_down_;
}; };
/// @relates multiplexer /// @relates multiplexer
......
...@@ -28,6 +28,7 @@ enum class operation { ...@@ -28,6 +28,7 @@ enum class operation {
read = 0x01, read = 0x01,
write = 0x02, write = 0x02,
read_write = 0x03, read_write = 0x03,
shutdown = 0x04,
}; };
constexpr operation operator|(operation x, operation y) { constexpr operation operator|(operation x, operation y) {
......
...@@ -29,7 +29,6 @@ ...@@ -29,7 +29,6 @@
#include "caf/net/pollset_updater.hpp" #include "caf/net/pollset_updater.hpp"
#include "caf/net/socket_manager.hpp" #include "caf/net/socket_manager.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/span.hpp"
#include "caf/variant.hpp" #include "caf/variant.hpp"
#ifndef CAF_WINDOWS #ifndef CAF_WINDOWS
...@@ -77,7 +76,7 @@ short to_bitmask(operation op) { ...@@ -77,7 +76,7 @@ short to_bitmask(operation op) {
} // namespace } // namespace
multiplexer::multiplexer() { multiplexer::multiplexer() : shutting_down_(false) {
// nop // nop
} }
...@@ -107,7 +106,9 @@ ptrdiff_t multiplexer::index_of(const socket_manager_ptr& mgr) { ...@@ -107,7 +106,9 @@ ptrdiff_t multiplexer::index_of(const socket_manager_ptr& mgr) {
void multiplexer::register_reading(const socket_manager_ptr& mgr) { void multiplexer::register_reading(const socket_manager_ptr& mgr) {
if (std::this_thread::get_id() == tid_) { if (std::this_thread::get_id() == tid_) {
if (mgr->mask() != operation::none) { if (shutting_down_) {
// discard
} else if (mgr->mask() != operation::none) {
CAF_ASSERT(index_of(mgr) != -1); CAF_ASSERT(index_of(mgr) != -1);
if (mgr->mask_add(operation::read)) { if (mgr->mask_add(operation::read)) {
auto& fd = pollset_[index_of(mgr)]; auto& fd = pollset_[index_of(mgr)];
...@@ -123,7 +124,9 @@ void multiplexer::register_reading(const socket_manager_ptr& mgr) { ...@@ -123,7 +124,9 @@ void multiplexer::register_reading(const socket_manager_ptr& mgr) {
void multiplexer::register_writing(const socket_manager_ptr& mgr) { void multiplexer::register_writing(const socket_manager_ptr& mgr) {
if (std::this_thread::get_id() == tid_) { if (std::this_thread::get_id() == tid_) {
if (mgr->mask() != operation::none) { if (shutting_down_) {
// discard
} else if (mgr->mask() != operation::none) {
CAF_ASSERT(index_of(mgr) != -1); CAF_ASSERT(index_of(mgr) != -1);
if (mgr->mask_add(operation::write)) { if (mgr->mask_add(operation::write)) {
auto& fd = pollset_[index_of(mgr)]; auto& fd = pollset_[index_of(mgr)];
...@@ -199,8 +202,7 @@ bool multiplexer::poll_once(bool blocking) { ...@@ -199,8 +202,7 @@ bool multiplexer::poll_once(bool blocking) {
auto new_events = handle(mgr, events, revents); auto new_events = handle(mgr, events, revents);
--presult; --presult;
if (new_events == 0) { if (new_events == 0) {
pollset_.erase(pollset_.begin() + i); del(i);
managers_.erase(managers_.begin() + i);
continue; continue;
} else if (new_events != events) { } else if (new_events != events) {
pollset_[i].events = new_events; pollset_[i].events = new_events;
...@@ -222,6 +224,27 @@ void multiplexer::run() { ...@@ -222,6 +224,27 @@ void multiplexer::run() {
poll_once(true); poll_once(true);
} }
void multiplexer::shutdown() {
if (std::this_thread::get_id() == tid_) {
shutting_down_ = true;
// First manager is the pollset_updater. Skip it and delete later.
for (size_t i = 1; i < managers_.size();) {
auto& mgr = managers_[i];
if (mgr->mask_del(operation::read)) {
auto& fd = pollset_[index_of(mgr)];
fd.events &= ~input_mask;
}
if (mgr->mask() == operation::none)
del(i);
else
++i;
}
close_pipe();
} else {
write_to_pipe(5, nullptr);
}
}
short multiplexer::handle(const socket_manager_ptr& mgr, short events, short multiplexer::handle(const socket_manager_ptr& mgr, short events,
short revents) { short revents) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle())); CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle()));
...@@ -262,10 +285,17 @@ void multiplexer::add(socket_manager_ptr mgr) { ...@@ -262,10 +285,17 @@ void multiplexer::add(socket_manager_ptr mgr) {
managers_.emplace_back(std::move(mgr)); managers_.emplace_back(std::move(mgr));
} }
void multiplexer::del(ptrdiff_t index) {
CAF_ASSERT(index != -1);
pollset_.erase(pollset_.begin() + index);
managers_.erase(managers_.begin() + index);
}
void multiplexer::write_to_pipe(uint8_t opcode, const socket_manager_ptr& mgr) { void multiplexer::write_to_pipe(uint8_t opcode, const socket_manager_ptr& mgr) {
CAF_ASSERT(opcode == 0 || opcode == 1); CAF_ASSERT(opcode == 0 || opcode == 1 || opcode == 4 || opcode == 5);
CAF_ASSERT(mgr != nullptr); CAF_ASSERT(mgr != nullptr || opcode == 4);
pollset_updater::msg_buf buf; pollset_updater::msg_buf buf;
if (opcode != 5)
mgr->ref(); mgr->ref();
buf[0] = static_cast<byte>(opcode); buf[0] = static_cast<byte>(opcode);
auto value = reinterpret_cast<intptr_t>(mgr.get()); auto value = reinterpret_cast<intptr_t>(mgr.get());
...@@ -278,7 +308,7 @@ void multiplexer::write_to_pipe(uint8_t opcode, const socket_manager_ptr& mgr) { ...@@ -278,7 +308,7 @@ void multiplexer::write_to_pipe(uint8_t opcode, const socket_manager_ptr& mgr) {
else else
res = sec::socket_invalid; res = sec::socket_invalid;
} }
if (holds_alternative<sec>(res)) if (holds_alternative<sec>(res) && opcode != 4)
mgr->deref(); mgr->deref();
} }
......
...@@ -44,6 +44,12 @@ error test::init() { ...@@ -44,6 +44,12 @@ error test::init() {
return none; return none;
} }
void test::stop() {
for (const auto& p : peers_)
proxies_.erase(p.first);
peers_.clear();
}
endpoint_manager_ptr test::peer(const node_id& id) { endpoint_manager_ptr test::peer(const node_id& id) {
return get_peer(id).second; return get_peer(id).second;
} }
......
...@@ -53,15 +53,19 @@ void middleman::start() { ...@@ -53,15 +53,19 @@ void middleman::start() {
} }
void middleman::stop() { void middleman::stop() {
mpx_->close_pipe(); for (const auto& backend : backends_)
backend->stop();
mpx_->shutdown();
if (mpx_thread_.joinable()) if (mpx_thread_.joinable())
mpx_thread_.join(); mpx_thread_.join();
else
mpx_->run();
} }
void middleman::init(actor_system_config& cfg) { void middleman::init(actor_system_config& cfg) {
if (auto err = mpx_->init()) { if (auto err = mpx_->init()) {
CAF_LOG_ERROR("mgr->init() failed: " << system().render(err)); CAF_LOG_ERROR("mgr->init() failed: " << system().render(err));
CAF_RAISE_ERROR("mpx->init failed"); CAF_RAISE_ERROR("mpx->init() failed");
} }
if (auto node_uri = get_if<uri>(&cfg, "middleman.this-node")) { if (auto node_uri = get_if<uri>(&cfg, "middleman.this-node")) {
auto this_node = make_node_id(std::move(*node_uri)); auto this_node = make_node_id(std::move(*node_uri));
......
...@@ -20,6 +20,7 @@ ...@@ -20,6 +20,7 @@
#include <cstring> #include <cstring>
#include "caf/logger.hpp"
#include "caf/net/multiplexer.hpp" #include "caf/net/multiplexer.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/span.hpp" #include "caf/span.hpp"
...@@ -51,11 +52,19 @@ bool pollset_updater::handle_read_event() { ...@@ -51,11 +52,19 @@ bool pollset_updater::handle_read_event() {
memcpy(&value, buf_.data() + 1, sizeof(intptr_t)); memcpy(&value, buf_.data() + 1, sizeof(intptr_t));
socket_manager_ptr mgr{reinterpret_cast<socket_manager*>(value), false}; socket_manager_ptr mgr{reinterpret_cast<socket_manager*>(value), false};
if (auto ptr = parent_.lock()) { if (auto ptr = parent_.lock()) {
if (opcode == 0) { switch (opcode) {
case 0:
ptr->register_reading(mgr); ptr->register_reading(mgr);
} else { break;
CAF_ASSERT(opcode == 1); case 1:
ptr->register_writing(mgr); ptr->register_writing(mgr);
break;
case 4:
ptr->shutdown();
break;
default:
CAF_LOG_DEBUG("opcode not recognized: " << CAF_ARG(opcode));
break;
} }
} }
} }
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment