Commit 9d21e4ec authored by Dominik Charousset's avatar Dominik Charousset

Move pollset_updater to caf::detail

The pollset_updater is an implementation detail of the multiplexer and
thus should not be part of the public API.
parent f411f91d
...@@ -29,6 +29,7 @@ caf_add_component( ...@@ -29,6 +29,7 @@ caf_add_component(
${CAF_NET_HEADERS} ${CAF_NET_HEADERS}
SOURCES SOURCES
src/detail/convert_ip_endpoint.cpp src/detail/convert_ip_endpoint.cpp
src/detail/pollset_updater.cpp
src/detail/rfc6455.cpp src/detail/rfc6455.cpp
src/net/abstract_actor_shell.cpp src/net/abstract_actor_shell.cpp
src/net/actor_shell.cpp src/net/actor_shell.cpp
...@@ -60,7 +61,6 @@ caf_add_component( ...@@ -60,7 +61,6 @@ caf_add_component(
src/net/octet_stream/transport.cpp src/net/octet_stream/transport.cpp
src/net/octet_stream/upper_layer.cpp src/net/octet_stream/upper_layer.cpp
src/net/pipe_socket.cpp src/net/pipe_socket.cpp
src/net/pollset_updater.cpp
src/net/prometheus/server.cpp src/net/prometheus/server.cpp
src/net/socket.cpp src/net/socket.cpp
src/net/socket_event_layer.cpp src/net/socket_event_layer.cpp
......
...@@ -14,13 +14,13 @@ ...@@ -14,13 +14,13 @@
#include <cstdint> #include <cstdint>
#include <mutex> #include <mutex>
namespace caf::net { namespace caf::detail {
class CAF_NET_EXPORT pollset_updater : public socket_event_layer { class CAF_NET_EXPORT pollset_updater : public net::socket_event_layer {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
using super = socket_manager; using super = net::socket_manager;
using msg_buf = std::array<std::byte, sizeof(intptr_t) + 1>; using msg_buf = std::array<std::byte, sizeof(intptr_t) + 1>;
...@@ -34,17 +34,17 @@ public: ...@@ -34,17 +34,17 @@ public:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
explicit pollset_updater(pipe_socket fd); explicit pollset_updater(net::pipe_socket fd);
// -- factories -------------------------------------------------------------- // -- factories --------------------------------------------------------------
static std::unique_ptr<pollset_updater> make(pipe_socket fd); static std::unique_ptr<pollset_updater> make(net::pipe_socket fd);
// -- implementation of socket_event_layer ----------------------------------- // -- implementation of socket_event_layer -----------------------------------
error start(socket_manager* owner) override; error start(net::socket_manager* owner) override;
socket handle() const override; net::socket handle() const override;
void handle_read_event() override; void handle_read_event() override;
...@@ -53,11 +53,11 @@ public: ...@@ -53,11 +53,11 @@ public:
void abort(const error& reason) override; void abort(const error& reason) override;
private: private:
pipe_socket fd_; net::pipe_socket fd_;
socket_manager* owner_ = nullptr; net::socket_manager* owner_ = nullptr;
multiplexer* mpx_ = nullptr; net::multiplexer* mpx_ = nullptr;
msg_buf buf_; msg_buf buf_;
size_t buf_size_ = 0; size_t buf_size_ = 0;
}; };
} // namespace caf::net } // namespace caf::detail
...@@ -89,3 +89,9 @@ class upper_layer; ...@@ -89,3 +89,9 @@ class upper_layer;
enum class errc; enum class errc;
} // namespace caf::net::octet_stream } // namespace caf::net::octet_stream
namespace caf::detail {
class pollset_updater;
} // namespace caf::detail
...@@ -27,8 +27,6 @@ struct pollfd; ...@@ -27,8 +27,6 @@ struct pollfd;
namespace caf::net { namespace caf::net {
class pollset_updater;
/// 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 detail::atomic_ref_counted,
public async::execution_context { public async::execution_context {
...@@ -48,7 +46,7 @@ public: ...@@ -48,7 +46,7 @@ public:
// -- friends ---------------------------------------------------------------- // -- friends ----------------------------------------------------------------
friend class pollset_updater; // Needs access to the `do_*` functions. friend class detail::pollset_updater; // Needs access to the `do_*` functions.
// -- static utility functions ----------------------------------------------- // -- static utility functions -----------------------------------------------
......
...@@ -2,7 +2,7 @@ ...@@ -2,7 +2,7 @@
// the main distribution directory for license terms and copyright or visit // the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE. // https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#include "caf/net/pollset_updater.hpp" #include "caf/detail/pollset_updater.hpp"
#include <cstring> #include <cstring>
...@@ -13,37 +13,37 @@ ...@@ -13,37 +13,37 @@
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/span.hpp" #include "caf/span.hpp"
namespace caf::net { namespace caf::detail {
// -- constructors, destructors, and assignment operators ---------------------- // -- constructors, destructors, and assignment operators ----------------------
pollset_updater::pollset_updater(pipe_socket fd) : fd_(fd) { pollset_updater::pollset_updater(net::pipe_socket fd) : fd_(fd) {
// nop // nop
} }
// -- factories ---------------------------------------------------------------- // -- factories ----------------------------------------------------------------
std::unique_ptr<pollset_updater> pollset_updater::make(pipe_socket fd) { std::unique_ptr<pollset_updater> pollset_updater::make(net::pipe_socket fd) {
return std::make_unique<pollset_updater>(fd); return std::make_unique<pollset_updater>(fd);
} }
// -- interface functions ------------------------------------------------------ // -- interface functions ------------------------------------------------------
error pollset_updater::start(socket_manager* owner) { error pollset_updater::start(net::socket_manager* owner) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
owner_ = owner; owner_ = owner;
mpx_ = owner->mpx_ptr(); mpx_ = owner->mpx_ptr();
return nonblocking(fd_, true); return net::nonblocking(fd_, true);
} }
socket pollset_updater::handle() const { net::socket pollset_updater::handle() const {
return fd_; return fd_;
} }
void pollset_updater::handle_read_event() { void pollset_updater::handle_read_event() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
auto as_mgr = [](intptr_t ptr) { auto as_mgr = [](intptr_t ptr) {
return intrusive_ptr{reinterpret_cast<socket_manager*>(ptr), false}; return intrusive_ptr{reinterpret_cast<net::socket_manager*>(ptr), false};
}; };
auto add_action = [this](intptr_t ptr) { auto add_action = [this](intptr_t ptr) {
auto f = action{intrusive_ptr{reinterpret_cast<action::impl*>(ptr), false}}; auto f = action{intrusive_ptr{reinterpret_cast<action::impl*>(ptr), false}};
...@@ -80,7 +80,7 @@ void pollset_updater::handle_read_event() { ...@@ -80,7 +80,7 @@ void pollset_updater::handle_read_event() {
CAF_LOG_DEBUG("pipe closed, assume shutdown"); CAF_LOG_DEBUG("pipe closed, assume shutdown");
owner_->deregister(); owner_->deregister();
return; return;
} else if (last_socket_error_is_temporary()) { } else if (net::last_socket_error_is_temporary()) {
return; return;
} else { } else {
CAF_LOG_ERROR("pollset updater failed to read from the pipe"); CAF_LOG_ERROR("pollset updater failed to read from the pipe");
...@@ -98,4 +98,4 @@ void pollset_updater::abort(const error&) { ...@@ -98,4 +98,4 @@ void pollset_updater::abort(const error&) {
// nop // nop
} }
} // namespace caf::net } // namespace caf::detail
...@@ -8,12 +8,12 @@ ...@@ -8,12 +8,12 @@
#include "caf/actor_system.hpp" #include "caf/actor_system.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/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/net/middleman.hpp" #include "caf/net/middleman.hpp"
#include "caf/net/pollset_updater.hpp"
#include "caf/net/socket_manager.hpp" #include "caf/net/socket_manager.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/span.hpp" #include "caf/span.hpp"
...@@ -97,7 +97,7 @@ error multiplexer::init() { ...@@ -97,7 +97,7 @@ 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());
auto updater = pollset_updater::make(pipe_handles->first); auto updater = detail::pollset_updater::make(pipe_handles->first);
auto mgr = socket_manager::make(this, std::move(updater)); auto mgr = socket_manager::make(this, std::move(updater));
if (auto err = mgr->start()) if (auto err = mgr->start())
return err; return err;
...@@ -153,7 +153,7 @@ void multiplexer::schedule(action what) { ...@@ -153,7 +153,7 @@ void multiplexer::schedule(action what) {
pending_actions_.push_back(what); pending_actions_.push_back(what);
} else { } else {
auto ptr = std::move(what).as_intrusive_ptr().release(); auto ptr = std::move(what).as_intrusive_ptr().release();
write_to_pipe(pollset_updater::code::run_action, ptr); write_to_pipe(detail::pollset_updater::code::run_action, ptr);
} }
} }
...@@ -168,7 +168,7 @@ void multiplexer::start(socket_manager_ptr mgr) { ...@@ -168,7 +168,7 @@ void multiplexer::start(socket_manager_ptr mgr) {
if (std::this_thread::get_id() == tid_) { if (std::this_thread::get_id() == tid_) {
do_start(mgr); do_start(mgr);
} else { } else {
write_to_pipe(pollset_updater::code::start_manager, mgr.release()); write_to_pipe(detail::pollset_updater::code::start_manager, mgr.release());
} }
} }
...@@ -178,7 +178,7 @@ void multiplexer::shutdown() { ...@@ -178,7 +178,7 @@ void multiplexer::shutdown() {
// thread, because do_shutdown calls apply_updates. This must only be called // thread, because do_shutdown calls apply_updates. This must only be called
// from the pollset_updater. // from the pollset_updater.
CAF_LOG_DEBUG("push shutdown event to pipe"); CAF_LOG_DEBUG("push shutdown event to pipe");
write_to_pipe(pollset_updater::code::shutdown, write_to_pipe(detail::pollset_updater::code::shutdown,
static_cast<socket_manager*>(nullptr)); static_cast<socket_manager*>(nullptr));
} }
...@@ -405,7 +405,7 @@ multiplexer::poll_update& multiplexer::update_for(socket_manager* mgr) { ...@@ -405,7 +405,7 @@ multiplexer::poll_update& multiplexer::update_for(socket_manager* mgr) {
template <class T> template <class T>
void multiplexer::write_to_pipe(uint8_t opcode, T* ptr) { void multiplexer::write_to_pipe(uint8_t opcode, T* ptr) {
pollset_updater::msg_buf buf; detail::pollset_updater::msg_buf buf;
// Note: no intrusive_ptr_add_ref(ptr) since we take ownership of `ptr`. // Note: no intrusive_ptr_add_ref(ptr) since we take ownership of `ptr`.
buf[0] = static_cast<std::byte>(opcode); buf[0] = static_cast<std::byte>(opcode);
auto value = reinterpret_cast<intptr_t>(ptr); auto value = reinterpret_cast<intptr_t>(ptr);
......
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