Commit 7c2f8d0a authored by Dominik Charousset's avatar Dominik Charousset

Fix ASIO multiplexer

parent 098a4343
...@@ -55,31 +55,33 @@ public: ...@@ -55,31 +55,33 @@ public:
friend class io::middleman; friend class io::middleman;
friend class supervisor; friend class supervisor;
connection_handle new_tcp_scribe(const std::string&, uint16_t) override; expected<connection_handle>
new_tcp_scribe(const std::string&, uint16_t) override;
void assign_tcp_scribe(abstract_broker*, connection_handle hdl) override; expected<void>
assign_tcp_scribe(abstract_broker*, connection_handle hdl) override;
template <class Socket> template <class Socket>
connection_handle add_tcp_scribe(abstract_broker*, Socket&& sock); connection_handle add_tcp_scribe(abstract_broker*, Socket&& sock);
connection_handle add_tcp_scribe(abstract_broker*, native_socket fd) override; connection_handle add_tcp_scribe(abstract_broker*, native_socket fd) override;
connection_handle add_tcp_scribe(abstract_broker*, const std::string& host, expected<connection_handle>
uint16_t port) override; add_tcp_scribe(abstract_broker*, const std::string&, uint16_t) override;
std::pair<accept_handle, uint16_t> new_tcp_doorman(uint16_t p, const char* in, expected<std::pair<accept_handle, uint16_t>>
bool rflag) override; new_tcp_doorman(uint16_t p, const char* in, bool rflag) override;
void assign_tcp_doorman(abstract_broker*, accept_handle hdl) override; expected<void>
assign_tcp_doorman(abstract_broker*, accept_handle hdl) override;
accept_handle add_tcp_doorman(abstract_broker*, accept_handle add_tcp_doorman(abstract_broker*,
asio_tcp_socket_acceptor&& sock); asio_tcp_socket_acceptor&& sock);
accept_handle add_tcp_doorman(abstract_broker*, native_socket fd) override; accept_handle add_tcp_doorman(abstract_broker*, native_socket fd) override;
std::pair<accept_handle, uint16_t> expected<std::pair<accept_handle, uint16_t>>
add_tcp_doorman(abstract_broker*, uint16_t port, add_tcp_doorman(abstract_broker*, uint16_t, const char*, bool) override;
const char* in, bool rflag) override;
void exec_later(resumable* ptr) override; void exec_later(resumable* ptr) override;
......
...@@ -43,74 +43,76 @@ private: ...@@ -43,74 +43,76 @@ private:
} // namespace anonymous } // namespace anonymous
asio_tcp_socket new_tcp_connection(io_service& backend, const std::string& host, expected<asio_tcp_socket>
uint16_t port) { new_tcp_connection(io_service& ios, const std::string& host, uint16_t port) {
asio_tcp_socket fd{backend}; asio_tcp_socket fd{ios};
using boost::asio::ip::tcp; using boost::asio::ip::tcp;
tcp::resolver r(fd.get_io_service()); tcp::resolver r(fd.get_io_service());
tcp::resolver::query q(host, std::to_string(port)); tcp::resolver::query q(host, std::to_string(port));
boost::system::error_code ec; boost::system::error_code ec;
auto i = r.resolve(q, ec); auto i = r.resolve(q, ec);
if (ec) { if (ec)
CAF_RAISE_ERROR("could not resolve host: " + host); return make_error(sec::cannot_connect_to_node);
}
boost::asio::connect(fd, i, ec); boost::asio::connect(fd, i, ec);
if (ec) { if (ec)
CAF_RAISE_ERROR("could not connect to host: " + host + return make_error(sec::cannot_connect_to_node);
":" + std::to_string(port)); return asio_tcp_socket{std::move(fd)};
}
return fd;
} }
void ip_bind(asio_tcp_socket_acceptor& fd, uint16_t port, error ip_bind(asio_tcp_socket_acceptor& fd, uint16_t port,
const char* addr, bool reuse_addr) { const char* addr, bool reuse_addr) {
CAF_LOG_TRACE(CAF_ARG(port) << CAF_ARG(reuse_addr)); CAF_LOG_TRACE(CAF_ARG(port) << CAF_ARG(reuse_addr));
using boost::asio::ip::tcp; using boost::asio::ip::tcp;
try { auto bind_and_listen = [&](tcp::endpoint& ep) -> error {
auto bind_and_listen = [&](tcp::endpoint& ep) { CAF_LOG_DEBUG("created IP endpoint:" << CAF_ARG(ep.address().to_string())
CAF_LOG_DEBUG("created IP endpoint:" << CAF_ARG(ep.address().to_string()) << CAF_ARG(ep.port()));
<< CAF_ARG(ep.port())); boost::system::error_code ec;
fd.open(ep.protocol()); fd.open(ep.protocol());
if (reuse_addr) if (reuse_addr) {
fd.set_option(tcp::acceptor::reuse_address(reuse_addr)); fd.set_option(tcp::acceptor::reuse_address(reuse_addr), ec);
fd.bind(ep); if (ec)
fd.listen(); return sec::cannot_open_port;
};
if (addr) {
CAF_LOG_DEBUG(CAF_ARG(addr));
tcp::endpoint ep(boost::asio::ip::address::from_string(addr), port);
CAF_LOG_DEBUG("got 'em");
bind_and_listen(ep);
} else {
CAF_LOG_DEBUG("addr = nullptr");
tcp::endpoint ep(tcp::v6(), port);
bind_and_listen(ep);
} }
} catch (boost::system::system_error& se) { fd.bind(ep, ec);
if (se.code() == boost::system::errc::address_in_use) { if (ec)
CAF_RAISE_ERROR(se.code().message()); return sec::cannot_open_port;
} fd.listen();
CAF_RAISE_ERROR(se.code().message()); return {};
};
if (addr) {
CAF_LOG_DEBUG(CAF_ARG(addr));
tcp::endpoint ep(boost::asio::ip::address::from_string(addr), port);
CAF_LOG_DEBUG("got 'em");
return bind_and_listen(ep);
} else {
CAF_LOG_DEBUG("addr = nullptr");
tcp::endpoint ep(tcp::v6(), port);
return bind_and_listen(ep);
} }
} }
connection_handle asio_multiplexer::new_tcp_scribe(const std::string& host, expected<connection_handle>
uint16_t port) { asio_multiplexer::new_tcp_scribe(const std::string& host, uint16_t port) {
asio_tcp_socket fd{new_tcp_connection(service(), host, 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()); auto id = int64_from_native_socket(fd.native_handle());
std::lock_guard<std::mutex> lock(mtx_sockets_); std::lock_guard<std::mutex> lock(mtx_sockets_);
unassigned_sockets_.insert(std::make_pair(id, std::move(fd))); unassigned_sockets_.insert(std::make_pair(id, std::move(fd)));
return connection_handle::from_int(id); return connection_handle::from_int(id);
} }
void asio_multiplexer::assign_tcp_scribe(abstract_broker* self, expected<void> asio_multiplexer::assign_tcp_scribe(abstract_broker* self,
connection_handle hdl) { connection_handle hdl) {
std::lock_guard<std::mutex> lock(mtx_sockets_); std::lock_guard<std::mutex> lock(mtx_sockets_);
auto itr = unassigned_sockets_.find(hdl.id()); auto itr = unassigned_sockets_.find(hdl.id());
if (itr != unassigned_sockets_.end()) { if (itr != unassigned_sockets_.end()) {
add_tcp_scribe(self, std::move(itr->second)); add_tcp_scribe(self, std::move(itr->second));
unassigned_sockets_.erase(itr); unassigned_sockets_.erase(itr);
return unit;
} }
return sec::failed_to_assign_scribe_from_handle;
} }
template <class Socket> template <class Socket>
...@@ -186,26 +188,31 @@ connection_handle asio_multiplexer::add_tcp_scribe(abstract_broker* self, ...@@ -186,26 +188,31 @@ connection_handle asio_multiplexer::add_tcp_scribe(abstract_broker* self,
return add_tcp_scribe(self, std::move(sock)); return add_tcp_scribe(self, std::move(sock));
} }
connection_handle asio_multiplexer::add_tcp_scribe(abstract_broker* self, expected<connection_handle>
const std::string& host, asio_multiplexer::add_tcp_scribe(abstract_broker* self,
uint16_t port) { const std::string& host, uint16_t port) {
CAF_LOG_TRACE(CAF_ARG(self) << ", " << CAF_ARG(host) << ":" << CAF_ARG(port)); CAF_LOG_TRACE(CAF_ARG(self) << ", " << CAF_ARG(host) << ":" << CAF_ARG(port));
return add_tcp_scribe(self, new_tcp_connection(service(), host, port)); auto conn = new_tcp_connection(service(), host, port);
if (! conn)
return std::move(conn.error());
return add_tcp_scribe(self, std::move(*conn));
} }
std::pair<accept_handle, uint16_t> expected<std::pair<accept_handle, uint16_t>>
asio_multiplexer::new_tcp_doorman(uint16_t port, const char* in, bool rflag) { asio_multiplexer::new_tcp_doorman(uint16_t port, const char* in, bool rflag) {
CAF_LOG_TRACE(CAF_ARG(port) << ", addr = " << (in ? in : "nullptr")); CAF_LOG_TRACE(CAF_ARG(port) << ", addr = " << (in ? in : "nullptr"));
asio_tcp_socket_acceptor fd{service()}; asio_tcp_socket_acceptor fd{service()};
ip_bind(fd, port, in, rflag); auto err = ip_bind(fd, port, in, rflag);
if (err)
return err;
auto id = int64_from_native_socket(fd.native_handle()); auto id = int64_from_native_socket(fd.native_handle());
auto assigned_port = fd.local_endpoint().port(); auto assigned_port = fd.local_endpoint().port();
std::lock_guard<std::mutex> lock(mtx_acceptors_); std::lock_guard<std::mutex> lock(mtx_acceptors_);
unassigned_acceptors_.insert(std::make_pair(id, std::move(fd))); unassigned_acceptors_.insert(std::make_pair(id, std::move(fd)));
return {accept_handle::from_int(id), assigned_port}; return std::make_pair(accept_handle::from_int(id), assigned_port);
} }
void asio_multiplexer::assign_tcp_doorman(abstract_broker* self, expected<void> asio_multiplexer::assign_tcp_doorman(abstract_broker* self,
accept_handle hdl) { accept_handle hdl) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
std::lock_guard<std::mutex> lock(mtx_acceptors_); std::lock_guard<std::mutex> lock(mtx_acceptors_);
...@@ -213,7 +220,9 @@ void asio_multiplexer::assign_tcp_doorman(abstract_broker* self, ...@@ -213,7 +220,9 @@ void asio_multiplexer::assign_tcp_doorman(abstract_broker* self,
if (itr != unassigned_acceptors_.end()) { if (itr != unassigned_acceptors_.end()) {
add_tcp_doorman(self, std::move(itr->second)); add_tcp_doorman(self, std::move(itr->second));
unassigned_acceptors_.erase(itr); unassigned_acceptors_.erase(itr);
return unit;
} }
return sec::failed_to_assign_doorman_from_handle;
} }
accept_handle accept_handle
...@@ -273,23 +282,26 @@ accept_handle asio_multiplexer::add_tcp_doorman(abstract_broker* self, ...@@ -273,23 +282,26 @@ accept_handle asio_multiplexer::add_tcp_doorman(abstract_broker* self,
return add_tcp_doorman(self, std::move(sock)); return add_tcp_doorman(self, std::move(sock));
} }
std::pair<accept_handle, uint16_t> expected<std::pair<accept_handle, uint16_t>>
asio_multiplexer::add_tcp_doorman(abstract_broker* self, uint16_t port, asio_multiplexer::add_tcp_doorman(abstract_broker* self, uint16_t port,
const char* in, bool rflag) { const char* in, bool rflag) {
CAF_LOG_TRACE(CAF_ARG(port) << CAF_ARG(in) << CAF_ARG(rflag)); CAF_LOG_TRACE(CAF_ARG(port) << CAF_ARG(in) << CAF_ARG(rflag));
asio_tcp_socket_acceptor fd{service()}; asio_tcp_socket_acceptor fd{service()};
ip_bind(fd, port, in, rflag); auto err = ip_bind(fd, port, in, rflag);
if (err)
return err;
auto p = fd.local_endpoint().port(); auto p = fd.local_endpoint().port();
return {add_tcp_doorman(self, std::move(fd)), p}; return std::make_pair(add_tcp_doorman(self, std::move(fd)), p);
} }
void asio_multiplexer::exec_later(resumable* rptr) { void asio_multiplexer::exec_later(resumable* rptr) {
auto mt = system().config().scheduler_max_throughput;
switch (rptr->subtype()) { switch (rptr->subtype()) {
case resumable::io_actor: case resumable::io_actor:
case resumable::function_object: { case resumable::function_object: {
intrusive_ptr<resumable> ptr{rptr, false}; intrusive_ptr<resumable> ptr{rptr, false};
service().post([=]() mutable { service().post([=]() mutable {
switch (ptr->resume(this, max_throughput())) { switch (ptr->resume(this, mt)) {
case resumable::resume_later: case resumable::resume_later:
exec_later(ptr.release()); exec_later(ptr.release());
break; break;
......
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