Commit 2c905303 authored by Dominik Charousset's avatar Dominik Charousset

Simplify lifetime management of socket managers

Our previous design for the endpoint managers required us to use rely on
a shared ownership model for the multiplexer, because actor proxies held
a pointer to the socket manager directly. In our revised design, proxies
and managers are going to use a shared queue instead. This change
enables us to simplify the ownership model to a simple parent child
relation between the middleman and the multiplexer.
parent 29f9b037
...@@ -34,8 +34,10 @@ endfunction() ...@@ -34,8 +34,10 @@ endfunction()
# -- add library targets ------------------------------------------------------- # -- add library targets -------------------------------------------------------
add_library(libcaf_net_obj OBJECT ${CAF_NET_HEADERS} add_library(libcaf_net_obj OBJECT ${CAF_NET_HEADERS}
src/actor_proxy_impl.cpp #src/net/backend/tcp.cpp
src/basp/application.cpp #src/net/backend/test.cpp
#src/actor_proxy_impl.cpp
#src/basp/application.cpp
src/basp/connection_state_strings.cpp src/basp/connection_state_strings.cpp
src/basp/ec_strings.cpp src/basp/ec_strings.cpp
src/basp/message_type_strings.cpp src/basp/message_type_strings.cpp
...@@ -43,16 +45,13 @@ add_library(libcaf_net_obj OBJECT ${CAF_NET_HEADERS} ...@@ -43,16 +45,13 @@ add_library(libcaf_net_obj OBJECT ${CAF_NET_HEADERS}
src/convert_ip_endpoint.cpp src/convert_ip_endpoint.cpp
src/datagram_socket.cpp src/datagram_socket.cpp
src/defaults.cpp src/defaults.cpp
src/defaults.cpp #src/endpoint_manager.cpp
src/endpoint_manager.cpp
src/header.cpp src/header.cpp
src/host.cpp src/host.cpp
src/ip.cpp src/ip.cpp
src/message_queue.cpp src/message_queue.cpp
src/multiplexer.cpp src/multiplexer.cpp
#src/net/backend/test.cpp #src/net/endpoint_manager_queue.cpp
#src/net/backend/tcp.cpp
src/net/endpoint_manager_queue.cpp
src/net/middleman.cpp src/net/middleman.cpp
src/net/middleman_backend.cpp src/net/middleman_backend.cpp
src/net/packet_writer.cpp src/net/packet_writer.cpp
...@@ -148,7 +147,7 @@ caf_incubator_add_test_suites(caf-net-test ...@@ -148,7 +147,7 @@ caf_incubator_add_test_suites(caf-net-test
stream_transport stream_transport
#string_application #string_application
tcp_sockets tcp_sockets
transport_worker #transport_worker
transport_worker_dispatcher #transport_worker_dispatcher
udp_datagram_socket udp_datagram_socket
) )
...@@ -29,6 +29,7 @@ ...@@ -29,6 +29,7 @@
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/net/fwd.hpp" #include "caf/net/fwd.hpp"
#include "caf/net/middleman_backend.hpp" #include "caf/net/middleman_backend.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/scoped_actor.hpp" #include "caf/scoped_actor.hpp"
namespace caf::net { namespace caf::net {
...@@ -128,7 +129,11 @@ public: ...@@ -128,7 +129,11 @@ public:
return sys_.config(); return sys_.config();
} }
const multiplexer_ptr& mpx() const noexcept { multiplexer& mpx() noexcept {
return mpx_;
}
const multiplexer& mpx() const noexcept {
return mpx_; return mpx_;
} }
...@@ -159,7 +164,7 @@ private: ...@@ -159,7 +164,7 @@ private:
actor_system& sys_; actor_system& sys_;
/// Stores the global socket I/O multiplexer. /// Stores the global socket I/O multiplexer.
multiplexer_ptr mpx_; multiplexer mpx_;
/// Stores all available backends for managing peers. /// Stores all available backends for managing peers.
middleman_backend_list backends_; middleman_backend_list backends_;
......
...@@ -38,8 +38,7 @@ struct pollfd; ...@@ -38,8 +38,7 @@ struct pollfd;
namespace caf::net { namespace caf::net {
/// Multiplexes any number of ::socket_manager objects with a ::socket. /// Multiplexes any number of ::socket_manager objects with a ::socket.
class CAF_NET_EXPORT multiplexer class CAF_NET_EXPORT multiplexer {
: public std::enable_shared_from_this<multiplexer> {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
...@@ -49,10 +48,16 @@ public: ...@@ -49,10 +48,16 @@ public:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
multiplexer(); /// @param parent Points to the owning middleman instance. May be `nullptr`
/// only for the purpose of unit testing if no @ref
/// socket_manager requires access to the @ref middleman or the
/// @ref actor_system.
explicit multiplexer(middleman* parent);
~multiplexer(); ~multiplexer();
// -- initialization ---------------------------------------------------------
error init(); error init();
// -- properties ------------------------------------------------------------- // -- properties -------------------------------------------------------------
...@@ -63,6 +68,12 @@ public: ...@@ -63,6 +68,12 @@ public:
/// Returns the index of `mgr` in the pollset or `-1`. /// Returns the index of `mgr` in the pollset or `-1`.
ptrdiff_t index_of(const socket_manager_ptr& mgr); ptrdiff_t index_of(const socket_manager_ptr& mgr);
/// Returns the owning @ref middleman instance.
middleman& owner();
/// Returns the enclosing @ref actor_system.
actor_system& system();
// -- thread-safe signaling -------------------------------------------------- // -- thread-safe signaling --------------------------------------------------
/// Registers `mgr` for read events. /// Registers `mgr` for read events.
...@@ -123,20 +134,17 @@ protected: ...@@ -123,20 +134,17 @@ protected:
/// calling `init()`. /// calling `init()`.
std::thread::id tid_; std::thread::id tid_;
/// Used for pushing updates to the multiplexer's thread.
pipe_socket write_handle_;
/// Guards `write_handle_`. /// Guards `write_handle_`.
std::mutex write_lock_; std::mutex write_lock_;
/// Signals shutdown has been requested. /// Used for pushing updates to the multiplexer's thread.
bool shutting_down_; pipe_socket write_handle_;
};
/// @relates multiplexer /// Points to the owning middleman.
using multiplexer_ptr = std::shared_ptr<multiplexer>; middleman* owner_;
/// @relates multiplexer /// Signals whether shutdown has been requested.
using weak_multiplexer_ptr = std::weak_ptr<multiplexer>; bool shutting_down_ = false;
};
} // namespace caf::net } // namespace caf::net
...@@ -38,7 +38,7 @@ public: ...@@ -38,7 +38,7 @@ public:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
pollset_updater(pipe_socket read_handle, const multiplexer_ptr& parent); pollset_updater(pipe_socket read_handle, multiplexer* parent);
~pollset_updater() override; ~pollset_updater() override;
...@@ -61,7 +61,7 @@ public: ...@@ -61,7 +61,7 @@ public:
private: private:
msg_buf buf_; msg_buf buf_;
size_t buf_size_; size_t buf_size_ = 0;
}; };
} // namespace caf::net } // namespace caf::net
...@@ -34,9 +34,9 @@ class CAF_NET_EXPORT socket_manager : public ref_counted { ...@@ -34,9 +34,9 @@ class CAF_NET_EXPORT socket_manager : public ref_counted {
public: public:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
/// @pre `parent != nullptr`
/// @pre `handle != invalid_socket` /// @pre `handle != invalid_socket`
socket_manager(socket handle, const multiplexer_ptr& parent); /// @pre `parent != nullptr`
socket_manager(socket handle, multiplexer* parent);
~socket_manager() override; ~socket_manager() override;
...@@ -52,9 +52,14 @@ public: ...@@ -52,9 +52,14 @@ public:
return socket_cast<Socket>(handle_); return socket_cast<Socket>(handle_);
} }
/// Returns a pointer to the multiplexer running this `socket_manager`. /// Returns the owning @ref multiplexer instance.
multiplexer_ptr multiplexer() const { multiplexer& mpx() noexcept {
return parent_.lock(); return *parent_;
}
/// Returns the owning @ref multiplexer instance.
const multiplexer& mpx() const noexcept {
return *parent_;
} }
/// Returns registered operations (read, write, or both). /// Returns registered operations (read, write, or both).
...@@ -114,7 +119,7 @@ protected: ...@@ -114,7 +119,7 @@ protected:
operation mask_; operation mask_;
weak_multiplexer_ptr parent_; multiplexer* parent_;
error abort_reason_; error abort_reason_;
}; };
...@@ -123,8 +128,8 @@ template <class Protocol> ...@@ -123,8 +128,8 @@ template <class Protocol>
class socket_manager_impl : public socket_manager { class socket_manager_impl : public socket_manager {
public: public:
template <class... Ts> template <class... Ts>
socket_manager_impl(typename Protocol::socket_type handle, socket_manager_impl(typename Protocol::socket_type handle, multiplexer* mpx,
const multiplexer_ptr& mpx, Ts&&... xs) Ts&&... xs)
: socket_manager{handle, mpx}, protocol_(std::forward<Ts>(xs)...) { : socket_manager{handle, mpx}, protocol_(std::forward<Ts>(xs)...) {
// nop // nop
} }
...@@ -177,7 +182,9 @@ template <class B, template <class> class Layer, ...@@ -177,7 +182,9 @@ template <class B, template <class> class Layer,
template <class> class... Layers> template <class> class... Layers>
struct socket_type_helper<B, Layer, Layers...> struct socket_type_helper<B, Layer, Layers...>
: socket_type_helper<Layer<B>, Layers...> { : socket_type_helper<Layer<B>, Layers...> {
// no content using upper_input = typename B::input_tag;
using lower_output = typename Layer<B>::output_tag;
static_assert(std::is_same<upper_input, lower_output>::value);
}; };
template <class B, template <class> class... Layers> template <class B, template <class> class... Layers>
...@@ -203,8 +210,8 @@ using make_socket_manager_helper_t = ...@@ -203,8 +210,8 @@ using make_socket_manager_helper_t =
typename make_socket_manager_helper<B, Layers...>::type; typename make_socket_manager_helper<B, Layers...>::type;
template <class App, template <class> class... Layers, class... Ts> template <class App, template <class> class... Layers, class... Ts>
auto make_socket_manager(socket_type_t<App, Layers...> handle, auto make_socket_manager(socket_type_t<App, Layers...> handle, multiplexer* mpx,
const multiplexer_ptr& mpx, Ts&&... xs) { Ts&&... xs) {
static_assert(std::is_base_of<socket, socket_type_t<App, Layers...>>::value); static_assert(std::is_base_of<socket, socket_type_t<App, Layers...>>::value);
using impl using impl
= make_socket_manager_helper_t<App, Layers..., socket_manager_impl>; = make_socket_manager_helper_t<App, Layers..., socket_manager_impl>;
......
...@@ -26,6 +26,7 @@ ...@@ -26,6 +26,7 @@
#include "caf/expected.hpp" #include "caf/expected.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/make_counted.hpp" #include "caf/make_counted.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/operation.hpp" #include "caf/net/operation.hpp"
#include "caf/net/pollset_updater.hpp" #include "caf/net/pollset_updater.hpp"
#include "caf/net/socket_manager.hpp" #include "caf/net/socket_manager.hpp"
...@@ -78,7 +79,9 @@ short to_bitmask(operation op) { ...@@ -78,7 +79,9 @@ short to_bitmask(operation op) {
} // namespace } // namespace
multiplexer::multiplexer() : shutting_down_(false) { // -- constructors, destructors, and assignment operators ----------------------
multiplexer::multiplexer(middleman* owner) : owner_(owner) {
// nop // nop
} }
...@@ -86,15 +89,19 @@ multiplexer::~multiplexer() { ...@@ -86,15 +89,19 @@ multiplexer::~multiplexer() {
// nop // nop
} }
// -- initialization -----------------------------------------------------------
error multiplexer::init() { error multiplexer::init() {
auto pipe_handles = make_pipe(); auto pipe_handles = make_pipe();
if (!pipe_handles) if (!pipe_handles)
return std::move(pipe_handles.error()); return std::move(pipe_handles.error());
add(make_counted<pollset_updater>(pipe_handles->first, shared_from_this())); add(make_counted<pollset_updater>(pipe_handles->first, this));
write_handle_ = pipe_handles->second; write_handle_ = pipe_handles->second;
return none; return none;
} }
// -- properties ---------------------------------------------------------------
size_t multiplexer::num_socket_managers() const noexcept { size_t multiplexer::num_socket_managers() const noexcept {
return managers_.size(); return managers_.size();
} }
...@@ -106,6 +113,17 @@ ptrdiff_t multiplexer::index_of(const socket_manager_ptr& mgr) { ...@@ -106,6 +113,17 @@ ptrdiff_t multiplexer::index_of(const socket_manager_ptr& mgr) {
return i == last ? -1 : std::distance(first, i); return i == last ? -1 : std::distance(first, i);
} }
middleman& multiplexer::owner() {
CAF_ASSERT(owner_ != nullptr);
return *owner_;
}
actor_system& multiplexer::system() {
return owner().system();
}
// -- thread-safe signaling ----------------------------------------------------
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 (shutting_down_) { if (shutting_down_) {
...@@ -150,6 +168,8 @@ void multiplexer::close_pipe() { ...@@ -150,6 +168,8 @@ void multiplexer::close_pipe() {
} }
} }
// -- control flow -------------------------------------------------------------
bool multiplexer::poll_once(bool blocking) { bool multiplexer::poll_once(bool blocking) {
if (pollset_.empty()) if (pollset_.empty())
return false; return false;
...@@ -216,6 +236,7 @@ bool multiplexer::poll_once(bool blocking) { ...@@ -216,6 +236,7 @@ bool multiplexer::poll_once(bool blocking) {
} }
void multiplexer::set_thread_id() { void multiplexer::set_thread_id() {
CAF_LOG_TRACE("");
tid_ = std::this_thread::get_id(); tid_ = std::this_thread::get_id();
} }
...@@ -227,6 +248,7 @@ void multiplexer::run() { ...@@ -227,6 +248,7 @@ void multiplexer::run() {
void multiplexer::shutdown() { void multiplexer::shutdown() {
if (std::this_thread::get_id() == tid_) { if (std::this_thread::get_id() == tid_) {
CAF_LOG_DEBUG("initiate shutdown");
shutting_down_ = true; shutting_down_ = true;
// First manager is the pollset_updater. Skip it and delete later. // First manager is the pollset_updater. Skip it and delete later.
for (size_t i = 1; i < managers_.size();) { for (size_t i = 1; i < managers_.size();) {
...@@ -242,10 +264,13 @@ void multiplexer::shutdown() { ...@@ -242,10 +264,13 @@ void multiplexer::shutdown() {
} }
close_pipe(); close_pipe();
} else { } else {
CAF_LOG_DEBUG("push shutdown event to pipe");
write_to_pipe(4, nullptr); write_to_pipe(4, nullptr);
} }
} }
// -- utility functions --------------------------------------------------------
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,6 +304,7 @@ short multiplexer::handle(const socket_manager_ptr& mgr, short events, ...@@ -279,6 +304,7 @@ short multiplexer::handle(const socket_manager_ptr& mgr, short events,
} }
void multiplexer::add(socket_manager_ptr mgr) { void multiplexer::add(socket_manager_ptr mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle()));
CAF_ASSERT(index_of(mgr) == -1); CAF_ASSERT(index_of(mgr) == -1);
pollfd new_entry{socket_cast<socket_id>(mgr->handle()), pollfd new_entry{socket_cast<socket_id>(mgr->handle()),
to_bitmask(mgr->mask()), 0}; to_bitmask(mgr->mask()), 0};
......
...@@ -25,7 +25,6 @@ ...@@ -25,7 +25,6 @@
#include "caf/net/basp/ec.hpp" #include "caf/net/basp/ec.hpp"
#include "caf/net/endpoint_manager.hpp" #include "caf/net/endpoint_manager.hpp"
#include "caf/net/middleman_backend.hpp" #include "caf/net/middleman_backend.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/raise_error.hpp" #include "caf/raise_error.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/send.hpp" #include "caf/send.hpp"
...@@ -37,8 +36,8 @@ void middleman::init_global_meta_objects() { ...@@ -37,8 +36,8 @@ void middleman::init_global_meta_objects() {
caf::init_global_meta_objects<id_block::net_module>(); caf::init_global_meta_objects<id_block::net_module>();
} }
middleman::middleman(actor_system& sys) : sys_(sys) { middleman::middleman(actor_system& sys) : sys_(sys), mpx_(this) {
mpx_ = std::make_shared<multiplexer>(); // nop
} }
middleman::~middleman() { middleman::~middleman() {
...@@ -47,15 +46,13 @@ middleman::~middleman() { ...@@ -47,15 +46,13 @@ middleman::~middleman() {
void middleman::start() { void middleman::start() {
if (!get_or(config(), "caf.middleman.manual-multiplexing", false)) { if (!get_or(config(), "caf.middleman.manual-multiplexing", false)) {
auto mpx = mpx_; mpx_thread_ = std::thread{[this] {
auto sys_ptr = &system(); CAF_SET_LOGGER_SYS(&sys_);
mpx_thread_ = std::thread{[mpx, sys_ptr] { detail::set_thread_name("caf.net.mpx");
CAF_SET_LOGGER_SYS(sys_ptr); sys_.thread_started();
detail::set_thread_name("caf.multiplexer"); mpx_.set_thread_id();
sys_ptr->thread_started(); mpx_.run();
mpx->set_thread_id(); sys_.thread_terminates();
mpx->run();
sys_ptr->thread_terminates();
}}; }};
} }
} }
...@@ -63,17 +60,17 @@ void middleman::start() { ...@@ -63,17 +60,17 @@ void middleman::start() {
void middleman::stop() { void middleman::stop() {
for (const auto& backend : backends_) for (const auto& backend : backends_)
backend->stop(); backend->stop();
mpx_->shutdown(); mpx_.shutdown();
if (mpx_thread_.joinable()) if (mpx_thread_.joinable())
mpx_thread_.join(); mpx_thread_.join();
else else
mpx_->run(); 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: " << err); CAF_LOG_ERROR("mpx_.init() failed: " << err);
CAF_RAISE_ERROR("mpx->init() failed"); CAF_RAISE_ERROR("mpx_.init() failed");
} }
if (auto node_uri = get_if<uri>(&cfg, "caf.middleman.this-node")) { if (auto node_uri = get_if<uri>(&cfg, "caf.middleman.this-node")) {
auto this_node = make_node_id(std::move(*node_uri)); auto this_node = make_node_id(std::move(*node_uri));
......
...@@ -28,12 +28,9 @@ ...@@ -28,12 +28,9 @@
namespace caf::net { namespace caf::net {
pollset_updater::pollset_updater(pipe_socket read_handle, pollset_updater::pollset_updater(pipe_socket read_handle, multiplexer* parent)
const multiplexer_ptr& parent) : super(read_handle, parent) {
: super(read_handle, parent), buf_size_(0) {
mask_ = operation::read; mask_ = operation::read;
if (auto err = nonblocking(read_handle, true))
CAF_LOG_ERROR("nonblocking failed: " << err);
} }
pollset_updater::~pollset_updater() { pollset_updater::~pollset_updater() {
...@@ -41,7 +38,7 @@ pollset_updater::~pollset_updater() { ...@@ -41,7 +38,7 @@ pollset_updater::~pollset_updater() {
} }
error pollset_updater::init(const settings&) { error pollset_updater::init(const settings&) {
return none; return nonblocking(handle(), true);
} }
bool pollset_updater::handle_read_event() { bool pollset_updater::handle_read_event() {
...@@ -56,25 +53,26 @@ bool pollset_updater::handle_read_event() { ...@@ -56,25 +53,26 @@ bool pollset_updater::handle_read_event() {
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}; socket_manager_ptr mgr{reinterpret_cast<socket_manager*>(value), false};
if (auto ptr = parent_.lock()) {
switch (opcode) { switch (opcode) {
case 0: case 0:
ptr->register_reading(mgr); parent_->register_reading(mgr);
break; break;
case 1: case 1:
ptr->register_writing(mgr); parent_->register_writing(mgr);
break; break;
case 4: case 4:
ptr->shutdown(); parent_->shutdown();
break; break;
default: default:
CAF_LOG_DEBUG("opcode not recognized: " << CAF_ARG(opcode)); CAF_LOG_ERROR("opcode not recognized: " << CAF_ARG(opcode));
break; break;
} }
} }
} } else if (num_bytes == 0) {
CAF_LOG_DEBUG("pipe closed, assume shutdown");
return false;
} else { } else {
return num_bytes < 0 && last_socket_error_is_temporary(); return last_socket_error_is_temporary();
} }
} }
} }
......
...@@ -23,10 +23,10 @@ ...@@ -23,10 +23,10 @@
namespace caf::net { namespace caf::net {
socket_manager::socket_manager(socket handle, const multiplexer_ptr& parent) socket_manager::socket_manager(socket handle, multiplexer* parent)
: handle_(handle), mask_(operation::none), parent_(parent) { : handle_(handle), mask_(operation::none), parent_(parent) {
CAF_ASSERT(parent != nullptr);
CAF_ASSERT(handle_ != invalid_socket); CAF_ASSERT(handle_ != invalid_socket);
CAF_ASSERT(parent != nullptr);
} }
socket_manager::~socket_manager() { socket_manager::~socket_manager() {
...@@ -54,17 +54,13 @@ bool socket_manager::mask_del(operation flag) noexcept { ...@@ -54,17 +54,13 @@ bool socket_manager::mask_del(operation flag) noexcept {
void socket_manager::register_reading() { void socket_manager::register_reading() {
if ((mask() & operation::read) == operation::read) if ((mask() & operation::read) == operation::read)
return; return;
auto ptr = parent_.lock(); parent_->register_reading(this);
if (ptr != nullptr)
ptr->register_reading(this);
} }
void socket_manager::register_writing() { void socket_manager::register_writing() {
if ((mask() & operation::write) == operation::write) if ((mask() & operation::write) == operation::write)
return; return;
auto ptr = parent_.lock(); parent_->register_writing(this);
if (ptr != nullptr)
ptr->register_writing(this);
} }
} // namespace caf::net } // namespace caf::net
...@@ -40,15 +40,15 @@ namespace { ...@@ -40,15 +40,15 @@ namespace {
class dummy_manager : public socket_manager { class dummy_manager : public socket_manager {
public: public:
dummy_manager(size_t& manager_count, stream_socket handle, dummy_manager(size_t& manager_count, stream_socket handle,
multiplexer_ptr parent) multiplexer* parent)
: socket_manager(handle, std::move(parent)), : socket_manager(handle, parent), count_(manager_count) {
count_(manager_count), CAF_MESSAGE("created new dummy manager");
rd_buf_pos_(0) {
++count_; ++count_;
rd_buf_.resize(1024); rd_buf_.resize(1024);
} }
~dummy_manager() { ~dummy_manager() {
CAF_MESSAGE("destroyed dummy manager");
--count_; --count_;
} }
...@@ -114,7 +114,7 @@ private: ...@@ -114,7 +114,7 @@ private:
size_t& count_; size_t& count_;
size_t rd_buf_pos_; size_t rd_buf_pos_ = 0;
byte_buffer wr_buf_; byte_buffer wr_buf_;
...@@ -124,23 +124,22 @@ private: ...@@ -124,23 +124,22 @@ private:
using dummy_manager_ptr = intrusive_ptr<dummy_manager>; using dummy_manager_ptr = intrusive_ptr<dummy_manager>;
struct fixture : host_fixture { struct fixture : host_fixture {
fixture() : manager_count(0), mpx(std::make_shared<multiplexer>()) { fixture() : mpx(nullptr) {
mpx->set_thread_id(); mpx.set_thread_id();
} }
~fixture() { ~fixture() {
mpx.reset();
CAF_REQUIRE_EQUAL(manager_count, 0u); CAF_REQUIRE_EQUAL(manager_count, 0u);
} }
void exhaust() { void exhaust() {
while (mpx->poll_once(false)) while (mpx.poll_once(false))
; // Repeat. ; // Repeat.
} }
size_t manager_count; size_t manager_count = 0;
multiplexer_ptr mpx; multiplexer mpx;
}; };
} // namespace } // namespace
...@@ -148,32 +147,36 @@ struct fixture : host_fixture { ...@@ -148,32 +147,36 @@ struct fixture : host_fixture {
CAF_TEST_FIXTURE_SCOPE(multiplexer_tests, fixture) CAF_TEST_FIXTURE_SCOPE(multiplexer_tests, fixture)
CAF_TEST(default construction) { CAF_TEST(default construction) {
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 0u); CAF_CHECK_EQUAL(mpx.num_socket_managers(), 0u);
} }
CAF_TEST(init) { CAF_TEST(init) {
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 0u); CAF_CHECK_EQUAL(mpx.num_socket_managers(), 0u);
CAF_REQUIRE_EQUAL(mpx->init(), none); CAF_REQUIRE_EQUAL(mpx.init(), none);
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 1u); CAF_CHECK_EQUAL(mpx.num_socket_managers(), 1u);
mpx->close_pipe(); mpx.close_pipe();
exhaust(); exhaust();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 0u); CAF_CHECK_EQUAL(mpx.num_socket_managers(), 0u);
// Calling run must have no effect now. // Calling run must have no effect now.
mpx->run(); mpx.run();
} }
CAF_TEST(send and receive) { CAF_TEST(send and receive) {
CAF_REQUIRE_EQUAL(mpx->init(), none); CAF_REQUIRE_EQUAL(mpx.init(), none);
auto sockets = unbox(make_stream_socket_pair()); auto sockets = unbox(make_stream_socket_pair());
auto alice = make_counted<dummy_manager>(manager_count, sockets.first, mpx); { // Lifetime scope of alice and bob.
auto bob = make_counted<dummy_manager>(manager_count, sockets.second, mpx); auto alice = make_counted<dummy_manager>(manager_count, sockets.first,
&mpx);
auto bob = make_counted<dummy_manager>(manager_count, sockets.second, &mpx);
alice->register_reading(); alice->register_reading();
bob->register_reading(); bob->register_reading();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 3u); CAF_CHECK_EQUAL(mpx.num_socket_managers(), 3u);
alice->send("hello bob"); alice->send("hello bob");
alice->register_writing(); alice->register_writing();
exhaust(); exhaust();
CAF_CHECK_EQUAL(bob->receive(), "hello bob"); CAF_CHECK_EQUAL(bob->receive(), "hello bob");
}
mpx.shutdown();
} }
CAF_TEST(shutdown) { CAF_TEST(shutdown) {
...@@ -181,26 +184,27 @@ CAF_TEST(shutdown) { ...@@ -181,26 +184,27 @@ CAF_TEST(shutdown) {
std::condition_variable cv; std::condition_variable cv;
bool thread_id_set = false; bool thread_id_set = false;
auto run_mpx = [&] { auto run_mpx = [&] {
std::unique_lock<std::mutex> lk(m); {
mpx->set_thread_id(); std::unique_lock<std::mutex> guard(m);
mpx.set_thread_id();
thread_id_set = true; thread_id_set = true;
lk.unlock();
cv.notify_one(); cv.notify_one();
mpx->run(); }
mpx.run();
}; };
CAF_REQUIRE_EQUAL(mpx->init(), none); CAF_REQUIRE_EQUAL(mpx.init(), none);
auto sockets = unbox(make_stream_socket_pair()); auto sockets = unbox(make_stream_socket_pair());
auto alice = make_counted<dummy_manager>(manager_count, sockets.first, mpx); auto alice = make_counted<dummy_manager>(manager_count, sockets.first, &mpx);
auto bob = make_counted<dummy_manager>(manager_count, sockets.second, mpx); auto bob = make_counted<dummy_manager>(manager_count, sockets.second, &mpx);
alice->register_reading(); alice->register_reading();
bob->register_reading(); bob->register_reading();
CAF_REQUIRE_EQUAL(mpx->num_socket_managers(), 3u); CAF_REQUIRE_EQUAL(mpx.num_socket_managers(), 3u);
std::thread mpx_thread{run_mpx}; std::thread mpx_thread{run_mpx};
std::unique_lock<std::mutex> lk(m); std::unique_lock<std::mutex> lk(m);
cv.wait(lk, [&] { return thread_id_set; }); cv.wait(lk, [&] { return thread_id_set; });
mpx->shutdown(); mpx.shutdown();
mpx_thread.join(); mpx_thread.join();
CAF_REQUIRE_EQUAL(mpx->num_socket_managers(), 0u); CAF_REQUIRE_EQUAL(mpx.num_socket_managers(), 0u);
} }
CAF_TEST_FIXTURE_SCOPE_END() CAF_TEST_FIXTURE_SCOPE_END()
...@@ -46,14 +46,14 @@ struct fixture : test_coordinator_fixture<>, host_fixture { ...@@ -46,14 +46,14 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
using byte_buffer_ptr = std::shared_ptr<byte_buffer>; using byte_buffer_ptr = std::shared_ptr<byte_buffer>;
fixture() fixture()
: recv_buf(1024), : mpx(nullptr),
recv_buf(1024),
shared_recv_buf{std::make_shared<byte_buffer>()}, shared_recv_buf{std::make_shared<byte_buffer>()},
shared_send_buf{std::make_shared<byte_buffer>()} { shared_send_buf{std::make_shared<byte_buffer>()} {
mpx = std::make_shared<multiplexer>(); if (auto err = mpx.init())
if (auto err = mpx->init()) CAF_FAIL("mpx.init failed: " << err);
CAF_FAIL("mpx->init failed: " << err); mpx.set_thread_id();
mpx->set_thread_id(); CAF_CHECK_EQUAL(mpx.num_socket_managers(), 1u);
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 1u);
auto sockets = unbox(make_stream_socket_pair()); auto sockets = unbox(make_stream_socket_pair());
send_socket_guard.reset(sockets.first); send_socket_guard.reset(sockets.first);
recv_socket_guard.reset(sockets.second); recv_socket_guard.reset(sockets.second);
...@@ -62,11 +62,11 @@ struct fixture : test_coordinator_fixture<>, host_fixture { ...@@ -62,11 +62,11 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
} }
bool handle_io_event() override { bool handle_io_event() override {
return mpx->poll_once(false); return mpx.poll_once(false);
} }
settings config; settings config;
multiplexer_ptr mpx; multiplexer mpx;
byte_buffer recv_buf; byte_buffer recv_buf;
socket_guard<stream_socket> send_socket_guard; socket_guard<stream_socket> send_socket_guard;
socket_guard<stream_socket> recv_socket_guard; socket_guard<stream_socket> recv_socket_guard;
...@@ -75,9 +75,11 @@ struct fixture : test_coordinator_fixture<>, host_fixture { ...@@ -75,9 +75,11 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
}; };
class dummy_application { class dummy_application {
public:
using byte_buffer_ptr = std::shared_ptr<byte_buffer>; using byte_buffer_ptr = std::shared_ptr<byte_buffer>;
public: using input_tag = tag::stream_oriented;
explicit dummy_application(byte_buffer_ptr recv_buf, byte_buffer_ptr send_buf) explicit dummy_application(byte_buffer_ptr recv_buf, byte_buffer_ptr send_buf)
: recv_buf_(std::move(recv_buf)), : recv_buf_(std::move(recv_buf)),
send_buf_(std::move(send_buf)){ send_buf_(std::move(send_buf)){
...@@ -164,9 +166,9 @@ CAF_TEST_FIXTURE_SCOPE(endpoint_manager_tests, fixture) ...@@ -164,9 +166,9 @@ CAF_TEST_FIXTURE_SCOPE(endpoint_manager_tests, fixture)
CAF_TEST(receive) { CAF_TEST(receive) {
auto mgr = make_socket_manager<dummy_application, stream_transport>( auto mgr = make_socket_manager<dummy_application, stream_transport>(
recv_socket_guard.release(), mpx, shared_recv_buf, shared_send_buf); recv_socket_guard.release(), &mpx, shared_recv_buf, shared_send_buf);
CAF_CHECK_EQUAL(mgr->init(config), none); CAF_CHECK_EQUAL(mgr->init(config), none);
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2u); CAF_CHECK_EQUAL(mpx.num_socket_managers(), 2u);
CAF_CHECK_EQUAL( CAF_CHECK_EQUAL(
static_cast<size_t>( static_cast<size_t>(
write(send_socket_guard.socket(), as_bytes(make_span(hello_manager)))), write(send_socket_guard.socket(), as_bytes(make_span(hello_manager)))),
...@@ -180,9 +182,9 @@ CAF_TEST(receive) { ...@@ -180,9 +182,9 @@ CAF_TEST(receive) {
CAF_TEST(send) { CAF_TEST(send) {
auto mgr = make_socket_manager<dummy_application, stream_transport>( auto mgr = make_socket_manager<dummy_application, stream_transport>(
recv_socket_guard.release(), mpx, shared_recv_buf, shared_send_buf); recv_socket_guard.release(), &mpx, shared_recv_buf, shared_send_buf);
CAF_CHECK_EQUAL(mgr->init(config), none); CAF_CHECK_EQUAL(mgr->init(config), none);
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2u); CAF_CHECK_EQUAL(mpx.num_socket_managers(), 2u);
mgr->register_writing(); mgr->register_writing();
while (handle_io_event()) while (handle_io_event())
; ;
......
...@@ -154,23 +154,23 @@ struct fixture : test_coordinator_fixture<>, host_fixture { ...@@ -154,23 +154,23 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
using worker_type = transport_worker<dummy_application, ip_endpoint>; using worker_type = transport_worker<dummy_application, ip_endpoint>;
fixture() fixture()
: transport_results{std::make_shared<transport_result>()}, : mpx(nullptr),
transport_results{std::make_shared<transport_result>()},
application_results{std::make_shared<application_result>()}, application_results{std::make_shared<application_result>()},
transport(sys, transport_results), transport(sys, transport_results),
worker{dummy_application{application_results}} { worker{dummy_application{application_results}} {
mpx = std::make_shared<multiplexer>(); if (auto err = mpx.init())
if (auto err = mpx->init()) CAF_FAIL("mpx.init failed: " << err);
CAF_FAIL("mpx->init failed: " << err);
if (auto err = parse("[::1]:12345", ep)) if (auto err = parse("[::1]:12345", ep))
CAF_FAIL("parse returned an error: " << err); CAF_FAIL("parse returned an error: " << err);
worker = worker_type{dummy_application{application_results}, ep}; worker = worker_type{dummy_application{application_results}, ep};
} }
bool handle_io_event() override { bool handle_io_event() override {
return mpx->poll_once(false); return mpx.poll_once(false);
} }
multiplexer_ptr mpx; multiplexer mpx;
std::shared_ptr<transport_result> transport_results; std::shared_ptr<transport_result> transport_results;
std::shared_ptr<application_result> application_results; std::shared_ptr<application_result> application_results;
dummy_transport transport; dummy_transport transport;
......
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