Commit dccf46db authored by Jakob Otto's avatar Jakob Otto

Reimplement shutdown routine

parent a35f2cd6
/******************************************************************************_opt /******************************************************************************
* ____ _ _____ * * ____ _ _____ *
* / ___| / \ | ___| C++ * * / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor * * | | / _ \ | |_ Actor *
......
...@@ -71,9 +71,9 @@ public: ...@@ -71,9 +71,9 @@ public:
/// @thread-safe /// @thread-safe
void register_writing(const socket_manager_ptr& mgr); void register_writing(const socket_manager_ptr& mgr);
/// Unregisters `mgr` for read events. /// Unregisters `mgr` for events.
/// @thread-safe /// @thread-safe
void unregister_reading(const socket_manager_ptr& mgr); void unregister_manager(const socket_manager_ptr& mgr);
/// Closes the pipe for signaling updates to the multiplexer. After closing /// Closes the pipe for signaling updates to the multiplexer. After closing
/// the pipe, calls to `update` no longer have any effect. /// the pipe, calls to `update` no longer have any effect.
...@@ -92,6 +92,9 @@ public: ...@@ -92,6 +92,9 @@ 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.
void shutdown();
protected: protected:
// -- utility functions ------------------------------------------------------ // -- utility functions ------------------------------------------------------
...@@ -101,6 +104,9 @@ protected: ...@@ -101,6 +104,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(const socket_manager_ptr& mgr);
/// 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);
...@@ -123,6 +129,9 @@ protected: ...@@ -123,6 +129,9 @@ protected:
/// Guards `write_handle_`. /// Guards `write_handle_`.
std::mutex write_lock_; std::mutex write_lock_;
/// Signals shutdown has been requested.
bool will_shutdown_;
}; };
/// @relates multiplexer /// @relates multiplexer
......
...@@ -28,8 +28,8 @@ enum class operation { ...@@ -28,8 +28,8 @@ enum class operation {
read = 0x01, read = 0x01,
write = 0x02, write = 0x02,
read_write = 0x03, read_write = 0x03,
unregister_read = 0x04, unregister = 0x04,
unregister_write = 0x05, shutdown = 0x05,
}; };
constexpr operation operator|(operation x, operation y) { constexpr operation operator|(operation x, operation y) {
......
...@@ -75,8 +75,6 @@ public: ...@@ -75,8 +75,6 @@ public:
void register_writing(); void register_writing();
void unregister_reading();
// -- pure virtual member functions ------------------------------------------ // -- pure virtual member functions ------------------------------------------
/// Called whenever the socket received new data. /// Called whenever the socket received new data.
......
/****************************************************************************** /******************************************************************************
* ____ _ _____ * / ___| / \ | ___| C++ * * ____ _ _____ *
* | | / _ \ | |_ Actor * | |___ / ___ \| _| * / ___| / \ | ___| C++ *
*Framework * * | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| * * \____/_/ \_|_| *
* * * *
* Copyright 2011-2019 Dominik Charousset * * Copyright 2011-2019 Dominik Charousset *
* * * *
* Distributed under the terms and conditions of the BSD 3-Clause License or * Distributed under the terms and conditions of the BSD 3-Clause License or *
** (at your option) under the terms and conditions of the Boost Software * * (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. * * License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* * * *
* If you did not receive a copy of the license files, see * * If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and * * http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#include "caf/net/multiplexer.hpp" #include "caf/net/multiplexer.hpp"
#include <algorithm> #include <algorithm>
...@@ -135,22 +137,13 @@ void multiplexer::register_writing(const socket_manager_ptr& mgr) { ...@@ -135,22 +137,13 @@ void multiplexer::register_writing(const socket_manager_ptr& mgr) {
} }
} }
void multiplexer::unregister_reading(const socket_manager_ptr& mgr) { void multiplexer::unregister_manager(const socket_manager_ptr& mgr) {
if (std::this_thread::get_id() == tid_) { if (std::this_thread::get_id() == tid_) {
if (mgr->mask() != operation::none) { del(mgr);
CAF_ASSERT(index_of(mgr) != -1); if (will_shutdown_ && managers_.size() == 1)
if (mgr->mask_del(operation::read)) { close_pipe();
auto& fd = pollset_[index_of(mgr)];
fd.events &= ~input_mask;
}
if (mgr->mask() == operation::none) {
auto mgr_index = index_of(mgr);
pollset_.erase(pollset_.begin() + mgr_index);
managers_.erase(managers_.begin() + mgr_index);
}
}
} else { } else {
write_to_pipe(0, mgr); write_to_pipe(4, mgr);
} }
} }
...@@ -239,6 +232,21 @@ void multiplexer::run() { ...@@ -239,6 +232,21 @@ void multiplexer::run() {
poll_once(true); poll_once(true);
} }
void multiplexer::shutdown() {
if (std::this_thread::get_id() == tid_) {
will_shutdown_ = true;
if (managers_.size() == 1) {
close_pipe();
} else {
// First manager is the pollset_updater. Skip it and delete later.
for (size_t i = 1; i < managers_.size(); ++i)
write_to_pipe(4, managers_[i]);
}
} 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()));
...@@ -279,10 +287,17 @@ void multiplexer::add(socket_manager_ptr mgr) { ...@@ -279,10 +287,17 @@ void multiplexer::add(socket_manager_ptr mgr) {
managers_.emplace_back(std::move(mgr)); managers_.emplace_back(std::move(mgr));
} }
void multiplexer::del(const socket_manager_ptr& mgr) {
CAF_ASSERT(index_of(mgr) != -1);
pollset_.erase(pollset_.begin() + index_of(mgr));
managers_.erase(managers_.begin() + index_of(mgr));
}
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 == 5);
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());
...@@ -295,7 +310,7 @@ void multiplexer::write_to_pipe(uint8_t opcode, const socket_manager_ptr& mgr) { ...@@ -295,7 +310,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 != 5)
mgr->deref(); mgr->deref();
} }
......
...@@ -45,8 +45,9 @@ error test::init() { ...@@ -45,8 +45,9 @@ error test::init() {
} }
void test::stop() { void test::stop() {
for (const auto& peer : peers_) for (const auto& pair : peers_)
peer.second.second->unregister_reading(); proxies_.erase(pair.first);
peers_.clear();
} }
endpoint_manager_ptr test::peer(const node_id& id) { endpoint_manager_ptr test::peer(const node_id& id) {
......
...@@ -55,7 +55,7 @@ void middleman::start() { ...@@ -55,7 +55,7 @@ void middleman::start() {
void middleman::stop() { void middleman::stop() {
for (const auto& backend : backends_) for (const auto& backend : backends_)
backend->stop(); backend->stop();
mpx_->close_pipe(); mpx_->shutdown();
if (mpx_thread_.joinable()) if (mpx_thread_.joinable())
mpx_thread_.join(); mpx_thread_.join();
} }
......
...@@ -49,13 +49,19 @@ bool pollset_updater::handle_read_event() { ...@@ -49,13 +49,19 @@ bool pollset_updater::handle_read_event() {
auto opcode = static_cast<uint8_t>(buf_[0]); auto opcode = static_cast<uint8_t>(buf_[0]);
intptr_t value; intptr_t value;
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};
if (auto ptr = parent_.lock()) { if (auto ptr = parent_.lock()) {
if (opcode == 0) { if (opcode == 0) {
ptr->register_reading(mgr); ptr->register_reading(
{reinterpret_cast<socket_manager*>(value), false});
} else if (opcode == 1) {
ptr->register_writing(
{reinterpret_cast<socket_manager*>(value), false});
} else if (opcode == 4) {
ptr->unregister_manager(
{reinterpret_cast<socket_manager*>(value), false});
} else { } else {
CAF_ASSERT(opcode == 1); CAF_ASSERT(opcode == 5);
ptr->register_writing(mgr); ptr->shutdown();
} }
} }
} }
......
...@@ -67,12 +67,4 @@ void socket_manager::register_writing() { ...@@ -67,12 +67,4 @@ void socket_manager::register_writing() {
ptr->register_writing(this); ptr->register_writing(this);
} }
void socket_manager::unregister_reading() {
if ((mask() & operation::read) != operation::read)
return;
auto ptr = parent_.lock();
if (ptr != nullptr)
ptr->unregister_reading(this);
}
} // namespace caf::net } // namespace caf::net
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