Commit 2e946561 authored by Jakob Otto's avatar Jakob Otto

Implement proper shutdown

parent 1e800f77
...@@ -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 {
......
...@@ -71,6 +71,10 @@ public: ...@@ -71,6 +71,10 @@ 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.
/// @thread-safe
void unregister_reading(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.
/// @thread-safe /// @thread-safe
......
...@@ -28,6 +28,8 @@ enum class operation { ...@@ -28,6 +28,8 @@ enum class operation {
read = 0x01, read = 0x01,
write = 0x02, write = 0x02,
read_write = 0x03, read_write = 0x03,
unregister_read = 0x04,
unregister_write = 0x05,
}; };
constexpr operation operator|(operation x, operation y) { constexpr operation operator|(operation x, operation y) {
......
...@@ -75,6 +75,8 @@ public: ...@@ -75,6 +75,8 @@ 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++ *
* / ___| / \ | ___| C++ * * | | / _ \ | |_ Actor * | |___ / ___ \| _|
* | | / _ \ | |_ Actor * *Framework *
* | |___ / ___ \| _| 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>
...@@ -137,6 +135,25 @@ void multiplexer::register_writing(const socket_manager_ptr& mgr) { ...@@ -137,6 +135,25 @@ void multiplexer::register_writing(const socket_manager_ptr& mgr) {
} }
} }
void multiplexer::unregister_reading(const socket_manager_ptr& mgr) {
if (std::this_thread::get_id() == tid_) {
if (mgr->mask() != operation::none) {
CAF_ASSERT(index_of(mgr) != -1);
if (mgr->mask_del(operation::read)) {
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 {
write_to_pipe(0, mgr);
}
}
void multiplexer::close_pipe() { void multiplexer::close_pipe() {
std::lock_guard<std::mutex> guard{write_lock_}; std::lock_guard<std::mutex> guard{write_lock_};
if (write_handle_ != invalid_socket) { if (write_handle_ != invalid_socket) {
......
...@@ -44,6 +44,11 @@ error test::init() { ...@@ -44,6 +44,11 @@ error test::init() {
return none; return none;
} }
void test::stop() {
for (const auto& peer : peers_)
peer.second.second->unregister_reading();
}
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,6 +53,8 @@ void middleman::start() { ...@@ -53,6 +53,8 @@ void middleman::start() {
} }
void middleman::stop() { void middleman::stop() {
for (const auto& backend : backends_)
backend->stop();
mpx_->close_pipe(); mpx_->close_pipe();
if (mpx_thread_.joinable()) if (mpx_thread_.joinable())
mpx_thread_.join(); mpx_thread_.join();
......
...@@ -67,4 +67,12 @@ void socket_manager::register_writing() { ...@@ -67,4 +67,12 @@ 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