Unverified Commit a19cb8e6 authored by Samir Halilčević's avatar Samir Halilčević Committed by GitHub

Refactor multiplexer to be an interface

parent bb22e7fe
...@@ -29,7 +29,6 @@ caf_add_component( ...@@ -29,7 +29,6 @@ caf_add_component(
${CAF_NET_HEADERS} ${CAF_NET_HEADERS}
SOURCES SOURCES
caf/detail/convert_ip_endpoint.cpp caf/detail/convert_ip_endpoint.cpp
caf/detail/pollset_updater.cpp
caf/detail/rfc6455.cpp caf/detail/rfc6455.cpp
caf/detail/rfc6455.test.cpp caf/detail/rfc6455.test.cpp
caf/net/abstract_actor_shell.cpp caf/net/abstract_actor_shell.cpp
......
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#include "caf/detail/pollset_updater.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/action.hpp"
#include "caf/actor_system.hpp"
#include "caf/logger.hpp"
#include "caf/sec.hpp"
#include "caf/span.hpp"
#include <cstring>
namespace caf::detail {
// -- constructors, destructors, and assignment operators ----------------------
pollset_updater::pollset_updater(net::pipe_socket fd) : fd_(fd) {
// nop
}
// -- factories ----------------------------------------------------------------
std::unique_ptr<pollset_updater> pollset_updater::make(net::pipe_socket fd) {
return std::make_unique<pollset_updater>(fd);
}
// -- interface functions ------------------------------------------------------
error pollset_updater::start(net::socket_manager* owner) {
CAF_LOG_TRACE("");
owner_ = owner;
mpx_ = owner->mpx_ptr();
return net::nonblocking(fd_, true);
}
net::socket pollset_updater::handle() const {
return fd_;
}
void pollset_updater::handle_read_event() {
CAF_LOG_TRACE("");
auto as_mgr = [](intptr_t ptr) {
return intrusive_ptr{reinterpret_cast<net::socket_manager*>(ptr), false};
};
auto add_action = [this](intptr_t ptr) {
auto f = action{intrusive_ptr{reinterpret_cast<action::impl*>(ptr), false}};
mpx_->pending_actions_.push_back(std::move(f));
};
for (;;) {
CAF_ASSERT((buf_.size() - buf_size_) > 0);
auto num_bytes
= read(fd_, make_span(buf_.data() + buf_size_, buf_.size() - buf_size_));
if (num_bytes > 0) {
buf_size_ += static_cast<size_t>(num_bytes);
if (buf_.size() == buf_size_) {
buf_size_ = 0;
auto opcode = static_cast<uint8_t>(buf_[0]);
intptr_t ptr;
memcpy(&ptr, buf_.data() + 1, sizeof(intptr_t));
switch (static_cast<code>(opcode)) {
case code::start_manager:
mpx_->do_start(as_mgr(ptr));
break;
case code::run_action:
add_action(ptr);
break;
case code::shutdown:
CAF_ASSERT(ptr == 0);
mpx_->do_shutdown();
break;
default:
CAF_LOG_ERROR("opcode not recognized: " << CAF_ARG(opcode));
break;
}
}
} else if (num_bytes == 0) {
CAF_LOG_DEBUG("pipe closed, assume shutdown");
owner_->deregister();
return;
} else if (net::last_socket_error_is_temporary()) {
return;
} else {
CAF_LOG_ERROR("pollset updater failed to read from the pipe");
owner_->deregister();
return;
}
}
}
void pollset_updater::handle_write_event() {
owner_->deregister_writing();
}
void pollset_updater::abort(const error&) {
// nop
}
} // namespace caf::detail
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/net/pipe_socket.hpp"
#include "caf/net/socket_event_layer.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/detail/net_export.hpp"
#include <array>
#include <cstddef>
#include <cstdint>
#include <mutex>
namespace caf::detail {
class CAF_NET_EXPORT pollset_updater : public net::socket_event_layer {
public:
// -- member types -----------------------------------------------------------
using super = net::socket_manager;
using msg_buf = std::array<std::byte, sizeof(intptr_t) + 1>;
enum class code : uint8_t {
start_manager,
shutdown_reading,
shutdown_writing,
run_action,
shutdown,
};
// -- constructors, destructors, and assignment operators --------------------
explicit pollset_updater(net::pipe_socket fd);
// -- factories --------------------------------------------------------------
static std::unique_ptr<pollset_updater> make(net::pipe_socket fd);
// -- implementation of socket_event_layer -----------------------------------
error start(net::socket_manager* owner) override;
net::socket handle() const override;
void handle_read_event() override;
void handle_write_event() override;
void abort(const error& reason) override;
private:
net::pipe_socket fd_;
net::socket_manager* owner_ = nullptr;
net::multiplexer* mpx_ = nullptr;
msg_buf buf_;
size_t buf_size_ = 0;
};
} // namespace caf::detail
...@@ -90,12 +90,6 @@ enum class errc; ...@@ -90,12 +90,6 @@ enum class errc;
} // namespace caf::net::octet_stream } // namespace caf::net::octet_stream
namespace caf::detail {
class pollset_updater;
} // namespace caf::detail
namespace caf::net::lp { namespace caf::net::lp {
class client; class client;
......
...@@ -4,23 +4,38 @@ ...@@ -4,23 +4,38 @@
#include "caf/net/multiplexer.hpp" #include "caf/net/multiplexer.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/middleman.hpp" #include "caf/net/middleman.hpp"
#include "caf/net/pipe_socket.hpp"
#include "caf/net/socket.hpp"
#include "caf/net/socket_event_layer.hpp"
#include "caf/net/socket_manager.hpp" #include "caf/net/socket_manager.hpp"
#include "caf/action.hpp" #include "caf/action.hpp"
#include "caf/actor_system.hpp" #include "caf/actor_system.hpp"
#include "caf/async/execution_context.hpp"
#include "caf/byte.hpp" #include "caf/byte.hpp"
#include "caf/config.hpp" #include "caf/config.hpp"
#include "caf/detail/pollset_updater.hpp" #include "caf/detail/atomic_ref_counted.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/error.hpp" #include "caf/error.hpp"
#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/ref_counted.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/span.hpp" #include "caf/span.hpp"
#include "caf/unordered_flat_map.hpp"
#include <algorithm> #include <algorithm>
#include <array>
#include <cstddef>
#include <cstdint>
#include <deque>
#include <memory>
#include <mutex>
#include <optional> #include <optional>
#include <thread>
#ifndef CAF_WINDOWS #ifndef CAF_WINDOWS
# include <poll.h> # include <poll.h>
...@@ -31,6 +46,10 @@ ...@@ -31,6 +46,10 @@
namespace caf::net { namespace caf::net {
namespace {
class default_multiplexer;
#ifndef POLLRDHUP #ifndef POLLRDHUP
# define POLLRDHUP POLLHUP # define POLLRDHUP POLLHUP
#endif #endif
...@@ -39,8 +58,6 @@ namespace caf::net { ...@@ -39,8 +58,6 @@ namespace caf::net {
# define POLLPRI POLLIN # define POLLPRI POLLIN
#endif #endif
namespace {
#ifdef CAF_WINDOWS #ifdef CAF_WINDOWS
// From the MSDN: If the POLLPRI flag is set on a socket for the Microsoft // From the MSDN: If the POLLPRI flag is set on a socket for the Microsoft
// Winsock provider, the WSAPoll function will fail. // Winsock provider, the WSAPoll function will fail.
...@@ -53,411 +70,591 @@ const short error_mask = POLLRDHUP | POLLERR | POLLHUP | POLLNVAL; ...@@ -53,411 +70,591 @@ const short error_mask = POLLRDHUP | POLLERR | POLLHUP | POLLNVAL;
const short output_mask = POLLOUT; const short output_mask = POLLOUT;
} // namespace class pollset_updater : public socket_event_layer {
public:
// -- member types -----------------------------------------------------------
// -- static utility functions ------------------------------------------------- using super = socket_manager;
#ifdef CAF_LINUX using msg_buf = std::array<std::byte, sizeof(intptr_t) + 1>;
void multiplexer::block_sigpipe() { enum class code : uint8_t {
sigset_t sigpipe_mask; start_manager,
sigemptyset(&sigpipe_mask); shutdown_reading,
sigaddset(&sigpipe_mask, SIGPIPE); shutdown_writing,
sigset_t saved_mask; run_action,
if (pthread_sigmask(SIG_BLOCK, &sigpipe_mask, &saved_mask) == -1) { shutdown,
perror("pthread_sigmask"); };
exit(1);
// -- constructors, destructors, and assignment operators --------------------
explicit pollset_updater(pipe_socket fd) : fd_(fd) {
// nop
} }
}
#else // -- factories --------------------------------------------------------------
void multiplexer::block_sigpipe() { static std::unique_ptr<pollset_updater> make(pipe_socket fd) {
// nop return std::make_unique<pollset_updater>(fd);
} }
#endif // -- implementation of socket_event_layer -----------------------------------
multiplexer* multiplexer::from(actor_system& sys) { error start(socket_manager* owner) override;
return sys.network_manager().mpx_ptr();
}
// -- constructors, destructors, and assignment operators ---------------------- socket handle() const override {
return fd_;
}
multiplexer::multiplexer(middleman* owner) : owner_(owner) { void handle_read_event() override;
// nop
}
multiplexer::~multiplexer() { void handle_write_event() override {
// nop owner_->deregister_writing();
} }
// -- initialization ----------------------------------------------------------- void abort(const error&) override {
// nop
error multiplexer::init() { }
auto pipe_handles = make_pipe();
if (!pipe_handles)
return std::move(pipe_handles.error());
auto updater = detail::pollset_updater::make(pipe_handles->first);
auto mgr = socket_manager::make(this, std::move(updater));
if (auto err = mgr->start())
return err;
write_handle_ = pipe_handles->second;
pollset_.emplace_back(pollfd{pipe_handles->first.id, input_mask, 0});
managers_.emplace_back(mgr);
return none;
}
// -- properties --------------------------------------------------------------- private:
pipe_socket fd_;
socket_manager* owner_ = nullptr;
default_multiplexer* mpx_ = nullptr;
msg_buf buf_;
size_t buf_size_ = 0;
};
size_t multiplexer::num_socket_managers() const noexcept { /// Multiplexes any number of ::socket_manager objects with a ::socket.
return managers_.size(); class default_multiplexer : public detail::atomic_ref_counted,
} public multiplexer {
public:
// -- member types -----------------------------------------------------------
ptrdiff_t multiplexer::index_of(const socket_manager_ptr& mgr) const noexcept { struct poll_update {
auto first = managers_.begin(); short events = 0;
auto last = managers_.end(); socket_manager_ptr mgr;
auto i = std::find(first, last, mgr); };
return i == last ? -1 : std::distance(first, i);
}
ptrdiff_t multiplexer::index_of(socket fd) const noexcept { using poll_update_map = unordered_flat_map<socket, poll_update>;
auto first = pollset_.begin();
auto last = pollset_.end();
auto i = std::find_if(first, last,
[fd](const pollfd& x) { return x.fd == fd.id; });
return i == last ? -1 : std::distance(first, i);
}
middleman& multiplexer::owner() { using pollfd_list = std::vector<pollfd>;
CAF_ASSERT(owner_ != nullptr);
return *owner_;
}
actor_system& multiplexer::system() { using manager_list = std::vector<socket_manager_ptr>;
return owner().system();
}
// -- implementation of execution_context -------------------------------------- // -- friends ----------------------------------------------------------------
void multiplexer::ref_execution_context() const noexcept { friend class pollset_updater; // Needs access to the `do_*` functions.
ref();
}
void multiplexer::deref_execution_context() const noexcept { // -- constructors, destructors, and assignment operators --------------------
deref();
}
void multiplexer::schedule(action what) { explicit default_multiplexer(middleman* parent) : owner_(parent) {
CAF_LOG_TRACE(""); // nop
if (std::this_thread::get_id() == tid_) {
pending_actions_.push_back(what);
} else {
auto ptr = std::move(what).as_intrusive_ptr().release();
write_to_pipe(detail::pollset_updater::code::run_action, ptr);
} }
}
void multiplexer::watch(disposable what) { // -- implementation of caf::net::multiplexer --------------------------------
watched_.emplace_back(what);
} error init() override {
auto pipe_handles = make_pipe();
if (!pipe_handles)
return std::move(pipe_handles.error());
auto updater = pollset_updater::make(pipe_handles->first);
auto mgr = socket_manager::make(this, std::move(updater));
if (auto err = mgr->start()) {
close(pipe_handles->second);
return err;
}
write_handle_ = pipe_handles->second;
pollset_.emplace_back(pollfd{pipe_handles->first.id, input_mask, 0});
managers_.emplace_back(mgr);
return none;
}
// -- thread-safe signaling ---------------------------------------------------- size_t num_socket_managers() const noexcept override {
return managers_.size();
}
void multiplexer::start(socket_manager_ptr mgr) { middleman& owner() override {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id)); CAF_ASSERT(owner_ != nullptr);
if (std::this_thread::get_id() == tid_) { return *owner_;
do_start(mgr); }
} else { actor_system& system() override {
write_to_pipe(detail::pollset_updater::code::start_manager, mgr.release()); return owner().system();
} }
}
void multiplexer::shutdown() { // -- implementation of execution_context ------------------------------------
CAF_LOG_TRACE("");
// Note: there is no 'shortcut' when calling the function in the multiplexer's
// thread, because do_shutdown calls apply_updates. This must only be called
// from the pollset_updater.
CAF_LOG_DEBUG("push shutdown event to pipe");
write_to_pipe(detail::pollset_updater::code::shutdown,
static_cast<socket_manager*>(nullptr));
}
// -- callbacks for socket managers -------------------------------------------- void ref_execution_context() const noexcept override {
ref();
}
void multiplexer::register_reading(socket_manager* mgr) { void deref_execution_context() const noexcept override {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id)); deref();
update_for(mgr).events |= input_mask; }
}
void multiplexer::register_writing(socket_manager* mgr) { void schedule(action what) override {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id)); CAF_LOG_TRACE("");
update_for(mgr).events |= output_mask; if (std::this_thread::get_id() == tid_) {
} pending_actions.push_back(what);
} else {
auto ptr = std::move(what).as_intrusive_ptr().release();
write_to_pipe(pollset_updater::code::run_action, ptr);
}
}
void multiplexer::deregister_reading(socket_manager* mgr) { void watch(disposable what) override {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id)); watched_.emplace_back(what);
update_for(mgr).events &= ~input_mask; }
}
void multiplexer::deregister_writing(socket_manager* mgr) { // -- thread-safe signaling --------------------------------------------------
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
update_for(mgr).events &= ~output_mask;
}
void multiplexer::deregister(socket_manager* mgr) { void start(socket_manager_ptr mgr) override {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id)); CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
update_for(mgr).events = 0; if (std::this_thread::get_id() == tid_) {
} do_start(mgr);
} else {
write_to_pipe(pollset_updater::code::start_manager, mgr.release());
}
}
bool multiplexer::is_reading(const socket_manager* mgr) const noexcept { void shutdown() override {
return (active_mask_of(mgr) & input_mask) != 0; CAF_LOG_TRACE("");
} // Note: there is no 'shortcut' when calling the function in the
// default_multiplexer's thread, because do_shutdown calls apply_updates.
// This must only be called from the pollset_updater.
CAF_LOG_DEBUG("push shutdown event to pipe");
write_to_pipe(pollset_updater::code::shutdown,
static_cast<socket_manager*>(nullptr));
}
bool multiplexer::is_writing(const socket_manager* mgr) const noexcept { // -- callbacks for socket managers ------------------------------------------
return (active_mask_of(mgr) & output_mask) != 0;
}
// -- control flow ------------------------------------------------------------- void register_reading(socket_manager* mgr) override {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
update_for(mgr).events |= input_mask;
}
bool multiplexer::poll_once(bool blocking) { void register_writing(socket_manager* mgr) override {
CAF_LOG_TRACE(CAF_ARG(blocking)); CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
if (pollset_.empty()) update_for(mgr).events |= output_mask;
return false; }
// We'll call poll() until poll() succeeds or fails.
for (;;) { void deregister_reading(socket_manager* mgr) override {
int presult = CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
update_for(mgr).events &= ~input_mask;
}
void deregister_writing(socket_manager* mgr) override {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
update_for(mgr).events &= ~output_mask;
}
void deregister(socket_manager* mgr) override {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
update_for(mgr).events = 0;
}
bool is_reading(const socket_manager* mgr) const noexcept override {
return (active_mask_of(mgr) & input_mask) != 0;
}
bool is_writing(const socket_manager* mgr) const noexcept override {
return (active_mask_of(mgr) & output_mask) != 0;
}
// -- control flow -----------------------------------------------------------
bool poll_once(bool blocking) override {
CAF_LOG_TRACE(CAF_ARG(blocking));
if (pollset_.empty())
return false;
// We'll call poll() until poll() succeeds or fails.
for (;;) {
int presult =
#ifdef CAF_WINDOWS #ifdef CAF_WINDOWS
::WSAPoll(pollset_.data(), static_cast<ULONG>(pollset_.size()), ::WSAPoll(pollset_.data(), static_cast<ULONG>(pollset_.size()),
blocking ? -1 : 0); blocking ? -1 : 0);
#else #else
::poll(pollset_.data(), static_cast<nfds_t>(pollset_.size()), ::poll(pollset_.data(), static_cast<nfds_t>(pollset_.size()),
blocking ? -1 : 0); blocking ? -1 : 0);
#endif #endif
if (presult > 0) { if (presult > 0) {
CAF_LOG_DEBUG("poll() on" << pollset_.size() << "sockets reported" CAF_LOG_DEBUG("poll() on" << pollset_.size() << "sockets reported"
<< presult << "event(s)"); << presult << "event(s)");
// Scan pollset for events. // Scan pollset for events.
if (auto revents = pollset_[0].revents; revents != 0) { if (auto revents = pollset_[0].revents; revents != 0) {
// Index 0 is always the pollset updater. This is the only handler that // Index 0 is always the pollset updater. This is the only handler
// is allowed to modify pollset_ and managers_. Since this may very well // that is allowed to modify pollset_ and managers_. Since this may
// mess with the for loop below, we process this handler first. // very well mess with the for loop below, we process this handler
auto mgr = managers_[0]; // first.
handle(mgr, pollset_[0].events, revents); auto mgr = managers_[0];
--presult; handle(mgr, pollset_[0].events, revents);
}
apply_updates();
for (size_t i = 1; i < pollset_.size() && presult > 0; ++i) {
if (auto revents = pollset_[i].revents; revents != 0) {
handle(managers_[i], pollset_[i].events, revents);
--presult; --presult;
} }
} apply_updates();
apply_updates(); for (size_t i = 1; i < pollset_.size() && presult > 0; ++i) {
return true; if (auto revents = pollset_[i].revents; revents != 0) {
} else if (presult == 0) { handle(managers_[i], pollset_[i].events, revents);
// No activity. --presult;
return false; }
} else {
auto code = last_socket_error();
switch (code) {
case std::errc::interrupted: {
// A signal was caught. Simply try again.
CAF_LOG_DEBUG("received errc::interrupted, try again");
break;
}
case std::errc::not_enough_memory: {
CAF_LOG_ERROR("poll() failed due to insufficient memory");
// There's not much we can do other than try again in hope someone
// else releases memory.
break;
} }
default: { apply_updates();
// Must not happen. return true;
auto int_code = static_cast<int>(code); } else if (presult == 0) {
auto msg = std::generic_category().message(int_code); // No activity.
std::string_view prefix = "poll() failed: "; return false;
msg.insert(msg.begin(), prefix.begin(), prefix.end()); } else {
CAF_CRITICAL(msg.c_str()); auto code = last_socket_error();
switch (code) {
case std::errc::interrupted: {
// A signal was caught. Simply try again.
CAF_LOG_DEBUG("received errc::interrupted, try again");
break;
}
case std::errc::not_enough_memory: {
CAF_LOG_ERROR("poll() failed due to insufficient memory");
// There's not much we can do other than try again in hope someone
// else releases memory.
break;
}
default: {
// Must not happen.
auto int_code = static_cast<int>(code);
auto msg = std::generic_category().message(int_code);
std::string_view prefix = "poll() failed: ";
msg.insert(msg.begin(), prefix.begin(), prefix.end());
CAF_CRITICAL(msg.c_str());
}
} }
} }
} }
} }
}
void multiplexer::poll() { void apply_updates() override {
while (poll_once(false)) CAF_LOG_DEBUG("apply" << updates_.size() << "updates");
; // repeat for (;;) {
} if (!updates_.empty()) {
for (auto& [fd, update] : updates_) {
void multiplexer::apply_updates() { if (auto index = index_of(fd); index == -1) {
CAF_LOG_DEBUG("apply" << updates_.size() << "updates"); if (update.events != 0) {
for (;;) { pollfd new_entry{socket_cast<socket_id>(fd), update.events, 0};
if (!updates_.empty()) { pollset_.emplace_back(new_entry);
for (auto& [fd, update] : updates_) { managers_.emplace_back(std::move(update.mgr));
if (auto index = index_of(fd); index == -1) { }
if (update.events != 0) { } else if (update.events != 0) {
pollfd new_entry{socket_cast<socket_id>(fd), update.events, 0}; pollset_[index].events = update.events;
pollset_.emplace_back(new_entry); managers_[index].swap(update.mgr);
managers_.emplace_back(std::move(update.mgr)); } else {
pollset_.erase(pollset_.begin() + index);
managers_.erase(managers_.begin() + index);
} }
} else if (update.events != 0) {
pollset_[index].events = update.events;
managers_[index].swap(update.mgr);
} else {
pollset_.erase(pollset_.begin() + index);
managers_.erase(managers_.begin() + index);
} }
updates_.clear();
} }
updates_.clear(); while (!pending_actions.empty()) {
auto next = std::move(pending_actions.front());
pending_actions.pop_front();
next.run();
}
if (updates_.empty())
return;
} }
while (!pending_actions_.empty()) { }
auto next = std::move(pending_actions_.front());
pending_actions_.pop_front(); void set_thread_id() override {
next.run(); CAF_LOG_TRACE("");
tid_ = std::this_thread::get_id();
}
void run() override {
CAF_LOG_TRACE("");
CAF_LOG_DEBUG("run default_multiplexer" << CAF_ARG(input_mask)
<< CAF_ARG(error_mask)
<< CAF_ARG(output_mask));
// On systems like Linux, we cannot disable sigpipe on the socket alone. We
// need to block the signal at thread level since some APIs (such as
// OpenSSL) are unsafe to call otherwise.
block_sigpipe();
while (!shutting_down_ || pollset_.size() > 1 || !watched_.empty()) {
poll_once(true);
disposable::erase_disposed(watched_);
}
// Close the pipe to block any future event.
std::lock_guard<std::mutex> guard{write_lock_};
if (write_handle_ != invalid_socket) {
close(write_handle_);
write_handle_ = pipe_socket{};
} }
if (updates_.empty())
return;
} }
}
void multiplexer::set_thread_id() { // -- internal callbacks the pollset updater ---------------------------------
CAF_LOG_TRACE("");
tid_ = std::this_thread::get_id(); void do_shutdown() {
} // Note: calling apply_updates here is only safe because we know that the
// pollset updater runs outside of the for-loop in run_once.
CAF_LOG_DEBUG("initiate shutdown");
shutting_down_ = true;
apply_updates();
// Skip the first manager (the pollset updater).
for (size_t i = 1; i < managers_.size(); ++i)
managers_[i]->dispose();
apply_updates();
}
void multiplexer::run() { void do_start(const socket_manager_ptr& mgr) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
CAF_LOG_DEBUG("run multiplexer" << CAF_ARG(input_mask) << CAF_ARG(error_mask) if (!shutting_down_) {
<< CAF_ARG(output_mask)); error err;
// On systems like Linux, we cannot disable sigpipe on the socket alone. We err = mgr->start();
// need to block the signal at thread level since some APIs (such as OpenSSL) if (err) {
// are unsafe to call otherwise. CAF_LOG_DEBUG("mgr->init failed: " << err);
block_sigpipe(); // The socket manager should not register itself for any events if
while (!shutting_down_ || pollset_.size() > 1 || !watched_.empty()) { // initialization fails. Purge any state just in case.
poll_once(true); update_for(mgr.get()).events = 0;
disposable::erase_disposed(watched_); }
} }
// Close the pipe to block any future event.
std::lock_guard<std::mutex> guard{write_lock_};
if (write_handle_ != invalid_socket) {
close(write_handle_);
write_handle_ = pipe_socket{};
} }
}
// -- utility functions -------------------------------------------------------- // -- utility functions ------------------------------------------------------
void multiplexer::handle(const socket_manager_ptr& mgr, /// Returns the index of `mgr` in the pollset or `-1`.
[[maybe_unused]] short events, short revents) { ptrdiff_t index_of(const socket_manager_ptr& mgr) const noexcept {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id) auto first = managers_.begin();
<< CAF_ARG(events) << CAF_ARG(revents)); auto last = managers_.end();
CAF_ASSERT(mgr != nullptr); auto i = std::find(first, last, mgr);
bool checkerror = true; return i == last ? -1 : std::distance(first, i);
CAF_LOG_DEBUG("handle event on socket" << mgr->handle().id << CAF_ARG(events)
<< CAF_ARG(revents));
// Note: we double-check whether the manager is actually reading because a
// previous action from the pipe may have disabled reading.
if ((revents & input_mask) != 0 && is_reading(mgr.get())) {
checkerror = false;
mgr->handle_read_event();
}
// Similar reasoning than before: double-check whether this event should still
// get dispatched.
if ((revents & output_mask) != 0 && is_writing(mgr.get())) {
checkerror = false;
mgr->handle_write_event();
}
if (checkerror && ((revents & error_mask) != 0)) {
if (revents & POLLNVAL)
mgr->handle_error(sec::socket_invalid);
else if (revents & POLLHUP)
mgr->handle_error(sec::socket_disconnected);
else
mgr->handle_error(sec::socket_operation_failed);
update_for(mgr.get()).events = 0;
} }
}
multiplexer::poll_update& multiplexer::update_for(ptrdiff_t index) { /// Returns the index of `fd` in the pollset or `-1`.
auto fd = socket{pollset_[index].fd}; ptrdiff_t index_of(socket fd) const noexcept {
if (auto i = updates_.find(fd); i != updates_.end()) { auto first = pollset_.begin();
return i->second; auto last = pollset_.end();
} else { auto i = std::find_if(first, last,
updates_.container().emplace_back(fd, poll_update{pollset_[index].events, [fd](const pollfd& x) { return x.fd == fd.id; });
managers_[index]}); return i == last ? -1 : std::distance(first, i);
return updates_.container().back().second; }
/// Handles an I/O event on given manager.
void handle(const socket_manager_ptr& mgr, short events, short revents) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id)
<< CAF_ARG(events) << CAF_ARG(revents));
CAF_ASSERT(mgr != nullptr);
bool checkerror = true;
CAF_LOG_DEBUG("handle event on socket"
<< mgr->handle().id << CAF_ARG(events) << CAF_ARG(revents));
// Note: we double-check whether the manager is actually reading because a
// previous action from the pipe may have disabled reading.
if ((revents & input_mask) != 0 && is_reading(mgr.get())) {
checkerror = false;
mgr->handle_read_event();
}
// Similar reasoning than before: double-check whether this event should
// still get dispatched.
if ((revents & output_mask) != 0 && is_writing(mgr.get())) {
checkerror = false;
mgr->handle_write_event();
}
if (checkerror && ((revents & error_mask) != 0)) {
if (revents & POLLNVAL)
mgr->handle_error(sec::socket_invalid);
else if (revents & POLLHUP)
mgr->handle_error(sec::socket_disconnected);
else
mgr->handle_error(sec::socket_operation_failed);
update_for(mgr.get()).events = 0;
}
}
/// Returns a change entry for the socket at given index. Lazily creates a new
/// entry before returning if necessary.
poll_update& update_for(ptrdiff_t index) {
auto fd = socket{pollset_[index].fd};
if (auto i = updates_.find(fd); i != updates_.end()) {
return i->second;
} else {
updates_.container().emplace_back(fd, poll_update{pollset_[index].events,
managers_[index]});
return updates_.container().back().second;
}
}
/// Returns a change entry for the socket of the manager.
poll_update& update_for(socket_manager* mgr) {
auto fd = mgr->handle();
if (auto i = updates_.find(fd); i != updates_.end()) {
return i->second;
} else if (auto index = index_of(fd); index != -1) {
updates_.container().emplace_back(
fd, poll_update{pollset_[index].events, socket_manager_ptr{mgr}});
return updates_.container().back().second;
} else {
updates_.container().emplace_back(fd, poll_update{0, mgr});
return updates_.container().back().second;
}
}
/// Writes `opcode` and pointer to `mgr` the the pipe for handling an event
/// later via the pollset updater.
/// @warning assumes ownership of @p ptr.
template <class T>
void write_to_pipe(uint8_t opcode, T* ptr) {
pollset_updater::msg_buf buf;
// Note: no intrusive_ptr_add_ref(ptr) since we take ownership of `ptr`.
buf[0] = static_cast<std::byte>(opcode);
auto value = reinterpret_cast<intptr_t>(ptr);
memcpy(buf.data() + 1, &value, sizeof(intptr_t));
ptrdiff_t res = -1;
{ // Lifetime scope of guard.
std::lock_guard<std::mutex> guard{write_lock_};
if (write_handle_ != invalid_socket)
res = write(write_handle_, buf);
}
if (res <= 0 && ptr)
intrusive_ptr_release(ptr);
} }
}
multiplexer::poll_update& multiplexer::update_for(socket_manager* mgr) { /// @copydoc write_to_pipe
auto fd = mgr->handle(); template <class Enum, class T>
if (auto i = updates_.find(fd); i != updates_.end()) { std::enable_if_t<std::is_enum_v<Enum>> write_to_pipe(Enum opcode, T* ptr) {
return i->second; write_to_pipe(static_cast<uint8_t>(opcode), ptr);
} else if (auto index = index_of(fd); index != -1) {
updates_.container().emplace_back(fd, poll_update{pollset_[index].events,
socket_manager_ptr{mgr}});
return updates_.container().back().second;
} else {
updates_.container().emplace_back(fd, poll_update{0, mgr});
return updates_.container().back().second;
} }
/// Queries the currently active event bitmask for `mgr`.
short active_mask_of(const socket_manager* mgr) const noexcept {
auto fd = mgr->handle();
if (auto i = updates_.find(fd); i != updates_.end()) {
return i->second.events;
} else if (auto index = index_of(fd); index != -1) {
return pollset_[index].events;
} else {
return 0;
}
}
/// Pending actions via `schedule`.
std::deque<action> pending_actions;
private:
// -- member variables -------------------------------------------------------
/// Bookkeeping data for managed sockets.
pollfd_list pollset_;
/// Maps sockets to their owning managers by storing the managers in the same
/// order as their sockets appear in `pollset_`.
manager_list managers_;
/// Caches changes to the events mask of managed sockets until they can safely
/// take place.
poll_update_map updates_;
/// Stores the ID of the thread this multiplexer is running in. Set when
/// calling `init()`.
std::thread::id tid_;
/// Guards `write_handle_`.
std::mutex write_lock_;
/// Used for pushing updates to the multiplexer's thread.
pipe_socket write_handle_;
/// Points to the owning middleman.
middleman* owner_;
/// Signals whether shutdown has been requested.
bool shutting_down_ = false;
/// Keeps track of watched disposables.
std::vector<disposable> watched_;
};
error pollset_updater::start(socket_manager* owner) {
CAF_LOG_TRACE("");
owner_ = owner;
mpx_ = static_cast<default_multiplexer*>(owner->mpx_ptr());
return nonblocking(fd_, true);
} }
template <class T> void pollset_updater::handle_read_event() {
void multiplexer::write_to_pipe(uint8_t opcode, T* ptr) { CAF_LOG_TRACE("");
detail::pollset_updater::msg_buf buf; auto as_mgr = [](intptr_t ptr) {
// Note: no intrusive_ptr_add_ref(ptr) since we take ownership of `ptr`. return intrusive_ptr{reinterpret_cast<socket_manager*>(ptr), false};
buf[0] = static_cast<std::byte>(opcode); };
auto value = reinterpret_cast<intptr_t>(ptr); auto add_action = [this](intptr_t ptr) {
memcpy(buf.data() + 1, &value, sizeof(intptr_t)); auto f = action{intrusive_ptr{reinterpret_cast<action::impl*>(ptr), false}};
ptrdiff_t res = -1; mpx_->pending_actions.push_back(std::move(f));
{ // Lifetime scope of guard. };
std::lock_guard<std::mutex> guard{write_lock_}; for (;;) {
if (write_handle_ != invalid_socket) CAF_ASSERT((buf_.size() - buf_size_) > 0);
res = write(write_handle_, buf); auto num_bytes
= read(fd_, make_span(buf_.data() + buf_size_, buf_.size() - buf_size_));
if (num_bytes > 0) {
buf_size_ += static_cast<size_t>(num_bytes);
if (buf_.size() == buf_size_) {
buf_size_ = 0;
auto opcode = static_cast<uint8_t>(buf_[0]);
intptr_t ptr;
memcpy(&ptr, buf_.data() + 1, sizeof(intptr_t));
switch (static_cast<code>(opcode)) {
case code::start_manager:
mpx_->do_start(as_mgr(ptr));
break;
case code::run_action:
add_action(ptr);
break;
case code::shutdown:
CAF_ASSERT(ptr == 0);
mpx_->do_shutdown();
break;
default:
CAF_LOG_ERROR("opcode not recognized: " << CAF_ARG(opcode));
break;
}
}
} else if (num_bytes == 0) {
CAF_LOG_DEBUG("pipe closed, assume shutdown");
owner_->deregister();
return;
} else if (last_socket_error_is_temporary()) {
return;
} else {
CAF_LOG_ERROR("pollset updater failed to read from the pipe");
owner_->deregister();
return;
}
} }
if (res <= 0 && ptr)
intrusive_ptr_release(ptr);
} }
short multiplexer::active_mask_of(const socket_manager* mgr) const noexcept { } // namespace
auto fd = mgr->handle();
if (auto i = updates_.find(fd); i != updates_.end()) { // -- multiplexer implementation -----------------------------------------------
return i->second.events;
} else if (auto index = index_of(fd); index != -1) { // -- static utility functions -------------------------------------------------
return pollset_[index].events;
} else { void multiplexer::block_sigpipe() {
return 0; #ifdef CAF_LINUX
sigset_t sigpipe_mask;
sigemptyset(&sigpipe_mask);
sigaddset(&sigpipe_mask, SIGPIPE);
sigset_t saved_mask;
if (pthread_sigmask(SIG_BLOCK, &sigpipe_mask, &saved_mask) == -1) {
perror("pthread_sigmask");
exit(1);
} }
#endif
} }
// -- internal callbacks the pollset updater ----------------------------------- multiplexer_ptr multiplexer::make(middleman* parent) {
return make_counted<default_multiplexer>(parent);
void multiplexer::do_shutdown() {
// Note: calling apply_updates here is only safe because we know that the
// pollset updater runs outside of the for-loop in run_once.
CAF_LOG_DEBUG("initiate shutdown");
shutting_down_ = true;
apply_updates();
// Skip the first manager (the pollset updater).
for (size_t i = 1; i < managers_.size(); ++i)
managers_[i]->dispose();
apply_updates();
} }
void multiplexer::do_start(const socket_manager_ptr& mgr) { multiplexer* multiplexer::from(actor_system& sys) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id)); return sys.network_manager().mpx_ptr();
if (!shutting_down_) { }
error err;
err = mgr->start(); // -- constructors, destructors, and assignment operators ----------------------
if (err) {
CAF_LOG_DEBUG("mgr->init failed: " << err); multiplexer::~multiplexer() {
// The socket manager should not register itself for any events if // nop
// initialization fails. Purge any state just in case.
update_for(mgr.get()).events = 0;
}
}
} }
} // namespace caf::net } // namespace caf::net
...@@ -5,54 +5,25 @@ ...@@ -5,54 +5,25 @@
#pragma once #pragma once
#include "caf/net/fwd.hpp" #include "caf/net/fwd.hpp"
#include "caf/net/pipe_socket.hpp"
#include "caf/net/socket.hpp" #include "caf/net/socket.hpp"
#include "caf/action.hpp"
#include "caf/async/execution_context.hpp" #include "caf/async/execution_context.hpp"
#include "caf/detail/atomic_ref_counted.hpp" #include "caf/detail/atomic_ref_counted.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/ref_counted.hpp"
#include "caf/unordered_flat_map.hpp"
#include <deque> #include <deque>
#include <memory> #include <memory>
#include <mutex> #include <mutex>
#include <thread> #include <thread>
extern "C" {
struct pollfd;
} // extern "C"
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 : public detail::atomic_ref_counted, class CAF_NET_EXPORT multiplexer : public async::execution_context {
public async::execution_context {
public: public:
// -- member types -----------------------------------------------------------
struct poll_update {
short events = 0;
socket_manager_ptr mgr;
};
using poll_update_map = unordered_flat_map<socket, poll_update>;
using pollfd_list = std::vector<pollfd>;
using manager_list = std::vector<socket_manager_ptr>;
// -- friends ----------------------------------------------------------------
friend class detail::pollset_updater; // Needs access to the `do_*` functions.
// -- static utility functions ----------------------------------------------- // -- static utility functions -----------------------------------------------
/// Blocks the PIPE signal on the current thread when running on a POSIX /// Blocks the PIPE signal on the current thread when running on Linux. Has no
/// windows. Has no effect when running on Windows. /// effect otherwise.
static void block_sigpipe(); static void block_sigpipe();
/// Returns a pointer to the multiplexer from the actor system. /// Returns a pointer to the multiplexer from the actor system.
...@@ -60,179 +31,79 @@ public: ...@@ -60,179 +31,79 @@ public:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
~multiplexer(); ~multiplexer() override;
// -- factories -------------------------------------------------------------- // -- factories --------------------------------------------------------------
/// Creates a new multiplexer instance with the default implementation.
/// @param parent Points to the owning middleman instance. May be `nullptr` /// @param parent Points to the owning middleman instance. May be `nullptr`
/// only for the purpose of unit testing if no @ref /// only for the purpose of unit testing if no @ref
/// socket_manager requires access to the @ref middleman or the /// socket_manager requires access to the @ref middleman or the
/// @ref actor_system. /// @ref actor_system.
static multiplexer_ptr make(middleman* parent) { static multiplexer_ptr make(middleman* parent);
auto ptr = new multiplexer(parent);
return multiplexer_ptr{ptr, false};
}
// -- initialization --------------------------------------------------------- // -- initialization ---------------------------------------------------------
error init(); virtual error init() = 0;
// -- properties ------------------------------------------------------------- // -- properties -------------------------------------------------------------
/// Returns the number of currently active socket managers. /// Returns the number of currently active socket managers.
size_t num_socket_managers() const noexcept; virtual size_t num_socket_managers() const noexcept = 0;
/// Returns the index of `mgr` in the pollset or `-1`.
ptrdiff_t index_of(const socket_manager_ptr& mgr) const noexcept;
/// Returns the index of `fd` in the pollset or `-1`.
ptrdiff_t index_of(socket fd) const noexcept;
/// Returns the owning @ref middleman instance. /// Returns the owning @ref middleman instance.
middleman& owner(); virtual middleman& owner() = 0;
/// Returns the enclosing @ref actor_system. /// Returns the enclosing @ref actor_system.
actor_system& system(); virtual actor_system& system() = 0;
// -- implementation of execution_context ------------------------------------
void ref_execution_context() const noexcept override;
void deref_execution_context() const noexcept override;
void schedule(action what) override;
void watch(disposable what) override;
// -- thread-safe signaling -------------------------------------------------- // -- thread-safe signaling --------------------------------------------------
/// Registers `mgr` for initialization in the multiplexer's thread. /// Registers `mgr` for initialization in the multiplexer's thread.
/// @thread-safe /// @thread-safe
void start(socket_manager_ptr mgr); virtual void start(socket_manager_ptr mgr) = 0;
/// Signals the multiplexer to initiate shutdown. /// Signals the multiplexer to initiate shutdown.
/// @thread-safe /// @thread-safe
void shutdown(); virtual void shutdown() = 0;
// -- callbacks for socket managers ------------------------------------------ // -- callbacks for socket managers ------------------------------------------
/// Registers `mgr` for read events. /// Registers `mgr` for read events.
void register_reading(socket_manager* mgr); virtual void register_reading(socket_manager* mgr) = 0;
/// Registers `mgr` for write events. /// Registers `mgr` for write events.
void register_writing(socket_manager* mgr); virtual void register_writing(socket_manager* mgr) = 0;
/// Deregisters `mgr` from read events. /// Deregisters `mgr` from read events.
void deregister_reading(socket_manager* mgr); virtual void deregister_reading(socket_manager* mgr) = 0;
/// Deregisters `mgr` from write events. /// Deregisters `mgr` from write events.
void deregister_writing(socket_manager* mgr); virtual void deregister_writing(socket_manager* mgr) = 0;
/// Deregisters @p mgr from read and write events. /// Deregisters @p mgr from read and write events.
void deregister(socket_manager* mgr); virtual void deregister(socket_manager* mgr) = 0;
/// Queries whether `mgr` is currently registered for reading. /// Queries whether `mgr` is currently registered for reading.
bool is_reading(const socket_manager* mgr) const noexcept; virtual bool is_reading(const socket_manager* mgr) const noexcept = 0;
/// Queries whether `mgr` is currently registered for writing. /// Queries whether `mgr` is currently registered for writing.
bool is_writing(const socket_manager* mgr) const noexcept; virtual bool is_writing(const socket_manager* mgr) const noexcept = 0;
// -- control flow ----------------------------------------------------------- // -- control flow -----------------------------------------------------------
/// Polls I/O activity once and runs all socket event handlers that become /// Polls I/O activity once and runs all socket event handlers that become
/// ready as a result. /// ready as a result.
bool poll_once(bool blocking); virtual bool poll_once(bool blocking) = 0;
/// Runs `poll_once(false)` until it returns `true`.`
void poll();
/// Applies all pending updates. /// Applies all pending updates.
void apply_updates(); virtual void apply_updates() = 0;
/// Runs all pending actions.
void run_actions();
/// Sets the thread ID to `std::this_thread::id()`. /// Sets the thread ID to `std::this_thread::id()`.
void set_thread_id(); virtual void set_thread_id() = 0;
/// Runs the multiplexer until no socket event handler remains active. /// Runs the multiplexer until no socket event handler remains active.
void run(); virtual void run() = 0;
protected:
// -- utility functions ------------------------------------------------------
/// Handles an I/O event on given manager.
void handle(const socket_manager_ptr& mgr, short events, short revents);
/// Returns a change entry for the socket at given index. Lazily creates a new
/// entry before returning if necessary.
poll_update& update_for(ptrdiff_t index);
/// Returns a change entry for the socket of the manager.
poll_update& update_for(socket_manager* mgr);
/// Writes `opcode` and pointer to `mgr` the the pipe for handling an event
/// later via the pollset updater.
/// @warning assumes ownership of @p ptr.
template <class T>
void write_to_pipe(uint8_t opcode, T* ptr);
/// @copydoc write_to_pipe
template <class Enum, class T>
std::enable_if_t<std::is_enum_v<Enum>> write_to_pipe(Enum opcode, T* ptr) {
write_to_pipe(static_cast<uint8_t>(opcode), ptr);
}
/// Queries the currently active event bitmask for `mgr`.
short active_mask_of(const socket_manager* mgr) const noexcept;
// -- member variables -------------------------------------------------------
/// Bookkeeping data for managed sockets.
pollfd_list pollset_;
/// Maps sockets to their owning managers by storing the managers in the same
/// order as their sockets appear in `pollset_`.
manager_list managers_;
/// Caches changes to the events mask of managed sockets until they can safely
/// take place.
poll_update_map updates_;
/// Stores the ID of the thread this multiplexer is running in. Set when
/// calling `init()`.
std::thread::id tid_;
/// Guards `write_handle_`.
std::mutex write_lock_;
/// Used for pushing updates to the multiplexer's thread.
pipe_socket write_handle_;
/// Points to the owning middleman.
middleman* owner_;
/// Signals whether shutdown has been requested.
bool shutting_down_ = false;
/// Pending actions via `schedule`.
std::deque<action> pending_actions_;
/// Keeps track of watched disposables.
std::vector<disposable> watched_;
private:
// -- constructors, destructors, and assignment operators --------------------
explicit multiplexer(middleman* parent);
// -- internal callbacks the pollset updater ---------------------------------
void do_shutdown();
void do_register_reading(const socket_manager_ptr& mgr);
void do_start(const socket_manager_ptr& mgr);
}; };
} // 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