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

Fix ASIO multiplexer

parent 97d51137
......@@ -13,7 +13,6 @@ set (LIBCAF_IO_SRCS
src/broker.cpp
src/default_multiplexer.cpp
src/doorman.cpp
src/max_msg_size.cpp
src/middleman.cpp
src/hook.cpp
src/interfaces.cpp
......
......@@ -40,7 +40,7 @@ namespace io {
namespace network {
/// Low-level backend for IO multiplexing.
using io_backend = boost::asio::io_service;
using io_service = boost::asio::io_service;
/// Low-level socket type used as default.
using default_socket = boost::asio::ip::tcp::socket;
......@@ -81,7 +81,7 @@ public:
std::pair<accept_handle, uint16_t>
add_tcp_doorman(abstract_broker*, uint16_t port, const char* in, bool rflag) override;
void dispatch_runnable(runnable_ptr ptr) override;
void exec_later(resumable* ptr) override;
asio_multiplexer(actor_system* sys);
......@@ -93,12 +93,12 @@ public:
boost::asio::io_service* pimpl() override;
private:
inline boost::asio::io_service& backend() {
return backend_;
inline boost::asio::io_service& service() {
return service_;
}
io_backend backend_;
private:
io_service service_;
std::mutex mtx_sockets_;
std::mutex mtx_acceptors_;
std::map<int64_t, default_socket> unassigned_sockets_;
......@@ -131,16 +131,27 @@ public:
/// A buffer class providing a compatible interface to `std::vector`.
using buffer_type = std::vector<char>;
stream(io_backend& backend) : writing_(false), fd_(backend) {
stream(asio_multiplexer& ref)
: writing_(false),
fd_(ref.service()),
backend_(ref) {
configure_read(receive_policy::at_most(1024));
}
/// Returns the IO socket.
Socket& socket_handle() { return fd_; }
const Socket& socket_handle() const { return fd_; }
Socket& socket_handle() {
return fd_;
}
/// Returns the IO socket.
const Socket& socket_handle() const {
return fd_;
}
/// Initializes this stream, setting the socket handle to `fd`.
void init(Socket fd) { fd_ = std::move(fd); }
void init(Socket fd) {
fd_ = std::move(fd);
}
/// Starts reading data from the socket, forwarding incoming data to `mgr`.
void start(const manager_ptr& mgr) {
......@@ -168,9 +179,13 @@ public:
/// Returns the write buffer of this stream.
/// @warning Must not be modified outside the IO multiplexers event loop
/// once the stream has been started.
buffer_type& wr_buf() { return wr_offline_buf_; }
buffer_type& wr_buf() {
return wr_offline_buf_;
}
buffer_type& rd_buf() { return rd_buf_; }
buffer_type& rd_buf() {
return rd_buf_;
}
/// Sends the content of the write buffer, calling the `io_failure`
/// member function of `mgr` in case of an error.
......@@ -186,26 +201,29 @@ public:
/// Closes the network connection, thus stopping this stream.
void stop() {
CAF_LOGMF(CAF_TRACE, "");
CAF_LOG_TRACE("");
fd_.close();
}
void stop_reading() {
CAF_LOGMF(CAF_TRACE, "");
CAF_LOG_TRACE("");
boost::system::error_code ec; // ignored
fd_.shutdown(boost::asio::ip::tcp::socket::shutdown_receive, ec);
}
asio_multiplexer& backend() {
return backend_;
}
private:
void read_loop(const manager_ptr& mgr) {
auto cb = [=](const boost::system::error_code& ec, size_t read_bytes) {
CAF_LOGC(CAF_TRACE, "caf::io::network::stream", "read_loop$cb",
CAF_ARG(this));
CAF_LOG_TRACE("");
if (! ec) {
mgr->consume(rd_buf_.data(), read_bytes);
mgr->consume(&backend(), rd_buf_.data(), read_bytes);
read_loop(mgr);
} else {
mgr->io_failure(operation::read);
mgr->io_failure(&backend(), operation::read);
}
};
switch (rd_flag_) {
......@@ -244,17 +262,14 @@ private:
boost::asio::async_write(
fd_, boost::asio::buffer(wr_buf_),
[=](const boost::system::error_code& ec, size_t nb) {
CAF_LOGC(CAF_TRACE, "caf::io::network::stream", "write_loop$lambda",
CAF_ARG(this));
CAF_LOG_TRACE("");
static_cast<void>(nb); // silence compiler warning
if (! ec) {
CAF_LOGC(CAF_DEBUG, "caf::io::network::stream", "write_loop$lambda",
nb << " bytes sent");
CAF_LOG_DEBUG(CAF_ARG(nb));
write_loop(mgr);
} else {
CAF_LOGC(CAF_DEBUG, "caf::io::network::stream", "write_loop$lambda",
"error during send: " << ec.message());
mgr->io_failure(operation::read);
CAF_LOG_DEBUG(CAF_ARG(ec.message()));
mgr->io_failure(&backend(), operation::read);
writing_ = false;
}
});
......@@ -264,18 +279,17 @@ private:
fd_.async_read_some(boost::asio::buffer(rd_buf_.data() + collected_bytes,
rd_buf_.size() - collected_bytes),
[=](const boost::system::error_code& ec, size_t nb) {
CAF_LOGC(CAF_TRACE, "caf::io::network::stream", "collect_data$lambda",
CAF_ARG(this));
CAF_LOG_TRACE(CAF_ARG(nb));
if (! ec) {
auto sum = collected_bytes + nb;
if (sum >= rd_size_) {
mgr->consume(rd_buf_.data(), sum);
mgr->consume(&backend(), rd_buf_.data(), sum);
read_loop(mgr);
} else {
collect_data(mgr, sum);
}
} else {
mgr->io_failure(operation::write);
mgr->io_failure(&backend(), operation::write);
}
});
}
......@@ -287,6 +301,7 @@ private:
buffer_type rd_buf_;
buffer_type wr_buf_;
buffer_type wr_offline_buf_;
asio_multiplexer& backend_;
};
/// An acceptor is responsible for accepting incoming connections.
......@@ -302,42 +317,60 @@ public:
/// A smart pointer to an acceptor manager.
using manager_ptr = intrusive_ptr<manager_type>;
acceptor(asio_multiplexer& am, io_backend& io)
: backend_(am), accept_fd_(io), fd_(io) {}
acceptor(asio_multiplexer& am, io_service& io)
: backend_(am),
accept_fd_(io),
fd_(io) {
// nop
}
/// Returns the `multiplexer` this acceptor belongs to.
inline asio_multiplexer& backend() { return backend_; }
asio_multiplexer& backend() { return backend_; }
/// Returns the IO socket.
SocketAcceptor& socket_handle() {
return accept_fd_;
}
/// Returns the IO socket.
inline SocketAcceptor& socket_handle() { return accept_fd_; }
inline const SocketAcceptor& socket_handle() const { return accept_fd_; }
const SocketAcceptor& socket_handle() const {
return accept_fd_;
}
/// Returns the accepted socket. This member function should
/// be called only from the `new_connection` callback.
inline socket_type& accepted_socket() { return fd_; }
inline socket_type& accepted_socket() {
return fd_;
}
/// Initializes this acceptor, setting the socket handle to `fd`.
void init(SocketAcceptor fd) { accept_fd_ = std::move(fd); }
void init(SocketAcceptor fd) {
accept_fd_ = std::move(fd);
}
/// Starts this acceptor, forwarding all incoming connections to
/// `manager`. The intrusive pointer will be released after the
/// acceptor has been closed or an IO error occured.
void start(const manager_ptr& mgr) { accept_loop(mgr); }
void start(const manager_ptr& mgr) {
accept_loop(mgr);
}
/// Closes the network connection, thus stopping this acceptor.
void stop() { accept_fd_.close(); }
void stop() {
accept_fd_.close();
}
private:
void accept_loop(const manager_ptr& mgr) {
accept_fd_.async_accept(fd_, [=](const boost::system::error_code& ec) {
CAF_LOGMF(CAF_TRACE, "");
CAF_LOG_TRACE("");
if (! ec) {
mgr->new_connection(); // probably moves fd_
// reset fd_ for next accept operation
fd_ = socket_type{accept_fd_.get_io_service()};
accept_loop(mgr);
} else {
mgr->io_failure(operation::read);
mgr->io_failure(&backend(), operation::read);
}
});
}
......
......@@ -23,6 +23,8 @@
#include "caf/io/broker.hpp"
#include "caf/io/middleman.hpp"
#include "caf/scheduler/abstract_coordinator.hpp"
#include "caf/io/network/asio_multiplexer.hpp"
namespace caf {
......@@ -33,7 +35,7 @@ namespace {
/// A wrapper for the supervisor backend provided by boost::asio.
struct asio_supervisor : public multiplexer::supervisor {
explicit asio_supervisor(io_backend& iob) : work(iob) {
explicit asio_supervisor(io_service& iob) : work(iob) {
// nop
}
......@@ -43,7 +45,7 @@ private:
} // namespace anonymous
default_socket new_tcp_connection(io_backend& backend, const std::string& host,
default_socket new_tcp_connection(io_service& backend, const std::string& host,
uint16_t port) {
default_socket fd{backend};
using boost::asio::ip::tcp;
......@@ -94,7 +96,7 @@ void ip_bind(default_socket_acceptor& fd, uint16_t port,
connection_handle asio_multiplexer::new_tcp_scribe(const std::string& host,
uint16_t port) {
default_socket fd{new_tcp_connection(backend(), host, port)};
default_socket fd{new_tcp_connection(service(), host, port)};
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)));
......@@ -117,10 +119,10 @@ connection_handle asio_multiplexer::add_tcp_scribe(abstract_broker* self,
CAF_LOG_TRACE("");
class impl : public scribe {
public:
impl(abstract_broker* ptr, Socket&& s)
impl(abstract_broker* ptr, asio_multiplexer& ref, Socket&& s)
: scribe(ptr, network::conn_hdl_from_socket(s)),
launched_(false),
stream_(s.get_io_service()) {
stream_(ref) {
stream_.init(std::move(s));
}
void configure_read(receive_policy::config config) override {
......@@ -139,7 +141,7 @@ connection_handle asio_multiplexer::add_tcp_scribe(abstract_broker* self,
void stop_reading() override {
CAF_LOG_TRACE("");
stream_.stop_reading();
detach(false);
detach(&stream_.backend(), false);
}
void flush() override {
CAF_LOG_TRACE("");
......@@ -162,7 +164,7 @@ connection_handle asio_multiplexer::add_tcp_scribe(abstract_broker* self,
bool launched_;
stream<Socket> stream_;
};
auto ptr = make_counted<impl>(self, std::move(sock));
auto ptr = make_counted<impl>(self, *this, std::move(sock));
self->add_scribe(ptr);
return ptr->hdl();
}
......@@ -171,7 +173,7 @@ connection_handle asio_multiplexer::add_tcp_scribe(abstract_broker* self,
native_socket fd) {
CAF_LOG_TRACE(CAF_ARG(self) << ", " << CAF_ARG(fd));
boost::system::error_code ec;
default_socket sock{backend()};
default_socket sock{service()};
sock.assign(boost::asio::ip::tcp::v6(), fd, ec);
if (ec) {
sock.assign(boost::asio::ip::tcp::v4(), fd, ec);
......@@ -186,13 +188,13 @@ 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));
return add_tcp_scribe(self, new_tcp_connection(backend(), host, port));
return add_tcp_scribe(self, new_tcp_connection(service(), host, port));
}
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"));
default_socket_acceptor fd{backend()};
default_socket_acceptor fd{service()};
ip_bind(fd, port, in, rflag);
auto id = int64_from_native_socket(fd.native_handle());
auto assigned_port = fd.local_endpoint().port();
......@@ -229,12 +231,12 @@ asio_multiplexer::add_tcp_doorman(abstract_broker* self,
auto& am = acceptor_.backend();
msg().handle
= am.add_tcp_scribe(parent(), std::move(acceptor_.accepted_socket()));
invoke_mailbox_element();
invoke_mailbox_element(&am);
}
void stop_reading() override {
CAF_LOG_TRACE("");
acceptor_.stop();
detach(false);
detach(&acceptor_.backend(), false);
}
void launch() override {
CAF_LOG_TRACE("");
......@@ -257,7 +259,7 @@ asio_multiplexer::add_tcp_doorman(abstract_broker* self,
accept_handle asio_multiplexer::add_tcp_doorman(abstract_broker* self,
native_socket fd) {
default_socket_acceptor sock{backend()};
default_socket_acceptor sock{service()};
boost::system::error_code ec;
sock.assign(boost::asio::ip::tcp::v6(), fd, ec);
if (ec) {
......@@ -272,16 +274,34 @@ accept_handle asio_multiplexer::add_tcp_doorman(abstract_broker* self,
std::pair<accept_handle, uint16_t>
asio_multiplexer::add_tcp_doorman(abstract_broker* self, uint16_t port, const char* in,
bool rflag) {
default_socket_acceptor fd{backend()};
default_socket_acceptor fd{service()};
ip_bind(fd, port, in, rflag);
auto p = fd.local_endpoint().port();
return {add_tcp_doorman(self, std::move(fd)), p};
}
void asio_multiplexer::dispatch_runnable(runnable_ptr ptr) {
backend().post([=]() {
ptr->run();
});
void asio_multiplexer::exec_later(resumable* rptr) {
switch (rptr->subtype()) {
case resumable::io_actor:
case resumable::function_object: {
intrusive_ptr<resumable> ptr{rptr};
service().post([=]() {
switch (ptr->resume(this, max_throughput())) {
case resumable::resume_later:
exec_later(ptr.get());
break;
case resumable::done:
intrusive_ptr_release(ptr.get());
break;
default:
; // ignored
}
});
break;
}
default:
system().scheduler().enqueue(rptr);
}
}
asio_multiplexer::asio_multiplexer(actor_system* sys) : multiplexer(sys) {
......@@ -293,20 +313,19 @@ asio_multiplexer::~asio_multiplexer() {
}
multiplexer::supervisor_ptr asio_multiplexer::make_supervisor() {
return std::unique_ptr<asio_supervisor>(new asio_supervisor(backend()));
return std::unique_ptr<asio_supervisor>(new asio_supervisor(service()));
}
void asio_multiplexer::run() {
CAF_LOG_TRACE("asio-based multiplexer");
boost::system::error_code ec;
backend().run(ec);
if (ec) {
service().run(ec);
if (ec)
throw std::runtime_error(ec.message());
}
}
boost::asio::io_service* asio_multiplexer::pimpl() {
return &backend_;
return &service_;
}
} // namesapce network
......
......@@ -219,8 +219,8 @@ CAF_TEST(test_broker) {
auto use_asio = r.opts.count("use-asio") > 0;
actor_system_config cfg;
# ifdef CAF_USE_ASIO
if (use_asio) {
cfg.load<io::middleman, io::network::asio_multiplexer>());
if (use_asio)
cfg.load<io::middleman, io::network::asio_multiplexer>();
else
# endif // CAF_USE_ASIO
cfg.load<io::middleman>();
......
......@@ -586,8 +586,8 @@ CAF_TEST(remote_actors) {
cfg.add_message_type<actor_vector>("actor_vector");
auto use_asio = r.opts.count("use-asio") > 0;
# ifdef CAF_USE_ASIO
if (use_asio) {
cfg.load<io::middleman, io::network::asio_multiplexer>());
if (use_asio)
cfg.load<io::middleman, io::network::asio_multiplexer>();
else
# endif // CAF_USE_ASIO
cfg.load<io::middleman>();
......
......@@ -121,7 +121,7 @@ CAF_TEST(remote_spawn) {
auto use_asio = r.opts.count("use-asio") > 0;
# ifdef CAF_USE_ASIO
if (use_asio)
cfg.load<io::middleman, io::network::asio_multiplexer>());
cfg.load<io::middleman, io::network::asio_multiplexer>();
else
# endif // CAF_USE_ASIO
cfg.load<io::middleman>();
......
......@@ -234,8 +234,8 @@ CAF_TEST(test_typed_broker) {
actor_system_config cfg;
auto use_asio = r.opts.count("use-asio") > 0;
# ifdef CAF_USE_ASIO
if (use_asio) {
cfg.load<io::middleman, io::network::asio_multiplexer>());
if (use_asio)
cfg.load<io::middleman, io::network::asio_multiplexer>();
else
# endif // CAF_USE_ASIO
cfg.load<io::middleman>();
......
......@@ -129,8 +129,8 @@ CAF_TEST(test_typed_remote_actor) {
.add_message_type<pong>("pong");
auto use_asio = r.opts.count("use-asio") > 0;
# ifdef CAF_USE_ASIO
if (use_asio) {
cfg.load<io::middleman, io::network::asio_multiplexer>());
if (use_asio)
cfg.load<io::middleman, io::network::asio_multiplexer>();
else
# endif // CAF_USE_ASIO
cfg.load<io::middleman>();
......
......@@ -74,7 +74,7 @@ CAF_TEST(unpublishing) {
auto argc = test::engine::argc();
auto argv = test::engine::argv();
if (argc == 1 && strcmp(argv[0], "--use-asio") == 0)
cfg.load<io::middleman, io::network::asio_multiplexer>());
cfg.load<io::middleman, io::network::asio_multiplexer>();
else
# endif // CAF_USE_ASIO
cfg.load<io::middleman>();
......
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