Commit 2e6411dc authored by Dominik Charousset's avatar Dominik Charousset

Modularize network backend

parent 89d97d9f
......@@ -3,7 +3,7 @@ project(caf_io C CXX)
# get header files; only needed by CMake generators,
# e.g., for creating proper Xcode projects
file(GLOB LIBCAF_IO_HDRS "caf/io/*.hpp")
file(GLOB LIBCAF_IO_HDRS "caf/io/*.hpp" "caf/io/network/*.hpp")
# list cpp files excluding platform-dependent files
set (LIBCAF_IO_SRCS
......@@ -12,10 +12,14 @@ set (LIBCAF_IO_SRCS
src/max_msg_size.cpp
src/middleman.cpp
src/hook.cpp
src/network.cpp
src/default_multiplexer.cpp
src/publish_local_groups.cpp
src/remote_actor_proxy.cpp
src/remote_group.cpp)
src/remote_group.cpp
src/manager.cpp
src/stream_manager.cpp
src/acceptor_manager.cpp
src/multiplexer.cpp)
# build shared library if not compiling static only
if (NOT "${CAF_BUILD_STATIC_ONLY}" STREQUAL "yes")
......
/******************************************************************************\
* *
* ____ _ _ _ *
* | __ ) ___ ___ ___| |_ / \ ___| |_ ___ _ __ *
* | _ \ / _ \ / _ \/ __| __| / _ \ / __| __/ _ \| '__| *
* | |_) | (_) | (_) \__ \ |_ _ / ___ \ (__| || (_) | | *
* |____/ \___/ \___/|___/\__(_)_/ \_\___|\__\___/|_| *
* *
* *
* *
* Copyright (C) 2011 - 2014 *
* 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
......@@ -52,11 +52,16 @@ class basp_broker : public broker, public actor_namespace::backend {
behavior make_behavior() override;
/*
template <class SocketAcceptor>
void publish(abstract_actor_ptr whom, SocketAcceptor fd) {
auto hdl = add_acceptor(std::move(fd));
announce_published_actor(hdl, whom);
}
*/
void announce_published_actor(accept_handle hdl,
const abstract_actor_ptr& whom);
actor_proxy_ptr make_proxy(const id_type&, actor_id) override;
......@@ -132,9 +137,6 @@ class basp_broker : public broker, public actor_namespace::backend {
optional<skip_message_t> kill_proxy(connection_context& ctx, actor_id aid,
std::uint32_t reason);
void announce_published_actor(accept_handle hdl,
const abstract_actor_ptr& whom);
void new_data(connection_context& ctx, buffer_type& buf);
void init_handshake_as_client(connection_context& ctx,
......
......@@ -27,17 +27,20 @@
#include "caf/extend.hpp"
#include "caf/local_actor.hpp"
#include "caf/io/network.hpp"
#include "caf/io/accept_handle.hpp"
#include "caf/io/system_messages.hpp"
#include "caf/io/connection_handle.hpp"
#include "caf/mixin/functor_based.hpp"
#include "caf/mixin/behavior_stack_based.hpp"
#include "caf/policy/not_prioritizing.hpp"
#include "caf/policy/sequential_invoke.hpp"
#include "caf/io/fwd.hpp"
#include "caf/io/accept_handle.hpp"
#include "caf/io/receive_policy.hpp"
#include "caf/io/system_messages.hpp"
#include "caf/io/connection_handle.hpp"
#include "caf/io/network/stream_manager.hpp"
#include "caf/io/network/acceptor_manager.hpp"
namespace caf {
namespace io {
......@@ -76,6 +79,10 @@ class broker : public extend<local_actor>::
virtual message disconnect_message() = 0;
inline broker* parent() {
return m_broker;
}
servant(broker* ptr);
void set_broker(broker* ptr);
......@@ -140,6 +147,8 @@ class broker : public extend<local_actor>::
message m_read_msg;
};
using scribe_pointer = intrusive_ptr<scribe>;
/**
* Manages incoming connections.
*/
......@@ -176,6 +185,8 @@ class broker : public extend<local_actor>::
message m_accept_msg;
};
using doorman_pointer = intrusive_ptr<doorman>;
class continuation;
// a broker needs friends
......@@ -241,6 +252,20 @@ class broker : public extend<local_actor>::
fun, hdl, std::forward<Ts>(vs)...);
}
inline void add_scribe(const scribe_pointer& ptr) {
m_scribes.insert(std::make_pair(ptr->hdl(), ptr));
}
inline void add_doorman(const doorman_pointer& ptr) {
m_doormen.insert(std::make_pair(ptr->hdl(), ptr));
if (is_initialized()) {
ptr->launch();
}
}
void invoke_message(const actor_addr& sender, message_id mid, message& msg);
/*
template <class Socket>
connection_handle add_connection(Socket sock) {
CAF_LOG_TRACE("");
......@@ -290,7 +315,7 @@ class broker : public extend<local_actor>::
template <class SocketAcceptor>
accept_handle add_acceptor(SocketAcceptor sock) {
CAF_LOG_TRACE("sock.fd = " << sock.fd());
CAF_REQUIRE(sock.fd() != network::invalid_socket);
CAF_REQUIRE(sock.fd() != network::invalid_native_socket);
class impl : public doorman {
public:
impl(broker* parent, SocketAcceptor&& s)
......@@ -322,6 +347,7 @@ class broker : public extend<local_actor>::
}
return ptr->hdl();
}
*/
void enqueue(const actor_addr&, message_id, message,
execution_unit*) override;
......@@ -381,10 +407,6 @@ class broker : public extend<local_actor>::
void cleanup(uint32_t reason);
using scribe_pointer = intrusive_ptr<scribe>;
using doorman_pointer = intrusive_ptr<doorman>;
virtual behavior make_behavior() = 0;
/** @endcond */
......@@ -413,9 +435,6 @@ class broker : public extend<local_actor>::
// throws on error
inline doorman& by_id(accept_handle hdl) { return by_id(hdl, m_doormen); }
void invoke_message(const actor_addr& sender, message_id mid,
message& msg);
bool invoke_message_from_cache();
void erase_io(int id);
......
......@@ -29,6 +29,12 @@ class middleman;
class receive_policy;
class remote_actor_proxy;
namespace network {
class multiplexer;
} // namespace network
} // namespace io
} // namespace caf
......
......@@ -32,7 +32,7 @@
#include "caf/io/hook.hpp"
#include "caf/io/broker.hpp"
#include "caf/io/network.hpp"
#include "caf/io/network/multiplexer.hpp"
namespace caf {
namespace io {
......@@ -78,14 +78,14 @@ class middleman : public detail::abstract_singleton {
*/
template <class F>
void run_later(F fun) {
m_backend.dispatch(fun);
m_backend->dispatch(fun);
}
/**
* Returns the IO backend used by this middleman.
*/
inline network::multiplexer& backend() {
return m_backend;
return *m_backend;
}
/**
......@@ -128,9 +128,9 @@ class middleman : public detail::abstract_singleton {
// guarded by singleton-getter `instance`
middleman();
// networking backend
network::multiplexer m_backend;
std::unique_ptr<network::multiplexer> m_backend;
// prevents backend from shutting down unless explicitly requested
network::supervisor* m_supervisor;
network::multiplexer::supervisor_ptr m_supervisor;
// runs the backend
std::thread m_thread;
// keeps track of "singleton-like" brokers
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2014 *
* Dominik Charousset <dominik.charousset (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 LICENCE_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_ACCEPTOR_MANAGER_HPP
#define CAF_IO_NETWORK_ACCEPTOR_MANAGER_HPP
#include "caf/io/network/manager.hpp"
namespace caf {
namespace io {
namespace network {
/**
* An acceptor manager configures an acceptor and provides
* callbacks for incoming connections as well as for error handling.
*/
class acceptor_manager : public manager {
public:
~acceptor_manager();
/**
* Called by the underlying IO device to indicate that
* a new connection is awaiting acceptance.
*/
virtual void new_connection() = 0;
};
} // namespace network
} // namespace io
} // namespace caf
#endif // CAF_IO_NETWORK_ACCEPTOR_MANAGER_HPP
......@@ -17,10 +17,11 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_IO_NETWORK_HPP
#define CAF_IO_NETWORK_HPP
#ifndef CAF_IO_NETWORK_DEFAULT_MULTIPLEXER_HPP
#define CAF_IO_NETWORK_DEFAULT_MULTIPLEXER_HPP
#include <thread>
#include <vector>
#include <string>
#include <cstdint>
......@@ -36,6 +37,12 @@
#include "caf/io/accept_handle.hpp"
#include "caf/io/receive_policy.hpp"
#include "caf/io/connection_handle.hpp"
#include "caf/io/network/operation.hpp"
#include "caf/io/network/multiplexer.hpp"
#include "caf/io/network/stream_manager.hpp"
#include "caf/io/network/acceptor_manager.hpp"
#include "caf/io/network/native_socket.hpp"
#include "caf/detail/logging.hpp"
......@@ -77,12 +84,10 @@ namespace network {
// annoying platform-dependent bootstrapping
#ifdef CAF_WINDOWS
using native_socket_t = SOCKET;
using setsockopt_ptr = const char*;
using socket_send_ptr = const char*;
using socket_recv_ptr = char*;
using socklen_t = int;
constexpr native_socket_t invalid_socket = INVALID_SOCKET;
inline int last_socket_error() { return WSAGetLastError(); }
inline bool would_block_or_temporarily_unavailable(int errcode) {
return errcode == WSAEWOULDBLOCK || errcode == WSATRY_AGAIN;
......@@ -90,12 +95,10 @@ namespace network {
constexpr int ec_out_of_memory = WSAENOBUFS;
constexpr int ec_interrupted_syscall = WSAEINTR;
#else
using native_socket_t = int;
using setsockopt_ptr = const void*;
using socket_send_ptr = const void*;
using socket_recv_ptr = void*;
constexpr native_socket_t invalid_socket = -1;
inline void closesocket(native_socket_t fd) { close(fd); }
inline void closesocket(int fd) { close(fd); }
inline int last_socket_error() { return errno; }
inline bool would_block_or_temporarily_unavailable(int errcode) {
return errcode == EAGAIN || errcode == EWOULDBLOCK;
......@@ -127,11 +130,6 @@ namespace network {
using multiplexer_poll_shadow_data = native_socket_t;
#endif
/**
* Platform-specific native socket type.
*/
using native_socket = native_socket_t;
/**
* Platform-specific native acceptor socket type.
*/
......@@ -141,7 +139,7 @@ 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_socket ? -1 : static_cast<int64_t>(sock);
return sock == invalid_native_socket ? -1 : static_cast<int64_t>(sock);
}
/**
......@@ -207,27 +205,14 @@ bool write_some(size_t& result, native_socket fd, const void* buf, size_t len);
*/
bool try_accept(native_socket& result, native_socket fd);
/**
* Identifies network IO operations, i.e., read or write.
*/
enum class operation {
read,
write,
propagate_error
};
class multiplexer;
class default_multiplexer;
/**
* A socket IO event handler.
*/
class event_handler {
friend class multiplexer;
public:
event_handler();
event_handler(default_multiplexer& dm);
virtual ~event_handler();
......@@ -244,6 +229,13 @@ class event_handler {
*/
virtual void removed_from_loop(operation op) = 0;
/**
* Returns the `multiplexer` this acceptor belongs to.
*/
inline default_multiplexer& backend() {
return m_backend;
}
/**
* Returns the bit field storing the subscribed events.
*/
......@@ -259,24 +251,52 @@ class event_handler {
}
protected: // used by the epoll implementation
virtual native_socket fd() const = 0;
default_multiplexer& m_backend;
int m_eventbf;
};
class supervisor;
/**
* Low-level socket type used as default.
*/
class default_socket {
public:
using socket_type = default_socket;
default_socket(default_multiplexer& parent,
native_socket sock = invalid_native_socket);
default_socket(default_socket&& other);
default_socket& operator=(default_socket&& other);
~default_socket();
void close_read();
inline native_socket fd() const {
return m_fd;
}
inline native_socket native_handle() const {
return m_fd;
}
inline default_multiplexer& backend() {
return m_parent;
}
private:
default_multiplexer& m_parent;
native_socket m_fd;
};
/**
* Low-level backend for IO multiplexing.
* Low-level socket type used as default.
*/
class multiplexer {
using default_socket_acceptor = default_socket;
struct runnable : extend<memory_managed>::with<mixin::memory_cached> {
virtual void run() = 0;
virtual ~runnable();
};
class default_multiplexer : public multiplexer {
friend class io::middleman; // disambiguate reference
friend class supervisor;
......@@ -289,9 +309,28 @@ class multiplexer {
event_handler* ptr;
};
multiplexer();
connection_handle add_tcp_scribe(broker*, default_socket&& sock);
~multiplexer();
connection_handle add_tcp_scribe(broker*, native_socket fd) override;
connection_handle add_tcp_scribe(broker*, const std::string& h,
uint16_t port) override;
accept_handle add_tcp_doorman(broker*, default_socket&& sock);
accept_handle add_tcp_doorman(broker*, native_socket fd) override;
accept_handle add_tcp_doorman(broker*, uint16_t p, const char* h) override;
void dispatch_runnable(runnable_ptr ptr) override;
default_multiplexer();
~default_multiplexer();
supervisor_ptr make_supervisor() override;
void run() override;
template <class F>
void dispatch(F fun, bool force_delayed_execution = false) {
......@@ -314,8 +353,6 @@ class multiplexer {
void del(operation op, native_socket fd, event_handler* ptr);
void run();
private:
// platform-dependent additional initialization code
......@@ -323,7 +360,7 @@ class multiplexer {
template <class F>
void new_event(F fun, operation op, native_socket fd, event_handler* ptr) {
CAF_REQUIRE(fd != invalid_socket);
CAF_REQUIRE(fd != invalid_native_socket);
CAF_REQUIRE(ptr != nullptr || fd == m_pipe.first);
// the only valid input where ptr == nullptr is our pipe
// read handle which is only registered for reading
......@@ -384,70 +421,7 @@ class multiplexer {
};
multiplexer& get_multiplexer_singleton();
/**
* 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.
*/
class supervisor {
public:
supervisor(multiplexer&);
virtual ~supervisor();
private:
multiplexer& m_multiplexer;
};
/**
* Low-level socket type used as default.
*/
class default_socket {
public:
using socket_type = default_socket;
default_socket(multiplexer& parent, native_socket sock = invalid_socket);
default_socket(default_socket&& other);
default_socket& operator=(default_socket&& other);
~default_socket();
void close_read();
inline native_socket fd() const {
return m_fd;
}
inline native_socket native_handle() const { // ASIO compatible signature
return m_fd;
}
inline multiplexer& backend() {
return m_parent;
}
private:
multiplexer& m_parent;
native_socket m_fd;
};
/**
* Low-level socket type used as default.
*/
using default_socket_acceptor = default_socket;
default_multiplexer& get_multiplexer_singleton();
template <class T>
inline connection_handle conn_hdl_from_socket(const T& sock) {
......@@ -462,59 +436,13 @@ inline accept_handle accept_hdl_from_socket(const T& sock) {
}
/**
* A manager configures an IO device and provides callbacks
* for various IO operations.
*/
class manager : public ref_counted {
public:
virtual ~manager();
/**
* Causes the manager to stop read operations on its IO device.
* Unwritten bytes are still send before the socket will be closed.
*/
virtual void stop_reading() = 0;
/**
* Called by the underlying IO device to report failures.
*/
virtual void io_failure(operation op) = 0;
};
/**
* @relates manager
*/
using manager_ptr = intrusive_ptr<manager>;
/**
* 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();
/**
* Called by the underlying IO device whenever it received data.
*/
virtual void consume(const void* data, size_t num_bytes) = 0;
};
/**
* 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 event_handler {
public:
/**
* A smart pointer to a stream manager.
*/
......@@ -526,15 +454,18 @@ class stream : public event_handler {
*/
using buffer_type = std::vector<char>;
stream(multiplexer& backend) : m_sock(backend), m_writing(false) {
stream(default_multiplexer& backend)
: event_handler(backend),
m_sock(backend),
m_writing(false) {
configure_read(receive_policy::at_most(1024));
}
/**
* Returns the `multiplexer` this stream belongs to.
*/
inline multiplexer& backend() {
return m_sock.backend();
inline default_multiplexer& backend() {
return static_cast<default_multiplexer&>(m_sock.backend());
}
/**
......@@ -679,13 +610,11 @@ class stream : public event_handler {
}
protected:
native_socket fd() const override {
return m_sock.fd();
}
private:
void read_loop() {
m_collected = 0;
switch (m_rd_flag) {
......@@ -729,7 +658,6 @@ class stream : public event_handler {
// reading & writing
Socket m_sock;
// reading
manager_ptr m_reader;
size_t m_threshold;
......@@ -737,32 +665,12 @@ class stream : public event_handler {
size_t m_max;
receive_policy_flag m_rd_flag;
buffer_type m_rd_buf;
// writing
manager_ptr m_writer;
bool m_writing;
size_t m_written;
buffer_type m_wr_buf;
buffer_type m_wr_offline_buf;
};
/**
* An acceptor manager configures an acceptor and provides
* callbacks for incoming connections as well as for error handling.
*/
class acceptor_manager : public manager {
public:
~acceptor_manager();
/**
* Called by the underlying IO device to indicate that
* a new connection is awaiting acceptance.
*/
virtual void new_connection() = 0;
};
/**
......@@ -785,18 +693,13 @@ class acceptor : public event_handler {
*/
using manager_ptr = intrusive_ptr<manager_type>;
acceptor(multiplexer& backend)
: m_backend(backend), m_accept_sock(backend), m_sock(backend) {
acceptor(default_multiplexer& backend)
: event_handler(backend),
m_accept_sock(backend),
m_sock(backend) {
// nop
}
/**
* Returns the `multiplexer` this acceptor belongs to.
*/
inline multiplexer& backend() {
return m_backend;
}
/**
* Returns the IO socket.
*/
......@@ -846,9 +749,9 @@ class acceptor : public event_handler {
CAF_LOG_TRACE("m_accept_sock.fd = " << m_accept_sock.fd()
<< ", op = " << static_cast<int>(op));
if (m_mgr && op == operation::read) {
native_socket fd = invalid_socket;
native_socket fd = invalid_native_socket;
if (try_accept(fd, m_accept_sock.fd())) {
if (fd != invalid_socket) {
if (fd != invalid_native_socket) {
m_sock = socket_type{backend(), fd};
m_mgr->new_connection();
}
......@@ -870,7 +773,6 @@ class acceptor : public event_handler {
private:
multiplexer& m_backend;
manager_ptr m_mgr;
SocketAcceptor m_accept_sock;
socket_type m_sock;
......@@ -903,4 +805,4 @@ void ipv4_bind(SocketAcceptor& sock,
} // namespace io
} // namespace caf
#endif // CAF_IO_NETWORK_HPP
#endif // CAF_IO_NETWORK_DEFAULT_MULTIPLEXER_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2014 *
* Dominik Charousset <dominik.charousset (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 LICENCE_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_MANAGER_HPP
#define CAF_IO_NETWORK_MANAGER_HPP
#include "caf/ref_counted.hpp"
#include "caf/intrusive_ptr.hpp"
#include "caf/io/network/operation.hpp"
namespace caf {
namespace io {
namespace network {
/**
* A manager configures an IO device and provides callbacks
* for various IO operations.
*/
class manager : public ref_counted {
public:
virtual ~manager();
/**
* Causes the manager to stop read operations on its IO device.
* Unwritten bytes are still send before the socket will be closed.
*/
virtual void stop_reading() = 0;
/**
* Called by the underlying IO device to report failures.
*/
virtual void io_failure(operation op) = 0;
};
/**
* @relates manager
*/
using manager_ptr = intrusive_ptr<manager>;
} // namespace network
} // namespace io
} // namespace caf
#endif // CAF_IO_NETWORK_MANAGER_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2014 *
* Dominik Charousset <dominik.charousset (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 LICENCE_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_MULTIPLEXER_HPP
#define CAF_IO_NETWORK_MULTIPLEXER_HPP
#include <string>
#include <thread>
#include <functional>
#include "caf/extend.hpp"
#include "caf/memory_managed.hpp"
#include "caf/io/fwd.hpp"
#include "caf/io/accept_handle.hpp"
#include "caf/io/connection_handle.hpp"
#include "caf/io/network/native_socket.hpp"
#include "caf/mixin/memory_cached.hpp"
#include "caf/detail/memory.hpp"
namespace boost {
namespace asio {
class io_service;
} // namespace asio
} // namespace boost
namespace caf {
namespace io {
namespace network {
/**
* Low-level backend for IO multiplexing.
*/
class multiplexer {
public:
virtual ~multiplexer();
/**
* Creates a new TCP doorman from a native socket handle.
*/
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.
*/
virtual connection_handle add_tcp_scribe(broker* ptr, const std::string& h,
uint16_t port) = 0;
/**
* Creates a new TCP doorman from a native socket handle.
*/
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 host `h`.
*/
virtual accept_handle add_tcp_doorman(broker* ptr, uint16_t p,
const char* h = nullptr) = 0;
/**
* Simple wrapper for runnables
*/
struct runnable : extend<memory_managed>::with<mixin::memory_cached> {
virtual void run() = 0;
virtual ~runnable();
};
using runnable_ptr = std::unique_ptr<runnable, detail::disposer>;
/**
* Makes sure the multipler does not exit its event loop until
* the destructor of `supervisor` has been called.
*/
struct supervisor {
public:
virtual ~supervisor();
};
using supervisor_ptr = std::unique_ptr<supervisor>;
/**
* Creates a supervisor to keep the event loop running.
*/
virtual supervisor_ptr make_supervisor() = 0;
/**
* Creates an instance using the networking backend compiled with CAF.
*/
static std::unique_ptr<multiplexer> make();
/**
* Runs the multiplexers event loop.
*/
virtual void run() = 0;
/**
* Invokes @p fun in the multiplexer's event loop.
*/
template <class F>
void dispatch(F fun) {
if (std::this_thread::get_id() == thread_id()) {
fun();
return;
}
struct impl : runnable {
F f;
impl(F&& mf) : f(std::move(mf)) { }
void run() override {
f();
}
};
dispatch_runnable(runnable_ptr{detail::memory::create<impl>(std::move(fun))});
}
/**
* Retrieves a pointer to the implementation or `nullptr` if CAF was
* compiled using the default backend.
*/
virtual boost::asio::io_service* pimpl();
inline const std::thread::id& thread_id() const {
return m_tid;
}
inline void thread_id(std::thread::id tid) {
m_tid = std::move(tid);
}
protected:
/**
* Implementation-specific dispatching to the multiplexer's thread.
*/
virtual void dispatch_runnable(runnable_ptr ptr) = 0;
/**
* Must be set by the subclass.
*/
std::thread::id m_tid;
};
using multiplexer_ptr = std::unique_ptr<multiplexer>;
} // namespace network
} // namespace io
} // namespace caf
#endif // CAF_IO_NETWORK_MULTIPLEXER_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2014 *
* Dominik Charousset <dominik.charousset (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 LICENCE_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_NATIVE_SOCKET_HPP
#define CAF_IO_NETWORK_NATIVE_SOCKET_HPP
#include "caf/config.hpp"
#ifdef CAF_WINDOWS
# include <winsock2.h>
#endif
namespace caf {
namespace io {
namespace network {
#ifdef CAF_WINDOWS
using native_socket = SOCKET;
constexpr native_socket invalid_native_socket = INVALID_SOCKET;
#else
using native_socket = int;
constexpr native_socket invalid_native_socket = -1;
#endif
} // namespace network
} // namespace io
} // namespace caf
#endif // CAF_IO_NETWORK_NATIVE_SOCKET_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2014 *
* Dominik Charousset <dominik.charousset (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 LICENCE_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_OPERATION_HPP
#define CAF_IO_NETWORK_OPERATION_HPP
namespace caf {
namespace io {
namespace network {
/**
* Identifies network IO operations, i.e., read or write.
*/
enum class operation {
read,
write,
propagate_error
};
} // namespace network
} // namespace io
} // namespace caf
#endif // CAF_IO_NETWORK_OPERATION_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2014 *
* Dominik Charousset <dominik.charousset (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 LICENCE_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_STREAM_MANAGER_HPP
#define CAF_IO_NETWORK_STREAM_MANAGER_HPP
#include <cstddef>
#include "caf/io/network/manager.hpp"
namespace caf {
namespace io {
namespace network {
/**
* 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();
/**
* Called by the underlying IO device whenever it received data.
*/
virtual void consume(const void* data, size_t num_bytes) = 0;
};
} // namespace network
} // namespace io
} // namespace caf
#endif // CAF_IO_NETWORK_STREAM_MANAGER_HPP
......@@ -20,38 +20,40 @@
#ifndef CAF_IO_PUBLISH_IMPL_HPP
#define CAF_IO_PUBLISH_IMPL_HPP
#include <future>
#include "caf/actor_cast.hpp"
#include "caf/abstract_actor.hpp"
#include "caf/detail/singletons.hpp"
#include "caf/detail/actor_registry.hpp"
#include "caf/io/network.hpp"
#include "caf/io/middleman.hpp"
#include "caf/io/basp_broker.hpp"
namespace caf {
namespace io {
template <class ActorHandle, class SocketAcceptor>
void publish_impl(ActorHandle whom, SocketAcceptor fd, uint16_t port) {
template <class... Ts>
void publish_impl(abstract_actor_ptr whom, uint16_t port, Ts&&... args) {
using namespace detail;
auto mm = middleman::instance();
// we can't move fd into our lambda in C++11 ...
using pair_type = std::pair<ActorHandle, SocketAcceptor>;
auto data = std::make_shared<pair_type>(whom, std::move(fd));
mm->run_later([mm, data, whom, port] {
std::promise<bool> res;
mm->run_later([&] {
auto bro = mm->get_named_broker<basp_broker>(atom("_BASP"));
bro->publish(std::move(data->first), std::move(data->second));
try {
auto hdl = mm->backend().add_tcp_doorman(bro.get(), port,
std::forward<Ts>(args)...);
bro->announce_published_actor(hdl, whom);
mm->notify<hook::actor_published>(whom->address(), port);
res.set_value(true);
}
catch (...) {
res.set_exception(std::current_exception());
}
});
}
inline void publish_impl(abstract_actor_ptr whom, uint16_t port,
const char* ipaddr) {
using namespace detail;
auto mm = middleman::instance();
network::default_socket_acceptor fd{mm->backend()};
network::ipv4_bind(fd, port, ipaddr);
publish_impl(std::move(whom), std::move(fd), port);
// block caller and re-throw exception here in case of an error
res.get_future().get();
}
} // namespace io
......
......@@ -56,7 +56,7 @@ inline actor remote_actor(Socket fd) {
* @throws std::invalid_argument Thrown when connecting to a typed actor.
*/
inline actor remote_actor(const char* host, uint16_t port) {
auto res = remote_actor_impl(host, port, std::set<std::string>{});
auto res = remote_actor_impl(std::set<std::string>{}, host, port);
return actor_cast<actor>(res);
}
......@@ -64,7 +64,7 @@ inline actor remote_actor(const char* host, uint16_t port) {
* @copydoc remote_actor(const char*, uint16_t)
*/
inline actor remote_actor(const std::string& host, uint16_t port) {
auto res = remote_actor_impl(host, port, std::set<std::string>{});
auto res = remote_actor_impl(std::set<std::string>{}, host, port);
return actor_cast<actor>(res);
}
......
......@@ -31,9 +31,9 @@
#include "caf/abstract_actor.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/io/network.hpp"
#include "caf/io/middleman.hpp"
#include "caf/io/basp_broker.hpp"
#include "caf/io/network/multiplexer.hpp"
#include "caf/detail/logging.hpp"
......@@ -44,36 +44,34 @@ constexpr uint32_t max_iface_size = 100;
constexpr uint32_t max_iface_clause_size = 500;
template <class Socket>
abstract_actor_ptr remote_actor_impl(Socket fd,
const std::set<std::string>& ifs) {
using string_set = std::set<std::string>;
template <class... Ts>
abstract_actor_ptr remote_actor_impl(const std::set<std::string>& ifs,
Ts&&... args) {
auto mm = middleman::instance();
std::string error_msg;
std::promise<abstract_actor_ptr> result_promise;
// we can't move fd into our lambda in C++11 ...
auto fd_ptr = std::make_shared<Socket>(std::move(fd));
basp_broker::client_handshake_data hdata{
invalid_node_id, &result_promise, &error_msg, &ifs};
auto hdata_ptr = &hdata;
mm->run_later([=] {
basp_broker::client_handshake_data hdata{invalid_node_id, &result_promise,
&error_msg, &ifs};
mm->run_later([&] {
try {
auto bro = mm->get_named_broker<basp_broker>(atom("_BASP"));
auto hdl = bro->add_connection(std::move(*fd_ptr));
bro->init_client(hdl, hdata_ptr);
auto hdl = mm->backend().add_tcp_scribe(bro.get(),
std::forward<Ts>(args)...);
bro->init_client(hdl, &hdata);
}
catch (...) {
result_promise.set_exception(std::current_exception());
}
});
auto result = result_promise.get_future().get();
if (!result) throw std::runtime_error(error_msg);
if (!result) {
throw std::runtime_error(error_msg);
}
return result;
}
inline abstract_actor_ptr
remote_actor_impl(const std::string& host, uint16_t port,
const std::set<std::string>& ifs) {
auto mm = middleman::instance();
network::default_socket fd{mm->backend()};
network::ipv4_connect(fd, host, port);
return remote_actor_impl(std::move(fd), ifs);
}
} // namespace io
} // namespace caf
......
......@@ -26,6 +26,8 @@
#include "caf/io/middleman.hpp"
#include "caf/io/connection_handle.hpp"
#include "caf/io/network/native_socket.hpp"
namespace caf {
namespace io {
......@@ -34,9 +36,8 @@ namespace io {
*/
template <spawn_options Os = no_spawn_options,
typename F = std::function<void(broker*)>, class... Ts>
actor spawn_io(F fun, Ts&&... args) {
return spawn<broker::functor_based>(std::move(fun),
std::forward<Ts>(args)...);
actor spawn_io(F fun, Ts&&... vs) {
return spawn<broker::functor_based>(std::move(fun), std::forward<Ts>(vs)...);
}
/**
......@@ -45,63 +46,36 @@ actor spawn_io(F fun, Ts&&... args) {
template <spawn_options Os = no_spawn_options,
typename F = std::function<void(broker*)>, class... Ts>
actor spawn_io_client(F fun, const std::string& host,
uint16_t port, Ts&&... args) {
network::default_socket sock{middleman::instance()->backend()};
network::ipv4_connect(sock, host, port);
auto hdl = network::conn_hdl_from_socket(sock);
uint16_t port, Ts&&... vs) {
// provoke compiler error early
using fun_res = decltype(fun((broker*) 0, hdl, std::forward<Ts>(args)...));
using fun_res = decltype(fun(static_cast<broker*>(nullptr),
connection_handle{},
std::forward<Ts>(vs)...));
// prevent warning about unused local type
static_assert(std::is_same<fun_res, fun_res>::value,
"your compiler is lying to you");
return spawn_class<broker::functor_based>(
nullptr,
[&](broker* ptr) {
auto hdl2 = ptr->add_connection(std::move(sock));
CAF_REQUIRE(hdl == hdl2);
static_cast<void>(hdl2); // prevent warning
},
std::move(fun), hdl, std::forward<Ts>(args)...);
[&](broker::functor_based* ptr) {
auto mm = middleman::instance();
auto hdl = mm->backend().add_tcp_scribe(ptr, host, port);
ptr->init(std::move(fun), hdl, std::forward<Ts>(vs)...);
});
}
struct is_socket_test {
template <class T>
static auto test(T* ptr) -> decltype(ptr->native_handle(),
std::true_type{});
template <class>
static auto test(...) -> std::false_type;
};
template <class T>
struct is_socket : decltype(is_socket_test::test<T>(0)) {
};
/**
* Spawns a new broker as server running on given `port`.
*/
template <spawn_options Os = no_spawn_options,
typename F = std::function<void(broker*)>,
typename Socket = network::default_socket_acceptor, class... Ts>
typename std::enable_if<is_socket<Socket>::value, actor>::type
spawn_io_server(F fun, Socket sock, Ts&&... args) {
class F = std::function<void(broker*)>, class... Ts>
actor spawn_io_server(F fun, uint16_t port, Ts&&... vs) {
return spawn_class<broker::functor_based>(
nullptr,
[&](broker* ptr) {
ptr->add_acceptor(std::move(sock));
},
std::move(fun), std::forward<Ts>(args)...);
}
/**
* Spawns a new broker as server running on given `port`.
*/
template <spawn_options Os = no_spawn_options,
typename F = std::function<void(broker*)>, class... Ts>
actor spawn_io_server(F fun, uint16_t port, Ts&&... args) {
network::default_socket_acceptor fd{middleman::instance()->backend()};
network::ipv4_bind(fd, port);
return spawn_io_server(std::move(fun), std::move(fd),
std::forward<Ts>(args)...);
[&](broker::functor_based* ptr) {
auto mm = middleman::instance();
mm->backend().add_tcp_doorman(ptr, port);
ptr->init(std::move(fun), std::forward<Ts>(vs)...);
});
}
} // namespace io
......
......@@ -41,7 +41,7 @@ struct typed_remote_actor_helper<List<Ts...>> {
template <class... Vs>
return_type operator()(Vs&&... vs) {
auto iface = return_type::message_types();
auto tmp = remote_actor_impl(std::forward<Vs>(vs)..., std::move(iface));
auto tmp = remote_actor_impl(std::move(iface), std::forward<Vs>(vs)...);
return actor_cast<return_type>(tmp);
}
};
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2014 *
* Dominik Charousset <dominik.charousset (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 LICENCE_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/io/network/acceptor_manager.hpp"
namespace caf {
namespace io {
namespace network {
acceptor_manager::~acceptor_manager() {
// nop
}
} // namespace network
} // namespace io
} // namespace caf
......@@ -17,10 +17,12 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/io/network/default_multiplexer.hpp"
#include "caf/config.hpp"
#include "caf/exception.hpp"
#include "caf/io/network.hpp"
#include "caf/io/broker.hpp"
#include "caf/io/middleman.hpp"
#ifdef CAF_WINDOWS
......@@ -151,9 +153,9 @@ namespace network {
\**************************************************************************/
std::pair<native_socket, native_socket> create_pipe() {
socklen_t addrlen = sizeof(sockaddr_in);
native_socket socks[2] = {invalid_socket, invalid_socket};
native_socket socks[2] = {invalid_native_socket, invalid_native_socket};
auto listener = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
if (listener == invalid_socket) {
if (listener == invalid_native_socket) {
throw_io_failure("socket() failed");
}
union {
......@@ -193,13 +195,13 @@ namespace network {
// create read-only end of the pipe
DWORD flags = 0;
auto read_fd = WSASocket(AF_INET, SOCK_STREAM, 0, NULL, 0, flags);
if (read_fd == invalid_socket) {
if (read_fd == invalid_native_socket) {
throw_io_failure("cannot create read handle: WSASocket() failed");
}
ccall("connect() failed", ::connect, read_fd, &a.addr, sizeof(a.inaddr));
// get write-only end of the pipe
auto write_fd = accept(listener, NULL, NULL);
if (write_fd == invalid_socket) {
if (write_fd == invalid_native_socket) {
throw_io_failure("cannot create write handle: accept() failed");
}
closesocket(listener);
......@@ -218,7 +220,9 @@ namespace network {
// In this implementation, m_shadow is the number of sockets we have
// registered to epoll.
multiplexer::multiplexer() : m_epollfd(invalid_socket), m_shadow(1) {
default_multiplexer::default_multiplexer()
: m_epollfd(invalid_native_socket),
m_shadow(1) {
init();
m_epollfd = epoll_create1(EPOLL_CLOEXEC);
if (m_epollfd == -1) {
......@@ -237,7 +241,7 @@ namespace network {
}
}
void multiplexer::run() {
void default_multiplexer::run() {
CAF_LOG_TRACE("epoll()-based multiplexer");
while (m_shadow > 0) {
int presult = epoll_wait(m_epollfd, m_pollset.data(),
......@@ -271,7 +275,7 @@ namespace network {
}
}
void multiplexer::handle(const multiplexer::event& e) {
void default_multiplexer::handle(const default_multiplexer::event& e) {
CAF_LOG_TRACE("e.fd = " << e.fd << ", mask = " << e.mask);
// ptr is only allowed to nullptr if fd is our pipe
// read handle which is only registered for input
......@@ -349,7 +353,7 @@ namespace network {
// are sorted by the file descriptor. This allows us to quickly,
// i.e., O(1), access the actual object when handling socket events.
multiplexer::multiplexer() : m_epollfd(-1) {
default_multiplexer::default_multiplexer() : m_epollfd(-1) {
init();
// initial setup
m_pipe = create_pipe();
......@@ -361,7 +365,7 @@ namespace network {
m_shadow.push_back(nullptr);
}
void multiplexer::run() {
void default_multiplexer::run() {
CAF_LOG_TRACE("poll()-based multiplexer; " << CAF_ARG(input_mask)
<< ", " << CAF_ARG(output_mask)
<< ", " << CAF_ARG(error_mask));
......@@ -369,7 +373,7 @@ namespace network {
// altering the pollset while traversing it is not exactly a
// bright idea ...
struct fd_event {
native_socket_t fd; // our file descriptor
native_socket fd; // our file descriptor
short mask; // the event mask returned by poll()
short fd_mask; // the mask associated with fd
event_handler* ptr; // nullptr in case of a pipe event
......@@ -431,8 +435,8 @@ namespace network {
}
}
void multiplexer::handle(const multiplexer::event& e) {
CAF_REQUIRE(e.fd != invalid_socket);
void default_multiplexer::handle(const default_multiplexer::event& e) {
CAF_REQUIRE(e.fd != invalid_native_socket);
CAF_REQUIRE(m_pollset.size() == m_shadow.size());
CAF_LOGF_TRACE("fd = " << e.fd
<< ", old mask = " << (e.ptr ? e.ptr->eventbf() : -1)
......@@ -521,8 +525,8 @@ int del_flag(operation op, int bf) {
return 0;
}
void multiplexer::add(operation op, native_socket fd, event_handler* ptr) {
CAF_REQUIRE(fd != invalid_socket);
void default_multiplexer::add(operation op, native_socket fd, event_handler* ptr) {
CAF_REQUIRE(fd != invalid_native_socket);
// ptr == nullptr is only allowed to store our pipe read handle
// and the pipe read handle is added in the ctor (not allowed here)
CAF_REQUIRE(ptr != nullptr);
......@@ -531,8 +535,8 @@ void multiplexer::add(operation op, native_socket fd, event_handler* ptr) {
new_event(add_flag, op, fd, ptr);
}
void multiplexer::del(operation op, native_socket fd, event_handler* ptr) {
CAF_REQUIRE(fd != invalid_socket);
void default_multiplexer::del(operation op, native_socket fd, event_handler* ptr) {
CAF_REQUIRE(fd != invalid_native_socket);
// ptr == nullptr is only allowed when removing our pipe read handle
CAF_REQUIRE(ptr != nullptr || fd == m_pipe.first);
CAF_LOG_TRACE(CAF_TARG(op, static_cast<int>)<< ", " << CAF_ARG(fd)
......@@ -540,7 +544,7 @@ void multiplexer::del(operation op, native_socket fd, event_handler* ptr) {
new_event(del_flag, op, fd, ptr);
}
void multiplexer::wr_dispatch_request(runnable* ptr) {
void default_multiplexer::wr_dispatch_request(runnable* ptr) {
intptr_t ptrval = reinterpret_cast<intptr_t>(ptr);
// on windows, we actually have sockets, otherwise we have file handles
# ifdef CAF_WINDOWS
......@@ -551,7 +555,7 @@ void multiplexer::wr_dispatch_request(runnable* ptr) {
# endif
}
multiplexer::runnable* multiplexer::rd_dispatch_request() {
default_multiplexer::runnable* default_multiplexer::rd_dispatch_request() {
intptr_t ptrval;
// on windows, we actually have sockets, otherwise we have file handles
# ifdef CAF_WINDOWS
......@@ -563,20 +567,32 @@ multiplexer::runnable* multiplexer::rd_dispatch_request() {
return reinterpret_cast<runnable*>(ptrval);;
}
multiplexer::runnable::~runnable() {
// nop
default_multiplexer& get_multiplexer_singleton() {
return static_cast<default_multiplexer&>(middleman::instance()->backend());
}
multiplexer& get_multiplexer_singleton() {
return middleman::instance()->backend();
multiplexer::supervisor_ptr default_multiplexer::make_supervisor() {
class impl : public multiplexer::supervisor {
public:
impl(default_multiplexer* thisptr) : m_this(thisptr) {
// nop
}
~impl() {
auto ptr = m_this;
ptr->dispatch([=] { ptr->close_pipe(); });
}
private:
default_multiplexer* m_this;
};
return supervisor_ptr{new impl(this)};
}
void multiplexer::close_pipe() {
void default_multiplexer::close_pipe() {
CAF_LOG_TRACE("");
del(operation::read, m_pipe.first, nullptr);
}
void multiplexer::handle_socket_event(native_socket fd, int mask,
void default_multiplexer::handle_socket_event(native_socket fd, int mask,
event_handler* ptr) {
CAF_LOG_TRACE(CAF_ARG(fd) << ", " << CAF_ARG(mask));
bool checkerror = true;
......@@ -621,7 +637,7 @@ void multiplexer::handle_socket_event(native_socket fd, int mask,
"handling error");
}
void multiplexer::init() {
void default_multiplexer::init() {
# ifdef CAF_WINDOWS
WSADATA WinsockData;
if (WSAStartup(MAKEWORD(2, 2), &WinsockData) != 0) {
......@@ -630,17 +646,123 @@ void multiplexer::init() {
# endif
}
multiplexer::~multiplexer() {
default_multiplexer::~default_multiplexer() {
# ifdef CAF_WINDOWS
WSACleanup();
# endif
if (m_epollfd != invalid_socket) {
if (m_epollfd != invalid_native_socket) {
closesocket(m_epollfd);
}
closesocket(m_pipe.first);
closesocket(m_pipe.second);
}
void default_multiplexer::dispatch_runnable(runnable_ptr ptr) {
wr_dispatch_request(ptr.release());
}
connection_handle default_multiplexer::add_tcp_scribe(broker* self,
default_socket&& sock) {
CAF_LOG_TRACE("");
class impl : public broker::scribe {
public:
impl(broker* parent, default_socket&& s)
: scribe(parent, network::conn_hdl_from_socket(s)),
m_launched(false),
m_stream(s.backend()) {
m_stream.init(std::move(s));
}
void configure_read(receive_policy::config config) override {
CAF_LOGM_TRACE("caf::io::broker::scribe", "");
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_LOGM_TRACE("caf::io::broker::scribe", "");
m_stream.stop_reading();
disconnect();
}
void flush() override {
CAF_LOGM_TRACE("caf::io::broker::scribe", "");
m_stream.flush(this);
}
void launch() {
CAF_LOGM_TRACE("caf::io::broker::scribe", "");
CAF_REQUIRE(!m_launched);
m_launched = true;
m_stream.start(this);
}
private:
bool m_launched;
stream<default_socket> m_stream;
};
broker::scribe_pointer ptr{new impl{self, std::move(sock)}};
self->add_scribe(ptr);
return ptr->hdl();
}
accept_handle default_multiplexer::add_tcp_doorman(broker* self,
default_socket_acceptor&& sock) {
CAF_LOG_TRACE("sock.fd = " << sock.fd());
CAF_REQUIRE(sock.fd() != network::invalid_native_socket);
class impl : public broker::doorman {
public:
impl(broker* parent, default_socket_acceptor&& s)
: doorman(parent, network::accept_hdl_from_socket(s)),
m_acceptor(s.backend()) {
m_acceptor.init(std::move(s));
}
void new_connection() override {
auto& dm = m_acceptor.backend();
accept_msg().handle = dm.add_tcp_scribe(parent(),
std::move(m_acceptor.accepted_socket()));
parent()->invoke_message(invalid_actor_addr,
message_id::invalid,
m_accept_msg);
}
void stop_reading() override {
m_acceptor.stop_reading();
disconnect();
}
void launch() override {
m_acceptor.start(this);
}
private:
network::acceptor<default_socket_acceptor> m_acceptor;
};
broker::doorman_pointer ptr{new impl{self, std::move(sock)}};
self->add_doorman(ptr);
return ptr->hdl();
}
connection_handle default_multiplexer::add_tcp_scribe(broker* self,
native_socket fd) {
return add_tcp_scribe(self, default_socket{*this, fd});
}
connection_handle default_multiplexer::add_tcp_scribe(broker* self,
const std::string& host,
uint16_t port) {
return add_tcp_scribe(self, new_ipv4_connection(host, port));
}
accept_handle default_multiplexer::add_tcp_doorman(broker* self,
native_socket fd) {
return add_tcp_doorman(self, default_socket_acceptor{*this, fd});
}
accept_handle default_multiplexer::add_tcp_doorman(broker* self,
uint16_t port,
const char* host) {
return add_tcp_doorman(self, new_ipv4_acceptor(port, host));
}
/******************************************************************************
* platform-independent implementations (finally) *
******************************************************************************/
......@@ -707,7 +829,7 @@ bool try_accept(native_socket& result, native_socket fd) {
result = ::accept(fd, &addr, &addrlen);
CAF_LOGF_DEBUG("tried to accept a new connection from from socket "
<< fd << ", accept returned " << result);
if (result == invalid_socket) {
if (result == invalid_native_socket) {
auto err = last_socket_error();
if (!would_block_or_temporarily_unavailable(err)) {
return false;
......@@ -716,7 +838,9 @@ bool try_accept(native_socket& result, native_socket fd) {
return true;
}
event_handler::event_handler() : m_eventbf(0) {
event_handler::event_handler(default_multiplexer& dm)
: m_backend(dm),
m_eventbf(0) {
// nop
}
......@@ -724,19 +848,11 @@ event_handler::~event_handler() {
// nop
}
supervisor::supervisor(multiplexer& mp) : m_multiplexer(mp) {
// nop
}
supervisor::~supervisor() {
auto mptr = &m_multiplexer;
m_multiplexer.dispatch([=] { mptr->close_pipe(); }, true);
}
default_socket::default_socket(multiplexer& parent, native_socket fd)
: m_parent(parent), m_fd(fd) {
default_socket::default_socket(default_multiplexer& parent, native_socket fd)
: m_parent(parent),
m_fd(fd) {
CAF_LOG_TRACE(CAF_ARG(fd));
if (fd != invalid_socket) {
if (fd != invalid_native_socket) {
// enable nonblocking IO & disable Nagle's algorithm
nonblocking(m_fd, true);
tcp_nodelay(m_fd, true);
......@@ -745,7 +861,7 @@ default_socket::default_socket(multiplexer& parent, native_socket fd)
default_socket::default_socket(default_socket&& other)
: m_parent(other.m_parent), m_fd(other.m_fd) {
other.m_fd = invalid_socket;
other.m_fd = invalid_native_socket;
}
default_socket& default_socket::operator=(default_socket&& other) {
......@@ -754,41 +870,29 @@ default_socket& default_socket::operator=(default_socket&& other) {
}
default_socket::~default_socket() {
if (m_fd != invalid_socket) {
if (m_fd != invalid_native_socket) {
CAF_LOG_DEBUG("close socket " << m_fd);
closesocket(m_fd);
}
}
void default_socket::close_read() {
if (m_fd != invalid_socket) {
if (m_fd != invalid_native_socket) {
::shutdown(m_fd, 0); // 0 identifyies the read channel on Win & UNIX
}
}
manager::~manager() {
// nop
}
stream_manager::~stream_manager() {
// nop
}
acceptor_manager::~acceptor_manager() {
// nop
}
struct socket_guard {
public:
socket_guard(native_socket fd) : m_fd(fd) {
// nop
}
~socket_guard() {
if (m_fd != invalid_socket)
if (m_fd != invalid_native_socket)
closesocket(m_fd);
}
void release() {
m_fd = invalid_socket;
m_fd = invalid_native_socket;
}
private:
native_socket m_fd;
......@@ -804,7 +908,7 @@ native_socket new_ipv4_connection_impl(const std::string& host, uint16_t port) {
sockaddr_in serv_addr;
hostent* server;
native_socket fd = socket(AF_INET, SOCK_STREAM, 0);
if (fd == invalid_socket) {
if (fd == invalid_native_socket) {
throw network_error("socket creation failed");
}
socket_guard sguard(fd);
......@@ -841,7 +945,7 @@ native_socket new_ipv4_acceptor_impl(uint16_t port, const char* addr) {
get_multiplexer_singleton();
# endif
native_socket fd = socket(AF_INET, SOCK_STREAM, 0);
if (fd == invalid_socket) {
if (fd == invalid_native_socket) {
throw network_error("could not create server socket");
}
// sguard closes the socket in case of exception
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2014 *
* Dominik Charousset <dominik.charousset (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 LICENCE_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/io/network/manager.hpp"
namespace caf {
namespace io {
namespace network {
manager::~manager() {
// nop
}
} // namespace network
} // namespace io
} // namespace caf
......@@ -201,12 +201,13 @@ void middleman::add_broker(broker_ptr bptr) {
void middleman::initialize() {
CAF_LOG_TRACE("");
m_supervisor = new network::supervisor{m_backend};
m_backend = network::multiplexer::make();
m_supervisor = m_backend->make_supervisor();
m_thread = std::thread([this] {
CAF_LOGC_TRACE("caf::io::middleman", "initialize$run", "");
m_backend.run();
m_backend->run();
});
m_backend.m_tid = m_thread.get_id();
m_backend->thread_id(m_thread.get_id());
// announce io-related types
announce_helper<new_data_msg, new_connection_msg,
acceptor_closed_msg, connection_closed_msg,
......@@ -217,10 +218,8 @@ void middleman::initialize() {
void middleman::stop() {
CAF_LOG_TRACE("");
m_backend.dispatch([=] {
m_backend->dispatch([=] {
CAF_LOGC_TRACE("caf::io::middleman", "stop$lambda", "");
delete m_supervisor;
m_supervisor = nullptr;
// m_managers will be modified while we are stopping each manager,
// because each manager will call remove(...)
std::vector<broker_ptr> brokers;
......@@ -231,6 +230,7 @@ void middleman::stop() {
bro->close_all();
}
});
m_supervisor.reset();
m_thread.join();
m_named_brokers.clear();
}
......@@ -239,7 +239,7 @@ void middleman::dispose() {
delete this;
}
middleman::middleman() : m_supervisor(nullptr) {
middleman::middleman() {
// nop
}
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2014 *
* Dominik Charousset <dominik.charousset (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 LICENCE_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/io/network/multiplexer.hpp"
#ifdef CAF_USE_ASIO
# include "caf/io/network/asio_multiplexer.hpp"
using caf_multiplexer_impl = caf::io::network::asio_multiplexer;
#else
# include "caf/io/network/default_multiplexer.hpp"
using caf_multiplexer_impl = caf::io::network::default_multiplexer;
#endif
namespace caf {
namespace io {
namespace network {
multiplexer::~multiplexer() {
// nop
}
boost::asio::io_service* pimpl() {
return nullptr;
}
multiplexer_ptr multiplexer::make() {
return multiplexer_ptr{new caf_multiplexer_impl};
}
boost::asio::io_service* multiplexer::pimpl() {
return nullptr;
}
multiplexer::supervisor::~supervisor() {
// nop
}
multiplexer::runnable::~runnable() {
// nop
}
} // namespace network
} // namespace io
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2014 *
* Dominik Charousset <dominik.charousset (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 LICENCE_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/io/network/stream_manager.hpp"
namespace caf {
namespace io {
namespace network {
stream_manager::~stream_manager() {
// nop
}
} // namespace 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