Commit 87cc725a authored by Dominik Charousset's avatar Dominik Charousset

Redesign multiplexer to allow exchanging scribes

Allow brokers to receive a `scribe_ptr` to a `scribe` created by some
other actor, e.g., the `middleman_actor`. This change is necessary for
integrating custom scribes with extended functionality. The motivating
use case for this change is the OpenSSL module.
parent 6b949592
......@@ -158,23 +158,30 @@ public:
return system().middleman();
}
/// Adds a `scribe` instance to this broker.
void add_scribe(const intrusive_ptr<scribe>& ptr);
/// Adds the unitialized `scribe` instance `ptr` to this broker.
void add_scribe(scribe_ptr ptr);
/// Creates and assigns a new `scribe` from given native socked `fd`.
connection_handle add_scribe(network::native_socket fd);
/// Tries to connect to `host` on given `port` and creates
/// a new scribe describing the connection afterwards.
/// @returns The handle of the new `scribe` on success.
expected<connection_handle> add_tcp_scribe(const std::string& hostname, uint16_t port);
expected<connection_handle> add_tcp_scribe(const std::string& host,
uint16_t port);
/// Assigns a detached `scribe` instance identified by `hdl`
/// from the `multiplexer` to this broker.
expected<void> assign_tcp_scribe(connection_handle hdl);
/// Moves the initialized `scribe` instance `ptr` from another broker to this
/// broker.
void move_scribe(scribe_ptr ptr);
/// Creates and assigns a new `scribe` from given native socked `fd`.
expected<connection_handle> add_tcp_scribe(network::native_socket fd);
/// Adds a `doorman` instance to this broker.
void add_doorman(doorman_ptr ptr);
/// Creates and assigns a new `doorman` from given native socked `fd`.
accept_handle add_doorman(network::native_socket fd);
/// Adds a `doorman` instance to this broker.
void add_doorman(const intrusive_ptr<doorman>& ptr);
void move_doorman(doorman_ptr ptr);
/// Tries to open a local port and creates a `doorman` managing
/// it on success. If `port == 0`, then the broker will ask
......@@ -184,13 +191,6 @@ public:
add_tcp_doorman(uint16_t port = 0, const char* in = nullptr,
bool reuse_addr = false);
/// Assigns a detached `doorman` instance identified by `hdl`
/// from the `multiplexer` to this broker.
expected<void> assign_tcp_doorman(accept_handle hdl);
/// Creates and assigns a new `doorman` from given native socked `fd`.
expected<accept_handle> add_tcp_doorman(network::native_socket fd);
/// Returns the remote address associated to `hdl`
/// or empty string if `hdl` is invalid.
std::string remote_addr(connection_handle hdl);
......@@ -316,6 +316,33 @@ protected:
}
private:
inline void launch_servant(scribe_ptr&) {
// nop
}
void launch_servant(doorman_ptr& ptr);
template <class T>
typename T::handle_type add_servant(intrusive_ptr<T>&& ptr) {
CAF_ASSERT(ptr != nullptr);
CAF_ASSERT(ptr->parent() == nullptr);
ptr->set_parent(this);
auto hdl = ptr->hdl();
launch_servant(ptr);
get_map(hdl).emplace(hdl, std::move(ptr));
return hdl;
}
template <class T>
void move_servant(intrusive_ptr<T>&& ptr) {
CAF_ASSERT(ptr != nullptr);
CAF_ASSERT(ptr->parent() != nullptr && ptr->parent() != this);
ptr->set_parent(this);
CAF_ASSERT(ptr->parent() == this);
auto hdl = ptr->hdl();
get_map(hdl).emplace(hdl, std::move(ptr));
}
scribe_map scribes_;
doorman_map doormen_;
detail::intrusive_partitioned_list<mailbox_element, detail::disposer> cache_;
......
......@@ -71,9 +71,7 @@ public:
cfg.init_fun = fac(std::move(fun), hdl, std::forward<Ts>(xs)...);
auto res = this->system().spawn_class<impl, no_spawn_options>(cfg);
auto forked = static_cast<impl*>(actor_cast<abstract_actor*>(res));
sptr->set_parent(forked);
CAF_ASSERT(sptr->parent() == forked);
forked->add_scribe(sptr);
forked->move_scribe(std::move(sptr));
return res;
}
......
......@@ -34,15 +34,16 @@ namespace io {
template <class Base, class Handle, class SysMsgType>
class broker_servant : public Base {
public:
broker_servant(abstract_broker* ptr, Handle x)
: Base(ptr),
hdl_(x),
using handle_type = Handle;
broker_servant(handle_type x)
: hdl_(x),
value_(strong_actor_ptr{}, message_id::make(),
mailbox_element::forwarding_stack{}, SysMsgType{x, {}}) {
// nop
}
Handle hdl() const {
handle_type hdl() const {
return hdl_;
}
......@@ -99,7 +100,7 @@ protected:
// producing, why we check the condition again afterwards
using passiv_t =
typename std::conditional<
std::is_same<Handle, connection_handle>::value,
std::is_same<handle_type, connection_handle>::value,
connection_passivated_msg,
acceptor_passivated_msg
>::type;
......@@ -116,7 +117,7 @@ protected:
return value_.template get_mutable_as<SysMsgType>(0);
}
Handle hdl_;
handle_type hdl_;
mailbox_element_vals<SysMsgType> value_;
optional<size_t> activity_tokens_;
};
......
......@@ -21,6 +21,7 @@
#define CAF_IO_DOORMAN_HPP
#include <cstddef>
#include <cstdint>
#include "caf/message.hpp"
#include "caf/mailbox_element.hpp"
......@@ -40,7 +41,7 @@ using doorman_base = broker_servant<network::acceptor_manager, accept_handle,
/// @ingroup Broker
class doorman : public doorman_base {
public:
doorman(abstract_broker* ptr, accept_handle acc_hdl);
doorman(accept_handle acc_hdl);
~doorman() override;
......@@ -50,14 +51,20 @@ public:
bool new_connection(execution_unit* ctx, connection_handle x);
// needs to be launched explicitly
/// Starts listening on the selected port.
virtual void launch() = 0;
protected:
message detach_message() override;
};
using doorman_ptr = intrusive_ptr<doorman>;
} // namespace io
} // namespace caf
// Allows the `middleman_actor` to create a `doorman` and then send it to the
// BASP broker.
CAF_ALLOW_UNSAFE_MESSAGE_TYPE(caf::io::doorman_ptr)
#endif // CAF_IO_DOORMAN_HPP
......@@ -21,8 +21,20 @@
#define CAF_IO_FWD_HPP
namespace caf {
// -- templates from the parent namespace necessary for defining aliases -------
template <class> class intrusive_ptr;
namespace io {
// -- variadic templates -------------------------------------------------------
template <class... Sigs>
class typed_broker;
// -- classes ------------------------------------------------------------------
class scribe;
class broker;
class doorman;
......@@ -31,8 +43,12 @@ class basp_broker;
class receive_policy;
class abstract_broker;
template <class... Sigs>
class typed_broker;
// -- aliases ------------------------------------------------------------------
using scribe_ptr = intrusive_ptr<scribe>;
using doorman_ptr = intrusive_ptr<doorman>;
// -- nested namespaces --------------------------------------------------------
namespace network {
......
......@@ -280,16 +280,17 @@ private:
template <spawn_options Os, class Impl, class F, class... Ts>
expected<typename infer_handle_from_class<Impl>::type>
spawn_client_impl(F fun, const std::string& host, uint16_t port, Ts&&... xs) {
auto ehdl = backend().new_tcp_scribe(host, port);
if (!ehdl)
return ehdl.error();
auto hdl = *ehdl;
auto eptr = backend().new_tcp_scribe(host, port);
if (!eptr)
return eptr.error();
auto ptr = std::move(*eptr);
CAF_ASSERT(ptr != nullptr);
detail::init_fun_factory<Impl, F> fac;
actor_config cfg{&backend()};
auto init_fun = fac(std::move(fun), hdl, std::forward<Ts>(xs)...);
cfg.init_fun = [hdl, init_fun](local_actor* ptr) -> behavior {
static_cast<abstract_broker*>(ptr)->assign_tcp_scribe(hdl);
return init_fun(ptr);
auto init_fun = fac(std::move(fun), ptr->hdl(), std::forward<Ts>(xs)...);
cfg.init_fun = [ptr, init_fun](local_actor* self) mutable -> behavior {
static_cast<abstract_broker*>(self)->add_scribe(std::move(ptr));
return init_fun(self);
};
return system().spawn_class<Impl, Os>(cfg);
}
......@@ -297,17 +298,17 @@ private:
template <spawn_options Os, class Impl, class F, class... Ts>
expected<typename infer_handle_from_class<Impl>::type>
spawn_server_impl(F fun, uint16_t& port, Ts&&... xs) {
auto eptr = backend().new_tcp_doorman(port);
if (!eptr)
return eptr.error();
auto ptr = std::move(*eptr);
detail::init_fun_factory<Impl, F> fac;
auto init_fun = fac(std::move(fun), std::forward<Ts>(xs)...);
auto ehdl = backend().new_tcp_doorman(port);
if (!ehdl)
return ehdl.error();
auto hdl = ehdl->first;
port = ehdl->second;
port = ptr->port();
actor_config cfg{&backend()};
cfg.init_fun = [hdl, init_fun](local_actor* ptr) -> behavior {
static_cast<abstract_broker*>(ptr)->assign_tcp_doorman(hdl);
return init_fun(ptr);
cfg.init_fun = [ptr, init_fun](local_actor* self) mutable -> behavior {
static_cast<abstract_broker*>(self)->add_doorman(std::move(ptr));
return init_fun(self);
};
return system().spawn_class<Impl, Os>(cfg);
}
......
......@@ -30,8 +30,6 @@ namespace network {
/// callbacks for incoming connections as well as for error handling.
class acceptor_manager : public manager {
public:
acceptor_manager(abstract_broker* ptr);
~acceptor_manager() override;
/// Called by the underlying I/O device to indicate that
......
......@@ -58,33 +58,19 @@ public:
friend class io::middleman;
friend class supervisor;
expected<connection_handle>
new_tcp_scribe(const std::string&, uint16_t) override;
scribe_ptr new_scribe(asio_tcp_socket&& sock);
expected<void>
assign_tcp_scribe(abstract_broker*, connection_handle hdl) override;
scribe_ptr new_scribe(native_socket fd) override;
template <class Socket>
connection_handle add_tcp_scribe(abstract_broker*, Socket&& sock);
expected<scribe_ptr> new_tcp_scribe(const std::string& host,
uint16_t port) override;
connection_handle add_tcp_scribe(abstract_broker*, native_socket fd) override;
doorman_ptr new_doorman(asio_tcp_socket_acceptor&& sock);
expected<connection_handle>
add_tcp_scribe(abstract_broker*, const std::string&, uint16_t) override;
doorman_ptr new_doorman(native_socket fd) override;
expected<std::pair<accept_handle, uint16_t>>
new_tcp_doorman(uint16_t p, const char* in, bool rflag) override;
expected<void>
assign_tcp_doorman(abstract_broker*, accept_handle hdl) override;
accept_handle add_tcp_doorman(abstract_broker*,
asio_tcp_socket_acceptor&& sock);
accept_handle add_tcp_doorman(abstract_broker*, native_socket fd) override;
expected<std::pair<accept_handle, uint16_t>>
add_tcp_doorman(abstract_broker*, uint16_t, const char*, bool) override;
expected<doorman_ptr> new_tcp_doorman(uint16_t port, const char* in,
bool reuse_addr) override;
void exec_later(resumable* ptr) override;
......@@ -104,10 +90,6 @@ public:
private:
io_service service_;
std::mutex mtx_sockets_;
std::mutex mtx_acceptors_;
std::map<int64_t, asio_tcp_socket> unassigned_sockets_;
std::map<int64_t, asio_tcp_socket_acceptor> unassigned_acceptors_;
};
template <class T>
......
......@@ -93,38 +93,12 @@ error ip_bind(asio_tcp_socket_acceptor& fd, uint16_t port,
}
}
expected<connection_handle>
asio_multiplexer::new_tcp_scribe(const std::string& host, uint16_t port) {
auto sck = new_tcp_connection(service(), host, port);
if (!sck)
return std::move(sck.error());
asio_tcp_socket fd{std::move(*sck)};
auto id = int64_from_native_socket(fd.native_handle());
std::lock_guard<std::mutex> lock(mtx_sockets_);
unassigned_sockets_.insert(std::make_pair(id, std::move(fd)));
return connection_handle::from_int(id);
}
expected<void> asio_multiplexer::assign_tcp_scribe(abstract_broker* self,
connection_handle hdl) {
std::lock_guard<std::mutex> lock(mtx_sockets_);
auto itr = unassigned_sockets_.find(hdl.id());
if (itr != unassigned_sockets_.end()) {
add_tcp_scribe(self, std::move(itr->second));
unassigned_sockets_.erase(itr);
return unit;
}
return sec::failed_to_assign_scribe_from_handle;
}
template <class Socket>
connection_handle asio_multiplexer::add_tcp_scribe(abstract_broker* self,
Socket&& sock) {
scribe_ptr asio_multiplexer::new_scribe(asio_tcp_socket&& sock) {
CAF_LOG_TRACE("");
class impl : public scribe {
public:
impl(abstract_broker* ptr, asio_multiplexer& am, Socket&& s)
: scribe(ptr, network::conn_hdl_from_socket(s)),
impl(asio_multiplexer& am, asio_tcp_socket&& s)
: scribe(network::conn_hdl_from_socket(s)),
launched_(false),
stream_(am) {
stream_.init(std::move(s));
......@@ -156,8 +130,7 @@ connection_handle asio_multiplexer::add_tcp_scribe(abstract_broker* self,
stream_.flush(this);
}
std::string addr() const override {
return
stream_.socket_handle().remote_endpoint().address().to_string();
return stream_.socket_handle().remote_endpoint().address().to_string();
}
uint16_t port() const override {
return stream_.socket_handle().remote_endpoint().port();
......@@ -174,17 +147,14 @@ connection_handle asio_multiplexer::add_tcp_scribe(abstract_broker* self,
void remove_from_loop() override {
stream_.passivate();
}
private:
private:
bool launched_;
asio_stream<Socket> stream_;
asio_stream<asio_tcp_socket> stream_;
};
auto ptr = make_counted<impl>(self, *this, std::move(sock));
self->add_scribe(ptr);
return ptr->hdl();
return make_counted<impl>(*this, std::move(sock));
}
connection_handle asio_multiplexer::add_tcp_scribe(abstract_broker* self,
native_socket fd) {
scribe_ptr asio_multiplexer::new_scribe(native_socket fd) {
CAF_LOG_TRACE(CAF_ARG(self) << ", " << CAF_ARG(fd));
boost::system::error_code ec;
asio_tcp_socket sock{service()};
......@@ -193,56 +163,24 @@ connection_handle asio_multiplexer::add_tcp_scribe(abstract_broker* self,
sock.assign(boost::asio::ip::tcp::v4(), fd, ec);
if (ec)
CAF_RAISE_ERROR(ec.message());
return add_tcp_scribe(self, std::move(sock));
return new_scribe(std::move(sock));
}
expected<connection_handle>
asio_multiplexer::add_tcp_scribe(abstract_broker* self,
const std::string& host, uint16_t port) {
CAF_LOG_TRACE(CAF_ARG(self) << ", " << CAF_ARG(host) << ":" << CAF_ARG(port));
auto conn = new_tcp_connection(service(), host, port);
if (!conn)
return std::move(conn.error());
return add_tcp_scribe(self, std::move(*conn));
}
expected<std::pair<accept_handle, uint16_t>>
asio_multiplexer::new_tcp_doorman(uint16_t port, const char* in, bool rflag) {
CAF_LOG_TRACE(CAF_ARG(port) << ", addr = " << (in ? in : "nullptr"));
asio_tcp_socket_acceptor fd{service()};
auto err = ip_bind(fd, port, in, rflag);
if (err)
return err;
auto id = int64_from_native_socket(fd.native_handle());
auto assigned_port = fd.local_endpoint().port();
std::lock_guard<std::mutex> lock(mtx_acceptors_);
unassigned_acceptors_.insert(std::make_pair(id, std::move(fd)));
return std::make_pair(accept_handle::from_int(id), assigned_port);
}
expected<void> asio_multiplexer::assign_tcp_doorman(abstract_broker* self,
accept_handle hdl) {
CAF_LOG_TRACE("");
std::lock_guard<std::mutex> lock(mtx_acceptors_);
auto itr = unassigned_acceptors_.find(hdl.id());
if (itr != unassigned_acceptors_.end()) {
add_tcp_doorman(self, std::move(itr->second));
unassigned_acceptors_.erase(itr);
return unit;
}
return sec::failed_to_assign_doorman_from_handle;
expected<scribe_ptr> asio_multiplexer::new_tcp_scribe(const std::string& host,
uint16_t port) {
auto sck = new_tcp_connection(service(), host, port);
if (!sck)
return std::move(sck.error());
return new_scribe(std::move(*sck));
}
accept_handle
asio_multiplexer::add_tcp_doorman(abstract_broker* self,
asio_tcp_socket_acceptor&& sock) {
doorman_ptr asio_multiplexer::new_doorman(asio_tcp_socket_acceptor&& sock) {
CAF_LOG_TRACE(CAF_ARG(sock.native_handle()));
CAF_ASSERT(sock.native_handle() != network::invalid_native_socket);
class impl : public doorman {
public:
impl(abstract_broker* ptr, asio_tcp_socket_acceptor&& s,
network::asio_multiplexer& am)
: doorman(ptr, network::accept_hdl_from_socket(s)),
impl(asio_tcp_socket_acceptor&& s, network::asio_multiplexer& am)
: doorman(network::accept_hdl_from_socket(s)),
acceptor_(am, s.get_io_service()) {
acceptor_.init(std::move(s));
}
......@@ -255,9 +193,10 @@ asio_multiplexer::add_tcp_doorman(abstract_broker* self,
// further activities for the broker
return false;
auto& am = acceptor_.backend();
auto x = am.add_tcp_scribe(parent(),
std::move(acceptor_.accepted_socket()));
return doorman::new_connection(&am, x);
auto sptr = am.new_scribe(std::move(acceptor_.accepted_socket()));
auto hdl = sptr->hdl();
parent()->add_scribe(std::move(sptr));
return doorman::new_connection(&am, hdl);
}
void stop_reading() override {
CAF_LOG_TRACE("");
......@@ -284,13 +223,10 @@ asio_multiplexer::add_tcp_doorman(abstract_broker* self,
private:
network::asio_acceptor<asio_tcp_socket_acceptor> acceptor_;
};
auto ptr = make_counted<impl>(self, std::move(sock), *this);
self->add_doorman(ptr);
return ptr->hdl();
return make_counted<impl>(std::move(sock), *this);
}
accept_handle asio_multiplexer::add_tcp_doorman(abstract_broker* self,
native_socket fd) {
doorman_ptr asio_multiplexer::new_doorman(native_socket fd) {
CAF_LOG_TRACE(CAF_ARG(fd));
asio_tcp_socket_acceptor sock{service()};
boost::system::error_code ec;
......@@ -299,19 +235,17 @@ accept_handle asio_multiplexer::add_tcp_doorman(abstract_broker* self,
sock.assign(boost::asio::ip::tcp::v4(), fd, ec);
if (ec)
CAF_RAISE_ERROR(ec.message());
return add_tcp_doorman(self, std::move(sock));
return new_doorman(std::move(sock));
}
expected<std::pair<accept_handle, uint16_t>>
asio_multiplexer::add_tcp_doorman(abstract_broker* self, uint16_t port,
const char* in, bool rflag) {
CAF_LOG_TRACE(CAF_ARG(port) << CAF_ARG(in) << CAF_ARG(rflag));
expected<doorman_ptr>
asio_multiplexer::new_tcp_doorman(uint16_t port, const char* in, bool rflag) {
CAF_LOG_TRACE(CAF_ARG(port) << ", addr = " << (in ? in : "nullptr"));
asio_tcp_socket_acceptor fd{service()};
auto err = ip_bind(fd, port, in, rflag);
if (err)
return err;
auto p = fd.local_endpoint().port();
return std::make_pair(add_tcp_doorman(self, std::move(fd)), p);
return new_doorman(std::move(fd));
}
void asio_multiplexer::exec_later(resumable* rptr) {
......
......@@ -270,29 +270,15 @@ public:
}
};
expected<connection_handle> new_tcp_scribe(const std::string &,
uint16_t) override;
scribe_ptr new_scribe(native_socket fd) override;
expected<void> assign_tcp_scribe(abstract_broker *self,
connection_handle hdl) override;
expected<scribe_ptr> new_tcp_scribe(const std::string& host,
uint16_t port) override;
connection_handle add_tcp_scribe(abstract_broker *,
native_socket fd) override;
doorman_ptr new_doorman(native_socket fd) override;
expected<connection_handle> add_tcp_scribe(abstract_broker *,
const std::string &host,
uint16_t port) override;
expected<std::pair<accept_handle, uint16_t>>
new_tcp_doorman(uint16_t port, const char *in, bool reuse_addr) override;
expected<void> assign_tcp_doorman(abstract_broker *ptr,
accept_handle hdl) override;
accept_handle add_tcp_doorman(abstract_broker*, native_socket fd) override;
expected<std::pair<accept_handle, uint16_t>>
add_tcp_doorman(abstract_broker *, uint16_t, const char *, bool) override;
expected<doorman_ptr> new_tcp_doorman(uint16_t port, const char* in,
bool reuse_addr) override;
void exec_later(resumable* ptr) override;
......@@ -498,8 +484,8 @@ expected<native_socket> new_tcp_connection(const std::string& host,
uint16_t port,
optional<protocol> preferred = none);
expected<std::pair<native_socket, uint16_t>>
new_tcp_acceptor_impl(uint16_t port, const char* addr, bool reuse_addr);
expected<native_socket> new_tcp_acceptor_impl(uint16_t port, const char* addr,
bool reuse_addr);
} // namespace network
} // namespace io
......
......@@ -35,7 +35,7 @@ namespace network {
/// for various I/O operations.
class manager : public ref_counted {
public:
manager(abstract_broker* ptr);
manager();
~manager() override;
......
......@@ -52,51 +52,26 @@ class multiplexer : public execution_unit {
public:
explicit multiplexer(actor_system* sys);
/// Tries to connect to `host` on given `port` and returns an unbound
/// connection handle on success.
/// Creates a new `scribe` from a native socket handle.
/// @threadsafe
virtual expected<connection_handle>
new_tcp_scribe(const std::string& host, uint16_t port) = 0;
virtual scribe_ptr new_scribe(native_socket fd) = 0;
/// Assigns an unbound scribe identified by `hdl` to `ptr`.
/// @warning Do not call from outside the multiplexer's event loop.
virtual expected<void>
assign_tcp_scribe(abstract_broker* ptr, connection_handle hdl) = 0;
/// Creates a new TCP doorman from a native socket handle.
/// @warning Do not call from outside the multiplexer's event loop.
virtual connection_handle add_tcp_scribe(abstract_broker* ptr,
native_socket fd) = 0;
/// Tries to connect to `host` on given `port` and returns a `scribe` instance
/// on success.
/// @threadsafe
virtual expected<scribe_ptr> new_tcp_scribe(const std::string& host,
uint16_t port) = 0;
/// Tries to connect to `host` on `port` and returns a
/// new scribe managing the connection on success.
/// @warning Do not call from outside the multiplexer's event loop.
virtual expected<connection_handle>
add_tcp_scribe(abstract_broker*, const std::string& host, uint16_t port) = 0;
/// Creates a new doorman from a native socket handle.
/// @threadsafe
virtual doorman_ptr new_doorman(native_socket fd) = 0;
/// Tries to create an unbound TCP doorman bound to `port`, optionally
/// accepting only connections from IP address `in`.
/// @warning Do not call from outside the multiplexer's event loop.
virtual expected<std::pair<accept_handle, uint16_t>>
new_tcp_doorman(uint16_t port, const char* in = nullptr,
bool reuse_addr = false) = 0;
/// Assigns an unbound doorman identified by `hdl` to `ptr`.
/// @warning Do not call from outside the multiplexer's event loop.
virtual expected<void>
assign_tcp_doorman(abstract_broker* ptr, accept_handle hdl) = 0;
/// Creates a new TCP doorman from a native socket handle.
/// @warning Do not call from outside the multiplexer's event loop.
virtual accept_handle add_tcp_doorman(abstract_broker* ptr,
native_socket fd) = 0;
/// Tries to create a new TCP doorman running on port `p`, optionally
/// accepting only connections from IP address `in`.
/// @warning Do not call from outside the multiplexer's event loop.
virtual expected<std::pair<accept_handle, uint16_t>>
add_tcp_doorman(abstract_broker* ptr, uint16_t port, const char* in = nullptr,
bool reuse_addr = false) = 0;
virtual expected<doorman_ptr> new_tcp_doorman(uint16_t port,
const char* in = nullptr,
bool reuse_addr = false) = 0;
/// Simple wrapper for runnables
class runnable : public resumable, public ref_counted {
......
......@@ -32,8 +32,6 @@ namespace network {
/// for incoming data as well as for error handling.
class stream_manager : public manager {
public:
stream_manager(abstract_broker* ptr);
~stream_manager() override;
/// Called by the underlying I/O device whenever it received data.
......
......@@ -37,35 +37,34 @@ public:
~test_multiplexer() override;
expected<connection_handle> new_tcp_scribe(const std::string& host,
uint16_t port_hint) override;
scribe_ptr new_scribe(native_socket) override;
expected<void> assign_tcp_scribe(abstract_broker* ptr,
connection_handle hdl) override;
expected<scribe_ptr> new_tcp_scribe(const std::string& host,
uint16_t port_hint) override;
connection_handle add_tcp_scribe(abstract_broker*, native_socket) override;
doorman_ptr new_doorman(native_socket) override;
expected<connection_handle> add_tcp_scribe(abstract_broker* ptr,
const std::string& host,
uint16_t desired_port) override;
expected<doorman_ptr> new_tcp_doorman(uint16_t prt, const char* in,
bool reuse_addr) override;
expected<std::pair<accept_handle, uint16_t>>
new_tcp_doorman(uint16_t desired_port, const char*, bool) override;
/// Checks whether `x` is assigned to any known doorman or is user-provided
/// for future assignment.
bool is_known_port(uint16_t x) const;
expected<void> assign_tcp_doorman(abstract_broker* ptr,
accept_handle hdl) override;
accept_handle add_tcp_doorman(abstract_broker*, native_socket) override;
expected<std::pair<accept_handle, uint16_t>>
add_tcp_doorman(abstract_broker* ptr, uint16_t prt,
const char* in, bool reuse_addr) override;
/// Checks whether `x` is assigned to any known doorman or is user-provided
/// for future assignment.
bool is_known_handle(accept_handle x) const;
supervisor_ptr make_supervisor() override;
void run() override;
void provide_scribe(std::string host, uint16_t desired_port, connection_handle hdl);
scribe_ptr new_scribe(connection_handle);
doorman_ptr new_doorman(accept_handle, uint16_t port);
void provide_scribe(std::string host, uint16_t desired_port,
connection_handle hdl);
void provide_acceptor(uint16_t desired_port, accept_handle hdl);
......@@ -97,7 +96,7 @@ public:
/// Returns `true` if this handle is inactive, otherwise `false`.
bool& passive_mode(connection_handle hdl);
intrusive_ptr<scribe>& impl_ptr(connection_handle hdl);
scribe_ptr& impl_ptr(connection_handle hdl);
uint16_t& port(accept_handle hdl);
......@@ -108,7 +107,7 @@ public:
/// Returns `true` if this handle is inactive, otherwise `false`.
bool& passive_mode(accept_handle hdl);
intrusive_ptr<doorman>& impl_ptr(accept_handle hdl);
doorman_ptr& impl_ptr(accept_handle hdl);
/// Stores `hdl` as a pending connection for `src`.
void add_pending_connect(accept_handle src, connection_handle hdl);
......@@ -128,6 +127,8 @@ public:
using pending_scribes_map = std::map<std::pair<std::string, uint16_t>,
connection_handle>;
using pending_doorman_map = std::unordered_map<uint16_t, accept_handle>;
bool has_pending_scribe(std::string x, uint16_t y);
/// Accepts a pending connect on `hdl`.
......@@ -182,20 +183,25 @@ private:
};
struct doorman_data {
doorman_ptr ptr;
uint16_t port;
bool stopped_reading = false;
bool passive_mode = false;
intrusive_ptr<doorman> ptr;
bool stopped_reading;
bool passive_mode;
doorman_data();
};
using scribe_data_map = std::unordered_map<connection_handle, scribe_data>;
using doorman_data_map = std::unordered_map<accept_handle, doorman_data>;
// guards resumables_ and scribes_
std::mutex mx_;
std::condition_variable cv_;
std::list<resumable_ptr> resumables_;
pending_scribes_map scribes_;
std::unordered_map<uint16_t, accept_handle> doormen_;
std::unordered_map<connection_handle, scribe_data> scribe_data_;
std::unordered_map<accept_handle, doorman_data> doorman_data_;
scribe_data_map scribe_data_;
doorman_data_map doorman_data_;
pending_connects_map pending_connects_;
// extra state for making sure the test multiplexer is not used in a
......
......@@ -39,7 +39,7 @@ using scribe_base = broker_servant<network::stream_manager, connection_handle,
/// @ingroup Broker
class scribe : public scribe_base {
public:
scribe(abstract_broker* ptr, connection_handle conn_hdl);
scribe(connection_handle conn_hdl);
~scribe() override;
......@@ -69,7 +69,13 @@ protected:
message detach_message() override;
};
using scribe_ptr = intrusive_ptr<scribe>;
} // namespace io
} // namespace caf
// Allows the `middleman_actor` to create a `scribe` and then send it to the
// BASP broker.
CAF_ALLOW_UNSAFE_MESSAGE_TYPE(caf::io::scribe_ptr)
#endif // CAF_IO_SCRIBE_HPP
......@@ -123,9 +123,7 @@ public:
cfg.init_fun = fac(std::move(fun), hdl, std::forward<Ts>(xs)...);
auto res = this->system().spawn_functor(cfg, fun, hdl, std::forward<Ts>(xs)...);
auto forked = static_cast<impl*>(actor_cast<abstract_actor*>(res));
sptr->set_parent(forked);
CAF_ASSERT(sptr->parent() == forked);
forked->add_scribe(sptr);
forked->move_scribe(std::move(sptr));
return res;
}
......
......@@ -118,48 +118,50 @@ std::vector<connection_handle> abstract_broker::connections() const {
return result;
}
void abstract_broker::add_scribe(const intrusive_ptr<scribe>& ptr) {
scribes_.emplace(ptr->hdl(), ptr);
void abstract_broker::add_scribe(scribe_ptr ptr) {
CAF_LOG_TRACE(CAF_ARG(ptr));
add_servant(std::move(ptr));
}
expected<connection_handle> abstract_broker::add_tcp_scribe(const std::string& hostname,
uint16_t port) {
CAF_LOG_TRACE(CAF_ARG(hostname) << ", " << CAF_ARG(port));
return backend().add_tcp_scribe(this, hostname, port);
connection_handle abstract_broker::add_scribe(network::native_socket fd) {
CAF_LOG_TRACE(CAF_ARG(fd));
return add_servant(backend().new_scribe(fd));
}
expected<void> abstract_broker::assign_tcp_scribe(connection_handle hdl) {
CAF_LOG_TRACE(CAF_ARG(hdl));
return backend().assign_tcp_scribe(this, hdl);
expected<connection_handle>
abstract_broker::add_tcp_scribe(const std::string& hostname, uint16_t port) {
CAF_LOG_TRACE(CAF_ARG(hostname) << ", " << CAF_ARG(port));
auto eptr = backend().new_tcp_scribe(hostname, port);
if (eptr)
return add_servant(std::move(*eptr));
return std::move(eptr.error());
}
void abstract_broker::move_scribe(scribe_ptr ptr) {
CAF_LOG_TRACE(CAF_ARG(ptr));
move_servant(std::move(ptr));
}
expected<connection_handle>
abstract_broker::add_tcp_scribe(network::native_socket fd) {
CAF_LOG_TRACE(CAF_ARG(fd));
return backend().add_tcp_scribe(this, fd);
void abstract_broker::add_doorman(doorman_ptr ptr) {
CAF_LOG_TRACE(CAF_ARG(ptr));
add_servant(std::move(ptr));
}
void abstract_broker::add_doorman(const intrusive_ptr<doorman>& ptr) {
doormen_.emplace(ptr->hdl(), ptr);
if (getf(is_initialized_flag))
ptr->launch();
accept_handle abstract_broker::add_doorman(network::native_socket fd) {
CAF_LOG_TRACE(CAF_ARG(fd));
return add_servant(backend().new_doorman(fd));
}
expected<std::pair<accept_handle, uint16_t>>
abstract_broker::add_tcp_doorman(uint16_t port, const char* in,
bool reuse_addr) {
CAF_LOG_TRACE(CAF_ARG(port) << CAF_ARG(in) << CAF_ARG(reuse_addr));
return backend().add_tcp_doorman(this, port, in, reuse_addr);
}
expected<void> abstract_broker::assign_tcp_doorman(accept_handle hdl) {
CAF_LOG_TRACE(CAF_ARG(hdl));
return backend().assign_tcp_doorman(this, hdl);
}
expected<accept_handle> abstract_broker::add_tcp_doorman(network::native_socket fd) {
CAF_LOG_TRACE(CAF_ARG(fd));
return backend().add_tcp_doorman(this, fd);
auto eptr = backend().new_tcp_doorman(port, in, reuse_addr);
if (eptr) {
auto ptr = std::move(*eptr);
auto p = ptr->port();
return std::make_pair(add_servant(std::move(ptr)), p);
}
return std::move(eptr.error());
}
std::string abstract_broker::remote_addr(connection_handle hdl) {
......@@ -234,5 +236,12 @@ network::multiplexer& abstract_broker::backend() {
return system().middleman().backend();
}
void abstract_broker::launch_servant(doorman_ptr& ptr) {
// A doorman needs to be launched in addition to being initialized. This
// allows CAF to assign doorman to uninitialized brokers.
if (getf(is_initialized_flag))
ptr->launch();
}
} // namespace io
} // namespace caf
......@@ -23,10 +23,6 @@ namespace caf {
namespace io {
namespace network {
acceptor_manager::acceptor_manager(abstract_broker* ptr) : manager(ptr) {
// nop
}
acceptor_manager::~acceptor_manager() {
// nop
}
......
......@@ -615,40 +615,30 @@ behavior basp_broker::make_behavior() {
state.instance.remove_published_actor(port);
},
// received from middleman actor
[=](publish_atom, accept_handle hdl, uint16_t port,
[=](publish_atom, doorman_ptr& ptr, uint16_t port,
const strong_actor_ptr& whom, std::set<std::string>& sigs) {
CAF_LOG_TRACE(CAF_ARG(hdl.id()) << CAF_ARG(whom) << CAF_ARG(port));
if (hdl.invalid()) {
CAF_LOG_WARNING("invalid handle");
return;
}
auto res = assign_tcp_doorman(hdl);
if (res) {
if (whom)
system().registry().put(whom->id(), whom);
state.instance.add_published_actor(port, whom, std::move(sigs));
} else {
CAF_LOG_DEBUG("failed to assign doorman from handle"
<< CAF_ARG(whom));
}
CAF_LOG_TRACE(CAF_ARG(ptr) << CAF_ARG(port)
<< CAF_ARG(whom) << CAF_ARG(sigs));
CAF_ASSERT(ptr != nullptr);
add_doorman(std::move(ptr));
if (whom)
system().registry().put(whom->id(), whom);
state.instance.add_published_actor(port, whom, std::move(sigs));
},
// received from middleman actor (delegated)
[=](connect_atom, connection_handle hdl, uint16_t port) {
CAF_LOG_TRACE(CAF_ARG(hdl.id()));
[=](connect_atom, scribe_ptr& ptr, uint16_t port) {
CAF_LOG_TRACE(CAF_ARG(ptr) << CAF_ARG(port));
CAF_ASSERT(ptr != nullptr);
auto rp = make_response_promise();
auto res = assign_tcp_scribe(hdl);
if (res) {
auto& ctx = state.ctx[hdl];
ctx.hdl = hdl;
ctx.remote_port = port;
ctx.cstate = basp::await_header;
ctx.callback = rp;
// await server handshake
configure_read(hdl, receive_policy::exactly(basp::header_size));
} else {
CAF_LOG_DEBUG("failed to assign scribe from handle" << CAF_ARG(res));
rp.deliver(std::move(res.error()));
}
auto hdl = ptr->hdl();
add_scribe(std::move(ptr));
auto& ctx = state.ctx[hdl];
ctx.hdl = hdl;
ctx.remote_port = port;
ctx.cstate = basp::await_header;
ctx.callback = rp;
// await server handshake
configure_read(hdl, receive_policy::exactly(basp::header_size));
},
[=](delete_atom, const node_id& nid, actor_id aid) {
CAF_LOG_TRACE(CAF_ARG(nid) << ", " << CAF_ARG(aid));
......
......@@ -731,13 +731,12 @@ void default_multiplexer::exec_later(resumable* ptr) {
}
}
connection_handle default_multiplexer::add_tcp_scribe(abstract_broker* self,
native_socket fd) {
scribe_ptr default_multiplexer::new_scribe(native_socket fd) {
CAF_LOG_TRACE("");
class impl : public scribe {
public:
impl(abstract_broker* ptr, default_multiplexer& mx, native_socket sockfd)
: scribe(ptr, network::conn_hdl_from_socket(sockfd)),
impl(default_multiplexer& mx, native_socket sockfd)
: scribe(network::conn_hdl_from_socket(sockfd)),
launched_(false),
stream_(mx, sockfd) {
// nop
......@@ -791,23 +790,28 @@ connection_handle default_multiplexer::add_tcp_scribe(abstract_broker* self,
void remove_from_loop() override {
stream_.passivate();
}
private:
private:
bool launched_;
stream stream_;
};
auto ptr = make_counted<impl>(self, *this, fd);
self->add_scribe(ptr);
return ptr->hdl();
return make_counted<impl>(*this, fd);
}
expected<scribe_ptr>
default_multiplexer::new_tcp_scribe(const std::string& host, uint16_t port) {
auto fd = new_tcp_connection(host, port);
if (!fd)
return std::move(fd.error());
return new_scribe(*fd);
}
accept_handle default_multiplexer::add_tcp_doorman(abstract_broker* self,
native_socket fd) {
doorman_ptr default_multiplexer::new_doorman(native_socket fd) {
CAF_LOG_TRACE(CAF_ARG(fd));
CAF_ASSERT(fd != network::invalid_native_socket);
class impl : public doorman {
public:
impl(abstract_broker* ptr, default_multiplexer& mx, native_socket sockfd)
: doorman(ptr, network::accept_hdl_from_socket(sockfd)),
impl(default_multiplexer& mx, native_socket sockfd)
: doorman(network::accept_hdl_from_socket(sockfd)),
acceptor_(mx, sockfd) {
// nop
}
......@@ -820,8 +824,9 @@ accept_handle default_multiplexer::add_tcp_doorman(abstract_broker* self,
// further activities for the broker
return false;
auto& dm = acceptor_.backend();
auto hdl = dm.add_tcp_scribe(parent(),
std::move(acceptor_.accepted_socket()));
auto sptr = dm.new_scribe(acceptor_.accepted_socket());
auto hdl = sptr->hdl();
parent()->add_scribe(std::move(sptr));
return doorman::new_connection(&dm, hdl);
}
void stop_reading() override {
......@@ -851,63 +856,19 @@ accept_handle default_multiplexer::add_tcp_doorman(abstract_broker* self,
void remove_from_loop() override {
acceptor_.passivate();
}
private:
private:
network::acceptor acceptor_;
};
auto ptr = make_counted<impl>(self, *this, fd);
self->add_doorman(ptr);
return ptr->hdl();
return make_counted<impl>(*this, fd);
}
expected<connection_handle>
default_multiplexer::new_tcp_scribe(const std::string& host, uint16_t port) {
auto fd = new_tcp_connection(host, port);
if (!fd)
return std::move(fd.error());
return connection_handle::from_int(int64_from_native_socket(*fd));
}
expected<void> default_multiplexer::assign_tcp_scribe(abstract_broker* self,
connection_handle hdl) {
CAF_LOG_TRACE(CAF_ARG(self->id()) << CAF_ARG(hdl));
add_tcp_scribe(self, static_cast<native_socket>(hdl.id()));
return unit;
}
expected<connection_handle>
default_multiplexer::add_tcp_scribe(abstract_broker* self,
const std::string& host, uint16_t port) {
CAF_LOG_TRACE(CAF_ARG(self->id()) << CAF_ARG(host) << CAF_ARG(port));
auto fd = new_tcp_connection(host, port);
if (!fd)
return std::move(fd.error());
return add_tcp_scribe(self, *fd);
}
expected<std::pair<accept_handle, uint16_t>>
default_multiplexer::new_tcp_doorman(uint16_t port, const char* in,
bool reuse_addr) {
auto res = new_tcp_acceptor_impl(port, in, reuse_addr);
if (!res)
return std::move(res.error());
return std::make_pair(accept_handle::from_int(int64_from_native_socket(res->first)),
res->second);
}
expected<void> default_multiplexer::assign_tcp_doorman(abstract_broker* ptr,
accept_handle hdl) {
add_tcp_doorman(ptr, static_cast<native_socket>(hdl.id()));
return unit;
}
expected<std::pair<accept_handle, uint16_t>>
default_multiplexer::add_tcp_doorman(abstract_broker* self, uint16_t port,
const char* host, bool reuse_addr) {
auto acceptor = new_tcp_acceptor_impl(port, host, reuse_addr);
if (!acceptor)
return std::move(acceptor.error());
auto bound_port = acceptor->second;
return std::make_pair(add_tcp_doorman(self, acceptor->first), bound_port);
expected<doorman_ptr> default_multiplexer::new_tcp_doorman(uint16_t port,
const char* in,
bool reuse_addr) {
auto fd = new_tcp_acceptor_impl(port, in, reuse_addr);
if (fd)
return new_doorman(*fd);
return std::move(fd.error());
}
/******************************************************************************
......@@ -1409,7 +1370,7 @@ expected<void> set_inaddr_any(native_socket fd, sockaddr_in6& sa) {
}
template <int Family>
expected<uint16_t> new_ip_acceptor_impl(native_socket fd, uint16_t port,
expected<void> new_ip_acceptor_impl(native_socket fd, uint16_t port,
const char* addr) {
static_assert(Family == AF_INET || Family == AF_INET6, "invalid family");
CAF_LOG_TRACE(CAF_ARG(port) << ", addr = " << (addr ? addr : "nullptr"));
......@@ -1432,12 +1393,11 @@ expected<uint16_t> new_ip_acceptor_impl(native_socket fd, uint16_t port,
CALL_CFUN(res, cc_zero, "bind",
bind(fd, reinterpret_cast<sockaddr*>(&sa),
static_cast<socklen_t>(sizeof(sa))));
read_port(fd, sa);
return ntohs(port_of(sa));
return unit;
}
expected<std::pair<native_socket, uint16_t>>
new_tcp_acceptor_impl(uint16_t port, const char* addr, bool reuse_addr) {
expected<native_socket> new_tcp_acceptor_impl(uint16_t port, const char* addr,
bool reuse_addr) {
CAF_LOG_TRACE(CAF_ARG(port) << ", addr = " << (addr ? addr : "nullptr"));
protocol proto = ipv6;
if (addr != nullptr) {
......@@ -1465,7 +1425,7 @@ new_tcp_acceptor_impl(uint16_t port, const char* addr, bool reuse_addr) {
CALL_CFUN(tmp2, cc_zero, "listen", listen(fd, SOMAXCONN));
// ok, no errors so far
CAF_LOG_DEBUG(CAF_ARG(fd) << CAF_ARG(p));
return std::make_pair(sguard.release(), *p);
return sguard.release();
}
expected<std::string> local_addr_of_fd(native_socket fd) {
......
......@@ -26,8 +26,7 @@
namespace caf {
namespace io {
doorman::doorman(abstract_broker* ptr, accept_handle acc_hdl)
: doorman_base(ptr, acc_hdl) {
doorman::doorman(accept_handle acc_hdl) : doorman_base(acc_hdl) {
// nop
}
......
......@@ -27,7 +27,7 @@ namespace caf {
namespace io {
namespace network {
manager::manager(abstract_broker* ptr) : parent_(ptr->ctrl()) {
manager::manager() : parent_(nullptr) {
// nop
}
......@@ -36,8 +36,7 @@ manager::~manager() {
}
void manager::set_parent(abstract_broker* ptr) {
if (!detached())
parent_ = ptr != nullptr ? ptr->ctrl() : nullptr;
parent_ = ptr != nullptr ? ptr->ctrl() : nullptr;
}
abstract_broker* manager::parent() {
......
......@@ -118,15 +118,16 @@ public:
return {};
}
// connect to endpoint and initiate handhsake etc.
auto y = system().middleman().backend().new_tcp_scribe(key.first, port);
if (!y) {
rp.deliver(std::move(y.error()));
auto r = system().middleman().backend().new_tcp_scribe(key.first, port);
if (!r) {
rp.deliver(std::move(r.error()));
return {};
}
auto hdl = *y;
auto& ptr = *r;
std::vector<response_promise> tmp{std::move(rp)};
pending_.emplace(key, std::move(tmp));
request(broker_, infinite, connect_atom::value, hdl, port).then(
request(broker_, infinite, connect_atom::value, std::move(ptr), port)
.then(
[=](node_id& nid, strong_actor_ptr& addr, mpi_set& sigs) {
auto i = pending_.find(key);
if (i == pending_.end())
......@@ -186,7 +187,6 @@ private:
bool reuse_addr = false) {
CAF_LOG_TRACE(CAF_ARG(port) << CAF_ARG(whom) << CAF_ARG(sigs)
<< CAF_ARG(in) << CAF_ARG(reuse_addr));
accept_handle hdl;
uint16_t actual_port;
// treat empty strings like nullptr
if (in != nullptr && in[0] == '\0')
......@@ -195,9 +195,9 @@ private:
reuse_addr);
if (!res)
return std::move(res.error());
hdl = res->first;
actual_port = res->second;
anon_send(broker_, publish_atom::value, hdl, actual_port,
auto& ptr = *res;
actual_port = ptr->port();
anon_send(broker_, publish_atom::value, std::move(ptr), actual_port,
std::move(whom), std::move(sigs));
return actual_port;
}
......
......@@ -24,8 +24,7 @@
namespace caf {
namespace io {
scribe::scribe(abstract_broker* ptr, connection_handle conn_hdl)
: scribe_base(ptr, conn_hdl) {
scribe::scribe(connection_handle conn_hdl) : scribe_base(conn_hdl) {
// nop
}
......
......@@ -23,10 +23,6 @@ namespace caf {
namespace io {
namespace network {
stream_manager::stream_manager(abstract_broker* ptr) : manager(ptr) {
// nop
}
stream_manager::~stream_manager() {
// nop
}
......
This diff is collapsed.
......@@ -140,7 +140,7 @@ public:
this_node_ = sys.node();
self_.reset(new scoped_actor{sys});
ahdl_ = accept_handle::from_int(1);
mpx_->assign_tcp_doorman(aut_, ahdl_);
aut_->add_doorman(mpx_->new_doorman(ahdl_, 1u));
registry_ = &sys.registry();
registry_->put((*self_)->id(), actor_cast<strong_actor_ptr>(*self_));
// first remote node is everything of this_node + 1, then +2, etc.
......@@ -285,7 +285,6 @@ public:
<< ", acceptor ID = " << src.id());
auto hdl = n.connection;
mpx_->add_pending_connect(src, hdl);
mpx_->assign_tcp_scribe(aut(), hdl);
CAF_REQUIRE(mpx_->accept_connection(src));
// technically, the server handshake arrives
// before we send the client handshake
......
......@@ -159,10 +159,10 @@ behavior http_worker(http_broker* self, connection_handle hdl) {
}
behavior server(broker* self) {
aout(self) << "server up and running" << endl;
CAF_MESSAGE("server up and running");
return {
[=](const new_connection_msg& ncm) {
aout(self) << "fork on new connection" << endl;
CAF_MESSAGE("fork on new connection");
self->fork(http_worker, ncm.handle);
}
};
......@@ -178,10 +178,9 @@ public:
aut_ = system.middleman().spawn_broker(server);
// assign the acceptor handle to the AUT
aut_ptr_ = static_cast<abstract_broker*>(actor_cast<abstract_actor*>(aut_));
mpx_->assign_tcp_doorman(aut_ptr_, acceptor_);
aut_ptr_->add_doorman(mpx_->new_doorman(acceptor_, 1u));
// "open" a new connection to our server
mpx_->add_pending_connect(acceptor_, connection_);
mpx_->assign_tcp_scribe(aut_ptr_, connection_);
CAF_REQUIRE(mpx_->accept_connection(acceptor_));
}
......
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