Unverified Commit 206a03e8 authored by Dominik Charousset's avatar Dominik Charousset Committed by GitHub

Merge pull request #714

Remove ASIO-based multiplexer
parents 2f562cfb 38a661d8
......@@ -389,23 +389,6 @@ to_int_value(CAF_NO_EXCEPTIONS)
to_int_value(CAF_NO_MEM_MANAGEMENT)
to_int_value(CAF_ENABLE_RUNTIME_CHECKS)
# find boost asio if the asio multiplexer should be used for testing
if(CAF_USE_ASIO)
set(Boost_USE_MULTITHREADED ON)
find_package(Boost COMPONENTS system)
if(Boost_FOUND)
include_directories(${Boost_INCLUDE_DIRS})
set(LD_DIRS ${LD_DIRS} ${Boost_LIBRARIES})
list(APPEND LDFLAGS ${Boost_SYSTEM_LIBRARY})
set(CAF_USE_ASIO_INT 1)
else()
set(CAF_USE_ASIO no)
set(CAF_USE_ASIO_INT -1)
endif()
else()
set(CAF_USE_ASIO_INT -1)
endif()
configure_file("${CMAKE_CURRENT_SOURCE_DIR}/cmake/build_config.hpp.in"
"${CMAKE_CURRENT_SOURCE_DIR}/libcaf_core/caf/detail/build_config.hpp"
IMMEDIATE @ONLY)
......@@ -639,11 +622,6 @@ if(NOT CAF_NO_UNIT_TESTS)
string(REPLACE " " "_" test_name ${suite})
set(caf_test ${EXECUTABLE_OUTPUT_PATH}/caf-test)
add_test(${test_name} ${caf_test} -n -v 5 -s "${suite}" ${ARGN})
# add some extra unit testing options when testing ASIO as well
if(CAF_USE_ASIO AND "${suite}" MATCHES "^io_.+$" AND NOT "${suite}" MATCHES ".+_udp$")
add_test(${test_name}_asio ${caf_test} -n -v 5 -s
"${suite}" ${ARGN} -- "--caf#middleman.network-backend=asio")
endif()
endmacro ()
list(LENGTH suites num_suites)
message(STATUS "Found ${num_suites} test suites")
......
......@@ -69,7 +69,6 @@ unixOpts = "-DCAF_NO_PROTOBUF_EXAMPLES:BOOL=yes " +
"-DCAF_MORE_WARNINGS:BOOL=yes " +
"-DCAF_ENABLE_ADDRESS_SANITIZER:BOOL=yes " +
"-DCAF_ENABLE_RUNTIME_CHECKS:BOOL=yes " +
"-DCAF_USE_ASIO:BOOL=yes " +
"-DCAF_NO_BENCHMARKS:BOOL=yes " +
"-DOPENSSL_ROOT_DIR=/usr/local/opt/openssl " +
"-DOPENSSL_INCLUDE_DIR=/usr/local/opt/openssl/include"
......
......@@ -33,10 +33,6 @@
#define CAF_ENABLE_RUNTIME_CHECKS
#endif
#if @CAF_USE_ASIO_INT@ != -1
#define CAF_USE_ASIO
#endif
#if @CAF_NO_EXCEPTIONS_INT@ != -1
#define CAF_NO_EXCEPTIONS
#endif
......
......@@ -76,9 +76,6 @@ Usage: $0 [OPTION]... [VAR=VALUE]...
--core-only sets no-examples, no-opencl, no-openssl,
no-tools, no-python, and no-io
Testing:
--with-asio use ASIO multiplexer in unit tests
Debugging:
--with-runtime-checks build with requirement checks at runtime
--with-log-level=LVL build with debugging output, possible values:
......@@ -287,9 +284,6 @@ while [ $# -ne 0 ]; do
append_cache_entry CMAKE_OSX_ARCHITECTURES STRING "\$(ARCHS_STANDARD_32_64_BIT)"
append_cache_entry CAF_IOS_DEPLOYMENT_TARGET STRING "$optarg"
;;
--with-asio)
append_cache_entry CAF_USE_ASIO BOOL yes
;;
--with-log-level=*)
level=`echo "$optarg" | tr '[:lower:]' '[:upper:]'`
case $level in
......
......@@ -37,8 +37,6 @@ relaxed-sleep-duration=10ms
[middleman]
; configures whether MMs try to span a full mesh
enable-automatic-connections=false
; accepted alternative: 'asio' (only when compiling CAF with ASIO)
network-backend='default'
; application identifier of this node, prevents connection to other CAF
; instances with different identifier
app-identifier=""
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* Raphael Hiesgen <raphael.hiesgen (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#include "caf/config.hpp"
CAF_PUSH_WARNINGS
#include "boost/asio.hpp"
CAF_POP_WARNINGS
#include "caf/logger.hpp"
#include "caf/io/receive_policy.hpp"
#include "caf/io/network/multiplexer.hpp"
#include "caf/io/network/native_socket.hpp"
#include "caf/io/network/stream_manager.hpp"
#include "caf/io/network/acceptor_manager.hpp"
namespace caf {
namespace io {
namespace network {
/// Low-level error code.
using error_code = boost::system::error_code;
/// Low-level backend for IO multiplexing.
using io_service = boost::asio::io_service;
/// Low-level socket type used as default.
using asio_tcp_socket = boost::asio::ip::tcp::socket;
/// Low-level socket type used as default.
using asio_tcp_socket_acceptor = boost::asio::ip::tcp::acceptor;
class multiplexer_backend : public io_service {
// nop
};
/// A wrapper for the boost::asio multiplexer
class asio_multiplexer : public multiplexer {
public:
friend class io::middleman;
friend class supervisor;
scribe_ptr new_scribe(asio_tcp_socket&& sock);
scribe_ptr new_scribe(native_socket fd) override;
expected<scribe_ptr> new_tcp_scribe(const std::string& host,
uint16_t port) override;
doorman_ptr new_doorman(asio_tcp_socket_acceptor&& sock);
doorman_ptr new_doorman(native_socket fd) override;
expected<doorman_ptr> new_tcp_doorman(uint16_t port, const char* in,
bool reuse_addr) override;
datagram_servant_ptr new_datagram_servant(native_socket fd) override;
datagram_servant_ptr
new_datagram_servant_for_endpoint(native_socket fd,
const ip_endpoint& ep) override;
expected<datagram_servant_ptr>
new_remote_udp_endpoint(const std::string& host, uint16_t port) override;
expected<datagram_servant_ptr>
new_local_udp_endpoint(uint16_t port, const char* in = nullptr,
bool reuse_addr = false) override;
void exec_later(resumable* ptr) override;
asio_multiplexer(actor_system* sys);
~asio_multiplexer();
supervisor_ptr make_supervisor() override;
bool try_run_once() override;
void run_once() override;
void run() override;
multiplexer_backend* pimpl() override;
inline boost::asio::io_service& service() {
return service_;
}
private:
multiplexer_backend service_;
};
template <class T>
connection_handle conn_hdl_from_socket(T& sock) {
return connection_handle::from_int(
int64_from_native_socket(sock.native_handle()));
}
template <class T>
accept_handle accept_hdl_from_socket(T& sock) {
return accept_handle::from_int(
int64_from_native_socket(sock.native_handle()));
}
/// @relates manager
using manager_ptr = intrusive_ptr<manager>;
/// A stream capable of both reading and writing. The stream's input
/// data is forwarded to its {@link stream_manager manager}.
template <class Socket>
class asio_stream {
public:
/// A smart pointer to a stream manager.
using manager_ptr = intrusive_ptr<stream_manager>;
/// A buffer class providing a compatible interface to `std::vector`.
using buffer_type = std::vector<char>;
asio_stream(asio_multiplexer& ref)
: reading_(false),
writing_(false),
ack_writes_(false),
fd_(ref.service()),
backend_(ref),
rd_buf_ready_(false),
async_read_pending_(false) {
configure_read(receive_policy::at_most(1024));
}
/// Returns the IO socket.
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);
}
/// Starts reading data from the socket, forwarding incoming data to `mgr`.
void start(stream_manager* mgr) {
CAF_ASSERT(mgr != nullptr);
activate(mgr);
}
/// Configures how much data will be provided for the next `consume` callback.
/// @warning Must not be called outside the IO multiplexers event loop
/// once the stream has been started.
void configure_read(receive_policy::config config) {
rd_flag_ = config.first;
rd_size_ = config.second;
}
void ack_writes(bool enable) {
CAF_LOG_TRACE(CAF_ARG(enable));
ack_writes_ = enable;
}
/// Copies data to the write buffer.
/// @note Not thread safe.
void write(const void* buf, size_t num_bytes) {
CAF_LOG_TRACE(CAF_ARG(num_bytes));
auto first = reinterpret_cast<const char*>(buf);
auto last = first + num_bytes;
wr_offline_buf_.insert(wr_offline_buf_.end(), first, last);
}
/// 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& 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.
/// @warning Must not be called outside the IO multiplexers event loop
/// once the stream has been started.
void flush(const manager_ptr& mgr) {
CAF_ASSERT(mgr != nullptr);
if (!wr_offline_buf_.empty() && !writing_) {
writing_ = true;
write_loop(mgr);
}
}
/// Closes the network connection, thus stopping this stream.
void stop() {
CAF_LOG_TRACE("");
fd_.close();
}
void stop_reading() {
CAF_LOG_TRACE("");
error_code ec; // ignored
fd_.shutdown(boost::asio::ip::tcp::socket::shutdown_receive, ec);
}
asio_multiplexer& backend() {
return backend_;
}
/// Activates the stream.
void activate(stream_manager* mgr) {
reading_ = true;
read_loop(mgr);
}
/// Stops activity of the stream.
void passivate() {
reading_ = false;
}
private:
bool read_one(stream_manager* ptr, size_t num_bytes) {
if (!reading_) {
// broker was passivated while async read was on its way
rd_buf_ready_ = true;
// make sure buf size matches read_bytes in case of async_read
if (rd_buf_.size() != num_bytes)
rd_buf_.resize(num_bytes);
return false;
}
if (ptr->consume(&backend(), rd_buf_.data(), num_bytes))
return reading_;
return false;
}
void read_loop(manager_ptr mgr) {
if (async_read_pending_)
return;
if (rd_buf_ready_) {
rd_buf_ready_ = false;
if (read_one(mgr.get(), rd_buf_.size()))
read_loop(std::move(mgr));
return;
}
auto cb = [=](const error_code& ec, size_t read_bytes) mutable {
async_read_pending_ = false;
CAF_LOG_TRACE("");
if (!ec) {
// bail out early in case broker passivated stream in the meantime
if (read_one(mgr.get(), read_bytes))
read_loop(std::move(mgr));
} else {
mgr->io_failure(&backend(), operation::read);
}
};
switch (rd_flag_) {
case receive_policy_flag::exactly:
if (rd_buf_.size() < rd_size_)
rd_buf_.resize(rd_size_);
async_read_pending_ = true;
boost::asio::async_read(fd_, boost::asio::buffer(rd_buf_, rd_size_),
cb);
break;
case receive_policy_flag::at_most:
if (rd_buf_.size() < rd_size_)
rd_buf_.resize(rd_size_);
async_read_pending_ = true;
fd_.async_read_some(boost::asio::buffer(rd_buf_, rd_size_), cb);
break;
case receive_policy_flag::at_least: {
// read up to 10% more, but at least allow 100 bytes more
auto min_size = rd_size_ + std::max<size_t>(100, rd_size_ / 10);
if (rd_buf_.size() < min_size) {
rd_buf_.resize(min_size);
}
collect_data(mgr, 0);
break;
}
}
}
void write_loop(const manager_ptr& mgr) {
if (wr_offline_buf_.empty()) {
writing_ = false;
return;
}
wr_buf_.clear();
wr_buf_.swap(wr_offline_buf_);
boost::asio::async_write(
fd_, boost::asio::buffer(wr_buf_),
[=](const error_code& ec, size_t nb) {
CAF_LOG_TRACE("");
if (ec) {
CAF_LOG_DEBUG(CAF_ARG(ec.message()));
mgr->io_failure(&backend(), operation::read);
writing_ = false;
return;
}
CAF_LOG_DEBUG(CAF_ARG(nb));
if (ack_writes_)
mgr->data_transferred(&backend(), nb, wr_offline_buf_.size());
write_loop(mgr);
});
}
void collect_data(manager_ptr mgr, size_t collected_bytes) {
async_read_pending_ = true;
fd_.async_read_some(boost::asio::buffer(rd_buf_.data() + collected_bytes,
rd_buf_.size() - collected_bytes),
[=](const error_code& ec, size_t nb) mutable {
async_read_pending_ = false;
CAF_LOG_TRACE(CAF_ARG(nb));
if (!ec) {
auto sum = collected_bytes + nb;
if (sum >= rd_size_) {
if (read_one(mgr.get(), sum))
read_loop(std::move(mgr));
} else {
collect_data(std::move(mgr), sum);
}
} else {
mgr->io_failure(&backend(), operation::write);
}
});
}
/// Set if read loop was started by user and unset if passivate is called.
bool reading_;
/// Set on flush, also indicates that an async_write is pending.
bool writing_;
/// Stores whether user requested ACK messages for async writes.
bool ack_writes_;
/// TCP socket for this connection.
Socket fd_;
/// Configures how chunk sizes are calculated.
receive_policy_flag rd_flag_;
/// Minimum, maximum, or exact size of a chunk, depending on `rd_flag_`.
size_t rd_size_;
/// Input buffer.
buffer_type rd_buf_;
/// Output buffer for ASIO.
buffer_type wr_buf_;
/// Swapped with `wr_buf_` before next write. Users write into this buffer
/// as long as `wr_buf_` is processed by ASIO.
buffer_type wr_offline_buf_;
/// Reference to our I/O backend.
asio_multiplexer& backend_;
/// Signalizes that a scribe was passivated while an async read was pending.
bool rd_buf_ready_;
/// Makes sure no more than one async_read is pending at any given time
bool async_read_pending_;
};
/// An acceptor is responsible for accepting incoming connections.
template <class SocketAcceptor>
class asio_acceptor {
using protocol_type = typename SocketAcceptor::protocol_type;
using socket_type = boost::asio::basic_stream_socket<protocol_type>;
public:
/// A manager providing the `accept` member function.
using manager_type = acceptor_manager;
/// A smart pointer to an acceptor manager.
using manager_ptr = intrusive_ptr<manager_type>;
asio_acceptor(asio_multiplexer& am, io_service& io)
: accepting_(false),
backend_(am),
accept_fd_(io),
fd_valid_(false),
fd_(io),
async_accept_pending_(false) {
// nop
}
/// Returns the `multiplexer` this acceptor belongs to.
asio_multiplexer& backend() { return backend_; }
/// Returns the IO socket.
SocketAcceptor& socket_handle() {
return accept_fd_;
}
/// Returns the IO socket.
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_;
}
/// Initializes this acceptor, setting the socket handle to `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(manager_type* mgr) {
activate(mgr);
}
/// Starts the accept loop.
void activate(manager_type* mgr) {
accepting_ = true;
accept_loop(mgr);
}
/// Starts the accept loop.
void passivate() {
accepting_ = false;
}
/// Closes the network connection, thus stopping this acceptor.
void stop() {
accept_fd_.close();
}
private:
bool accept_one(manager_type* mgr) {
auto res = mgr->new_connection(); // moves fd_
// reset fd_ for next accept operation
fd_ = socket_type{accept_fd_.get_io_service()};
return res && accepting_;
}
void accept_loop(manager_ptr mgr) {
if (async_accept_pending_)
return;
// accept "cached" connection first
if (fd_valid_) {
fd_valid_ = false;
if (accept_one(mgr.get()))
accept_loop(std::move(mgr));
return;
}
async_accept_pending_ = true;
accept_fd_.async_accept(fd_, [=](const error_code& ec) mutable {
CAF_LOG_TRACE("");
async_accept_pending_ = false;
if (!ec) {
// if broker has passivated this in the meantime, cache fd_ for later
if (!accepting_) {
fd_valid_ = true;
return;
}
if (accept_one(mgr.get()))
accept_loop(std::move(mgr));
} else {
mgr->io_failure(&backend(), operation::read);
}
});
}
bool accepting_;
asio_multiplexer& backend_;
SocketAcceptor accept_fd_;
bool fd_valid_;
socket_type fd_;
/// Makes sure no more than one async_accept is pending at any given time
bool async_accept_pending_;
};
} // namesapce network
} // namespace io
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* Raphael Hiesgen <raphael.hiesgen (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#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 {
namespace io {
namespace network {
namespace {
/// A wrapper for the supervisor backend provided by boost::asio.
struct asio_supervisor : public multiplexer::supervisor {
explicit asio_supervisor(io_service& iob) : work(iob) {
// nop
}
private:
boost::asio::io_service::work work;
};
} // namespace anonymous
expected<asio_tcp_socket>
new_tcp_connection(io_service& ios, const std::string& host, uint16_t port) {
asio_tcp_socket fd{ios};
using boost::asio::ip::tcp;
tcp::resolver r(fd.get_io_service());
tcp::resolver::query q(host, std::to_string(port));
boost::system::error_code ec;
auto i = r.resolve(q, ec);
if (ec)
return make_error(sec::cannot_connect_to_node);
boost::asio::connect(fd, i, ec);
if (ec)
return make_error(sec::cannot_connect_to_node);
return asio_tcp_socket{std::move(fd)};
}
error ip_bind(asio_tcp_socket_acceptor& fd, uint16_t port,
const char* addr, bool reuse_addr) {
CAF_LOG_TRACE(CAF_ARG(port) << CAF_ARG(reuse_addr));
using boost::asio::ip::tcp;
auto bind_and_listen = [&](tcp::endpoint& ep) -> error {
CAF_LOG_DEBUG("created IP endpoint:" << CAF_ARG(ep.address().to_string())
<< CAF_ARG(ep.port()));
boost::system::error_code ec;
fd.open(ep.protocol());
if (reuse_addr) {
fd.set_option(tcp::acceptor::reuse_address(reuse_addr), ec);
if (ec)
return sec::cannot_open_port;
}
fd.bind(ep, ec);
if (ec)
return sec::cannot_open_port;
fd.listen(asio_tcp_socket_acceptor::max_connections, ec);
if (ec)
return sec::cannot_open_port;
return none;
};
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);
}
}
scribe_ptr asio_multiplexer::new_scribe(asio_tcp_socket&& sock) {
CAF_LOG_TRACE("");
class impl : public scribe {
public:
impl(asio_multiplexer& am, asio_tcp_socket&& s)
: scribe(network::conn_hdl_from_socket(s)),
launched_(false),
stream_(am) {
stream_.init(std::move(s));
}
void configure_read(receive_policy::config config) override {
CAF_LOG_TRACE("");
stream_.configure_read(config);
if (!launched_) {
launch();
}
}
void ack_writes(bool enable) override {
CAF_LOG_TRACE(CAF_ARG(enable));
stream_.ack_writes(enable);
}
std::vector<char>& wr_buf() override {
return stream_.wr_buf();
}
std::vector<char>& rd_buf() override {
return stream_.rd_buf();
}
void stop_reading() override {
CAF_LOG_TRACE("");
stream_.stop_reading();
detach(&stream_.backend(), false);
}
void flush() override {
CAF_LOG_TRACE("");
stream_.flush(this);
}
std::string addr() const override {
return stream_.socket_handle().remote_endpoint().address().to_string();
}
uint16_t port() const override {
return stream_.socket_handle().remote_endpoint().port();
}
void launch() {
CAF_LOG_TRACE("");
CAF_ASSERT(!launched_);
launched_ = true;
stream_.start(this);
}
void add_to_loop() override {
stream_.activate(this);
}
void remove_from_loop() override {
stream_.passivate();
}
private:
bool launched_;
asio_stream<asio_tcp_socket> stream_;
};
return make_counted<impl>(*this, std::move(sock));
}
scribe_ptr asio_multiplexer::new_scribe(native_socket fd) {
CAF_LOG_TRACE(CAF_ARG(fd));
boost::system::error_code ec;
asio_tcp_socket sock{service()};
sock.assign(boost::asio::ip::tcp::v6(), fd, ec);
if (ec)
sock.assign(boost::asio::ip::tcp::v4(), fd, ec);
if (ec)
CAF_RAISE_ERROR(ec.message());
return new_scribe(std::move(sock));
}
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));
}
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(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));
}
bool new_connection() override {
CAF_LOG_TRACE("");
if (detached())
// we are already disconnected from the broker while the multiplexer
// did not yet remove the socket, this can happen if an I/O event causes
// the broker to call close_all() while the pollset contained
// further activities for the broker
return false;
auto& am = acceptor_.backend();
auto sptr = am.new_scribe(std::move(acceptor_.accepted_socket()));
auto shdl = sptr->hdl();
parent()->add_scribe(std::move(sptr));
return doorman::new_connection(&am, shdl);
}
void stop_reading() override {
CAF_LOG_TRACE("");
acceptor_.stop();
detach(&acceptor_.backend(), false);
}
void launch() override {
CAF_LOG_TRACE("");
acceptor_.start(this);
}
std::string addr() const override {
return
acceptor_.socket_handle().local_endpoint().address().to_string();
}
uint16_t port() const override {
return acceptor_.socket_handle().local_endpoint().port();
}
void add_to_loop() override {
acceptor_.activate(this);
}
void remove_from_loop() override {
acceptor_.passivate();
}
private:
network::asio_acceptor<asio_tcp_socket_acceptor> acceptor_;
};
return make_counted<impl>(std::move(sock), *this);
}
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;
sock.assign(boost::asio::ip::tcp::v6(), fd, ec);
if (ec)
sock.assign(boost::asio::ip::tcp::v4(), fd, ec);
if (ec)
CAF_RAISE_ERROR(ec.message());
return new_doorman(std::move(sock));
}
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;
return new_doorman(std::move(fd));
}
datagram_servant_ptr asio_multiplexer::new_datagram_servant(native_socket) {
CAF_RAISE_ERROR("UDP is not implemented for asio");
return nullptr;
}
datagram_servant_ptr
asio_multiplexer::new_datagram_servant_for_endpoint(native_socket,
const ip_endpoint&) {
CAF_RAISE_ERROR("UDP is not implemented for asio");
return nullptr;
}
expected<datagram_servant_ptr>
asio_multiplexer::new_remote_udp_endpoint(const std::string&, uint16_t) {
CAF_RAISE_ERROR("UDP is not implemented for asio");
return sec::bad_function_call;
}
expected<datagram_servant_ptr>
asio_multiplexer::new_local_udp_endpoint(uint16_t, const char*, bool) {
CAF_RAISE_ERROR("UDP is not implemented for asio");
return sec::bad_function_call;
}
void asio_multiplexer::exec_later(resumable* rptr) {
auto mt = system().config().scheduler_max_throughput;
switch (rptr->subtype()) {
case resumable::io_actor:
case resumable::function_object: {
intrusive_ptr<resumable> ptr{rptr, false};
service().post([=]() mutable {
switch (ptr->resume(this, mt)) {
case resumable::resume_later:
exec_later(ptr.release());
break;
case resumable::done:
case resumable::awaiting_message:
break;
default:
; // ignored
}
});
break;
}
default:
system().scheduler().enqueue(rptr);
}
}
asio_multiplexer::asio_multiplexer(actor_system* sys) : multiplexer(sys) {
// nop
}
asio_multiplexer::~asio_multiplexer() {
//nop
}
multiplexer::supervisor_ptr asio_multiplexer::make_supervisor() {
return std::unique_ptr<asio_supervisor>(new asio_supervisor(service()));
}
bool asio_multiplexer::try_run_once() {
boost::system::error_code ec;
auto num = service().poll(ec);
if (ec)
CAF_RAISE_ERROR(ec.message());
return num > 0;
}
void asio_multiplexer::run_once() {
boost::system::error_code ec;
service().run_one(ec);
if (ec)
CAF_RAISE_ERROR(ec.message());
}
void asio_multiplexer::run() {
CAF_LOG_TRACE("asio-based multiplexer");
boost::system::error_code ec;
service().run(ec);
if (ec)
CAF_RAISE_ERROR(ec.message());
}
multiplexer_backend* asio_multiplexer::pimpl() {
return &service_;
}
} // namesapce network
} // namespace io
} // namespace caf
......@@ -57,11 +57,6 @@
#include "caf/detail/get_root_uuid.hpp"
#include "caf/detail/get_mac_addresses.hpp"
#ifdef CAF_USE_ASIO
#include "caf/io/network/asio_multiplexer.hpp"
#include "caf/io/network/asio_multiplexer_impl.hpp"
#endif // CAF_USE_ASIO
#ifdef CAF_WINDOWS
#include <io.h>
#include <fcntl.h>
......@@ -93,10 +88,6 @@ actor_system::module* middleman::make(actor_system& sys, detail::type_list<>) {
auto atm = get_or(sys.config(), "middleman.network-backend",
defaults::middleman::network_backend);
switch (atom_uint(atm)) {
# ifdef CAF_USE_ASIO
case atom_uint(atom("asio")):
return new mm_impl<network::asio_multiplexer>(sys);
# endif // CAF_USE_ASIO
case atom_uint(atom("testing")):
return new mm_impl<network::test_multiplexer>(sys);
default:
......
......@@ -63,8 +63,8 @@ public:
bool authentication_enabled();
/// Returns an OpenSSL manager using the default network backend.
/// @warning Creating an OpenSSL manager will fail when using the ASIO
/// network backend or any other custom implementation.
/// @warning Creating an OpenSSL manager will fail when using
// a custom implementation.
/// @throws `logic_error` if the middleman is not loaded or is not using the
/// default network backend.
static actor_system::module* make(actor_system&, detail::type_list<>);
......
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