Commit c3ab8dbf authored by Dominik Charousset's avatar Dominik Charousset

Disentangle middleman from default_multiplexer

parent 4e811fc1
......@@ -255,6 +255,8 @@ class broker : public extend<local_actor>::
connection_handle add_tcp_scribe(const std::string& host, uint16_t port);
void assign_tcp_scribe(connection_handle hdl);
connection_handle add_tcp_scribe(network::native_socket fd);
inline void add_doorman(const doorman_pointer& ptr) {
......@@ -268,6 +270,8 @@ class broker : public extend<local_actor>::
const char* in = nullptr,
bool reuse_addr = false);
void assign_tcp_doorman(accept_handle hdl);
accept_handle add_tcp_doorman(network::native_socket fd);
void invoke_message(const actor_addr& sender, message_id mid, message& msg);
......
......@@ -135,13 +135,6 @@ namespace network {
*/
using native_socket_acceptor = native_socket;
inline int64_t int64_from_native_socket(native_socket sock) {
// on Windows, SOCK is an unsigned value;
// hence, static_cast<> alone would yield the wrong result,
// as our io_handle assumes -1 as invalid value
return sock == invalid_native_socket ? -1 : static_cast<int64_t>(sock);
}
/**
* Returns the last socket error as human-readable string.
*/
......@@ -325,6 +318,10 @@ class default_multiplexer : public multiplexer {
}
};
connection_handle new_tcp_scribe(const std::string&, uint16_t) override;
void assign_tcp_scribe(broker* ptr, connection_handle hdl) override;
connection_handle add_tcp_scribe(broker*, default_socket_acceptor&& sock);
connection_handle add_tcp_scribe(broker*, native_socket fd) override;
......@@ -332,12 +329,17 @@ class default_multiplexer : public multiplexer {
connection_handle add_tcp_scribe(broker*, const std::string& h,
uint16_t port) override;
std::pair<accept_handle, uint16_t>
new_tcp_doorman(uint16_t p, const char* in, bool rflag) override;
void assign_tcp_doorman(broker* ptr, accept_handle hdl) override;
accept_handle add_tcp_doorman(broker*, default_socket_acceptor&& sock);
accept_handle add_tcp_doorman(broker*, native_socket fd) override;
std::pair<accept_handle, uint16_t>
add_tcp_doorman(broker*, uint16_t p, const char* h, bool reuse_addr) override;
add_tcp_doorman(broker*, uint16_t p, const char* in, bool rflag) override;
void dispatch_runnable(runnable_ptr ptr) override;
......
......@@ -30,6 +30,8 @@
#include "caf/io/fwd.hpp"
#include "caf/io/accept_handle.hpp"
#include "caf/io/connection_handle.hpp"
#include "caf/io/network/protocol.hpp"
#include "caf/io/network/native_socket.hpp"
#include "caf/mixin/memory_cached.hpp"
......@@ -53,26 +55,59 @@ class multiplexer {
public:
virtual ~multiplexer();
/**
* Tries to connect to `host` on given `port` and returns an unbound
* connection handle on success.
* @threadsafe
*/
virtual connection_handle new_tcp_scribe(const std::string& host,
uint16_t port) = 0;
/**
* Assigns an unbound scribe identified by `hdl` to `ptr`.
* @warning Do not call from outside the multiplexer's event loop.
*/
virtual void assign_tcp_scribe(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(broker* ptr, native_socket fd) = 0;
/**
* Tries to connect to host `h` on given `port` and returns a
* new scribe managing the connection on success.
* @warning Do not call from outside the multiplexer's event loop.
*/
virtual connection_handle add_tcp_scribe(broker* ptr, const std::string& host,
uint16_t port) = 0;
/**
* Tries to create an unbound TCP doorman running `port`, optionally
* accepting only connections from IP address `in`.
* @warning Do not call from outside the multiplexer's event loop.
*/
virtual 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 void assign_tcp_doorman(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(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 std::pair<accept_handle, uint16_t>
add_tcp_doorman(broker* ptr, uint16_t port, const char* in = nullptr,
......@@ -116,6 +151,7 @@ class multiplexer {
/**
* Invokes @p fun in the multiplexer's event loop.
* @threadsafe
*/
template <class F>
void dispatch(F fun) {
......@@ -128,6 +164,7 @@ class multiplexer {
/**
* Invokes @p fun in the multiplexer's event loop.
* @threadsafe
*/
template <class F>
void post(F fun) {
......
......@@ -29,9 +29,15 @@ namespace network {
#ifdef CAF_WINDOWS
using native_socket = unsigned;
constexpr native_socket invalid_native_socket = static_cast<unsigned>(-1);
inline int64_t int64_from_native_socket(native_socket sock) {
return sock == invalid_native_socket ? -1 : static_cast<uint64_t>(sock);
}
#else
using native_socket = int;
constexpr native_socket invalid_native_socket = -1;
inline int64_t int64_from_native_socket(native_socket sock) {
return static_cast<int64_t>(sock);
}
#endif
} // namespace network
......
......@@ -170,15 +170,15 @@ behavior basp_broker::make_behavior() {
erase_proxy(nid, aid);
},
// received from middleman actor
[=](put_atom, network::native_socket fd,
[=](put_atom, accept_handle hdl,
const actor_addr& whom, uint16_t port) {
auto hdl = add_tcp_doorman(fd);
assign_tcp_doorman(hdl);
add_published_actor(hdl, actor_cast<abstract_actor_ptr>(whom), port);
parent().notify<hook::actor_published>(whom, port);
},
[=](get_atom, network::native_socket fd, int64_t request_id,
[=](get_atom, connection_handle hdl, int64_t request_id,
actor client, std::set<std::string>& expected_ifs) {
auto hdl = add_tcp_scribe(fd);
assign_tcp_scribe(hdl);
auto& ctx = m_ctx[hdl];
ctx.hdl = hdl;
// PODs are not movable, so passing expected_ifs to the ctor would cause
......
......@@ -361,10 +361,18 @@ connection_handle broker::add_tcp_scribe(const std::string& hst, uint16_t prt) {
return backend().add_tcp_scribe(this, hst, prt);
}
void broker::assign_tcp_scribe(connection_handle hdl) {
backend().assign_tcp_scribe(this, hdl);
}
connection_handle broker::add_tcp_scribe(network::native_socket fd) {
return backend().add_tcp_scribe(this, fd);
}
void broker::assign_tcp_doorman(accept_handle hdl) {
backend().assign_tcp_doorman(this, hdl);
}
std::pair<accept_handle, uint16_t>
broker::add_tcp_doorman(uint16_t port, const char* in, bool reuse_addr) {
return backend().add_tcp_doorman(this, port, in, reuse_addr);
......
......@@ -764,6 +764,17 @@ accept_handle default_multiplexer::add_tcp_doorman(broker* self,
return ptr->hdl();
}
connection_handle default_multiplexer::new_tcp_scribe(const std::string& host,
uint16_t port) {
auto fd = new_ipv4_connection_impl(host, port);
return connection_handle::from_int(int64_from_native_socket(fd));
}
void default_multiplexer::assign_tcp_scribe(broker* ptr,
connection_handle hdl) {
add_tcp_scribe(ptr, static_cast<native_socket>(hdl.id()));
}
connection_handle default_multiplexer::add_tcp_scribe(broker* self,
native_socket fd) {
return add_tcp_scribe(self, default_socket{*this, fd});
......@@ -775,6 +786,19 @@ connection_handle default_multiplexer::add_tcp_scribe(broker* self,
return add_tcp_scribe(self, new_ipv4_connection(host, port));
}
std::pair<accept_handle, uint16_t>
default_multiplexer::new_tcp_doorman(uint16_t port, const char* in,
bool reuse_addr) {
auto res = new_ipv4_acceptor_impl(port, in, reuse_addr);
return {accept_handle::from_int(int64_from_native_socket(res.first)),
res.second};
}
void default_multiplexer::assign_tcp_doorman(broker* ptr, accept_handle hdl) {
add_tcp_doorman(ptr, static_cast<native_socket>(hdl.id()));
}
accept_handle default_multiplexer::add_tcp_doorman(broker* self,
native_socket fd) {
return add_tcp_doorman(self, default_socket_acceptor{*this, fd});
......
......@@ -37,7 +37,6 @@
#include "caf/io/middleman.hpp"
#include "caf/io/basp_broker.hpp"
#include "caf/io/system_messages.hpp"
#include "caf/io/network/default_multiplexer.hpp"
#include "caf/detail/logging.hpp"
#include "caf/detail/ripemd_160.hpp"
......@@ -183,9 +182,10 @@ using middleman_actor_base = middleman_actor::extend<
class middleman_actor_impl : public middleman_actor_base::base {
public:
middleman_actor_impl(actor default_broker)
middleman_actor_impl(middleman& mref, actor default_broker)
: m_broker(default_broker),
m_next_request_id(0) {
m_next_request_id(0),
m_parent(mref) {
// nop
}
......@@ -244,15 +244,15 @@ class middleman_actor_impl : public middleman_actor_base::base {
either<ok_atom, uint16_t>::or_else<error_atom, std::string>
put(const actor_addr& whom, uint16_t port,
const char* in = nullptr, bool reuse_addr = false) {
network::native_socket fd;
accept_handle hdl;
uint16_t actual_port;
try {
// treat empty strings like nullptr
if (in != nullptr && in[0] == '\0') {
in = nullptr;
}
auto res = network::new_ipv4_acceptor_impl(port, in, reuse_addr);
fd = res.first;
auto res = m_parent.backend().new_tcp_doorman(port, in, reuse_addr);
hdl = res.first;
actual_port = res.second;
}
catch (bind_failure& err) {
......@@ -261,7 +261,7 @@ class middleman_actor_impl : public middleman_actor_base::base {
catch (network_error& err) {
return {error_atom{}, std::string("network_error: ") + err.what()};
}
send(m_broker, put_atom{}, fd, whom, actual_port);
send(m_broker, put_atom{}, hdl, whom, actual_port);
return {ok_atom{}, actual_port};
}
......@@ -269,9 +269,9 @@ class middleman_actor_impl : public middleman_actor_base::base {
std::set<std::string> expected_ifs) {
get_op_promise result = make_response_promise();
try {
auto fd = network::new_ipv4_connection_impl(hostname, port);
auto hdl = m_parent.backend().new_tcp_scribe(hostname, port);
auto req_id = m_next_request_id++;
send(m_broker, get_atom{}, fd, req_id,
send(m_broker, get_atom{}, hdl, req_id,
actor{this}, std::move(expected_ifs));
m_pending_requests.insert(std::make_pair(req_id, result));
}
......@@ -286,6 +286,7 @@ class middleman_actor_impl : public middleman_actor_base::base {
actor m_broker;
int64_t m_next_request_id;
middleman& m_parent;
std::map<int64_t, get_op_promise> m_pending_requests;
};
......@@ -326,7 +327,7 @@ void middleman::initialize() {
do_announce<new_connection_msg>("caf::io::new_connection_msg");
do_announce<new_data_msg>("caf::io::new_data_msg");
actor mgr = get_named_broker<basp_broker>(atom("_BASP"));
m_manager = spawn_typed<middleman_actor_impl, detached + hidden>(mgr);
m_manager = spawn_typed<middleman_actor_impl, detached + hidden>(*this, mgr);
}
void middleman::stop() {
......
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