Commit b45136ea authored by Dominik Charousset's avatar Dominik Charousset

Merge pull request #299 from actor-framework/topic/asio

Add boost asio as an optional multiplexer backend
parents 53e67b5d e3bea9a2
......@@ -28,6 +28,10 @@ endif()
if(NOT CAF_NO_SUMMARY)
set(CAF_NO_SUMMARY no)
endif()
if(NOT CAF_USE_ASIO)
set(CAF_USE_ASIO no)
endif()
################################################################################
# get version of CAF #
......@@ -243,6 +247,10 @@ if(CAF_NO_MEM_MANAGEMENT)
add_definitions(-DCAF_NO_MEM_MANAGEMENT)
endif()
if (CAF_USE_ASIO)
add_definitions(-DCAF_USE_ASIO)
endif()
################################################################################
# check for biicode build pipeline #
################################################################################
......@@ -320,6 +328,20 @@ endif()
# all projects need the headers of the core components
include_directories("${LIBCAF_INCLUDE_DIRS}")
# Find boost asio if the asio multiplexer should be used
if (CAF_USE_ASIO)
if (CAF_BUILD_STATIC_ONLY)
set(Boost_USE_STATIC_LIBS ON) # only find static libs
endif()
set(Boost_USE_MULTITHREADED ON)
find_package(Boost REQUIRED COMPONENTS system)
if (Boost_FOUND)
set(INCLUDE_DIRS "${INCLUDE_DIRS} ${Boost_INCLUDE_DIRS}")
set(LD_DIRS ${LD_DIRS} ${Boost_LIBRARIES})
set(LD_FLAGS ${LD_FLAGS} ${Boost_SYSTEM_LIBRARY})
endif()
endif()
################################################################################
# add subprojects #
......@@ -563,6 +585,7 @@ if(NOT CAF_NO_SUMMARY)
"\nRuntime checks: ${CAF_ENABLE_RUNTIME_CHECKS}"
"\nLog level: ${LOG_LEVEL_STR}"
"\nWith mem. mgmt.: ${CAF_BUILD_MEM_MANAGEMENT}"
"\nUse Boost ASIO: ${CAF_USE_ASIO}"
"\n"
"\nBuild examples: ${CAF_BUILD_EXAMPLES}"
"\nBuild unit tests: ${CAF_BUILD_UNIT_TESTS}"
......
......@@ -49,6 +49,7 @@ Usage: $0 [OPTION]... [VAR=VALUE]...
--no-compiler-check disable compiler version check
--no-auto-libc++ do not automatically enable libc++ for Clang
--warnings-as-errors enables -Werror
--with-asio enable ASIO multiplexer
Installation Directories:
--prefix=PREFIX installation directory [/usr/local]
......@@ -241,6 +242,9 @@ 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
......
......@@ -13,7 +13,6 @@ set (LIBCAF_IO_SRCS
src/middleman.cpp
src/hook.cpp
src/interfaces.cpp
src/default_multiplexer.cpp
src/publish.cpp
src/publish_local_groups.cpp
src/remote_actor.cpp
......@@ -26,6 +25,12 @@ set (LIBCAF_IO_SRCS
add_custom_target(libcaf_io)
if (CAF_USE_ASIO)
set(LIBCAF_IO_SRCS ${LIBCAF_IO_SRCS} src/asio_multiplexer.cpp)
else()
set(LIBCAF_IO_SRCS ${LIBCAF_IO_SRCS} src/default_multiplexer.cpp)
endif()
# build shared library if not compiling static only
if (NOT CAF_BUILD_STATIC_ONLY)
add_library(libcaf_io_shared SHARED ${LIBCAF_IO_SRCS} ${LIBCAF_IO_HDRS})
......
/******************************************************************************\
* *
* ____ _ _ _ *
* | __ ) ___ ___ ___| |_ / \ ___| |_ ___ _ __ *
* | _ \ / _ \ / _ \/ __| __| / _ \ / __| __/ _ \| '__| *
* | |_) | (_) | (_) \__ \ |_ _ / ___ \ (__| || (_) | | *
* |____/ \___/ \___/|___/\__(_)_/ \_\___|\__\___/|_| *
* *
* *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the Boost Software License, Version 1.0. See *
* accompanying file LICENSE or copy at http://www.boost.org/LICENSE_1_0.txt *
\******************************************************************************/
#ifndef CAF_IO_ASIO_NETWORK_HPP
#define CAF_IO_ASIO_NETWORK_HPP
#include <vector>
#include <string>
#include <cstdint>
#include "boost/asio.hpp"
#include "caf/exception.hpp"
#include "caf/ref_counted.hpp"
#include "caf/io/receive_policy.hpp"
#include "caf/detail/logging.hpp"
namespace boost {
namespace actor_io {
namespace network {
/**
* @brief Low-level backend for IO multiplexing.
*/
using multiplexer = asio::io_service;
/**
* @brief Gets the multiplexer singleton.
*/
multiplexer& get_multiplexer_singleton();
/**
* @brief Makes sure a {@link multiplexer} does not stop its event loop
* before the application requests a shutdown.
*
* The supervisor informs the multiplexer in its constructor that it
* must not exit the event loop until the destructor of the supervisor
* has been called.
*/
using supervisor = asio::io_service::work;
/**
* @brief Low-level socket type used as default.
*/
using default_socket = asio::ip::tcp::socket;
/**
* @brief Low-level socket type used as default.
*/
using default_socket_acceptor = asio::ip::tcp::acceptor;
/**
* @brief Platform-specific native socket type.
*/
using native_socket = typename default_socket::native_handle_type;
/**
* @brief Platform-specific native acceptor socket type.
*/
using native_socket_acceptor = typename default_socket_acceptor::native_handle_type;
/**
* @brief Identifies network IO operations, i.e., read or write.
*/
enum class operation {
read,
write
};
/**
* @brief A manager configures an IO device and provides callbacks
* for various IO operations.
*/
class manager : public actor::ref_counted {
public:
virtual ~manager();
/**
* @brief Called during application shutdown, indicating that the manager
* should cause its underlying IO device to stop read IO operations.
*/
virtual void stop_reading() = 0;
/**
* @brief Causes the manager to stop all IO operations on its IO device.
*/
virtual void stop() = 0;
/**
* @brief Called by the underlying IO device to report failures.
*/
virtual void io_failure(operation op, const std::string& error_message) = 0;
};
// functions from ref_counted cannot be found by ADL => provide new overload
inline void intrusive_ptr_add_ref(manager* p) {
p->ref();
}
// functions from ref_counted cannot be found by ADL => provide new overload
inline void intrusive_ptr_release(manager* p) {
p->deref();
}
/**
* @relates manager
*/
using manager_ptr = intrusive_ptr<manager>;
/**
* @brief A stream manager configures an IO stream and provides callbacks
* for incoming data as well as for error handling.
*/
class stream_manager : public manager {
public:
virtual ~stream_manager();
/**
* @brief Called by the underlying IO device whenever it received data.
*/
virtual void consume(const void* data, size_t num_bytes) = 0;
};
/**
* @brief 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 stream {
public:
/**
* @brief A smart pointer to a stream manager.
*/
using manager_ptr = intrusive_ptr<stream_manager>;
/**
* @brief A buffer class providing a compatible
* interface to @p std::vector.
*/
using buffer_type = std::vector<char>;
stream(multiplexer& backend) : m_writing(false), m_fd(backend) {
configure_read(receive_policy::at_most(1024));
}
/**
* @brief Returns the @p multiplexer this stream belongs to.
*/
inline multiplexer& backend() {
return m_fd.get_io_service();
}
/**
* @brief Returns the IO socket.
*/
inline Socket& socket_handle() {
return m_fd;
}
/**
* @brief Initializes this stream, setting the socket handle to @p fd.
*/
void init(Socket fd) {
m_fd = std::move(fd);
}
/**
* @brief Starts reading data from the socket, forwarding incoming
* data to @p mgr.
*/
void start(const manager_ptr& mgr) {
BOOST_ACTOR_REQUIRE(mgr != nullptr);
read_loop(mgr);
}
/**
* @brief Configures how much data will be provided
* for the next @p 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) {
m_rd_flag = config.first;
m_rd_size = config.second;
}
/**
* @brief Copies data to the write buffer.
* @note Not thread safe.
*/
void write(const void* buf, size_t num_bytes) {
auto first = reinterpret_cast<const char*>(buf);
auto last = first + num_bytes;
m_wr_offline_buf.insert(first, last);
}
/**
* @brief 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 m_wr_offline_buf;
}
buffer_type& rd_buf() {
return m_rd_buf;
}
/**
* @brief Sends the content of the write buffer, calling
* the @p io_failure member function of @p 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) {
BOOST_ACTOR_REQUIRE(mgr != nullptr);
if (!m_wr_offline_buf.empty() && !m_writing) {
m_writing = true;
write_loop(mgr);
}
}
/**
* @brief Closes the network connection, thus stopping this stream.
*/
void stop() {
BOOST_ACTOR_LOGM_TRACE("boost::actor_io::network::stream", "");
m_fd.close();
}
void stop_reading() {
BOOST_ACTOR_LOGM_TRACE("boost::actor_io::network::stream", "");
system::error_code ec; // ignored
m_fd.shutdown(asio::ip::tcp::socket::shutdown_receive, ec);
}
private:
void write_loop(const manager_ptr& mgr) {
if (m_wr_offline_buf.empty()) {
m_writing = false;
return;
}
m_wr_buf.clear();
m_wr_buf.swap(m_wr_offline_buf);
asio::async_write(m_fd, asio::buffer(m_wr_buf),
[=](const system::error_code& ec, size_t nb) {
BOOST_ACTOR_LOGC_TRACE("boost::actor_io::network::stream",
"write_loop$lambda",
BOOST_ACTOR_ARG(this));
static_cast<void>(nb); // silence compiler warning
if (!ec) {
BOOST_ACTOR_LOGC_DEBUG("boost::actor_io::network::stream",
"write_loop$lambda",
nb << " bytes sent");
write_loop(mgr);
}
else {
BOOST_ACTOR_LOGC_DEBUG("boost::actor_io::network::stream",
"write_loop$lambda",
"error during send: " << ec.message());
mgr->io_failure(operation::read, ec.message());
m_writing = false;
}
});
}
void read_loop(const manager_ptr& mgr) {
auto cb = [=](const system::error_code& ec, size_t read_bytes) {
BOOST_ACTOR_LOGC_TRACE("boost::actor_io::network::stream",
"read_loop$cb",
BOOST_ACTOR_ARG(this));
if (!ec) {
mgr->consume(m_rd_buf.data(), read_bytes);
read_loop(mgr);
}
else mgr->io_failure(operation::read, ec.message());
};
switch (m_rd_flag) {
case receive_policy_flag::exactly:
if (m_rd_buf.size() < m_rd_size) m_rd_buf.resize(m_rd_size);
asio::async_read(m_fd, asio::buffer(m_rd_buf, m_rd_size), cb);
break;
case receive_policy_flag::at_most:
if (m_rd_buf.size() < m_rd_size) m_rd_buf.resize(m_rd_size);
m_fd.async_read_some(asio::buffer(m_rd_buf, m_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 = m_rd_size
+ std::max<size_t>(100, m_rd_size / 10);
if (m_rd_buf.size() < min_size) m_rd_buf.resize(min_size);
collect_data(mgr, 0);
break;
}
}
}
void collect_data(const manager_ptr& mgr, size_t collected_bytes) {
m_fd.async_read_some(asio::buffer(m_rd_buf.data() + collected_bytes,
m_rd_buf.size() - collected_bytes),
[=](const system::error_code& ec, size_t nb) {
BOOST_ACTOR_LOGC_TRACE("boost::actor_io::network::stream",
"collect_data$lambda",
BOOST_ACTOR_ARG(this));
if (!ec) {
auto sum = collected_bytes + nb;
if (sum >= m_rd_size) {
mgr->consume(m_rd_buf.data(), sum);
read_loop(mgr);
}
else collect_data(mgr, sum);
}
else mgr->io_failure(operation::write, ec.message());
});
}
bool m_writing;
Socket m_fd;
receive_policy_flag m_rd_flag;
size_t m_rd_size;
buffer_type m_rd_buf;
buffer_type m_wr_buf;
buffer_type m_wr_offline_buf;
};
/**
* @brief An acceptor manager configures an acceptor and provides
* callbacks for incoming connections as well as for error handling.
*/
class acceptor_manager : public manager {
public:
/**
* @brief Called by the underlying IO device to indicate that
* a new connection is awaiting acceptance.
*/
virtual void new_connection() = 0;
};
/**
* @brief An acceptor is responsible for accepting incoming connections.
*/
template<class SocketAcceptor>
class acceptor {
using protocol_type = typename SocketAcceptor::protocol_type;
using socket_type = asio::basic_stream_socket<protocol_type>;
public:
/**
* @brief A manager providing the @p accept member function.
*/
using manager_type = acceptor_manager;
/**
* @brief A smart pointer to an acceptor manager.
*/
using manager_ptr = intrusive_ptr<manager_type>;
acceptor(multiplexer& backend)
: m_backend(backend), m_accept_fd(backend), m_fd(backend) { }
/**
* @brief Returns the @p multiplexer this acceptor belongs to.
*/
inline multiplexer& backend() {
return m_backend;
}
/**
* @brief Returns the IO socket.
*/
inline SocketAcceptor& socket_handle() {
return m_accept_fd;
}
/**
* @brief Returns the accepted socket. This member function should
* be called only from the @p new_connection callback.
*/
inline socket_type& accepted_socket() {
return m_fd;
}
/**
* @brief Initializes this acceptor, setting the socket handle to @p fd.
*/
void init(SocketAcceptor fd) {
m_accept_fd = std::move(fd);
}
/**
* @brief Starts this acceptor, forwarding all incoming connections to
* @p manager. The intrusive pointer will be released after the
* acceptor has been closed or an IO error occured.
*/
void start(const manager_ptr& mgr) {
BOOST_ACTOR_REQUIRE(mgr != nullptr);
accept_loop(mgr);
}
/**
* @brief Closes the network connection, thus stopping this acceptor.
*/
void stop() {
m_accept_fd.close();
}
private:
void accept_loop(const manager_ptr& mgr) {
m_accept_fd.async_accept(m_fd, [=](const system::error_code& ec) {
BOOST_ACTOR_LOGM_TRACE("boost::actor_io::network::acceptor", "");
if (!ec) {
mgr->new_connection(); // probably moves m_fd
// reset m_fd for next accept operation
m_fd = socket_type{m_accept_fd.get_io_service()};
accept_loop(mgr);
}
else mgr->io_failure(operation::read, ec.message());
});
}
multiplexer& m_backend;
SocketAcceptor m_accept_fd;
socket_type m_fd;
};
template<class Socket>
void ipv4_connect(Socket& fd, const std::string& host, uint16_t port) {
CAF_LOGF_TRACE(CAF_ARG(host) << ", " CAF_ARG(port));
using asio::ip::tcp;
try {
tcp::resolver r(fd.get_io_service());
tcp::resolver::query q(tcp::v4(), host, std::to_string(port));
auto i = r.resolve(q);
asio::connect(fd, i);
}
catch (system::system_error& se) {
throw actor::network_error(se.code().message());
}
}
inline default_socket new_ipv4_connection(const std::string& host,
uint16_t port) {
default_socket fd{get_multiplexer_singleton()};
ipv4_connect(fd, host, port);
return fd;
}
template<class SocketAcceptor>
void ipv4_bind(SocketAcceptor& fd,
uint16_t port,
const char* addr = nullptr) {
BOOST_ACTOR_LOGF_TRACE(BOOST_ACTOR_ARG(port));
using asio::ip::tcp;
try {
auto bind_and_listen = [&](tcp::endpoint& ep) {
fd.open(ep.protocol());
fd.set_option(tcp::acceptor::reuse_address(true));
fd.bind(ep);
fd.listen();
};
if (addr) {
tcp::endpoint ep(asio::ip::address::from_string(addr), port);
bind_and_listen(ep);
}
else {
tcp::endpoint ep(tcp::v4(), port);
bind_and_listen(ep);
}
}
catch (system::system_error& se) {
if (se.code() == system::errc::address_in_use) {
throw actor::bind_failure(se.code().message());
}
throw actor::network_error(se.code().message());
}
}
inline default_socket_acceptor new_ipv4_acceptor(uint16_t port,
const char* addr = nullptr) {
default_socket_acceptor fd{get_multiplexer_singleton()};
ipv4_bind(fd, port, addr);
return fd;
}
} // namespace network
} // namespace actor_io
} // namespace boost
#endif // CAF_IO_ASIO_NETWORK_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* 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. *
******************************************************************************/
#ifndef CAF_IO_NETWORK_ASIO_MULTIPLEXER_HPP
#define CAF_IO_NETWORK_ASIO_MULTIPLEXER_HPP
#include "boost/asio.hpp"
#include "caf/detail/logging.hpp"
#include "caf/io/receive_policy.hpp"
#include "caf/io/network/multiplexer.hpp"
#include "caf/io/network/stream_manager.hpp"
#include "caf/io/network/acceptor_manager.hpp"
namespace caf {
namespace io {
namespace network {
/**
* Low-level backend for IO multiplexing.
*/
using io_backend = boost::asio::io_service;
/**
* Low-level socket type used as default.
*/
using default_socket = boost::asio::ip::tcp::socket;
/**
* Low-level socket type used as default.
*/
using default_socket_acceptor = boost::asio::ip::tcp::acceptor;
/**
* Platform-specific native socket type.
*/
using native_socket = typename default_socket::native_handle_type;
/**
* Platform-specific native acceptor socket type.
*/
using native_socket_acceptor
= typename default_socket_acceptor::native_handle_type;
/**
* A wrapper for the supervisor backend provided by boost::asio.
*/
struct asio_supervisor : public multiplexer::supervisor {
asio_supervisor(io_backend& iob) : work(iob) {}
private:
boost::asio::io_service::work work;
};
/**
* A wrapper for the boost::asio multiplexer
*/
class asio_multiplexer : public multiplexer {
public:
friend class io::middleman;
friend class supervisor;
connection_handle new_tcp_scribe(const std::string&, uint16_t) override;
void assign_tcp_scribe(broker*, connection_handle hdl) override;
template <class Socket>
connection_handle add_tcp_scribe(broker*, Socket&& sock);
connection_handle add_tcp_scribe(broker*, native_socket fd) override;
connection_handle add_tcp_scribe(broker*, const std::string& host,
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*, 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 port, const char* in, bool rflag) override;
void dispatch_runnable(runnable_ptr ptr) override;
asio_multiplexer();
~asio_multiplexer();
supervisor_ptr make_supervisor() override;
void run() override;
private:
inline boost::asio::io_service& backend() { return m_backend; }
io_backend m_backend;
std::mutex m_mtx_sockets;
std::mutex m_mtx_acceptors;
std::map<int64_t, default_socket> m_unassigned_sockets;
std::map<int64_t, default_socket_acceptor> m_unassigned_acceptors;
};
asio_multiplexer& get_multiplexer_singleton();
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 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>;
stream(io_backend& backend) : m_writing(false), m_fd(backend) {
configure_read(receive_policy::at_most(1024));
}
/**
* Returns the IO socket.
*/
Socket& socket_handle() { return m_fd; }
/**
* Initializes this stream, setting the socket handle to `fd`.
*/
void init(Socket fd) { m_fd = std::move(fd); }
/**
* Starts reading data from the socket, forwarding incoming data to `mgr`.
*/
void start(const manager_ptr& mgr) {
CAF_ASSERT(mgr != nullptr);
read_loop(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) {
m_rd_flag = config.first;
m_rd_size = config.second;
}
/**
* Copies data to the write buffer.
* @note Not thread safe.
*/
void write(const void* buf, size_t num_bytes) {
CAF_LOG_TRACE("num_bytes: " << num_bytes);
auto first = reinterpret_cast<const char*>(buf);
auto last = first + num_bytes;
m_wr_offline_buf.insert(m_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 m_wr_offline_buf; }
buffer_type& rd_buf() { return m_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 (!m_wr_offline_buf.empty() && !m_writing) {
m_writing = true;
write_loop(mgr);
}
}
/**
* Closes the network connection, thus stopping this stream.
*/
void stop() {
CAF_LOGMF(CAF_TRACE, "");
m_fd.close();
}
void stop_reading() {
CAF_LOGMF(CAF_TRACE, "");
boost::system::error_code ec; // ignored
m_fd.shutdown(boost::asio::ip::tcp::socket::shutdown_receive, ec);
}
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));
if (!ec) {
mgr->consume(m_rd_buf.data(), read_bytes);
read_loop(mgr);
} else {
mgr->io_failure(operation::read);
}
};
switch (m_rd_flag) {
case receive_policy_flag::exactly:
if (m_rd_buf.size() < m_rd_size) {
m_rd_buf.resize(m_rd_size);
}
boost::asio::async_read(m_fd, boost::asio::buffer(m_rd_buf, m_rd_size),
cb);
break;
case receive_policy_flag::at_most:
if (m_rd_buf.size() < m_rd_size) {
m_rd_buf.resize(m_rd_size);
}
m_fd.async_read_some(boost::asio::buffer(m_rd_buf, m_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 = m_rd_size + std::max<size_t>(100, m_rd_size / 10);
if (m_rd_buf.size() < min_size) {
m_rd_buf.resize(min_size);
}
collect_data(mgr, 0);
break;
}
}
}
void write_loop(const manager_ptr& mgr) {
if (m_wr_offline_buf.empty()) {
m_writing = false;
return;
}
m_wr_buf.clear();
m_wr_buf.swap(m_wr_offline_buf);
boost::asio::async_write(
m_fd, boost::asio::buffer(m_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));
static_cast<void>(nb); // silence compiler warning
if (!ec) {
CAF_LOGC(CAF_DEBUG, "caf::io::network::stream", "write_loop$lambda",
nb << " bytes sent");
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);
m_writing = false;
}
});
}
void collect_data(const manager_ptr& mgr, size_t collected_bytes) {
m_fd.async_read_some(boost::asio::buffer(m_rd_buf.data() + collected_bytes,
m_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));
if (!ec) {
auto sum = collected_bytes + nb;
if (sum >= m_rd_size) {
mgr->consume(m_rd_buf.data(), sum);
read_loop(mgr);
} else {
collect_data(mgr, sum);
}
} else {
mgr->io_failure(operation::write);
}
});
}
bool m_writing;
Socket m_fd;
receive_policy_flag m_rd_flag;
size_t m_rd_size;
buffer_type m_rd_buf;
buffer_type m_wr_buf;
buffer_type m_wr_offline_buf;
};
/**
* An acceptor is responsible for accepting incoming connections.
*/
template <class SocketAcceptor>
class 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>;
acceptor(asio_multiplexer& backend, io_backend& io)
: m_backend(backend), m_accept_fd(io), m_fd(io) {}
/**
* Returns the `multiplexer` this acceptor belongs to.
*/
inline asio_multiplexer& backend() { return m_backend; }
/**
* Returns the IO socket.
*/
inline SocketAcceptor& socket_handle() { return m_accept_fd; }
/**
* Returns the accepted socket. This member function should
* be called only from the `new_connection` callback.
*/
inline socket_type& accepted_socket() { return m_fd; }
/**
* Initializes this acceptor, setting the socket handle to `fd`.
*/
void init(SocketAcceptor fd) { m_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); }
/**
* Closes the network connection, thus stopping this acceptor.
*/
void stop() { m_accept_fd.close(); }
private:
void accept_loop(const manager_ptr& mgr) {
m_accept_fd.async_accept(m_fd, [=](const boost::system::error_code& ec) {
CAF_LOGMF(CAF_TRACE, "");
if (!ec) {
mgr->new_connection(); // probably moves m_fd
// reset m_fd for next accept operation
m_fd = socket_type{m_accept_fd.get_io_service()};
accept_loop(mgr);
} else {
mgr->io_failure(operation::read);
}
});
}
asio_multiplexer& m_backend;
SocketAcceptor m_accept_fd;
socket_type m_fd;
};
} // namesapce network
} // namespace io
} // namespace caf
#endif // CAF_IO_NETWORK_ASIO_MULTIPLEXER_HPP
......@@ -4,6 +4,8 @@
#include <cstddef>
#include <utility>
#include "caf/config.hpp"
namespace caf {
namespace io {
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* 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. *
******************************************************************************/
#include "caf/exception.hpp"
#include "caf/io/broker.hpp"
#include "caf/io/middleman.hpp"
#include "caf/io/network/asio_multiplexer.hpp"
namespace caf {
namespace io {
namespace network {
default_socket new_tcp_connection(io_backend& backend, const std::string& host,
uint16_t port) {
default_socket fd{backend};
using boost::asio::ip::tcp;
try {
tcp::resolver r(fd.get_io_service());
tcp::resolver::query q(host, std::to_string(port));
auto i = r.resolve(q);
boost::asio::connect(fd, i);
} catch (boost::system::system_error& se) {
throw network_error(se.code().message());
}
return fd;
}
void ip_bind(default_socket_acceptor& fd, uint16_t port,
const char* addr = nullptr, bool reuse_addr = true) {
CAF_LOGF_TRACE(CAF_ARG(port));
using boost::asio::ip::tcp;
try {
auto bind_and_listen = [&](tcp::endpoint& ep) {
fd.open(ep.protocol());
fd.set_option(tcp::acceptor::reuse_address(reuse_addr));
fd.bind(ep);
fd.listen();
};
if (addr) {
tcp::endpoint ep(boost::asio::ip::address::from_string(addr), port);
bind_and_listen(ep);
CAF_LOGF_DEBUG("created IPv6 endpoint: " << ep.address() << ":"
<< fd.local_endpoint().port());
} else {
tcp::endpoint ep(tcp::v6(), port);
bind_and_listen(ep);
CAF_LOGF_DEBUG("created IPv6 endpoint: " << ep.address() << ":"
<< fd.local_endpoint().port());
}
} catch (boost::system::system_error& se) {
if (se.code() == boost::system::errc::address_in_use) {
throw bind_failure(se.code().message());
}
throw network_error(se.code().message());
}
}
connection_handle asio_multiplexer::new_tcp_scribe(const std::string& host,
uint16_t port) {
default_socket fd{new_tcp_connection(backend(), host, port)};
auto id = int64_from_native_socket(fd.native_handle());
std::lock_guard<std::mutex> lock(m_mtx_sockets);
m_unassigned_sockets.insert(std::make_pair(id, std::move(fd)));
return connection_handle::from_int(id);
}
void asio_multiplexer::assign_tcp_scribe(broker* self, connection_handle hdl) {
std::lock_guard<std::mutex> lock(m_mtx_sockets);
auto itr = m_unassigned_sockets.find(hdl.id());
if (itr != m_unassigned_sockets.end()) {
add_tcp_scribe(self, std::move(itr->second));
m_unassigned_sockets.erase(itr);
}
}
template <class Socket>
connection_handle asio_multiplexer::add_tcp_scribe(broker* self,
Socket&& sock) {
CAF_LOG_TRACE("");
class impl : public broker::scribe {
public:
impl(broker* ptr, Socket&& s)
: scribe(ptr, network::conn_hdl_from_socket(s)),
m_launched(false),
m_stream(s.get_io_service()) {
m_stream.init(std::move(s));
}
void configure_read(receive_policy::config config) override {
CAF_LOG_TRACE("");
m_stream.configure_read(config);
if (!m_launched) {
launch();
}
}
broker::buffer_type& wr_buf() override { return m_stream.wr_buf(); }
broker::buffer_type& rd_buf() override { return m_stream.rd_buf(); }
void stop_reading() override {
CAF_LOG_TRACE("");
m_stream.stop_reading();
disconnect(false);
}
void flush() override {
CAF_LOG_TRACE("");
m_stream.flush(this);
}
void launch() {
CAF_LOG_TRACE("");
CAF_ASSERT(!m_launched);
m_launched = true;
m_stream.start(this);
}
private:
bool m_launched;
stream<Socket> m_stream;
};
broker::scribe_pointer ptr = make_counted<impl>(self, std::move(sock));
self->add_scribe(ptr);
return ptr->hdl();
}
connection_handle asio_multiplexer::add_tcp_scribe(broker* self,
native_socket fd) {
CAF_LOG_TRACE(CAF_ARG(self) << ", " << CAF_ARG(fd));
boost::system::error_code ec;
default_socket sock{backend()};
sock.assign(boost::asio::ip::tcp::v6(), fd, ec);
if (ec) {
sock.assign(boost::asio::ip::tcp::v4(), fd, ec);
}
if (ec) {
throw network_error(ec.message());
}
return add_tcp_scribe(self, std::move(sock));
}
connection_handle asio_multiplexer::add_tcp_scribe(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));
}
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()};
ip_bind(fd, port, in, rflag);
auto id = int64_from_native_socket(fd.native_handle());
auto assigned_port = fd.local_endpoint().port();
std::lock_guard<std::mutex> lock(m_mtx_acceptors);
m_unassigned_acceptors.insert(std::make_pair(id, std::move(fd)));
return {accept_handle::from_int(id), assigned_port};
}
void asio_multiplexer::assign_tcp_doorman(broker* self, accept_handle hdl) {
CAF_LOG_TRACE("");
std::lock_guard<std::mutex> lock(m_mtx_acceptors);
auto itr = m_unassigned_acceptors.find(hdl.id());
if (itr != m_unassigned_acceptors.end()) {
add_tcp_doorman(self, std::move(itr->second));
m_unassigned_acceptors.erase(itr);
}
}
accept_handle
asio_multiplexer::add_tcp_doorman(broker* self,
default_socket_acceptor&& sock) {
CAF_LOG_TRACE("sock.fd = " << sock.native_handle());
CAF_ASSERT(sock.native_handle() != network::invalid_native_socket);
class impl : public broker::doorman {
public:
impl(broker* ptr, default_socket_acceptor&& s,
network::asio_multiplexer& am)
: doorman(ptr, network::accept_hdl_from_socket(s)),
m_acceptor(am, s.get_io_service()) {
m_acceptor.init(std::move(s));
}
void new_connection() override {
CAF_LOG_TRACE("");
auto& am = m_acceptor.backend();
accept_msg().handle
= am.add_tcp_scribe(parent(), std::move(m_acceptor.accepted_socket()));
parent()->invoke_message(invalid_actor_addr, invalid_message_id,
m_accept_msg);
}
void stop_reading() override {
CAF_LOG_TRACE("");
m_acceptor.stop();
disconnect(false);
}
void launch() override {
CAF_LOG_TRACE("");
m_acceptor.start(this);
}
private:
network::acceptor<default_socket_acceptor> m_acceptor;
};
broker::doorman_pointer ptr
= make_counted<impl>(self, std::move(sock), *this);
self->add_doorman(ptr);
return ptr->hdl();
}
accept_handle asio_multiplexer::add_tcp_doorman(broker* self,
native_socket fd) {
default_socket_acceptor sock{backend()};
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) {
throw network_error(ec.message());
}
return add_tcp_doorman(self, std::move(sock));
}
std::pair<accept_handle, uint16_t>
asio_multiplexer::add_tcp_doorman(broker* self, uint16_t port, const char* in,
bool rflag) {
default_socket_acceptor fd{backend()};
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) {
auto r = ptr.release();
backend().post([=]() {
r->run();
r->request_deletion(false);
});
}
asio_multiplexer::asio_multiplexer() {
// nop
}
asio_multiplexer::~asio_multiplexer() {
//nop
}
multiplexer::supervisor_ptr asio_multiplexer::make_supervisor() {
return std::unique_ptr<asio_supervisor>(new asio_supervisor(backend()));
}
void asio_multiplexer::run() {
CAF_LOG_TRACE("asio-based multiplexer");
boost::system::error_code ec;
backend().run(ec);
if (ec) {
throw std::runtime_error(ec.message());
}
}
asio_multiplexer& get_multiplexer_singleton() {
return static_cast<asio_multiplexer&>(middleman::instance()->backend());
}
} // namesapce network
} // namespace io
} // namespace caf
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