Commit e0e0012e authored by Joseph Noir's avatar Joseph Noir

Split up newb test file

parent 7673ab3f
...@@ -51,7 +51,8 @@ add(remoting group_chat) ...@@ -51,7 +51,8 @@ add(remoting group_chat)
add(remoting group_server) add(remoting group_server)
add(remoting remote_spawn) add(remoting remote_spawn)
add(remoting distributed_calculator) add(remoting distributed_calculator)
add(remoting newbc) add(remoting udp_newb)
add(remoting tcp_newb)
# basic I/O with brokers # basic I/O with brokers
add(broker simple_broker) add(broker simple_broker)
......
#include <mutex>
#include <thread>
#include <iostream>
#include <condition_variable>
#include "caf/all.hpp"
#include "caf/io/all.hpp"
#include "caf/io/network/newb.hpp"
#include "caf/io/network/default_multiplexer.hpp"
using namespace caf;
using namespace caf::io::network;
namespace {
class config : public actor_system_config {
public:
uint16_t port = 0;
std::string host = "localhost";
bool server_mode = false;
config() {
opt_group{custom_options_, "global"}
.add(port, "port,p", "set port")
.add(host, "host,H", "set host (ignored in server mode)")
.add(server_mode, "server-mode,s", "enable server mode");
}
};
void caf_main(actor_system& system, const config&) {
default_multiplexer mpx{&system};
// Setup thread to run the multiplexer.
std::thread t;
std::atomic<bool> init_done{false};
std::mutex mtx;
std::condition_variable cv;
t = std::thread{[&] {
system.thread_started();
std::cout << "starting multiplexer" << std::endl;
{
std::unique_lock<std::mutex> guard{mtx};
mpx.thread_id(std::this_thread::get_id());
init_done = true;
cv.notify_one();
}
mpx.run();
system.thread_terminates();
}};
std::unique_lock<std::mutex> guard{mtx};
while (init_done == false)
cv.wait(guard);
// Create an event handling actor to run in the multiplexer.
actor_config cfg{&mpx};
auto n = make_newb<detail::protocol_policy,
detail::mutating_policy>(system, cfg, mpx, -1);
anon_send(n, 1);
t.join();
}
} // namespace <anonymous>
CAF_MAIN(io::middleman)
#include "caf/io/network/newb.hpp"
#include "caf/config.hpp"
#include "caf/logger.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/detail/call_cfun.hpp"
#ifdef CAF_WINDOWS
# ifndef WIN32_LEAN_AND_MEAN
# define WIN32_LEAN_AND_MEAN
# endif // WIN32_LEAN_AND_MEAN
# ifndef NOMINMAX
# define NOMINMAX
# endif
# ifdef CAF_MINGW
# undef _WIN32_WINNT
# undef WINVER
# define _WIN32_WINNT WindowsVista
# define WINVER WindowsVista
# include <w32api.h>
# endif
# include <io.h>
# include <windows.h>
# include <winsock2.h>
# include <ws2ipdef.h>
# include <ws2tcpip.h>
#else
# include <unistd.h>
# include <arpa/inet.h>
# include <cerrno>
# include <fcntl.h>
# include <netdb.h>
# include <netinet/in.h>
# include <netinet/ip.h>
# include <netinet/tcp.h>
# include <sys/socket.h>
# include <sys/types.h>
# ifdef CAF_POLL_MULTIPLEXER
# include <poll.h>
# elif defined(CAF_EPOLL_MULTIPLEXER)
# include <sys/epoll.h>
# else
# error "neither CAF_POLL_MULTIPLEXER nor CAF_EPOLL_MULTIPLEXER defined"
# endif
#endif
using namespace caf;
using io::network::native_socket;
using io::network::invalid_native_socket;
using io::network::default_multiplexer;
using io::network::last_socket_error_as_string;
namespace {
constexpr auto ipv4 = caf::io::network::protocol::ipv4;
//constexpr auto ipv6 = caf::io::network::protocol::ipv6;
auto addr_of(sockaddr_in& what) -> decltype(what.sin_addr)& {
return what.sin_addr;
}
auto family_of(sockaddr_in& what) -> decltype(what.sin_family)& {
return what.sin_family;
}
auto port_of(sockaddr_in& what) -> decltype(what.sin_port)& {
return what.sin_port;
}
auto addr_of(sockaddr_in6& what) -> decltype(what.sin6_addr)& {
return what.sin6_addr;
}
auto family_of(sockaddr_in6& what) -> decltype(what.sin6_family)& {
return what.sin6_family;
}
auto port_of(sockaddr_in6& what) -> decltype(what.sin6_port)& {
return what.sin6_port;
}
// -- atoms --------------------------------------------------------------------
using ordering_atom = atom_constant<atom("ordering")>;
using send_atom = atom_constant<atom("send")>;
using quit_atom = atom_constant<atom("quit")>;
// -- network code -------------------------------------------------------------
expected<void> set_inaddr_any(native_socket, sockaddr_in& sa) {
sa.sin_addr.s_addr = INADDR_ANY;
return unit;
}
expected<void> set_inaddr_any(native_socket fd, sockaddr_in6& sa) {
sa.sin6_addr = in6addr_any;
// also accept ipv4 requests on this socket
int off = 0;
CALL_CFUN(res, detail::cc_zero, "setsockopt",
setsockopt(fd, IPPROTO_IPV6, IPV6_V6ONLY,
reinterpret_cast<io::network::setsockopt_ptr>(&off),
static_cast<io::network::socket_size_type>(sizeof(off))));
return unit;
}
template <int Family, int SockType = SOCK_STREAM>
expected<native_socket> new_ip_acceptor_impl(uint16_t port, const char* addr,
bool reuse_addr, bool any) {
static_assert(Family == AF_INET || Family == AF_INET6, "invalid family");
CAF_LOG_TRACE(CAF_ARG(port) << ", addr = " << (addr ? addr : "nullptr"));
CALL_CFUN(fd, detail::cc_valid_socket, "socket", socket(Family, SockType, 0));
// sguard closes the socket in case of exception
detail::socket_guard sguard{fd};
if (reuse_addr) {
int on = 1;
CALL_CFUN(tmp1, detail::cc_zero, "setsockopt",
setsockopt(fd, SOL_SOCKET, SO_REUSEADDR,
reinterpret_cast<io::network::setsockopt_ptr>(&on),
static_cast<io::network::socket_size_type>(sizeof(on))));
}
using sockaddr_type =
typename std::conditional<
Family == AF_INET,
sockaddr_in,
sockaddr_in6
>::type;
sockaddr_type sa;
memset(&sa, 0, sizeof(sockaddr_type));
family_of(sa) = Family;
if (any)
set_inaddr_any(fd, sa);
CALL_CFUN(tmp, detail::cc_one, "inet_pton",
inet_pton(Family, addr, &addr_of(sa)));
port_of(sa) = htons(port);
CALL_CFUN(res, detail::cc_zero, "bind",
bind(fd, reinterpret_cast<sockaddr*>(&sa),
static_cast<io::network::socket_size_type>(sizeof(sa))));
return sguard.release();
}
expected<native_socket> new_tcp_acceptor_impl(uint16_t port, const char* addr,
bool reuse_addr) {
CAF_LOG_TRACE(CAF_ARG(port) << ", addr = " << (addr ? addr : "nullptr"));
auto addrs = io::network::interfaces::server_address(port, addr);
auto addr_str = std::string{addr == nullptr ? "" : addr};
if (addrs.empty())
return make_error(sec::cannot_open_port, "No local interface available",
addr_str);
bool any = addr_str.empty() || addr_str == "::" || addr_str == "0.0.0.0";
auto fd = invalid_native_socket;
for (auto& elem : addrs) {
auto hostname = elem.first.c_str();
auto p = elem.second == ipv4
? new_ip_acceptor_impl<AF_INET>(port, hostname, reuse_addr, any)
: new_ip_acceptor_impl<AF_INET6>(port, hostname, reuse_addr, any);
if (!p) {
CAF_LOG_DEBUG(p.error());
continue;
}
fd = *p;
break;
}
if (fd == invalid_native_socket) {
CAF_LOG_WARNING("could not open tcp socket on:" << CAF_ARG(port)
<< CAF_ARG(addr_str));
return make_error(sec::cannot_open_port, "tcp socket creation failed",
port, addr_str);
}
detail::socket_guard sguard{fd};
CALL_CFUN(tmp2, detail::cc_zero, "listen", listen(fd, SOMAXCONN));
// ok, no errors so far
CAF_LOG_DEBUG(CAF_ARG(fd));
return sguard.release();
}
// -- tcp impls ----------------------------------------------------------------
struct tcp_basp_header {
uint32_t payload_len;
actor_id from;
actor_id to;
};
constexpr size_t tcp_basp_header_len = sizeof(uint32_t) + sizeof(actor_id) * 2;
template <class Inspector>
typename Inspector::result_type inspect(Inspector& fun, tcp_basp_header& hdr) {
return fun(meta::type_name("tcp_basp_header"),
hdr.payload_len, hdr.from, hdr.to);
}
struct new_tcp_basp_message {
tcp_basp_header header;
char* payload;
size_t payload_len;
};
template <class Inspector>
typename Inspector::result_type inspect(Inspector& fun,
new_tcp_basp_message& msg) {
return fun(meta::type_name("new_tcp_basp_message"), msg.header,
msg.payload_len);
}
struct tcp_basp {
static constexpr size_t header_size = sizeof(tcp_basp_header);
using message_type = new_tcp_basp_message;
using result_type = optional<message_type>;
io::network::newb<message_type>* parent;
message_type msg;
bool expecting_header = true;
tcp_basp(io::network::newb<message_type>* parent) : parent(parent) {
// nop
}
error read_header(char* bytes, size_t count) {
if (count < tcp_basp_header_len) {
CAF_LOG_DEBUG("buffer contains " << count << " bytes of expected "
<< tcp_basp_header_len);
return sec::unexpected_message;
}
binary_deserializer bd{&parent->backend(), bytes, count};
bd(msg.header);
CAF_LOG_DEBUG("read header " << CAF_ARG(msg.header));
size_t size = static_cast<size_t>(msg.header.payload_len);
parent->configure_read(io::receive_policy::exactly(size));
expecting_header = false;
return none;
}
error read_payload(char* bytes, size_t count) {
if (count < msg.header.payload_len) {
CAF_LOG_DEBUG("buffer contains " << count << " bytes of expected "
<< msg.header.payload_len);
return sec::unexpected_message;
}
msg.payload = bytes;
msg.payload_len = msg.header.payload_len;
parent->handle(msg);
expecting_header = true;
parent->configure_read(io::receive_policy::exactly(tcp_basp_header_len));
return none;
}
error read(char* bytes, size_t count) {
if (expecting_header)
return read_header(bytes, count);
else
return read_payload(bytes, count);
}
error timeout(atom_value, uint32_t) {
return none;
}
size_t write_header(io::network::byte_buffer& buf,
io::network::header_writer* hw) {
CAF_ASSERT(hw != nullptr);
(*hw)(buf);
return header_size;
}
void prepare_for_sending(io::network::byte_buffer& buf,
size_t hstart, size_t plen) {
stream_serializer<charbuf> out{&parent->backend(), buf.data() + hstart,
sizeof(uint32_t)};
auto len = static_cast<uint32_t>(plen);
out(len);
}
};
struct tcp_transport_policy : public io::network::transport_policy {
tcp_transport_policy()
: read_threshold{0},
collected{0},
maximum{0},
writing{false},
written{0} {
// nop
}
error read_some(io::network::event_handler* parent) override {
CAF_LOG_TRACE("");
size_t len = receive_buffer.size() - collected;
receive_buffer.resize(len);
void* buf = receive_buffer.data() + collected;
auto sres = ::recv(parent->fd(),
reinterpret_cast<io::network::socket_recv_ptr>(buf),
len, io::network::no_sigpipe_io_flag);
if (io::network::is_error(sres, true) || sres == 0) {
// recv returns 0 when the peer has performed an orderly shutdown
return sec::runtime_error;
}
size_t result = (sres > 0) ? static_cast<size_t>(sres) : 0;
collected += result;
received_bytes = collected;
return none;
}
bool should_deliver() override {
CAF_LOG_DEBUG(CAF_ARG(collected) << CAF_ARG(read_threshold));
return collected >= read_threshold;
}
void prepare_next_read(io::network::event_handler*) override {
collected = 0;
received_bytes = 0;
switch (rd_flag) {
case io::receive_policy_flag::exactly:
if (receive_buffer.size() != maximum)
receive_buffer.resize(maximum);
read_threshold = maximum;
break;
case io::receive_policy_flag::at_most:
if (receive_buffer.size() != maximum)
receive_buffer.resize(maximum);
read_threshold = 1;
break;
case io::receive_policy_flag::at_least: {
// read up to 10% more, but at least allow 100 bytes more
auto maximumsize = maximum + std::max<size_t>(100, maximum / 10);
if (receive_buffer.size() != maximumsize)
receive_buffer.resize(maximumsize);
read_threshold = maximum;
break;
}
}
}
void configure_read(io::receive_policy::config config) override {
rd_flag = config.first;
maximum = config.second;
}
error write_some(io::network::event_handler* parent) override {
CAF_LOG_TRACE("");
const void* buf = send_buffer.data() + written;
auto len = send_buffer.size() - written;
auto sres = ::send(parent->fd(),
reinterpret_cast<io::network::socket_send_ptr>(buf),
len, io::network::no_sigpipe_io_flag);
if (io::network::is_error(sres, true))
return sec::runtime_error;
size_t result = (sres > 0) ? static_cast<size_t>(sres) : 0;
written += result;
auto remaining = send_buffer.size() - written;
if (remaining == 0)
prepare_next_write(parent);
return none;
}
void prepare_next_write(io::network::event_handler* parent) override {
written = 0;
send_buffer.clear();
if (offline_buffer.empty()) {
parent->backend().del(io::network::operation::write,
parent->fd(), parent);
writing = false;
} else {
send_buffer.swap(offline_buffer);
}
}
io::network::byte_buffer& wr_buf() {
return offline_buffer;
}
void flush(io::network::event_handler* parent) override {
CAF_ASSERT(parent != nullptr);
CAF_LOG_TRACE(CAF_ARG(offline_buffer.size()));
if (!offline_buffer.empty() && !writing) {
parent->backend().add(io::network::operation::write,
parent->fd(), parent);
writing = true;
prepare_next_write(parent);
}
}
// State for reading.
size_t read_threshold;
size_t collected;
size_t maximum;
io::receive_policy_flag rd_flag;
// State for writing.
bool writing;
size_t written;
};
template <class T>
struct tcp_protocol_policy
: public io::network::protocol_policy<typename T::message_type> {
T impl;
tcp_protocol_policy(io::network::newb<typename T::message_type>* parent)
: impl(parent) {
// nop
}
error read(char* bytes, size_t count) override {
return impl.read(bytes, count);
}
error timeout(atom_value atm, uint32_t id) override {
return impl.timeout(atm, id);
}
void write_header(io::network::byte_buffer& buf,
io::network::header_writer* hw) override {
impl.write_header(buf, hw);
}
void prepare_for_sending(io::network::byte_buffer& buf, size_t hstart,
size_t plen) override {
impl.prepare_for_sending(buf, hstart, plen);
}
};
struct tcp_basp_newb : io::network::newb<new_tcp_basp_message> {
tcp_basp_newb(caf::actor_config& cfg, default_multiplexer& dm,
native_socket sockfd)
: newb<new_tcp_basp_message>(cfg, dm, sockfd) {
// nop
}
void handle(new_tcp_basp_message& msg) override {
CAF_PUSH_AID_FROM_PTR(this);
CAF_LOG_TRACE("");
std::string res;
binary_deserializer bd(&backend(), msg.payload, msg.payload_len);
bd(res);
send(responder, res);
}
behavior make_behavior() override {
set_default_handler(print_and_drop);
return {
// Must be implemented at the moment, will be cought by the broker in a
// later implementation.
[=](atom_value atm, uint32_t id) {
protocol->timeout(atm, id);
},
[=](send_atom, actor_id sender, actor_id receiver, std::string payload) {
auto hw = caf::make_callback([&](io::network::byte_buffer& buf) -> error {
binary_serializer bs(&backend(), buf);
bs(tcp_basp_header{0, sender, receiver});
return none;
});
auto whdl = wr_buf(&hw);
CAF_ASSERT(whdl.buf != nullptr);
CAF_ASSERT(whdl.protocol != nullptr);
binary_serializer bs(&backend(), *whdl.buf);
bs(payload);
},
[=](quit_atom) {
// Remove from multiplexer loop.
stop();
// Quit actor.
quit();
}
};
}
actor responder;
};
struct tcp_accept_policy
: public io::network::accept_policy<new_tcp_basp_message> {
virtual std::pair<native_socket, io::network::transport_policy_ptr>
accept(io::network::event_handler* parent) {
using namespace io::network;
sockaddr_storage addr;
std::memset(&addr, 0, sizeof(addr));
socket_size_type addrlen = sizeof(addr);
auto result = ::accept(parent->fd(), reinterpret_cast<sockaddr*>(&addr),
&addrlen);
if (result == invalid_native_socket) {
auto err = last_socket_error();
if (!would_block_or_temporarily_unavailable(err)) {
return {invalid_native_socket, nullptr};
}
}
transport_policy_ptr ptr{new tcp_transport_policy};
return {result, std::move(ptr)};
}
virtual void init(io::network::newb<new_tcp_basp_message>& n) {
n.start();
}
};
template <class ProtocolPolicy>
struct tcp_basp_acceptor
: public io::network::newb_acceptor<typename ProtocolPolicy::message_type> {
using super = io::network::newb_acceptor<typename ProtocolPolicy::message_type>;
static expected<native_socket> create_socket(uint16_t port, const char* host,
bool reuse = false) {
return new_tcp_acceptor_impl(port, host, reuse);
}
tcp_basp_acceptor(default_multiplexer& dm, native_socket sockfd)
: super(dm, sockfd) {
// nop
}
expected<actor> create_newb(native_socket sockfd,
io::network::transport_policy_ptr pol) override {
CAF_LOG_DEBUG("creating new basp tcp newb");
auto n = io::network::make_newb<tcp_basp_newb>(this->backend().system(),
sockfd);
auto ptr = caf::actor_cast<caf::abstract_actor*>(n);
if (ptr == nullptr)
return sec::runtime_error;
auto& ref = dynamic_cast<tcp_basp_newb&>(*ptr);
ref.transport = std::move(pol);
ref.protocol.reset(new ProtocolPolicy(&ref));
ref.responder = responder;
// This should happen somewhere else?
ref.configure_read(io::receive_policy::exactly(tcp_basp_header_len));
anon_send(responder, n);
return n;
}
actor responder;
};
struct tcp_test_broker_state {
tcp_basp_header hdr;
bool expecting_header = true;
};
void caf_main(actor_system& sys, const actor_system_config&) {
using tcp_protocol_policy_t = tcp_protocol_policy<tcp_basp>;
using tcp_accept_policy_t = tcp_accept_policy;
using tcp_newb_acceptor_t = tcp_basp_acceptor<tcp_protocol_policy_t>;
using tcp_transport_policy_t = tcp_transport_policy;
const char* host = "localhost";
const uint16_t port = 12345;
scoped_actor main_actor{sys};
actor newb_actor;
auto testing = [&](io::stateful_broker<tcp_test_broker_state>* self,
io::connection_handle hdl, actor) -> behavior {
CAF_ASSERT(hdl != io::invalid_connection_handle);
self->configure_read(hdl, io::receive_policy::exactly(tcp_basp_header_len));
self->state.expecting_header = true;
return {
[=](send_atom, std::string str) {
CAF_LOG_DEBUG("sending '" << str << "'");
io::network::byte_buffer buf;
binary_serializer bs(self->system(), buf);
tcp_basp_header hdr{0, 1, 2};
bs(hdr);
auto header_len = buf.size();
CAF_ASSERT(header_len == tcp_basp_header_len);
bs(str);
hdr.payload_len = static_cast<uint32_t>(buf.size() - header_len);
stream_serializer<charbuf> out{self->system(), buf.data(),
sizeof(hdr.payload_len)};
out(hdr.payload_len);
CAF_LOG_DEBUG("header len: " << header_len
<< ", packet_len: " << buf.size()
<< ", header: " << to_string(hdr));
self->write(hdl, buf.size(), buf.data());
self->flush(hdl);
},
[=](quit_atom) {
CAF_LOG_DEBUG("test broker shutting down");
self->quit();
},
[=](io::new_data_msg& msg) {
auto& s = self->state;
size_t next_len = tcp_basp_header_len;
binary_deserializer bd(self->system(), msg.buf);
if (s.expecting_header) {
bd(s.hdr);
next_len = s.hdr.payload_len;
s.expecting_header = false;
} else {
std::string str;
bd(str);
CAF_LOG_DEBUG("received '" << str << "'");
std::reverse(std::begin(str), std::end(str));
io::network::byte_buffer buf;
binary_serializer bs(self->system(), buf);
tcp_basp_header hdr{0, 1, 2};
bs(hdr);
auto header_len = buf.size();
CAF_ASSERT(header_len == tcp_basp_header_len);
bs(str);
hdr.payload_len = static_cast<uint32_t>(buf.size() - header_len);
stream_serializer<charbuf> out{self->system(), buf.data(),
sizeof(hdr.payload_len)};
out(hdr.payload_len);
CAF_LOG_DEBUG("header len: " << header_len
<< ", packet_len: " << buf.size()
<< ", header: " << to_string(hdr));
self->write(hdl, buf.size(), buf.data());
self->flush(hdl);
// self->send(m, quit_atom::value);
}
self->configure_read(msg.handle, io::receive_policy::exactly(next_len));
}
};
};
auto helper_actor = sys.spawn([&](event_based_actor* self, actor m) -> behavior {
return {
[=](const std::string& str) {
CAF_LOG_DEBUG("received '" << str << "'");
self->send(m, quit_atom::value);
},
[=](actor a) {
CAF_LOG_DEBUG("got new newb handle");
self->send(m, a);
},
[=](quit_atom) {
CAF_LOG_DEBUG("helper shutting down");
self->quit();
}
};
}, main_actor);
CAF_LOG_DEBUG("creating new acceptor");
auto newb_acceptor_ptr
= io::network::make_newb_acceptor<tcp_newb_acceptor_t, tcp_accept_policy_t>(sys, port);
dynamic_cast<tcp_newb_acceptor_t*>(newb_acceptor_ptr.get())->responder
= helper_actor;
CAF_LOG_DEBUG("connecting from 'old-style' broker");
auto exp = sys.middleman().spawn_client(testing, host, port, main_actor);
CAF_ASSERT(exp);
auto test_broker = std::move(*exp);
main_actor->receive(
[&](actor a) {
newb_actor = a;
}
);
CAF_LOG_DEBUG("sending message to newb");
main_actor->send(test_broker, send_atom::value, "hello world");
std::this_thread::sleep_for(std::chrono::seconds(1));
main_actor->receive(
[](quit_atom) {
CAF_LOG_DEBUG("check");
}
);
CAF_LOG_DEBUG("sending message from newb");
main_actor->send(newb_actor, send_atom::value, actor_id{3}, actor_id{4},
"dlrow olleh");
main_actor->receive(
[](quit_atom) {
CAF_LOG_DEBUG("check");
}
);
CAF_LOG_DEBUG("shutting everything down");
newb_acceptor_ptr->stop();
anon_send(newb_actor, quit_atom::value);
anon_send(helper_actor, quit_atom::value);
anon_send(test_broker, quit_atom::value);
sys.await_all_actors_done();
CAF_LOG_DEBUG("done");
}
} // namespace anonymous
CAF_MAIN(io::middleman);
#include "caf/io/network/newb.hpp"
#include "caf/config.hpp"
#include "caf/logger.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/detail/call_cfun.hpp"
#ifdef CAF_WINDOWS
# ifndef WIN32_LEAN_AND_MEAN
# define WIN32_LEAN_AND_MEAN
# endif // WIN32_LEAN_AND_MEAN
# ifndef NOMINMAX
# define NOMINMAX
# endif
# ifdef CAF_MINGW
# undef _WIN32_WINNT
# undef WINVER
# define _WIN32_WINNT WindowsVista
# define WINVER WindowsVista
# include <w32api.h>
# endif
# include <io.h>
# include <windows.h>
# include <winsock2.h>
# include <ws2ipdef.h>
# include <ws2tcpip.h>
#else
# include <unistd.h>
# include <arpa/inet.h>
# include <cerrno>
# include <fcntl.h>
# include <netdb.h>
# include <netinet/in.h>
# include <netinet/ip.h>
# include <netinet/tcp.h>
# include <sys/socket.h>
# include <sys/types.h>
# ifdef CAF_POLL_MULTIPLEXER
# include <poll.h>
# elif defined(CAF_EPOLL_MULTIPLEXER)
# include <sys/epoll.h>
# else
# error "neither CAF_POLL_MULTIPLEXER nor CAF_EPOLL_MULTIPLEXER defined"
# endif
#endif
using namespace caf;
using io::network::native_socket;
using io::network::invalid_native_socket;
using io::network::default_multiplexer;
using io::network::last_socket_error_as_string;
namespace {
constexpr auto ipv4 = caf::io::network::protocol::ipv4;
//constexpr auto ipv6 = caf::io::network::protocol::ipv6;
auto addr_of(sockaddr_in& what) -> decltype(what.sin_addr)& {
return what.sin_addr;
}
auto family_of(sockaddr_in& what) -> decltype(what.sin_family)& {
return what.sin_family;
}
auto port_of(sockaddr_in& what) -> decltype(what.sin_port)& {
return what.sin_port;
}
auto addr_of(sockaddr_in6& what) -> decltype(what.sin6_addr)& {
return what.sin6_addr;
}
auto family_of(sockaddr_in6& what) -> decltype(what.sin6_family)& {
return what.sin6_family;
}
auto port_of(sockaddr_in6& what) -> decltype(what.sin6_port)& {
return what.sin6_port;
}
// -- atoms --------------------------------------------------------------------
using ordering_atom = atom_constant<atom("ordering")>;
using send_atom = atom_constant<atom("send")>;
using quit_atom = atom_constant<atom("quit")>;
// -- network code -------------------------------------------------------------
expected<void> set_inaddr_any(native_socket, sockaddr_in& sa) {
sa.sin_addr.s_addr = INADDR_ANY;
return unit;
}
expected<void> set_inaddr_any(native_socket fd, sockaddr_in6& sa) {
sa.sin6_addr = in6addr_any;
// also accept ipv4 requests on this socket
int off = 0;
CALL_CFUN(res, detail::cc_zero, "setsockopt",
setsockopt(fd, IPPROTO_IPV6, IPV6_V6ONLY,
reinterpret_cast<io::network::setsockopt_ptr>(&off),
static_cast<io::network::socket_size_type>(sizeof(off))));
return unit;
}
template <int Family, int SockType = SOCK_STREAM>
expected<native_socket> new_ip_acceptor_impl(uint16_t port, const char* addr,
bool reuse_addr, bool any) {
static_assert(Family == AF_INET || Family == AF_INET6, "invalid family");
CAF_LOG_TRACE(CAF_ARG(port) << ", addr = " << (addr ? addr : "nullptr"));
CALL_CFUN(fd, detail::cc_valid_socket, "socket", socket(Family, SockType, 0));
// sguard closes the socket in case of exception
detail::socket_guard sguard{fd};
if (reuse_addr) {
int on = 1;
CALL_CFUN(tmp1, detail::cc_zero, "setsockopt",
setsockopt(fd, SOL_SOCKET, SO_REUSEADDR,
reinterpret_cast<io::network::setsockopt_ptr>(&on),
static_cast<io::network::socket_size_type>(sizeof(on))));
}
using sockaddr_type =
typename std::conditional<
Family == AF_INET,
sockaddr_in,
sockaddr_in6
>::type;
sockaddr_type sa;
memset(&sa, 0, sizeof(sockaddr_type));
family_of(sa) = Family;
if (any)
set_inaddr_any(fd, sa);
CALL_CFUN(tmp, detail::cc_one, "inet_pton",
inet_pton(Family, addr, &addr_of(sa)));
port_of(sa) = htons(port);
CALL_CFUN(res, detail::cc_zero, "bind",
bind(fd, reinterpret_cast<sockaddr*>(&sa),
static_cast<io::network::socket_size_type>(sizeof(sa))));
return sguard.release();
}
expected<native_socket> new_udp_endpoint_impl(uint16_t port, const char* addr,
bool reuse) {
CAF_LOG_TRACE(CAF_ARG(port) << ", addr = " << (addr ? addr : "nullptr"));
auto addrs = io::network::interfaces::server_address(port, addr);
auto addr_str = std::string{addr == nullptr ? "" : addr};
if (addrs.empty())
return make_error(sec::cannot_open_port, "No local interface available",
addr_str);
bool any = addr_str.empty() || addr_str == "::" || addr_str == "0.0.0.0";
auto fd = invalid_native_socket;
for (auto& elem : addrs) {
auto host = elem.first.c_str();
auto p = elem.second == ipv4
? new_ip_acceptor_impl<AF_INET, SOCK_DGRAM>(port, host, reuse, any)
: new_ip_acceptor_impl<AF_INET6, SOCK_DGRAM>(port, host, reuse, any);
if (!p) {
CAF_LOG_DEBUG(p.error());
continue;
}
fd = *p;
break;
}
if (fd == invalid_native_socket) {
CAF_LOG_WARNING("could not open udp socket on:" << CAF_ARG(port)
<< CAF_ARG(addr_str));
return make_error(sec::cannot_open_port, "udp socket creation failed",
port, addr_str);
}
CAF_LOG_DEBUG(CAF_ARG(fd));
return fd;
}
// -- udp impls ----------------------------------------------------------------
struct udp_basp_header {
uint32_t payload_len;
actor_id from;
actor_id to;
};
template <class Inspector>
typename Inspector::result_type inspect(Inspector& fun, udp_basp_header& hdr) {
return fun(meta::type_name("udp_basp_header"), hdr.payload_len,
hdr.from, hdr.to);
}
constexpr size_t udp_basp_header_len = sizeof(uint32_t) + sizeof(actor_id) * 2;
using sequence_type = uint16_t;
struct udp_ordering_header {
sequence_type seq;
};
template <class Inspector>
typename Inspector::result_type inspect(Inspector& fun, udp_ordering_header& hdr) {
return fun(meta::type_name("udp_ordering_header"), hdr.seq);
}
constexpr size_t udp_ordering_header_len = sizeof(sequence_type);
struct new_udp_basp_message {
udp_basp_header header;
char* payload;
size_t payload_len;
};
template <class Inspector>
typename Inspector::result_type inspect(Inspector& fun,
new_udp_basp_message& msg) {
return fun(meta::type_name("new_udp_basp_message"), msg.header,
msg.payload_len);
}
struct udp_basp {
static constexpr size_t header_size = udp_basp_header_len;
using message_type = new_udp_basp_message;
using result_type = optional<message_type>;
io::network::newb<message_type>* parent;
message_type msg;
udp_basp(io::network::newb<message_type>* parent) : parent(parent) {
// nop
}
error read(char* bytes, size_t count) {
CAF_LOG_DEBUG("reading basp udp header");
// Read header.
if (count < udp_basp_header_len) {
CAF_LOG_DEBUG("not enought bytes for basp header");
CAF_LOG_DEBUG("buffer contains " << count << " bytes of expected "
<< udp_basp_header_len);
return sec::unexpected_message;
}
binary_deserializer bd{&parent->backend(), bytes, count};
bd(msg.header);
CAF_LOG_DEBUG("read basp header " << CAF_ARG(msg.header));
size_t payload_len = static_cast<size_t>(msg.header.payload_len);
// Read payload.
auto remaining = count - udp_basp_header_len;
// TODO: Could be `!=` ?
if (remaining < payload_len) {
CAF_LOG_DEBUG("only " << remaining << " bytes remaining of expected "
<< msg.header.payload_len);
return sec::unexpected_message;
}
msg.payload = bytes + udp_basp_header_len;
msg.payload_len = msg.header.payload_len;
parent->handle(msg);
return none;
}
error timeout(atom_value, uint32_t) {
return none;
}
size_t write_header(io::network::byte_buffer& buf,
io::network::header_writer* hw) {
CAF_ASSERT(hw != nullptr);
(*hw)(buf);
return header_size;
}
void prepare_for_sending(io::network::byte_buffer& buf,
size_t hstart, size_t plen) {
stream_serializer<charbuf> out{&parent->backend(), buf.data() + hstart,
sizeof(uint32_t)};
auto len = static_cast<uint32_t>(plen);
out(len);
}
};
template <class Next>
struct udp_ordering {
static constexpr size_t header_size = udp_ordering_header_len;
using message_type = typename Next::message_type;
using result_type = typename Next::result_type;
sequence_type seq_read = 0;
sequence_type seq_write = 0;
size_t max_pending_messages = 10;
std::chrono::milliseconds pending_to = std::chrono::milliseconds(100);
io::network::newb<message_type>* parent;
Next next;
std::unordered_map<sequence_type, std::vector<char>> pending;
udp_ordering(io::network::newb<message_type>* parent)
: parent(parent),
next(parent) {
// nop
}
error deliver_pending() {
if (pending.empty())
return none;
while (pending.count(seq_read) > 0) {
auto& buf = pending[seq_read];
auto res = next.read(buf.data(), buf.size());
pending.erase(seq_read);
// TODO: Cancel timeout.
if (res)
return res;
}
return none;
}
error add_pending(char* bytes, size_t count, sequence_type seq) {
pending[seq] = std::vector<char>(bytes + header_size, bytes + count);
parent->set_timeout(pending_to, ordering_atom::value, seq);
if (pending.size() > max_pending_messages) {
seq_read = pending.begin()->first;
return deliver_pending();
}
return none;
}
error read(char* bytes, size_t count) {
if (count < header_size)
return sec::unexpected_message;
udp_ordering_header hdr;
binary_deserializer bd(&parent->backend(), bytes, count);
bd(hdr);
CAF_LOG_DEBUG("read udp ordering header: " << to_string(hdr));
// TODO: Use the comparison function from BASP instance.
if (hdr.seq == seq_read) {
seq_read += 1;
auto res = next.read(bytes + header_size, count - header_size);
if (res)
return res;
return deliver_pending();
} else if (hdr.seq > seq_read) {
add_pending(bytes, count, hdr.seq);
return none;
}
return none;
}
error timeout(atom_value atm, uint32_t id) {
if (atm == ordering_atom::value) {
error err = none;
sequence_type seq = static_cast<sequence_type>(id);
if (pending.count(seq) > 0) {
seq_read = static_cast<sequence_type>(seq);
err = deliver_pending();
}
return err;
}
return next.timeout(atm, id);
}
void write_header(io::network::byte_buffer& buf,
io::network::header_writer* hw) {
binary_serializer bs(&parent->backend(), buf);
bs(udp_ordering_header{seq_write});
seq_write += 1;
next.write_header(buf, hw);
return;
}
void prepare_for_sending(io::network::byte_buffer& buf,
size_t hstart, size_t plen) {
next.prepare_for_sending(buf, hstart, plen);
}
};
struct udp_transport_policy : public io::network::transport_policy {
udp_transport_policy()
: maximum{std::numeric_limits<uint16_t>::max()},
first_message{true},
writing{false},
written{0},
offline_sum{0} {
// nop
}
error read_some(io::network::event_handler* parent) override {
CAF_LOG_TRACE(CAF_ARG(parent->fd()));
memset(sender.address(), 0, sizeof(sockaddr_storage));
io::network::socket_size_type len = sizeof(sockaddr_storage);
auto buf_ptr = static_cast<io::network::socket_recv_ptr>(receive_buffer.data());
auto buf_len = receive_buffer.size();
auto sres = ::recvfrom(parent->fd(), buf_ptr, buf_len,
0, sender.address(), &len);
if (io::network::is_error(sres, true)) {
CAF_LOG_ERROR("recvfrom returned" << CAF_ARG(sres));
return sec::runtime_error;
} else if (io::network::would_block_or_temporarily_unavailable(
io::network::last_socket_error())) {
CAF_LOG_DEBUG("try later");
return sec::end_of_stream;
}
if (sres == 0)
CAF_LOG_INFO("Received empty datagram");
else if (sres > static_cast<io::network::signed_size_type>(buf_len))
CAF_LOG_WARNING("recvfrom cut of message, only received "
<< CAF_ARG(buf_len) << " of " << CAF_ARG(sres) << " bytes");
received_bytes = (sres > 0) ? static_cast<size_t>(sres) : 0;
*sender.length() = static_cast<size_t>(len);
if (first_message) {
endpoint = sender;
first_message = false;
}
return none;
}
bool should_deliver() override {
CAF_LOG_TRACE("");
return received_bytes != 0 && sender == endpoint;
}
void prepare_next_read(io::network::event_handler*) override {
received_bytes = 0;
receive_buffer.resize(maximum);
}
void configure_read(io::receive_policy::config) override {
// nop
}
error write_some(io::network::event_handler* parent) override {
using namespace caf::io::network;
CAF_LOG_TRACE(CAF_ARG(parent->fd()) << CAF_ARG(send_buffer.size()));
socket_size_type len = static_cast<socket_size_type>(*endpoint.clength());
auto buf_ptr = reinterpret_cast<socket_send_ptr>(send_buffer.data() + written);
auto buf_len = send_sizes.front();
auto sres = ::sendto(parent->fd(), buf_ptr, buf_len,
0, endpoint.caddress(), len);
if (is_error(sres, true)) {
CAF_LOG_ERROR("sendto returned" << CAF_ARG(sres));
return sec::runtime_error;
}
send_sizes.pop_front();
written += (sres > 0) ? static_cast<size_t>(sres) : 0;
auto remaining = send_buffer.size() - written;
if (remaining == 0)
prepare_next_write(parent);
return none;
}
void prepare_next_write(io::network::event_handler* parent) override {
written = 0;
send_buffer.clear();
send_sizes.clear();
if (offline_buffer.empty()) {
writing = false;
parent->backend().del(io::network::operation::write,
parent->fd(), parent);
} else {
// Add size of last chunk.
offline_sizes.push_back(offline_buffer.size() - offline_sum);
// Switch buffers.
send_buffer.swap(offline_buffer);
send_sizes.swap(offline_sizes);
// Reset sum.
offline_sum = 0;
}
}
io::network::byte_buffer& wr_buf() {
if (!offline_buffer.empty()) {
auto chunk_size = offline_buffer.size() - offline_sum;
offline_sizes.push_back(chunk_size);
offline_sum += chunk_size;
}
return offline_buffer;
}
void flush(io::network::event_handler* parent) override {
CAF_ASSERT(parent != nullptr);
CAF_LOG_TRACE(CAF_ARG(offline_buffer.size()));
if (!offline_buffer.empty() && !writing) {
parent->backend().add(io::network::operation::write,
parent->fd(), parent);
writing = true;
prepare_next_write(parent);
}
}
// State for reading.
size_t maximum;
bool first_message;
// State for writing.
bool writing;
size_t written;
size_t offline_sum;
std::deque<size_t> send_sizes;
std::deque<size_t> offline_sizes;
// UDP endpoints.
io::network::ip_endpoint endpoint;
io::network::ip_endpoint sender;
};
template <class T>
struct udp_protocol_policy
: public io::network::protocol_policy<typename T::message_type> {
T impl;
udp_protocol_policy(io::network::newb<typename T::message_type>* parent)
: impl(parent) {
// nop
}
error read(char* bytes, size_t count) override {
return impl.read(bytes, count);
}
error timeout(atom_value atm, uint32_t id) override {
return impl.timeout(atm, id);
}
void write_header(io::network::byte_buffer& buf,
io::network::header_writer* hw) override {
impl.write_header(buf, hw);
}
void prepare_for_sending(io::network::byte_buffer& buf,
size_t hstart, size_t plen) override {
impl.prepare_for_sending(buf, hstart, plen);
}
};
struct udp_basp_newb : public io::network::newb<new_udp_basp_message> {
udp_basp_newb(caf::actor_config& cfg, default_multiplexer& dm,
native_socket sockfd)
: newb<new_udp_basp_message>(cfg, dm, sockfd) {
// nop
}
void handle(new_udp_basp_message& msg) override {
CAF_PUSH_AID_FROM_PTR(this);
CAF_LOG_TRACE("");
std::string res;
binary_deserializer bd(&backend(), msg.payload, msg.payload_len);
bd(res);
send(responder, res);
}
behavior make_behavior() override {
set_default_handler(print_and_drop);
return {
// Must be implemented at the moment, will be cought by the broker in a
// later implementation.
[=](atom_value atm, uint32_t id) {
protocol->timeout(atm, id);
},
[=](send_atom, actor_id sender, actor_id receiver, std::string payload) {
auto hw = caf::make_callback([&](io::network::byte_buffer& buf) -> error {
binary_serializer bs(&backend(), buf);
bs(udp_basp_header{0, sender, receiver});
return none;
});
auto whdl = wr_buf(&hw);
CAF_ASSERT(whdl.buf != nullptr);
CAF_ASSERT(whdl.protocol != nullptr);
binary_serializer bs(&backend(), *whdl.buf);
bs(payload);
},
[=](quit_atom) {
// Remove from multiplexer loop.
stop();
// Quit actor.
quit();
}
};
}
actor responder;
};
struct udp_accept_policy
: public io::network::accept_policy<new_udp_basp_message> {
std::pair<native_socket, io::network::transport_policy_ptr>
accept(io::network::event_handler*) override {
auto esock = new_udp_endpoint_impl(0, nullptr, false);
if (!esock)
return {invalid_native_socket, nullptr};
auto result = std::move(*esock);
io::network::transport_policy_ptr ptr{new udp_transport_policy};
return {result, std::move(ptr)};
}
void init(io::network::newb<new_udp_basp_message>& n) override {
n.start();
}
};
template <class ProtocolPolicy>
struct udp_basp_acceptor
: public io::network::newb_acceptor<typename ProtocolPolicy::message_type> {
using super = io::network::newb_acceptor<typename ProtocolPolicy::message_type>;
udp_basp_acceptor(default_multiplexer& dm, native_socket sockfd)
: super(dm, sockfd) {
// nop
}
static expected<native_socket> create_socket(uint16_t port, const char* host,
bool reuse = false) {
return new_udp_endpoint_impl(port, host, reuse);
}
expected<actor> create_newb(native_socket sockfd,
io::network::transport_policy_ptr pol) override {
CAF_LOG_DEBUG("creating new basp udp newb");
auto n = io::network::make_newb<udp_basp_newb>(this->backend().system(),
sockfd);
auto ptr = caf::actor_cast<caf::abstract_actor*>(n);
if (ptr == nullptr)
return sec::runtime_error;
auto& ref = dynamic_cast<udp_basp_newb&>(*ptr);
ref.transport = std::move(pol);
ref.protocol.reset(new ProtocolPolicy(&ref));
ref.responder = responder;
// Read first message from this socket.
ref.transport->prepare_next_read(this);
ref.transport->read_some(this, *ref.protocol.get());
// Subsequent messages will be read from `sockfd`.
anon_send(responder, n);
return n;
}
actor responder;
};
struct udp_test_broker_state {
io::datagram_handle hdl;
};
// -- main ---------------------------------------------------------------------
void caf_main(actor_system& sys, const actor_system_config&) {
using udp_protocol_policy_t = udp_protocol_policy<udp_ordering<udp_basp>>;
using udp_accept_policy_t = udp_accept_policy;
using udp_newb_acceptor_t = udp_basp_acceptor<udp_protocol_policy_t>;
using udp_transport_policy_t = udp_transport_policy;
const char* host = "localhost";
const uint16_t port = 12345;
scoped_actor main_actor{sys};
actor newb_actor;
auto testing = [&](io::stateful_broker<udp_test_broker_state>* self,
std::string host, uint16_t port, actor m) -> behavior {
auto ehdl = self->add_udp_datagram_servant(host, port);
CAF_ASSERT(ehdl);
self->state.hdl = std::move(*ehdl);
return {
[=](send_atom, std::string str) {
CAF_LOG_DEBUG("sending '" << str << "'");
io::network::byte_buffer buf;
udp_ordering_header ohdr{0};
udp_basp_header bhdr{0, 1, 2};
binary_serializer bs(self->system(), buf);
bs(ohdr);
auto ordering_header_len = buf.size();
CAF_ASSERT(ordering_header_len == udp_ordering_header_len);
bs(bhdr);
auto header_len = buf.size();
CAF_ASSERT(header_len == udp_ordering_header_len + udp_basp_header_len);
bs(str);
bhdr.payload_len = static_cast<uint32_t>(buf.size() - header_len);
stream_serializer<charbuf> out{self->system(),
buf.data() + ordering_header_len,
sizeof(bhdr.payload_len)};
out(bhdr.payload_len);
CAF_LOG_DEBUG("header len: " << header_len
<< ", packet_len: " << buf.size()
<< ", ordering header: " << to_string(ohdr)
<< ", basp header: " << to_string(bhdr));
self->enqueue_datagram(self->state.hdl, std::move(buf));
self->flush(self->state.hdl);
},
[=](quit_atom) {
CAF_LOG_DEBUG("test broker shutting down");
self->quit();
},
[=](io::new_datagram_msg& msg) {
binary_deserializer bd(self->system(), msg.buf);
udp_ordering_header ohdr;
udp_basp_header bhdr;
std::string str;
bd(ohdr);
bd(bhdr);
bd(str);
CAF_LOG_DEBUG("received '" << str << "'");
self->send(m, quit_atom::value);
}
};
};
auto helper_actor = sys.spawn([&](event_based_actor* self, actor m) -> behavior {
return {
[=](const std::string& str) {
CAF_LOG_DEBUG("received '" << str << "'");
self->send(m, quit_atom::value);
},
[=](actor a) {
CAF_LOG_DEBUG("got new newb handle");
self->send(m, a);
},
[=](quit_atom) {
CAF_LOG_DEBUG("helper shutting down");
self->quit();
}
};
}, main_actor);
CAF_LOG_DEBUG("creating new acceptor");
auto newb_acceptor_ptr
= io::network::make_newb_acceptor<udp_newb_acceptor_t,
udp_accept_policy_t>(sys, port);
dynamic_cast<udp_newb_acceptor_t*>(newb_acceptor_ptr.get())->responder
= helper_actor;
CAF_LOG_DEBUG("contacting from 'old-style' broker");
auto test_broker = sys.middleman().spawn_broker(testing, host, port, main_actor);
main_actor->send(test_broker, send_atom::value, "hello world");
std::this_thread::sleep_for(std::chrono::seconds(1));
main_actor->receive(
[&](actor a) {
newb_actor = a;
}
);
CAF_LOG_DEBUG("new newb was created");
main_actor->receive(
[](quit_atom) {
CAF_LOG_DEBUG("check");
}
);
CAF_LOG_DEBUG("sending message from newb");
main_actor->send(newb_actor, send_atom::value, actor_id{3}, actor_id{4},
"dlrow olleh");
main_actor->receive(
[](quit_atom) {
CAF_LOG_DEBUG("check");
}
);
CAF_LOG_DEBUG("shutting everything down");
newb_acceptor_ptr->stop();
anon_send(newb_actor, quit_atom::value);
anon_send(helper_actor, quit_atom::value);
anon_send(test_broker, quit_atom::value);
sys.await_all_actors_done();
CAF_LOG_DEBUG("done");
}
} // namespace anonymous
CAF_MAIN(io::middleman);
...@@ -18,108 +18,202 @@ ...@@ -18,108 +18,202 @@
#pragma once #pragma once
#include "caf/scheduled_actor.hpp" #include "caf/config.hpp"
#include "caf/detail/type_list.hpp"
#include "caf/mixin/sender.hpp" #include <cstdint>
#include "caf/mixin/requester.hpp" #include <cstring>
#include "caf/mixin/behavior_changer.hpp" #include <tuple>
#include "caf/io/network/event_handler.hpp" #include "caf/test/dsl.hpp"
#include "caf/callback.hpp"
#include "caf/config.hpp"
#include "caf/detail/call_cfun.hpp"
#include "caf/detail/enum_to_string.hpp"
#include "caf/detail/socket_guard.hpp"
#include "caf/io/broker.hpp"
#include "caf/io/middleman.hpp"
#include "caf/io/network/default_multiplexer.hpp" #include "caf/io/network/default_multiplexer.hpp"
#include "caf/io/network/event_handler.hpp"
#include "caf/io/network/interfaces.hpp"
#include "caf/io/network/native_socket.hpp"
#include "caf/mixin/behavior_changer.hpp"
#include "caf/mixin/requester.hpp"
#include "caf/mixin/sender.hpp"
#include "caf/scheduled_actor.hpp"
namespace caf { namespace caf {
namespace io { namespace io {
namespace network { namespace network {
template <class T, class... Xs> // -- forward declarations -----------------------------------------------------
class newb;
template <class T>
struct protocol_policy;
template <class T>
struct newb;
} // namespace network
} // namespace io } // namespace io
} // namespace network
// -- required to make newb an actor ------------------------------------------
template <class T, class... Xs> template <class T>
class behavior_type_of<io::network::newb<T, Xs...>> { class behavior_type_of<io::network::newb<T>> {
public: public:
using type = behavior; using type = behavior;
}; };
namespace io {
namespace network {
namespace detail { // -- aliases ------------------------------------------------------------------
using byte_buffer = std::vector<char>;
using header_writer = caf::callback<byte_buffer&>;
// -- transport policy ---------------------------------------------------------
struct transport_policy {
transport_policy() : received_bytes{0}, max_consecutive_reads{50} {
// nop
}
// Policies provide a newb with functionality. A protocol policy implements a virtual ~transport_policy() {
// network protocol and provides implementations to read and write data as well
// as for forking to new event handlers.
//
// After receiving data the policy stack of a newb is called in order to
struct protocol_policy {
protocol_policy(int i)
: idx{i} {
// nop // nop
} }
void read() { virtual error write_some(network::event_handler*) {
return none;
}
virtual error read_some(network::event_handler*) {
return none;
}
virtual bool should_deliver() {
return true;
}
virtual void prepare_next_read(network::event_handler*) {
// nop
}
virtual void prepare_next_write(network::event_handler*) {
// nop
} }
void write() { virtual void configure_read(receive_policy::config) {
// nop
}
virtual void flush(network::event_handler*) {
// nop
} }
void fork() { inline byte_buffer& wr_buf() {
return offline_buffer;
}
template <class T>
error read_some(network::event_handler* parent, protocol_policy<T>& policy) {
CAF_LOG_TRACE("");
auto mcr = max_consecutive_reads;
for (size_t i = 0; i < mcr; ++i) {
auto res = read_some(parent);
// The return statements seems weird, needs cleanup.
if (res)
return res;
if (should_deliver()) {
res = policy.read(receive_buffer.data(), received_bytes);
prepare_next_read(parent);
if (!res)
return res;
}
}
return none;
} }
int idx;
size_t received_bytes;
size_t max_consecutive_reads;
byte_buffer offline_buffer;
byte_buffer receive_buffer;
byte_buffer send_buffer;
}; };
struct mutating_policy { using transport_policy_ptr = std::unique_ptr<transport_policy>;
mutating_policy(int i)
: idx{i} { // -- accept policy ------------------------------------------------------------
template <class Message>
struct accept_policy {
virtual ~accept_policy() {
// nop // nop
} }
// Call on incoming data. Should return either a bool or a enum to virtual std::pair<native_socket, transport_policy_ptr>
// know if furhter policies are called and how to continue processing. accept(network::event_handler*) = 0;
void handle() {
virtual void init(newb<Message>&) = 0;
};
// -- protocol policy ----------------------------------------------------------
struct protocol_policy_base {
virtual ~protocol_policy_base() {
// nop
} }
int idx; virtual error read(char* bytes, size_t count) = 0;
virtual error timeout(atom_value, uint32_t) = 0;
virtual void write_header(byte_buffer&, header_writer*) = 0;
virtual void prepare_for_sending(byte_buffer&, size_t, size_t) = 0;
}; };
template <class T> template <class T>
struct is_protocol_policy_type : std::is_base_of<protocol_policy, T> {}; struct protocol_policy : protocol_policy_base {
using message_type = T;
virtual ~protocol_policy() override {
// nop
}
};
template <class T> template <class T>
struct is_mutating_policy_type : std::is_base_of<mutating_policy, T> {}; using protocol_policy_ptr = std::unique_ptr<protocol_policy<T>>;
} // namespace detail // -- new broker classes -------------------------------------------------------
namespace io { /// @relates newb
namespace network { /// Returned by funtion wr_buf of newb.
template <class Message>
struct write_handle {
newb<Message>* parent;
protocol_policy_base* protocol;
byte_buffer* buf;
size_t header_start;
size_t header_len;
~write_handle() {
// Can we calculate added bytes for datagram things?
auto payload_size = buf->size() - (header_start + header_len);
protocol->prepare_for_sending(*buf, header_start, payload_size);
parent->flush();
}
};
/** template <class Message>
* TODO: struct newb : public extend<scheduled_actor, newb<Message>>::template
* - [x] create a class `newb` that is an event handler and actor with<mixin::sender, mixin::requester,
* - [x] get it running in the multiplexer mixin::behavior_changer>,
* - [ ] create a base policy class public dynamically_typed_actor_base,
* - [x] build `make_policy` which creates a `newb` with multiple policies public network::event_handler {
* - [ ] is there a difference between a protocol policy and a guarantee? using super = typename extend<scheduled_actor, newb<Message>>::
* - [ ] get a call hierarchy in both directions
* - [ ] what should policy their constrcutors do?
* - [ ] ...
*/
template <class Protocol, class... Policies>
class newb : public extend<scheduled_actor,
newb<Protocol, Policies...>>::template
with<mixin::sender, mixin::requester,
mixin::behavior_changer>,
public dynamically_typed_actor_base,
public event_handler {
public:
using super = typename extend<scheduled_actor, newb<Protocol, Policies...>>::
template with<mixin::sender, mixin::requester, mixin::behavior_changer>; template with<mixin::sender, mixin::requester, mixin::behavior_changer>;
using signatures = none_t; using signatures = none_t;
...@@ -128,9 +222,14 @@ public: ...@@ -128,9 +222,14 @@ public:
newb(actor_config& cfg, default_multiplexer& dm, native_socket sockfd) newb(actor_config& cfg, default_multiplexer& dm, native_socket sockfd)
: super(cfg), : super(cfg),
event_handler(dm, sockfd), event_handler(dm, sockfd) {
protocol_{0}, // nop
policies_{Policies{1}...} { }
newb() = default;
newb(newb<Message>&&) = default;
~newb() override {
// nop // nop
} }
...@@ -147,6 +246,10 @@ public: ...@@ -147,6 +246,10 @@ public:
&backend()); &backend());
} }
resumable::subtype_t subtype() const override {
return resumable::io_actor;
}
// -- overridden modifiers of local_actor ------------------------------------ // -- overridden modifiers of local_actor ------------------------------------
void launch(execution_unit* eu, bool lazy, bool hide) override { void launch(execution_unit* eu, bool lazy, bool hide) override {
...@@ -176,8 +279,6 @@ public: ...@@ -176,8 +279,6 @@ public:
} }
} }
// -- overridden modifiers of abstract_broker --------------------------------
bool cleanup(error&& reason, execution_unit* host) override { bool cleanup(error&& reason, execution_unit* host) override {
CAF_LOG_TRACE(CAF_ARG(reason)); CAF_LOG_TRACE(CAF_ARG(reason));
// TODO: Ask policies, close socket. // TODO: Ask policies, close socket.
...@@ -186,8 +287,9 @@ public: ...@@ -186,8 +287,9 @@ public:
// -- overridden modifiers of resumable -------------------------------------- // -- overridden modifiers of resumable --------------------------------------
multiplexer::runnable::resume_result resume(execution_unit* ctx, network::multiplexer::runnable::resume_result resume(execution_unit* ctx,
size_t mt) override { size_t mt) override {
CAF_PUSH_AID_FROM_PTR(this);
CAF_ASSERT(ctx != nullptr); CAF_ASSERT(ctx != nullptr);
CAF_ASSERT(ctx == &backend()); CAF_ASSERT(ctx == &backend());
return scheduled_actor::resume(ctx, mt); return scheduled_actor::resume(ctx, mt);
...@@ -195,34 +297,115 @@ public: ...@@ -195,34 +297,115 @@ public:
// -- overridden modifiers of event handler ---------------------------------- // -- overridden modifiers of event handler ----------------------------------
void handle_event(operation op) override { void handle_event(network::operation op) override {
std::cout << "handling event " << to_string(op) << std::endl; CAF_PUSH_AID_FROM_PTR(this);
CAF_LOG_TRACE("");
switch (op) {
case io::network::operation::read:
read_event();
break;
case io::network::operation::write:
write_event();
break;
case io::network::operation::propagate_error:
handle_error();
}
} }
void removed_from_loop(operation op) override { void removed_from_loop(network::operation op) override {
std::cout << "removing myself from the loop for " CAF_LOG_DEBUG("newb removed from loop: " << to_string(op));
<< to_string(op) << std::endl; CAF_PUSH_AID_FROM_PTR(this);
CAF_LOG_TRACE(CAF_ARG(op));
switch (op) {
case network::operation::read: break;
case network::operation::write: break;
case network::operation::propagate_error: ; // nop
}
// nop
} }
// -- members ---------------------------------------------------------------- // -- members ----------------------------------------------------------------
void init_newb() {
CAF_LOG_TRACE("");
super::setf(super::is_initialized_flag);
}
void start() {
CAF_PUSH_AID_FROM_PTR(this);
CAF_LOG_TRACE("");
CAF_LOG_DEBUG("starting newb");
activate();
if (transport)
transport->prepare_next_read(this);
}
void stop() {
CAF_PUSH_AID_FROM_PTR(this);
CAF_LOG_TRACE("");
close_read_channel();
passivate();
}
write_handle<Message> wr_buf(header_writer* hw) {
// TODO: We somehow need to tell the transport policy how much we've
// written to enable it to split the buffer into datagrams.
auto& buf = transport->wr_buf();
auto hstart = buf.size();
protocol->write_header(buf, hw);
auto hlen = buf.size() - hstart;
return {this, protocol.get(), &buf, hstart, hlen};
}
void flush() {
transport->flush(this);
}
error read_event() {
return transport->read_some(this, *protocol);
}
void write_event() {
transport->write_some(this);
}
void handle_error() {
CAF_CRITICAL("got error to handle: not implemented");
}
// Protocol policies can set timeouts using a custom message.
template<class Rep = int, class Period = std::ratio<1>>
void set_timeout(std::chrono::duration<Rep, Period> timeout,
atom_value atm, uint32_t id) {
this->delayed_send(this, timeout, atm, id);
// TODO: Use actor clock.
// TODO: Make this system messages and handle them separately.
}
// Allow protocol policies to enqueue a data for sending. Probably required for
// reliability to send ACKs. The problem is that only the headers of policies
// lower, well closer to the transport, should write their headers. So we're
// facing a similiar porblem to slicing here.
// void (char* bytes, size_t count);
/// Returns the `multiplexer` running this broker. /// Returns the `multiplexer` running this broker.
network::multiplexer& backend() { network::multiplexer& backend() {
return super::system().middleman().backend(); return event_handler::backend();
} }
behavior make_behavior() { virtual void handle(Message& msg) = 0;
std::cout << "creating newb behavior" << std::endl;
// Currently has to handle timeouts as well, see handler below.
virtual behavior make_behavior() = 0; /*{
return { return {
[](int i) { [=](atom_value atm, uint32_t id) {
std::cout << "got message " << i << std::endl; protocol->timeout(atm, id);
} }
}; };
} }*/
void init_newb() { void configure_read(receive_policy::config config) {
CAF_LOG_TRACE(""); transport->configure_read(config);
super::setf(super::is_initialized_flag);
} }
/// @cond PRIVATE /// @cond PRIVATE
...@@ -237,24 +420,115 @@ public: ...@@ -237,24 +420,115 @@ public:
/// @endcond /// @endcond
private: // -- policies ---------------------------------------------------------------
Protocol protocol_;
std::tuple<Policies...> policies_; std::unique_ptr<transport_policy> transport;
std::unique_ptr<protocol_policy<Message>> protocol;
};
// -- new broker acceptor ------------------------------------------------------
template <class Message>
struct newb_acceptor : public network::event_handler {
// -- constructors and destructors -------------------------------------------
newb_acceptor(default_multiplexer& dm, native_socket sockfd)
: event_handler(dm, sockfd) {
// nop
}
// -- overridden modifiers of event handler ----------------------------------
void handle_event(network::operation op) override {
CAF_LOG_DEBUG("new event: " << to_string(op));
switch (op) {
case network::operation::read:
read_event();
break;
case network::operation::write:
// nop
break;
case network::operation::propagate_error:
CAF_LOG_DEBUG("acceptor got error operation");
break;
}
}
void removed_from_loop(network::operation op) override {
CAF_LOG_TRACE(CAF_ARG(op));
switch (op) {
case network::operation::read: break;
case network::operation::write: break;
case network::operation::propagate_error: ; // nop
}
}
// -- members ----------------------------------------------------------------
error read_event() {
native_socket sock;
transport_policy_ptr transport;
std::tie(sock, transport) = acceptor->accept(this);
auto en = create_newb(sock, std::move(transport));
if (!en)
return std::move(en.error());
auto ptr = caf::actor_cast<caf::abstract_actor*>(*en);
CAF_ASSERT(ptr != nullptr);
auto& ref = dynamic_cast<newb<Message>&>(*ptr);
acceptor->init(ref);
return none;
}
void start() {
activate();
}
void stop() {
CAF_LOG_TRACE(CAF_ARG2("fd", fd()));
close_read_channel();
passivate();
}
virtual expected<actor> create_newb(native_socket sock,
transport_policy_ptr pol) = 0;
// TODO: Has to implement a static create socket function ...
std::unique_ptr<accept_policy<Message>> acceptor;
}; };
template <class Protocol, class... Policies> // -- factories ----------------------------------------------------------------
actor make_newb(actor_system& sys, actor_config& cfg, default_multiplexer& mpx,
native_socket sockfd) { template <class Newb>
using policy_types = detail::type_list<Policies...>; actor make_newb(actor_system& sys, native_socket sockfd) {
static_assert(detail::tl_forall<policy_types, detail::is_mutating_policy_type>::value, auto& mpx = dynamic_cast<default_multiplexer&>(sys.middleman().backend());
"Only network policies allowed as template parameters"); actor_config acfg{&mpx};
static_assert(std::is_base_of<detail::protocol_policy, Protocol>::value, auto res = sys.spawn_impl<Newb, hidden + lazy_init>(acfg, mpx, sockfd);
"First template argument must be a protocol policy");
using newb_t = newb<Protocol, Policies...>;
auto res = sys.spawn_impl<newb_t, hidden>(cfg, mpx, sockfd);
return actor_cast<actor>(res); return actor_cast<actor>(res);
} }
// TODO: I feel like this should include the ProtocolPolicy somehow.
template <class NewbAcceptor, class AcceptPolicy>
std::unique_ptr<NewbAcceptor> make_newb_acceptor(actor_system& sys,
uint16_t port,
const char* addr = nullptr,
bool reuse_addr = false) {
auto sockfd = NewbAcceptor::create_socket(port, addr, reuse_addr);
// new_tcp_acceptor_impl(port, addr, reuse_addr);
if (!sockfd) {
CAF_LOG_DEBUG("Could not open " << CAF_ARG(port) << CAF_ARG(addr));
return nullptr;
}
auto& mpx = dynamic_cast<default_multiplexer&>(sys.middleman().backend());
std::unique_ptr<NewbAcceptor> ptr{new NewbAcceptor(mpx, *sockfd)};
ptr->acceptor.reset(new AcceptPolicy);
ptr->start();
return ptr;
}
} // namespace network } // namespace network
} // namespace io } // namespace io
} // namespace caf } // namespace caf
...@@ -26,121 +26,17 @@ ...@@ -26,121 +26,17 @@
#include "caf/test/dsl.hpp" #include "caf/test/dsl.hpp"
#include "caf/callback.hpp" #include "caf/io/network/newb.hpp"
#include "caf/config.hpp"
#include "caf/detail/call_cfun.hpp"
#include "caf/detail/enum_to_string.hpp"
#include "caf/detail/socket_guard.hpp"
#include "caf/io/broker.hpp"
#include "caf/io/middleman.hpp"
#include "caf/io/network/default_multiplexer.hpp"
#include "caf/io/network/event_handler.hpp"
#include "caf/io/network/interfaces.hpp"
#include "caf/io/network/native_socket.hpp"
#include "caf/mixin/behavior_changer.hpp"
#include "caf/mixin/requester.hpp"
#include "caf/mixin/sender.hpp"
#include "caf/scheduled_actor.hpp"
#ifdef CAF_WINDOWS
# ifndef WIN32_LEAN_AND_MEAN
# define WIN32_LEAN_AND_MEAN
# endif // WIN32_LEAN_AND_MEAN
# ifndef NOMINMAX
# define NOMINMAX
# endif
# ifdef CAF_MINGW
# undef _WIN32_WINNT
# undef WINVER
# define _WIN32_WINNT WindowsVista
# define WINVER WindowsVista
# include <w32api.h>
# endif
# include <io.h>
# include <windows.h>
# include <winsock2.h>
# include <ws2ipdef.h>
# include <ws2tcpip.h>
#else
# include <unistd.h>
# include <arpa/inet.h>
# include <cerrno>
# include <fcntl.h>
# include <netdb.h>
# include <netinet/in.h>
# include <netinet/ip.h>
# include <netinet/tcp.h>
# include <sys/socket.h>
# include <sys/types.h>
# ifdef CAF_POLL_MULTIPLEXER
# include <poll.h>
# elif defined(CAF_EPOLL_MULTIPLEXER)
# include <sys/epoll.h>
# else
# error "neither CAF_POLL_MULTIPLEXER nor CAF_EPOLL_MULTIPLEXER defined"
# endif
#endif
using namespace caf; using namespace caf;
using namespace caf::io; using namespace caf::io;
using network::native_socket; using io::network::byte_buffer;
using network::invalid_native_socket; using io::network::header_writer;
using network::default_multiplexer; using network::default_multiplexer;
using network::invalid_native_socket;
using network::last_socket_error_as_string; using network::last_socket_error_as_string;
using network::native_socket;
namespace {
// -- forward declarations -----------------------------------------------------
struct dummy_device;
struct newb_base;
struct protocol_policy_base;
template <class T>
struct protocol_policy;
template <class T>
struct newb;
constexpr auto ipv4 = caf::io::network::protocol::ipv4;
//constexpr auto ipv6 = caf::io::network::protocol::ipv6;
auto addr_of(sockaddr_in& what) -> decltype(what.sin_addr)& {
return what.sin_addr;
}
auto family_of(sockaddr_in& what) -> decltype(what.sin_family)& {
return what.sin_family;
}
auto port_of(sockaddr_in& what) -> decltype(what.sin_port)& {
return what.sin_port;
}
auto addr_of(sockaddr_in6& what) -> decltype(what.sin6_addr)& {
return what.sin6_addr;
}
auto family_of(sockaddr_in6& what) -> decltype(what.sin6_family)& {
return what.sin6_family;
}
auto port_of(sockaddr_in6& what) -> decltype(what.sin6_port)& {
return what.sin6_port;
}
} // namespace <anonymous>
namespace caf {
template <class T>
class behavior_type_of<newb<T>> {
public:
using type = behavior;
};
} // namespace caf
namespace { namespace {
...@@ -152,11 +48,6 @@ using send_atom = atom_constant<atom("send")>; ...@@ -152,11 +48,6 @@ using send_atom = atom_constant<atom("send")>;
using shutdown_atom = atom_constant<atom("shutdown")>; using shutdown_atom = atom_constant<atom("shutdown")>;
using quit_atom = atom_constant<atom("quit")>; using quit_atom = atom_constant<atom("quit")>;
// -- aliases ------------------------------------------------------------------
using byte_buffer = std::vector<char>;
using header_writer = caf::callback<byte_buffer&>;
// -- dummy headers ------------------------------------------------------------ // -- dummy headers ------------------------------------------------------------
struct basp_header { struct basp_header {
...@@ -178,7 +69,7 @@ typename Inspector::result_type inspect(Inspector& fun, ordering_header& hdr) { ...@@ -178,7 +69,7 @@ typename Inspector::result_type inspect(Inspector& fun, ordering_header& hdr) {
return fun(meta::type_name("ordering_header"), hdr.seq_nr); return fun(meta::type_name("ordering_header"), hdr.seq_nr);
} }
// -- message types ------------------------------------------------------------ // -- message type -------------------------------------------------------------
struct new_basp_message { struct new_basp_message {
basp_header header; basp_header header;
...@@ -194,1145 +85,58 @@ typename Inspector::result_type inspect(Inspector& f, new_basp_message& x) { ...@@ -194,1145 +85,58 @@ typename Inspector::result_type inspect(Inspector& f, new_basp_message& x) {
} // namespace anonymous } // namespace anonymous
CAF_ALLOW_UNSAFE_MESSAGE_TYPE(new_basp_message); CAF_ALLOW_UNSAFE_MESSAGE_TYPE(new_basp_message);
namespace {
// -- transport policy ---------------------------------------------------------
struct transport_policy {
transport_policy() : received_bytes{0}, max_consecutive_reads{50} {
}
virtual ~transport_policy() {
// nop
}
virtual error write_some(network::event_handler*) {
return none;
}
virtual error read_some(network::event_handler*) {
return none;
}
virtual bool should_deliver() {
return true;
}
virtual void prepare_next_read(network::event_handler*) {
// nop
}
virtual void prepare_next_write(network::event_handler*) {
// nop
}
virtual void configure_read(receive_policy::config) {
// nop
}
virtual void flush(network::event_handler*) {
// nop
}
byte_buffer& wr_buf() {
return offline_buffer;
}
template <class T>
error read_some(network::event_handler* parent, protocol_policy<T>& policy) {
CAF_LOG_TRACE("");
auto mcr = max_consecutive_reads;
for (size_t i = 0; i < mcr; ++i) {
auto res = read_some(parent);
// The return statements seems weird, needs cleanup.
if (res)
return res;
if (should_deliver()) {
res = policy.read(receive_buffer.data(), received_bytes);
prepare_next_read(parent);
if (!res)
return res;
}
}
return none;
}
size_t received_bytes;
size_t max_consecutive_reads;
byte_buffer offline_buffer;
byte_buffer receive_buffer;
byte_buffer send_buffer;
};
using transport_policy_ptr = std::unique_ptr<transport_policy>;
// -- accept policy ------------------------------------------------------------
template <class Message>
struct accept_policy {
virtual ~accept_policy() {
// nop
}
virtual std::pair<native_socket, transport_policy_ptr>
accept(network::event_handler*) = 0;
virtual void init(newb<Message>&) = 0;
};
// -- protocol policies --------------------------------------------------------
struct protocol_policy_base {
virtual ~protocol_policy_base() {
// nop
}
virtual void write_header(byte_buffer&, header_writer*) = 0;
virtual void prepare_for_sending(byte_buffer&, size_t, size_t) = 0;
};
template <class T>
struct protocol_policy : protocol_policy_base {
using message_type = T;
virtual ~protocol_policy() override {
// nop
}
virtual error read(char* bytes, size_t count) = 0;
virtual error timeout(atom_value, uint32_t) = 0;
};
template <class T>
using protocol_policy_ptr = std::unique_ptr<protocol_policy<T>>;
template <class T>
struct protocol_policy_impl : protocol_policy<typename T::message_type> {
T impl;
protocol_policy_impl(newb<typename T::message_type>* parent) : impl(parent) {
// nop
}
error read(char* bytes, size_t count) override {
return impl.read(bytes, count);
}
error timeout(atom_value atm, uint32_t id) override {
return impl.timeout(atm, id);
}
void write_header(byte_buffer& buf, header_writer* hw) override {
return impl.write_header(buf, hw);
}
void prepare_for_sending(byte_buffer&, size_t, size_t) override {
return;
}
};
// -- new broker classes -------------------------------------------------------
/// @relates newb
/// Returned by funtion wr_buf of newb.
template <class Message>
struct write_handle {
newb<Message>* parent;
protocol_policy_base* protocol;
byte_buffer* buf;
size_t header_start;
size_t header_len;
~write_handle() {
// Can we calculate added bytes for datagram things?
auto payload_size = buf->size() - (header_start + header_len);
protocol->prepare_for_sending(*buf, header_start, payload_size);
parent->flush();
}
};
template <class Message>
struct newb : public extend<scheduled_actor, newb<Message>>::template
with<mixin::sender, mixin::requester,
mixin::behavior_changer>,
public dynamically_typed_actor_base,
public network::event_handler {
using super = typename extend<scheduled_actor, newb<Message>>::
template with<mixin::sender, mixin::requester, mixin::behavior_changer>;
using signatures = none_t;
// -- constructors and destructors -------------------------------------------
newb(actor_config& cfg, default_multiplexer& dm, native_socket sockfd)
: super(cfg),
event_handler(dm, sockfd) {
// nop
}
newb() = default;
newb(newb<Message>&&) = default;
~newb() override {
// nop
}
// -- overridden modifiers of abstract_actor ---------------------------------
void enqueue(mailbox_element_ptr ptr, execution_unit*) override {
CAF_PUSH_AID(this->id());
scheduled_actor::enqueue(std::move(ptr), &backend());
}
void enqueue(strong_actor_ptr src, message_id mid, message msg,
execution_unit*) override {
enqueue(make_mailbox_element(std::move(src), mid, {}, std::move(msg)),
&backend());
}
resumable::subtype_t subtype() const override {
return resumable::io_actor;
}
// -- overridden modifiers of local_actor ------------------------------------
void launch(execution_unit* eu, bool lazy, bool hide) override {
CAF_PUSH_AID_FROM_PTR(this);
CAF_ASSERT(eu != nullptr);
CAF_ASSERT(eu == &backend());
CAF_LOG_TRACE(CAF_ARG(lazy) << CAF_ARG(hide));
// add implicit reference count held by middleman/multiplexer
if (!hide)
super::register_at_system();
if (lazy && super::mailbox().try_block())
return;
intrusive_ptr_add_ref(super::ctrl());
eu->exec_later(this);
}
void initialize() override {
CAF_LOG_TRACE("");
init_newb();
auto bhvr = make_behavior();
CAF_LOG_DEBUG_IF(!bhvr, "make_behavior() did not return a behavior:"
<< CAF_ARG(this->has_behavior()));
if (bhvr) {
// make_behavior() did return a behavior instead of using become()
CAF_LOG_DEBUG("make_behavior() did return a valid behavior");
this->become(std::move(bhvr));
}
}
bool cleanup(error&& reason, execution_unit* host) override {
CAF_LOG_TRACE(CAF_ARG(reason));
// TODO: Ask policies, close socket.
return local_actor::cleanup(std::move(reason), host);
}
// -- overridden modifiers of resumable --------------------------------------
network::multiplexer::runnable::resume_result resume(execution_unit* ctx,
size_t mt) override {
CAF_PUSH_AID_FROM_PTR(this);
CAF_ASSERT(ctx != nullptr);
CAF_ASSERT(ctx == &backend());
return scheduled_actor::resume(ctx, mt);
}
// -- overridden modifiers of event handler ----------------------------------
void handle_event(network::operation op) override {
CAF_PUSH_AID_FROM_PTR(this);
CAF_LOG_TRACE("");
switch (op) {
case io::network::operation::read:
read_event();
break;
case io::network::operation::write:
write_event();
break;
case io::network::operation::propagate_error:
handle_error();
}
}
void removed_from_loop(network::operation op) override {
CAF_MESSAGE("newb removed from loop: " << to_string(op));
CAF_PUSH_AID_FROM_PTR(this);
CAF_LOG_TRACE(CAF_ARG(op));
switch (op) {
case network::operation::read: break;
case network::operation::write: break;
case network::operation::propagate_error: ; // nop
}
// nop
}
// -- members ----------------------------------------------------------------
void init_newb() {
CAF_LOG_TRACE("");
super::setf(super::is_initialized_flag);
}
void start() {
CAF_PUSH_AID_FROM_PTR(this);
CAF_LOG_TRACE("");
CAF_MESSAGE("starting newb");
activate();
if (transport)
transport->prepare_next_read(this);
}
void stop() {
CAF_PUSH_AID_FROM_PTR(this);
CAF_LOG_TRACE("");
close_read_channel();
passivate();
}
write_handle<Message> wr_buf(header_writer* hw) {
// TODO: We somehow need to tell the transport policy how much we've
// written to enable it to split the buffer into datagrams.
auto& buf = transport->wr_buf();
auto hstart = buf.size();
protocol->write_header(buf, hw);
auto hlen = buf.size() - hstart;
return {this, protocol.get(), &buf, hstart, hlen};
}
void flush() {
transport->flush(this);
}
error read_event() {
return transport->read_some(this, *protocol);
}
void write_event() {
transport->write_some(this);
}
void handle_error() {
CAF_CRITICAL("got error to handle: not implemented");
}
// Protocol policies can set timeouts using a custom message.
template<class Rep = int, class Period = std::ratio<1>>
void set_timeout(std::chrono::duration<Rep, Period> timeout,
atom_value atm, uint32_t id) {
this->delayed_send(this, timeout, atm, id);
// TODO: Use actor clock.
// TODO: Make this system messages and handle them separately.
}
// Allow protocol policies to enqueue a data for sending. Probably required for
// reliability to send ACKs. The problem is that only the headers of policies
// lower, well closer to the transport, should write their headers. So we're
// facing a similiar porblem to slicing here.
// void (char* bytes, size_t count);
/// Returns the `multiplexer` running this broker.
network::multiplexer& backend() {
return event_handler::backend();
}
virtual void handle(Message& msg) = 0;
// Currently has to handle timeouts as well, see handler below.
virtual behavior make_behavior() = 0; /*{
return {
[=](atom_value atm, uint32_t id) {
protocol->timeout(atm, id);
}
};
}*/
void configure_read(receive_policy::config config) {
transport->configure_read(config);
}
/// @cond PRIVATE
template <class... Ts>
void eq_impl(message_id mid, strong_actor_ptr sender,
execution_unit* ctx, Ts&&... xs) {
enqueue(make_mailbox_element(std::move(sender), mid,
{}, std::forward<Ts>(xs)...),
ctx);
}
/// @endcond
// -- policies ---------------------------------------------------------------
std::unique_ptr<transport_policy> transport;
std::unique_ptr<protocol_policy<Message>> protocol;
};
template <class Message>
struct newb_acceptor : public network::event_handler {
// -- constructors and destructors -------------------------------------------
newb_acceptor(default_multiplexer& dm, native_socket sockfd)
: event_handler(dm, sockfd) {
// nop
}
// -- overridden modifiers of event handler ----------------------------------
void handle_event(network::operation op) override {
CAF_MESSAGE("new event: " << to_string(op));
switch (op) {
case network::operation::read:
read_event();
break;
case network::operation::write:
// nop
break;
case network::operation::propagate_error:
CAF_MESSAGE("acceptor got error operation");
break;
}
}
void removed_from_loop(network::operation op) override {
CAF_LOG_TRACE(CAF_ARG(op));
switch (op) {
case network::operation::read: break;
case network::operation::write: break;
case network::operation::propagate_error: ; // nop
}
}
// -- members ----------------------------------------------------------------
error read_event() {
native_socket sock;
transport_policy_ptr transport;
std::tie(sock, transport) = acceptor->accept(this);
auto en = create_newb(sock, std::move(transport));
if (!en)
return std::move(en.error());
auto ptr = caf::actor_cast<caf::abstract_actor*>(*en);
CAF_ASSERT(ptr != nullptr);
auto& ref = dynamic_cast<newb<Message>&>(*ptr);
acceptor->init(ref);
return none;
}
void start() {
activate();
}
void stop() {
CAF_LOG_TRACE(CAF_ARG2("fd", fd()));
close_read_channel();
passivate();
}
virtual expected<actor> create_newb(native_socket sock,
transport_policy_ptr pol) = 0;
std::unique_ptr<accept_policy<Message>> acceptor;
};
// -- policies -----------------------------------------------------------------
/// @relates protocol_policy
/// Protocol policy layer for the BASP application protocol.
struct basp_policy {
static constexpr size_t header_size = sizeof(basp_header);
static constexpr size_t offset = header_size;
using message_type = new_basp_message;
using result_type = optional<new_basp_message>;
newb<message_type>* parent;
basp_policy(newb<message_type>* parent) : parent(parent) {
// nop
}
error read(char* bytes, size_t count) {
if (count < header_size) {
CAF_MESSAGE("data left in packet to small to contain the basp header");
return sec::unexpected_message;
}
new_basp_message msg;
binary_deserializer bd(&parent->backend(), bytes, count);
bd(msg.header);
msg.payload = bytes + header_size;
msg.payload_size = count - header_size;
parent->handle(msg);
return none;
}
error timeout(atom_value, uint32_t) {
return none;
}
void write_header(byte_buffer& buf, header_writer* hw) {
CAF_ASSERT(hw != nullptr);
(*hw)(buf);
return;
}
};
/// @relates protocol_policy
/// Protocol policy layer for ordering.
template <class Next>
struct ordering {
static constexpr size_t header_size = sizeof(ordering_header);
static constexpr size_t offset = Next::offset + header_size;
using message_type = typename Next::message_type;
using result_type = typename Next::result_type;
uint32_t seq_read = 0;
uint32_t seq_write = 0;
newb<message_type>* parent;
Next next;
std::unordered_map<uint32_t, std::vector<char>> pending;
ordering(newb<message_type>* parent) : parent(parent), next(parent) {
// nop
}
error deliver_pending() {
if (pending.empty())
return none;
while (pending.count(seq_read) > 0) {
auto& buf = pending[seq_read];
auto res = next.read(buf.data(), buf.size());
pending.erase(seq_read);
if (res)
return res;
}
return none;
}
error read(char* bytes, size_t count) {
CAF_MESSAGE("ordering: read (" << count << ")");
if (count < header_size)
return sec::unexpected_message;
ordering_header hdr;
binary_deserializer bd(&parent->backend(), bytes, count);
bd(hdr);
// TODO: Use the comparison function from BASP instance.
if (hdr.seq_nr == seq_read) {
seq_read += 1;
auto res = next.read(bytes + header_size, count - header_size);
if (res)
return res;
return deliver_pending();
} else if (hdr.seq_nr > seq_read) {
pending[hdr.seq_nr] = std::vector<char>(bytes + header_size,
bytes + count);
parent->set_timeout(std::chrono::seconds(2), ordering_atom::value,
hdr.seq_nr);
return none;
}
// Is late, drop it. TODO: Should this return an error?
return none;
}
error timeout(atom_value atm, uint32_t id) {
if (atm == ordering_atom::value) {
error err;
if (pending.count(id) > 0) {
auto& buf = pending[id];
err = next.read(buf.data(), buf.size());
seq_read = id + 1;
if (!err)
err = deliver_pending();
}
return err;
}
return next.timeout(atm, id);
}
void write_header(byte_buffer& buf, header_writer* hw) {
binary_serializer bs(&parent->backend(), buf);
bs(ordering_header{seq_write});
seq_write += 1;
next.write_header(buf, hw);
return;
}
};
// -- network code -------------------------------------------------------------
expected<void> set_inaddr_any(native_socket, sockaddr_in& sa) {
sa.sin_addr.s_addr = INADDR_ANY;
return unit;
}
expected<void> set_inaddr_any(native_socket fd, sockaddr_in6& sa) {
sa.sin6_addr = in6addr_any;
// also accept ipv4 requests on this socket
int off = 0;
CALL_CFUN(res, detail::cc_zero, "setsockopt",
setsockopt(fd, IPPROTO_IPV6, IPV6_V6ONLY,
reinterpret_cast<network::setsockopt_ptr>(&off),
static_cast<network::socket_size_type>(sizeof(off))));
return unit;
}
template <int Family, int SockType = SOCK_STREAM>
expected<native_socket> new_ip_acceptor_impl(uint16_t port, const char* addr,
bool reuse_addr, bool any) {
static_assert(Family == AF_INET || Family == AF_INET6, "invalid family");
CAF_LOG_TRACE(CAF_ARG(port) << ", addr = " << (addr ? addr : "nullptr"));
CALL_CFUN(fd, detail::cc_valid_socket, "socket", socket(Family, SockType, 0));
// sguard closes the socket in case of exception
detail::socket_guard sguard{fd};
if (reuse_addr) {
int on = 1;
CALL_CFUN(tmp1, detail::cc_zero, "setsockopt",
setsockopt(fd, SOL_SOCKET, SO_REUSEADDR,
reinterpret_cast<network::setsockopt_ptr>(&on),
static_cast<network::socket_size_type>(sizeof(on))));
}
using sockaddr_type =
typename std::conditional<
Family == AF_INET,
sockaddr_in,
sockaddr_in6
>::type;
sockaddr_type sa;
memset(&sa, 0, sizeof(sockaddr_type));
family_of(sa) = Family;
if (any)
set_inaddr_any(fd, sa);
CALL_CFUN(tmp, detail::cc_one, "inet_pton",
inet_pton(Family, addr, &addr_of(sa)));
port_of(sa) = htons(port);
CALL_CFUN(res, detail::cc_zero, "bind",
bind(fd, reinterpret_cast<sockaddr*>(&sa),
static_cast<network::socket_size_type>(sizeof(sa))));
return sguard.release();
}
expected<native_socket> new_tcp_acceptor_impl(uint16_t port, const char* addr,
bool reuse_addr) {
CAF_LOG_TRACE(CAF_ARG(port) << ", addr = " << (addr ? addr : "nullptr"));
auto addrs = network::interfaces::server_address(port, addr);
auto addr_str = std::string{addr == nullptr ? "" : addr};
if (addrs.empty())
return make_error(sec::cannot_open_port, "No local interface available",
addr_str);
bool any = addr_str.empty() || addr_str == "::" || addr_str == "0.0.0.0";
auto fd = invalid_native_socket;
for (auto& elem : addrs) {
auto hostname = elem.first.c_str();
auto p = elem.second == ipv4
? new_ip_acceptor_impl<AF_INET>(port, hostname, reuse_addr, any)
: new_ip_acceptor_impl<AF_INET6>(port, hostname, reuse_addr, any);
if (!p) {
CAF_LOG_DEBUG(p.error());
continue;
}
fd = *p;
break;
}
if (fd == invalid_native_socket) {
CAF_LOG_WARNING("could not open tcp socket on:" << CAF_ARG(port)
<< CAF_ARG(addr_str));
return make_error(sec::cannot_open_port, "tcp socket creation failed",
port, addr_str);
}
detail::socket_guard sguard{fd};
CALL_CFUN(tmp2, detail::cc_zero, "listen", listen(fd, SOMAXCONN));
// ok, no errors so far
CAF_LOG_DEBUG(CAF_ARG(fd));
return sguard.release();
}
expected<native_socket> new_udp_endpoint_impl(uint16_t port, const char* addr,
bool reuse) {
CAF_LOG_TRACE(CAF_ARG(port) << ", addr = " << (addr ? addr : "nullptr"));
auto addrs = network::interfaces::server_address(port, addr);
auto addr_str = std::string{addr == nullptr ? "" : addr};
if (addrs.empty())
return make_error(sec::cannot_open_port, "No local interface available",
addr_str);
bool any = addr_str.empty() || addr_str == "::" || addr_str == "0.0.0.0";
auto fd = invalid_native_socket;
for (auto& elem : addrs) {
auto host = elem.first.c_str();
auto p = elem.second == ipv4
? new_ip_acceptor_impl<AF_INET, SOCK_DGRAM>(port, host, reuse, any)
: new_ip_acceptor_impl<AF_INET6, SOCK_DGRAM>(port, host, reuse, any);
if (!p) {
CAF_LOG_DEBUG(p.error());
continue;
}
fd = *p;
break;
}
if (fd == invalid_native_socket) {
CAF_LOG_WARNING("could not open udp socket on:" << CAF_ARG(port)
<< CAF_ARG(addr_str));
return make_error(sec::cannot_open_port, "udp socket creation failed",
port, addr_str);
}
CAF_LOG_DEBUG(CAF_ARG(fd));
return fd;
}
// -- create newbs -------------------------------------------------------------
template <class Newb>
actor make_newb(actor_system& sys, native_socket sockfd) {
auto& mpx = dynamic_cast<default_multiplexer&>(sys.middleman().backend());
actor_config acfg{&mpx};
auto res = sys.spawn_impl<Newb, hidden + lazy_init>(acfg, mpx, sockfd);
return actor_cast<actor>(res);
}
// TODO: I feel like this should include the ProtocolPolicy somehow.
template <class NewbAcceptor, class AcceptPolicy>
std::unique_ptr<NewbAcceptor> make_newb_acceptor(actor_system& sys,
uint16_t port,
const char* addr = nullptr,
bool reuse_addr = false) {
auto sockfd = NewbAcceptor::create_socket(port, addr, reuse_addr);
// new_tcp_acceptor_impl(port, addr, reuse_addr);
if (!sockfd) {
CAF_LOG_DEBUG("Could not open " << CAF_ARG(port) << CAF_ARG(addr));
return nullptr;
}
auto& mpx = dynamic_cast<default_multiplexer&>(sys.middleman().backend());
std::unique_ptr<NewbAcceptor> ptr{new NewbAcceptor(mpx, *sockfd)};
ptr->acceptor.reset(new AcceptPolicy);
ptr->start();
return ptr;
}
// -- tcp impls ----------------------------------------------------------------
struct tcp_basp_header {
uint32_t payload_len;
actor_id from;
actor_id to;
};
constexpr size_t tcp_basp_header_len = sizeof(uint32_t) + sizeof(actor_id) * 2;
template <class Inspector>
typename Inspector::result_type inspect(Inspector& fun, tcp_basp_header& hdr) {
return fun(meta::type_name("tcp_basp_header"),
hdr.payload_len, hdr.from, hdr.to);
}
struct new_tcp_basp_message {
tcp_basp_header header;
char* payload;
size_t payload_len;
};
template <class Inspector>
typename Inspector::result_type inspect(Inspector& fun,
new_tcp_basp_message& msg) {
return fun(meta::type_name("new_tcp_basp_message"), msg.header,
msg.payload_len);
}
struct tcp_basp {
static constexpr size_t header_size = sizeof(tcp_basp_header);
using message_type = new_tcp_basp_message;
using result_type = optional<message_type>;
newb<message_type>* parent;
message_type msg;
bool expecting_header = true;
tcp_basp(newb<message_type>* parent) : parent(parent) {
// nop
}
error read_header(char* bytes, size_t count) {
if (count < tcp_basp_header_len) {
CAF_LOG_DEBUG("buffer contains " << count << " bytes of expected "
<< tcp_basp_header_len);
return sec::unexpected_message;
}
binary_deserializer bd{&parent->backend(), bytes, count};
bd(msg.header);
CAF_LOG_DEBUG("read header " << CAF_ARG(msg.header));
size_t size = static_cast<size_t>(msg.header.payload_len);
parent->configure_read(receive_policy::exactly(size));
expecting_header = false;
return none;
}
error read_payload(char* bytes, size_t count) {
if (count < msg.header.payload_len) {
CAF_LOG_DEBUG("buffer contains " << count << " bytes of expected "
<< msg.header.payload_len);
return sec::unexpected_message;
}
msg.payload = bytes;
msg.payload_len = msg.header.payload_len;
parent->handle(msg);
expecting_header = true;
parent->configure_read(receive_policy::exactly(tcp_basp_header_len));
return none;
}
error read(char* bytes, size_t count) {
if (expecting_header)
return read_header(bytes, count);
else
return read_payload(bytes, count);
}
error timeout(atom_value, uint32_t) {
return none;
}
size_t write_header(byte_buffer& buf, header_writer* hw) {
CAF_ASSERT(hw != nullptr);
(*hw)(buf);
return header_size;
}
void prepare_for_sending(byte_buffer& buf, size_t hstart, size_t plen) {
stream_serializer<charbuf> out{&parent->backend(), buf.data() + hstart,
sizeof(uint32_t)};
auto len = static_cast<uint32_t>(plen);
out(len);
}
};
struct tcp_transport_policy : public transport_policy {
tcp_transport_policy()
: read_threshold{0},
collected{0},
maximum{0},
writing{false},
written{0} {
// nop
}
error read_some(network::event_handler* parent) override {
CAF_LOG_TRACE("");
size_t len = receive_buffer.size() - collected;
receive_buffer.resize(len);
void* buf = receive_buffer.data() + collected;
auto sres = ::recv(parent->fd(),
reinterpret_cast<network::socket_recv_ptr>(buf),
len, network::no_sigpipe_io_flag);
if (network::is_error(sres, true) || sres == 0) {
// recv returns 0 when the peer has performed an orderly shutdown
return sec::runtime_error;
}
size_t result = (sres > 0) ? static_cast<size_t>(sres) : 0;
collected += result;
received_bytes = collected;
return none;
}
bool should_deliver() override {
CAF_LOG_DEBUG(CAF_ARG(collected) << CAF_ARG(read_threshold));
return collected >= read_threshold;
}
void prepare_next_read(network::event_handler*) override {
collected = 0;
received_bytes = 0;
switch (rd_flag) {
case receive_policy_flag::exactly:
if (receive_buffer.size() != maximum)
receive_buffer.resize(maximum);
read_threshold = maximum;
break;
case receive_policy_flag::at_most:
if (receive_buffer.size() != maximum)
receive_buffer.resize(maximum);
read_threshold = 1;
break;
case receive_policy_flag::at_least: {
// read up to 10% more, but at least allow 100 bytes more
auto maximumsize = maximum + std::max<size_t>(100, maximum / 10);
if (receive_buffer.size() != maximumsize)
receive_buffer.resize(maximumsize);
read_threshold = maximum;
break;
}
}
}
void configure_read(receive_policy::config config) override {
rd_flag = config.first;
maximum = config.second;
}
error write_some(network::event_handler* parent) override {
CAF_LOG_TRACE("");
const void* buf = send_buffer.data() + written;
auto len = send_buffer.size() - written;
auto sres = ::send(parent->fd(),
reinterpret_cast<network::socket_send_ptr>(buf),
len, network::no_sigpipe_io_flag);
if (network::is_error(sres, true))
return sec::runtime_error;
size_t result = (sres > 0) ? static_cast<size_t>(sres) : 0;
written += result;
auto remaining = send_buffer.size() - written;
if (remaining == 0)
prepare_next_write(parent);
return none;
}
void prepare_next_write(network::event_handler* parent) override {
written = 0;
send_buffer.clear();
if (offline_buffer.empty()) {
writing = false;
parent->backend().del(network::operation::write, parent->fd(), parent);
} else {
send_buffer.swap(offline_buffer);
}
}
byte_buffer& wr_buf() {
return offline_buffer;
}
void flush(network::event_handler* parent) override {
CAF_ASSERT(parent != nullptr);
CAF_LOG_TRACE(CAF_ARG(offline_buffer.size()));
if (!offline_buffer.empty() && !writing) {
parent->backend().add(network::operation::write, parent->fd(), parent);
writing = true;
prepare_next_write(parent);
}
}
// State for reading.
size_t read_threshold;
size_t collected;
size_t maximum;
receive_policy_flag rd_flag;
// State for writing.
bool writing;
size_t written;
};
template <class T>
struct tcp_protocol_policy : protocol_policy<typename T::message_type> {
T impl;
tcp_protocol_policy(newb<typename T::message_type>* parent) : impl(parent) {
// nop
}
error read(char* bytes, size_t count) override {
return impl.read(bytes, count);
}
error timeout(atom_value atm, uint32_t id) override {
return impl.timeout(atm, id);
}
void write_header(byte_buffer& buf, header_writer* hw) override {
impl.write_header(buf, hw);
}
void prepare_for_sending(byte_buffer& buf, size_t hstart,
size_t plen) override {
impl.prepare_for_sending(buf, hstart, plen);
}
};
struct tcp_basp_newb : newb<new_tcp_basp_message> {
tcp_basp_newb(caf::actor_config& cfg, default_multiplexer& dm,
native_socket sockfd)
: newb<new_tcp_basp_message>(cfg, dm, sockfd) {
// nop
}
void handle(new_tcp_basp_message& msg) override {
CAF_PUSH_AID_FROM_PTR(this);
CAF_LOG_TRACE("");
std::string res;
binary_deserializer bd(&backend(), msg.payload, msg.payload_len);
bd(res);
send(responder, res);
}
behavior make_behavior() override {
set_default_handler(print_and_drop);
return {
// Must be implemented at the moment, will be cought by the broker in a
// later implementation.
[=](atom_value atm, uint32_t id) {
protocol->timeout(atm, id);
},
[=](send_atom, actor_id sender, actor_id receiver, std::string payload) {
auto hw = caf::make_callback([&](byte_buffer& buf) -> error {
binary_serializer bs(&backend(), buf);
bs(tcp_basp_header{0, sender, receiver});
return none;
});
auto whdl = wr_buf(&hw);
CAF_CHECK(whdl.buf != nullptr);
CAF_CHECK(whdl.protocol != nullptr);
binary_serializer bs(&backend(), *whdl.buf);
bs(payload);
},
[=](quit_atom) {
// Remove from multiplexer loop.
stop();
// Quit actor.
quit();
}
};
}
actor responder;
};
struct tcp_accept_policy : public accept_policy<new_tcp_basp_message> {
virtual std::pair<native_socket, transport_policy_ptr>
accept(network::event_handler* parent) {
using namespace io::network;
sockaddr_storage addr;
std::memset(&addr, 0, sizeof(addr));
socket_size_type addrlen = sizeof(addr);
auto result = ::accept(parent->fd(), reinterpret_cast<sockaddr*>(&addr),
&addrlen);
if (result == invalid_native_socket) {
auto err = last_socket_error();
if (!would_block_or_temporarily_unavailable(err)) {
return {invalid_native_socket, nullptr};
}
}
transport_policy_ptr ptr{new tcp_transport_policy};
return {result, std::move(ptr)};
}
virtual void init(newb<new_tcp_basp_message>& n) {
n.start();
}
};
template <class ProtocolPolicy>
struct tcp_basp_acceptor
: public newb_acceptor<typename ProtocolPolicy::message_type> {
using super = newb_acceptor<typename ProtocolPolicy::message_type>;
static expected<native_socket> create_socket(uint16_t port, const char* host,
bool reuse = false) {
return new_tcp_acceptor_impl(port, host, reuse);
}
tcp_basp_acceptor(default_multiplexer& dm, native_socket sockfd)
: super(dm, sockfd) {
// nop
}
expected<actor> create_newb(native_socket sockfd,
transport_policy_ptr pol) override {
CAF_MESSAGE("creating new basp tcp newb");
auto n = make_newb<tcp_basp_newb>(this->backend().system(), sockfd);
auto ptr = caf::actor_cast<caf::abstract_actor*>(n);
if (ptr == nullptr)
return sec::runtime_error;
auto& ref = dynamic_cast<tcp_basp_newb&>(*ptr);
ref.transport = std::move(pol);
ref.protocol.reset(new ProtocolPolicy(&ref));
ref.responder = responder;
// This should happen somewhere else?
ref.configure_read(receive_policy::exactly(tcp_basp_header_len));
anon_send(responder, n);
return n;
}
actor responder;
};
// -- udp impls ----------------------------------------------------------------
struct udp_basp_header {
uint32_t payload_len;
actor_id from;
actor_id to;
};
template <class Inspector> namespace {
typename Inspector::result_type inspect(Inspector& fun, udp_basp_header& hdr) {
return fun(meta::type_name("udp_basp_header"), hdr.payload_len,
hdr.from, hdr.to);
}
constexpr size_t udp_basp_header_len = sizeof(uint32_t) + sizeof(actor_id) * 2; template <class T>
struct protocol_policy_impl
: public network::protocol_policy<typename T::message_type> {
T impl;
using sequence_type = uint16_t; protocol_policy_impl(network::newb<typename T::message_type>* parent)
: impl(parent) {
// nop
}
struct udp_ordering_header { error read(char* bytes, size_t count) override {
sequence_type seq; return impl.read(bytes, count);
}; }
template <class Inspector> error timeout(atom_value atm, uint32_t id) override {
typename Inspector::result_type inspect(Inspector& fun, udp_ordering_header& hdr) { return impl.timeout(atm, id);
return fun(meta::type_name("udp_ordering_header"), hdr.seq); }
}
constexpr size_t udp_ordering_header_len = sizeof(sequence_type); void write_header(byte_buffer& buf, header_writer* hw) override {
return impl.write_header(buf, hw);
}
struct new_udp_basp_message { void prepare_for_sending(byte_buffer&, size_t, size_t) override {
udp_basp_header header; return;
char* payload; }
size_t payload_len;
}; };
template <class Inspector> struct basp_policy {
typename Inspector::result_type inspect(Inspector& fun, static constexpr size_t header_size = sizeof(basp_header);
new_udp_basp_message& msg) { static constexpr size_t offset = header_size;
return fun(meta::type_name("new_udp_basp_message"), msg.header, using message_type = new_basp_message;
msg.payload_len); using result_type = optional<new_basp_message>;
} network::newb<message_type>* parent;
struct udp_basp {
static constexpr size_t header_size = udp_basp_header_len;
using message_type = new_udp_basp_message;
using result_type = optional<message_type>;
newb<message_type>* parent;
message_type msg;
udp_basp(newb<message_type>* parent) : parent(parent) { basp_policy(network::newb<message_type>* parent) : parent(parent) {
// nop // nop
} }
error read(char* bytes, size_t count) { error read(char* bytes, size_t count) {
CAF_MESSAGE("reading basp udp header"); if (count < header_size) {
// Read header. CAF_MESSAGE("data left in packet to small to contain the basp header");
if (count < udp_basp_header_len) {
CAF_MESSAGE("not enought bytes for basp header");
CAF_LOG_DEBUG("buffer contains " << count << " bytes of expected "
<< udp_basp_header_len);
return sec::unexpected_message; return sec::unexpected_message;
} }
binary_deserializer bd{&parent->backend(), bytes, count}; new_basp_message msg;
binary_deserializer bd(&parent->backend(), bytes, count);
bd(msg.header); bd(msg.header);
CAF_MESSAGE("read basp header " << CAF_ARG(msg.header)); msg.payload = bytes + header_size;
size_t payload_len = static_cast<size_t>(msg.header.payload_len); msg.payload_size = count - header_size;
// Read payload.
auto remaining = count - udp_basp_header_len;
// TODO: Could be `!=` ?
if (remaining < payload_len) {
CAF_LOG_DEBUG("only " << remaining << " bytes remaining of expected "
<< msg.header.payload_len);
return sec::unexpected_message;
}
msg.payload = bytes + udp_basp_header_len;
msg.payload_len = msg.header.payload_len;
parent->handle(msg); parent->handle(msg);
return none; return none;
} }
...@@ -1341,34 +145,27 @@ struct udp_basp { ...@@ -1341,34 +145,27 @@ struct udp_basp {
return none; return none;
} }
size_t write_header(byte_buffer& buf, header_writer* hw) { void write_header(byte_buffer& buf, header_writer* hw) {
CAF_ASSERT(hw != nullptr); CAF_ASSERT(hw != nullptr);
(*hw)(buf); (*hw)(buf);
return header_size; return;
}
void prepare_for_sending(byte_buffer& buf, size_t hstart, size_t plen) {
stream_serializer<charbuf> out{&parent->backend(), buf.data() + hstart,
sizeof(uint32_t)};
auto len = static_cast<uint32_t>(plen);
out(len);
} }
}; };
template <class Next> template <class Next>
struct udp_ordering { struct ordering {
static constexpr size_t header_size = udp_ordering_header_len; static constexpr size_t header_size = sizeof(ordering_header);
static constexpr size_t offset = Next::offset + header_size;
using message_type = typename Next::message_type; using message_type = typename Next::message_type;
using result_type = typename Next::result_type; using result_type = typename Next::result_type;
sequence_type seq_read = 0; uint32_t seq_read = 0;
sequence_type seq_write = 0; uint32_t seq_write = 0;
size_t max_pending_messages = 10; network::newb<message_type>* parent;
std::chrono::milliseconds pending_to = std::chrono::milliseconds(100);
newb<message_type>* parent;
Next next; Next next;
std::unordered_map<sequence_type, std::vector<char>> pending; std::unordered_map<uint32_t, std::vector<char>> pending;
udp_ordering(newb<message_type>* parent) : parent(parent), next(parent) { ordering(network::newb<message_type>* parent)
: parent(parent), next(parent) {
// nop // nop
} }
...@@ -1379,51 +176,46 @@ struct udp_ordering { ...@@ -1379,51 +176,46 @@ struct udp_ordering {
auto& buf = pending[seq_read]; auto& buf = pending[seq_read];
auto res = next.read(buf.data(), buf.size()); auto res = next.read(buf.data(), buf.size());
pending.erase(seq_read); pending.erase(seq_read);
// TODO: Cancel timeout.
if (res) if (res)
return res; return res;
} }
return none; return none;
} }
error add_pending(char* bytes, size_t count, sequence_type seq) {
pending[seq] = std::vector<char>(bytes + header_size, bytes + count);
parent->set_timeout(pending_to, ordering_atom::value, seq);
if (pending.size() > max_pending_messages) {
seq_read = pending.begin()->first;
return deliver_pending();
}
return none;
}
error read(char* bytes, size_t count) { error read(char* bytes, size_t count) {
CAF_MESSAGE("ordering: read (" << count << ")");
if (count < header_size) if (count < header_size)
return sec::unexpected_message; return sec::unexpected_message;
udp_ordering_header hdr; ordering_header hdr;
binary_deserializer bd(&parent->backend(), bytes, count); binary_deserializer bd(&parent->backend(), bytes, count);
bd(hdr); bd(hdr);
CAF_MESSAGE("read udp ordering header: " << to_string(hdr));
// TODO: Use the comparison function from BASP instance. // TODO: Use the comparison function from BASP instance.
if (hdr.seq == seq_read) { if (hdr.seq_nr == seq_read) {
seq_read += 1; seq_read += 1;
auto res = next.read(bytes + header_size, count - header_size); auto res = next.read(bytes + header_size, count - header_size);
if (res) if (res)
return res; return res;
return deliver_pending(); return deliver_pending();
} else if (hdr.seq > seq_read) { } else if (hdr.seq_nr > seq_read) {
add_pending(bytes, count, hdr.seq); pending[hdr.seq_nr] = std::vector<char>(bytes + header_size,
bytes + count);
parent->set_timeout(std::chrono::seconds(2), ordering_atom::value,
hdr.seq_nr);
return none; return none;
} }
// Is late, drop it. TODO: Should this return an error?
return none; return none;
} }
error timeout(atom_value atm, uint32_t id) { error timeout(atom_value atm, uint32_t id) {
if (atm == ordering_atom::value) { if (atm == ordering_atom::value) {
error err = none; error err;
sequence_type seq = static_cast<sequence_type>(id); if (pending.count(id) > 0) {
if (pending.count(seq) > 0) { auto& buf = pending[id];
seq_read = static_cast<sequence_type>(seq); err = next.read(buf.data(), buf.size());
err = deliver_pending(); seq_read = id + 1;
if (!err)
err = deliver_pending();
} }
return err; return err;
} }
...@@ -1432,274 +224,14 @@ struct udp_ordering { ...@@ -1432,274 +224,14 @@ struct udp_ordering {
void write_header(byte_buffer& buf, header_writer* hw) { void write_header(byte_buffer& buf, header_writer* hw) {
binary_serializer bs(&parent->backend(), buf); binary_serializer bs(&parent->backend(), buf);
bs(udp_ordering_header{seq_write}); bs(ordering_header{seq_write});
seq_write += 1; seq_write += 1;
next.write_header(buf, hw); next.write_header(buf, hw);
return; return;
} }
void prepare_for_sending(byte_buffer& buf, size_t hstart, size_t plen) {
next.prepare_for_sending(buf, hstart, plen);
}
};
struct udp_transport_policy : public transport_policy {
udp_transport_policy()
: maximum{std::numeric_limits<uint16_t>::max()},
first_message{true},
writing{false},
written{0},
offline_sum{0} {
// nop
}
error read_some(network::event_handler* parent) override {
CAF_LOG_TRACE(CAF_ARG(parent->fd()));
memset(sender.address(), 0, sizeof(sockaddr_storage));
network::socket_size_type len = sizeof(sockaddr_storage);
auto buf_ptr = static_cast<network::socket_recv_ptr>(receive_buffer.data());
auto buf_len = receive_buffer.size();
auto sres = ::recvfrom(parent->fd(), buf_ptr, buf_len,
0, sender.address(), &len);
if (network::is_error(sres, true)) {
CAF_LOG_ERROR("recvfrom returned" << CAF_ARG(sres));
return sec::runtime_error;
} else if (network::would_block_or_temporarily_unavailable(network::last_socket_error())) {
CAF_LOG_DEBUG("try later");
return sec::end_of_stream;
}
if (sres == 0)
CAF_LOG_INFO("Received empty datagram");
else if (sres > static_cast<network::signed_size_type>(buf_len))
CAF_LOG_WARNING("recvfrom cut of message, only received "
<< CAF_ARG(buf_len) << " of " << CAF_ARG(sres) << " bytes");
received_bytes = (sres > 0) ? static_cast<size_t>(sres) : 0;
*sender.length() = static_cast<size_t>(len);
if (first_message) {
endpoint = sender;
first_message = false;
}
return none;
}
bool should_deliver() override {
CAF_LOG_TRACE("");
return received_bytes != 0 && sender == endpoint;
}
void prepare_next_read(network::event_handler*) override {
received_bytes = 0;
receive_buffer.resize(maximum);
}
void configure_read(receive_policy::config) override {
// nop
}
error write_some(network::event_handler* parent) override {
using namespace caf::io::network;
CAF_LOG_TRACE(CAF_ARG(parent->fd()) << CAF_ARG(send_buffer.size()));
socket_size_type len = static_cast<socket_size_type>(*endpoint.clength());
auto buf_ptr = reinterpret_cast<socket_send_ptr>(send_buffer.data() + written);
auto buf_len = send_sizes.front();
auto sres = ::sendto(parent->fd(), buf_ptr, buf_len,
0, endpoint.caddress(), len);
if (is_error(sres, true)) {
CAF_LOG_ERROR("sendto returned" << CAF_ARG(sres));
return sec::runtime_error;
}
send_sizes.pop_front();
written += (sres > 0) ? static_cast<size_t>(sres) : 0;
auto remaining = send_buffer.size() - written;
if (remaining == 0)
prepare_next_write(parent);
return none;
}
void prepare_next_write(network::event_handler* parent) override {
written = 0;
send_buffer.clear();
send_sizes.clear();
if (offline_buffer.empty()) {
writing = false;
parent->backend().del(network::operation::write, parent->fd(), parent);
} else {
// Add size of last chunk.
offline_sizes.push_back(offline_buffer.size() - offline_sum);
// Switch buffers.
send_buffer.swap(offline_buffer);
send_sizes.swap(offline_sizes);
// Reset sum.
offline_sum = 0;
}
}
byte_buffer& wr_buf() {
if (!offline_buffer.empty()) {
auto chunk_size = offline_buffer.size() - offline_sum;
offline_sizes.push_back(chunk_size);
offline_sum += chunk_size;
}
return offline_buffer;
}
void flush(network::event_handler* parent) override {
CAF_ASSERT(parent != nullptr);
CAF_LOG_TRACE(CAF_ARG(offline_buffer.size()));
if (!offline_buffer.empty() && !writing) {
parent->backend().add(network::operation::write, parent->fd(), parent);
writing = true;
prepare_next_write(parent);
}
}
// State for reading.
size_t maximum;
bool first_message;
// State for writing.
bool writing;
size_t written;
size_t offline_sum;
std::deque<size_t> send_sizes;
std::deque<size_t> offline_sizes;
// UDP endpoints.
network::ip_endpoint endpoint;
network::ip_endpoint sender;
};
template <class T>
struct udp_protocol_policy : protocol_policy<typename T::message_type> {
T impl;
udp_protocol_policy(newb<typename T::message_type>* parent) : impl(parent) {
// nop
}
error read(char* bytes, size_t count) override {
return impl.read(bytes, count);
}
error timeout(atom_value atm, uint32_t id) override {
return impl.timeout(atm, id);
}
void write_header(byte_buffer& buf, header_writer* hw) override {
impl.write_header(buf, hw);
}
void prepare_for_sending(byte_buffer& buf, size_t hstart,
size_t plen) override {
impl.prepare_for_sending(buf, hstart, plen);
}
}; };
struct udp_basp_newb : newb<new_udp_basp_message> { struct dummy_basp_newb : network::newb<new_basp_message> {
udp_basp_newb(caf::actor_config& cfg, default_multiplexer& dm,
native_socket sockfd)
: newb<new_udp_basp_message>(cfg, dm, sockfd) {
// nop
}
void handle(new_udp_basp_message& msg) override {
CAF_PUSH_AID_FROM_PTR(this);
CAF_LOG_TRACE("");
std::string res;
binary_deserializer bd(&backend(), msg.payload, msg.payload_len);
bd(res);
send(responder, res);
}
behavior make_behavior() override {
set_default_handler(print_and_drop);
return {
// Must be implemented at the moment, will be cought by the broker in a
// later implementation.
[=](atom_value atm, uint32_t id) {
protocol->timeout(atm, id);
},
[=](send_atom, actor_id sender, actor_id receiver, std::string payload) {
auto hw = caf::make_callback([&](byte_buffer& buf) -> error {
binary_serializer bs(&backend(), buf);
bs(udp_basp_header{0, sender, receiver});
return none;
});
auto whdl = wr_buf(&hw);
CAF_CHECK(whdl.buf != nullptr);
CAF_CHECK(whdl.protocol != nullptr);
binary_serializer bs(&backend(), *whdl.buf);
bs(payload);
},
[=](quit_atom) {
// Remove from multiplexer loop.
stop();
// Quit actor.
quit();
}
};
}
actor responder;
};
struct udp_accept_policy : public accept_policy<new_udp_basp_message> {
virtual std::pair<native_socket, transport_policy_ptr>
accept(network::event_handler*) {
auto esock = new_udp_endpoint_impl(0, nullptr, false);
if (!esock)
return {invalid_native_socket, nullptr};
auto result = std::move(*esock);
transport_policy_ptr ptr{new udp_transport_policy};
return {result, std::move(ptr)};
}
virtual void init(newb<new_udp_basp_message>& n) {
n.start();
}
};
template <class ProtocolPolicy>
struct udp_basp_acceptor
: public newb_acceptor<typename ProtocolPolicy::message_type> {
using super = newb_acceptor<typename ProtocolPolicy::message_type>;
udp_basp_acceptor(default_multiplexer& dm, native_socket sockfd)
: super(dm, sockfd) {
// nop
}
static expected<native_socket> create_socket(uint16_t port, const char* host,
bool reuse = false) {
return new_udp_endpoint_impl(port, host, reuse);
}
expected<actor> create_newb(native_socket sockfd,
transport_policy_ptr pol) override {
CAF_MESSAGE("creating new basp udp newb");
auto n = make_newb<udp_basp_newb>(this->backend().system(), sockfd);
auto ptr = caf::actor_cast<caf::abstract_actor*>(n);
if (ptr == nullptr)
return sec::runtime_error;
auto& ref = dynamic_cast<udp_basp_newb&>(*ptr);
ref.transport = std::move(pol);
ref.protocol.reset(new ProtocolPolicy(&ref));
ref.responder = responder;
// Read first message from this socket.
ref.transport->prepare_next_read(this);
ref.transport->read_some(this, *ref.protocol.get());
// Subsequent messages will be read from `sockfd`.
anon_send(responder, n);
return n;
}
actor responder;
};
// -- test classes -------------------------------------------------------------
struct dummy_basp_newb : newb<new_basp_message> {
std::vector<std::pair<atom_value, uint32_t>> timeout_messages; std::vector<std::pair<atom_value, uint32_t>> timeout_messages;
std::vector<std::pair<new_basp_message, std::vector<char>>> messages; std::vector<std::pair<new_basp_message, std::vector<char>>> messages;
std::deque<std::pair<basp_header, int>> expected; std::deque<std::pair<basp_header, int>> expected;
...@@ -1770,24 +302,24 @@ struct dummy_basp_newb : newb<new_basp_message> { ...@@ -1770,24 +302,24 @@ struct dummy_basp_newb : newb<new_basp_message> {
} }
}; };
struct accept_policy_impl : accept_policy<new_basp_message> { struct accept_policy_impl : public network::accept_policy<new_basp_message> {
std::pair<native_socket, transport_policy_ptr> std::pair<native_socket, network::transport_policy_ptr>
accept(network::event_handler*) override { accept(network::event_handler*) override {
// TODO: For UDP read the message into a buffer. Create a new socket. // TODO: For UDP read the message into a buffer. Create a new socket.
// Move the buffer into the transport policy as the new receive buffer. // Move the buffer into the transport policy as the new receive buffer.
transport_policy_ptr ptr{new transport_policy}; network::transport_policy_ptr ptr{new network::transport_policy};
return {invalid_native_socket, std::move(ptr)}; return {invalid_native_socket, std::move(ptr)};
} }
void init(newb<new_basp_message>& n) override { void init(network::newb<new_basp_message>& n) override {
n.handle_event(network::operation::read); n.handle_event(network::operation::read);
} }
}; };
template <class ProtocolPolicy> template <class ProtocolPolicy>
struct dummy_basp_newb_acceptor struct dummy_basp_newb_acceptor
: newb_acceptor<typename ProtocolPolicy::message_type> { : public network::newb_acceptor<typename ProtocolPolicy::message_type> {
using super = newb_acceptor<typename ProtocolPolicy::message_type>; using super = network::newb_acceptor<typename ProtocolPolicy::message_type>;
using message_tuple_t = std::tuple<ordering_header, basp_header, int>; using message_tuple_t = std::tuple<ordering_header, basp_header, int>;
dummy_basp_newb_acceptor(default_multiplexer& dm, native_socket sockfd) dummy_basp_newb_acceptor(default_multiplexer& dm, native_socket sockfd)
...@@ -1796,9 +328,10 @@ struct dummy_basp_newb_acceptor ...@@ -1796,9 +328,10 @@ struct dummy_basp_newb_acceptor
} }
expected<actor> create_newb(native_socket sockfd, expected<actor> create_newb(native_socket sockfd,
transport_policy_ptr pol) override { network::transport_policy_ptr pol) override {
spawned.emplace_back(make_newb<dummy_basp_newb>(this->backend().system(), spawned.emplace_back(
sockfd)); network::make_newb<dummy_basp_newb>(this->backend().system(), sockfd)
);
auto ptr = caf::actor_cast<caf::abstract_actor*>(spawned.back()); auto ptr = caf::actor_cast<caf::abstract_actor*>(spawned.back());
if (ptr == nullptr) if (ptr == nullptr)
return sec::runtime_error; return sec::runtime_error;
...@@ -1827,70 +360,11 @@ public: ...@@ -1827,70 +360,11 @@ public:
set("logger.inline-output", true); set("logger.inline-output", true);
set("middleman.manual-multiplexing", true); set("middleman.manual-multiplexing", true);
set("middleman.attach-utility-actors", true); set("middleman.attach-utility-actors", true);
load<io::middleman>(); load<middleman>();
}
};
class io_config : public actor_system_config {
public:
io_config() {
load<io::middleman>();
} }
}; };
struct tcp_test_broker_state {
tcp_basp_header hdr;
bool expecting_header = true;
};
struct udp_test_broker_state {
datagram_handle hdl;
};
struct fixture { struct fixture {
// TCP testing.
using tcp_protocol_policy_t = tcp_protocol_policy<tcp_basp>;
using tcp_accept_policy_t = tcp_accept_policy;
using tcp_newb_acceptor_t = tcp_basp_acceptor<tcp_protocol_policy_t>;
using tcp_transport_policy_t = tcp_transport_policy;
// UDP testing.
using udp_protocol_policy_t = udp_protocol_policy<udp_ordering<udp_basp>>;
using udp_accept_policy_t = udp_accept_policy;
using udp_newb_acceptor_t = udp_basp_acceptor<udp_protocol_policy_t>;
using udp_transport_policy_t = udp_transport_policy;
io_config cfg;
actor_system sys;
default_multiplexer& mpx;
// scheduler::test_coordinator& sched;
const char* host = "localhost";
const uint16_t port = 12345;
// -- constructor ------------------------------------------------------------
fixture()
: sys(cfg.parse(test::engine::argc(), test::engine::argv())),
mpx(dynamic_cast<default_multiplexer&>(sys.middleman().backend())) {
// sched(dynamic_cast<caf::scheduler::test_coordinator&>(sys.scheduler())) {
// nop
}
// -- supporting -------------------------------------------------------------
// void exec_all() {
// while (mpx.try_run_once()) {
// // rince and repeat
// }
// }
// void run_all() {
// sched.run_dispatch_loop();
// }
};
struct dm_fixture {
using policy_t = protocol_policy_impl<ordering<basp_policy>>; using policy_t = protocol_policy_impl<ordering<basp_policy>>;
using acceptor_t = dummy_basp_newb_acceptor<policy_t>; using acceptor_t = dummy_basp_newb_acceptor<policy_t>;
config cfg; config cfg;
...@@ -1900,15 +374,17 @@ struct dm_fixture { ...@@ -1900,15 +374,17 @@ struct dm_fixture {
actor self; actor self;
std::unique_ptr<acceptor_t> na; std::unique_ptr<acceptor_t> na;
dm_fixture() fixture()
: sys(cfg.parse(test::engine::argc(), test::engine::argv())), : sys(cfg.parse(test::engine::argc(), test::engine::argv())),
mpx(dynamic_cast<default_multiplexer&>(sys.middleman().backend())), mpx(dynamic_cast<default_multiplexer&>(sys.middleman().backend())),
sched(dynamic_cast<caf::scheduler::test_coordinator&>(sys.scheduler())) { sched(dynamic_cast<caf::scheduler::test_coordinator&>(sys.scheduler())) {
// Create newb. // Create newb.
self = make_newb<dummy_basp_newb>(sys, network::invalid_native_socket); self = network::make_newb<dummy_basp_newb>(sys,
auto& ref = deref<newb<new_basp_message>>(self); network::invalid_native_socket);
ref.transport.reset(new transport_policy); auto& ref = deref<network::newb<new_basp_message>>(self);
ref.protocol.reset(new protocol_policy_impl<ordering<basp_policy>>(&ref)); network::newb<new_basp_message>* self_ptr = &ref;
ref.transport.reset(new network::transport_policy);
ref.protocol.reset(new protocol_policy_impl<ordering<basp_policy>>(self_ptr));
// Create acceptor. // Create acceptor.
auto& mpx = dynamic_cast<default_multiplexer&>(sys.middleman().backend()); auto& mpx = dynamic_cast<default_multiplexer&>(sys.middleman().backend());
std::unique_ptr<acceptor_t> ptr{new acceptor_t(mpx, invalid_native_socket)}; std::unique_ptr<acceptor_t> ptr{new acceptor_t(mpx, invalid_native_socket)};
...@@ -1930,53 +406,13 @@ struct dm_fixture { ...@@ -1930,53 +406,13 @@ struct dm_fixture {
CAF_REQUIRE(ptr != nullptr); CAF_REQUIRE(ptr != nullptr);
return dynamic_cast<T&>(*ptr); return dynamic_cast<T&>(*ptr);
} }
// -- serialization ----------------------------------------------------------
template <class T>
void to_buffer(ordering_header& hdr, T& x) {
binary_serializer bs(sys, x);
bs(hdr);
}
template <class T>
void to_buffer(basp_header& hdr, T& x) {
binary_serializer bs(sys, x);
bs(hdr);
}
template <class T, class U>
void to_buffer(U value, T& x) {
binary_serializer bs(sys, x);
bs(value);
}
/*
template <class T>
void from_buffer(T& x, size_t offset, ordering_header& hdr) {
binary_deserializer bd(sys, x.data() + offset, sizeof(ordering_header));
bd(hdr);
}
template <class T>
void from_buffer(T& x, size_t offset, basp_header& hdr) {
binary_deserializer bd(sys, x.data() + offset, sizeof(basp_header));
bd(hdr);
}
*/
template <class T>
void from_buffer(char* x, T& value) {
binary_deserializer bd(sys, x, sizeof(T));
bd(value);
}
}; };
} // namespace <anonymous> } // namespace <anonymous>
CAF_TEST_FIXTURE_SCOPE(test_newb_creation, dm_fixture) CAF_TEST_FIXTURE_SCOPE(newb_basics, fixture)
CAF_TEST(ordering and basp read event) { CAF_TEST(read event) {
exec_all(); exec_all();
CAF_MESSAGE("create some values for our buffer"); CAF_MESSAGE("create some values for our buffer");
ordering_header ohdr{0}; ordering_header ohdr{0};
...@@ -2003,11 +439,12 @@ CAF_TEST(ordering and basp read event) { ...@@ -2003,11 +439,12 @@ CAF_TEST(ordering and basp read event) {
CAF_CHECK_EQUAL(msg.header.from, bhdr.from); CAF_CHECK_EQUAL(msg.header.from, bhdr.from);
CAF_CHECK_EQUAL(msg.header.to, bhdr.to); CAF_CHECK_EQUAL(msg.header.to, bhdr.to);
int return_payload = 0; int return_payload = 0;
from_buffer(msg.payload, return_payload); binary_deserializer bd(sys, msg.payload, msg.payload_size);
bd(return_payload);
CAF_CHECK_EQUAL(return_payload, payload); CAF_CHECK_EQUAL(return_payload, payload);
} }
CAF_TEST(ordering and basp message passing) { CAF_TEST(message passing) {
exec_all(); exec_all();
ordering_header ohdr{0}; ordering_header ohdr{0};
basp_header bhdr{13, 42}; basp_header bhdr{13, 42};
...@@ -2024,11 +461,12 @@ CAF_TEST(ordering and basp message passing) { ...@@ -2024,11 +461,12 @@ CAF_TEST(ordering and basp message passing) {
CAF_CHECK_EQUAL(msg.header.from, bhdr.from); CAF_CHECK_EQUAL(msg.header.from, bhdr.from);
CAF_CHECK_EQUAL(msg.header.to, bhdr.to); CAF_CHECK_EQUAL(msg.header.to, bhdr.to);
int return_payload = 0; int return_payload = 0;
from_buffer(msg.payload, return_payload); binary_deserializer bd(sys, msg.payload, msg.payload_size);
bd(return_payload);
CAF_CHECK_EQUAL(return_payload, payload); CAF_CHECK_EQUAL(return_payload, payload);
} }
CAF_TEST(ordering and basp read event with timeout) { CAF_TEST(timeouts) {
// Should be an unexpected sequence number and lead to an error. Since // Should be an unexpected sequence number and lead to an error. Since
// we start with 0, the 1 below should be out of order. // we start with 0, the 1 below should be out of order.
ordering_header ohdr{1}; ordering_header ohdr{1};
...@@ -2051,7 +489,7 @@ CAF_TEST(ordering and basp read event with timeout) { ...@@ -2051,7 +489,7 @@ CAF_TEST(ordering and basp read event with timeout) {
// Message handler will check if the expected message was received. // Message handler will check if the expected message was received.
} }
CAF_TEST(ordering and basp multiple messages) { CAF_TEST(message ordering) {
CAF_MESSAGE("create data for two messges"); CAF_MESSAGE("create data for two messges");
// Message one. // Message one.
ordering_header ohdr_first{0}; ordering_header ohdr_first{0};
...@@ -2089,7 +527,7 @@ CAF_TEST(ordering and basp multiple messages) { ...@@ -2089,7 +527,7 @@ CAF_TEST(ordering and basp multiple messages) {
dummy.read_event(); dummy.read_event();
} }
CAF_TEST(ordering and basp write buf) { CAF_TEST(write buf) {
exec_all(); exec_all();
basp_header bhdr{13, 42}; basp_header bhdr{13, 42};
int payload = 1337; int payload = 1337;
...@@ -2103,7 +541,7 @@ CAF_TEST(ordering and basp write buf) { ...@@ -2103,7 +541,7 @@ CAF_TEST(ordering and basp write buf) {
// Message handler will check if the expected message was received. // Message handler will check if the expected message was received.
} }
CAF_TEST(ordering and basp acceptor) { CAF_TEST(newb acceptor) {
CAF_MESSAGE("trigger read event on acceptor"); CAF_MESSAGE("trigger read event on acceptor");
// This will write a message into the receive buffer and trigger // This will write a message into the receive buffer and trigger
// a read event on the newly created newb. // a read event on the newly created newb.
...@@ -2113,231 +551,3 @@ CAF_TEST(ordering and basp acceptor) { ...@@ -2113,231 +551,3 @@ CAF_TEST(ordering and basp acceptor) {
} }
CAF_TEST_FIXTURE_SCOPE_END() CAF_TEST_FIXTURE_SCOPE_END()
CAF_TEST_FIXTURE_SCOPE(tcp_newbs, fixture)
CAF_TEST(tcp basp newb) {
scoped_actor main_actor{sys};
actor newb_actor;
auto testing = [&](stateful_broker<tcp_test_broker_state>* self,
connection_handle hdl, actor) -> behavior {
CAF_CHECK(hdl != invalid_connection_handle);
self->configure_read(hdl, receive_policy::exactly(tcp_basp_header_len));
self->state.expecting_header = true;
return {
[=](send_atom, std::string str) {
CAF_MESSAGE("sending '" << str << "'");
byte_buffer buf;
binary_serializer bs(sys, buf);
tcp_basp_header hdr{0, 1, 2};
bs(hdr);
auto header_len = buf.size();
CAF_REQUIRE(header_len == tcp_basp_header_len);
bs(str);
hdr.payload_len = static_cast<uint32_t>(buf.size() - header_len);
stream_serializer<charbuf> out{sys, buf.data(), sizeof(hdr.payload_len)};
out(hdr.payload_len);
CAF_MESSAGE("header len: " << header_len
<< ", packet_len: " << buf.size()
<< ", header: " << to_string(hdr));
self->write(hdl, buf.size(), buf.data());
self->flush(hdl);
},
[=](quit_atom) {
CAF_MESSAGE("test broker shutting down");
self->quit();
},
[=](new_data_msg& msg) {
auto& s = self->state;
size_t next_len = tcp_basp_header_len;
binary_deserializer bd(self->system(), msg.buf);
if (s.expecting_header) {
bd(s.hdr);
next_len = s.hdr.payload_len;
s.expecting_header = false;
} else {
std::string str;
bd(str);
CAF_MESSAGE("received '" << str << "'");
std::reverse(std::begin(str), std::end(str));
byte_buffer buf;
binary_serializer bs(sys, buf);
tcp_basp_header hdr{0, 1, 2};
bs(hdr);
auto header_len = buf.size();
CAF_REQUIRE(header_len == tcp_basp_header_len);
bs(str);
hdr.payload_len = static_cast<uint32_t>(buf.size() - header_len);
stream_serializer<charbuf> out{sys, buf.data(), sizeof(hdr.payload_len)};
out(hdr.payload_len);
CAF_MESSAGE("header len: " << header_len
<< ", packet_len: " << buf.size()
<< ", header: " << to_string(hdr));
self->write(hdl, buf.size(), buf.data());
self->flush(hdl);
// self->send(m, quit_atom::value);
}
self->configure_read(msg.handle, receive_policy::exactly(next_len));
}
};
};
auto helper_actor = sys.spawn([&](event_based_actor* self, actor m) -> behavior {
return {
[=](const std::string& str) {
CAF_MESSAGE("received '" << str << "'");
self->send(m, quit_atom::value);
},
[=](actor a) {
CAF_MESSAGE("got new newb handle");
self->send(m, a);
},
[=](quit_atom) {
CAF_MESSAGE("helper shutting down");
self->quit();
}
};
}, main_actor);
CAF_MESSAGE("creating new acceptor");
auto newb_acceptor_ptr
= make_newb_acceptor<tcp_newb_acceptor_t, tcp_accept_policy_t>(sys, port);
dynamic_cast<tcp_newb_acceptor_t*>(newb_acceptor_ptr.get())->responder
= helper_actor;
CAF_MESSAGE("connecting from 'old-style' broker");
auto exp = sys.middleman().spawn_client(testing, host, port, main_actor);
CAF_CHECK(exp);
auto test_broker = std::move(*exp);
main_actor->receive(
[&](actor a) {
newb_actor = a;
}
);
CAF_MESSAGE("sending message to newb");
main_actor->send(test_broker, send_atom::value, "hello world");
std::this_thread::sleep_for(std::chrono::seconds(1));
main_actor->receive(
[](quit_atom) {
CAF_MESSAGE("check");
}
);
CAF_MESSAGE("sending message from newb");
main_actor->send(newb_actor, send_atom::value, actor_id{3}, actor_id{4},
"dlrow olleh");
main_actor->receive(
[](quit_atom) {
CAF_MESSAGE("check");
}
);
CAF_MESSAGE("shutting everything down");
newb_acceptor_ptr->stop();
anon_send(newb_actor, quit_atom::value);
anon_send(helper_actor, quit_atom::value);
anon_send(test_broker, quit_atom::value);
sys.await_all_actors_done();
CAF_MESSAGE("done");
}
CAF_TEST(udp basp newb) {
scoped_actor main_actor{sys};
actor newb_actor;
auto testing = [&](stateful_broker<udp_test_broker_state>* self,
std::string host, uint16_t port, actor m) -> behavior {
auto ehdl = self->add_udp_datagram_servant(host, port);
CAF_REQUIRE(ehdl);
self->state.hdl = std::move(*ehdl);
return {
[=](send_atom, std::string str) {
CAF_MESSAGE("sending '" << str << "'");
byte_buffer buf;
udp_ordering_header ohdr{0};
udp_basp_header bhdr{0, 1, 2};
binary_serializer bs(sys, buf);
bs(ohdr);
auto ordering_header_len = buf.size();
CAF_REQUIRE(ordering_header_len == udp_ordering_header_len);
bs(bhdr);
auto header_len = buf.size();
CAF_REQUIRE(header_len == udp_ordering_header_len + udp_basp_header_len);
bs(str);
bhdr.payload_len = static_cast<uint32_t>(buf.size() - header_len);
stream_serializer<charbuf> out{sys, buf.data() + ordering_header_len,
sizeof(bhdr.payload_len)};
out(bhdr.payload_len);
CAF_MESSAGE("header len: " << header_len
<< ", packet_len: " << buf.size()
<< ", ordering header: " << to_string(ohdr)
<< ", basp header: " << to_string(bhdr));
self->enqueue_datagram(self->state.hdl, std::move(buf));
self->flush(self->state.hdl);
},
[=](quit_atom) {
CAF_MESSAGE("test broker shutting down");
self->quit();
},
[=](new_datagram_msg& msg) {
binary_deserializer bd(self->system(), msg.buf);
udp_ordering_header ohdr;
udp_basp_header bhdr;
std::string str;
bd(ohdr);
bd(bhdr);
bd(str);
CAF_MESSAGE("received '" << str << "'");
self->send(m, quit_atom::value);
}
};
};
auto helper_actor = sys.spawn([&](event_based_actor* self, actor m) -> behavior {
return {
[=](const std::string& str) {
CAF_MESSAGE("received '" << str << "'");
self->send(m, quit_atom::value);
},
[=](actor a) {
CAF_MESSAGE("got new newb handle");
self->send(m, a);
},
[=](quit_atom) {
CAF_MESSAGE("helper shutting down");
self->quit();
}
};
}, main_actor);
CAF_MESSAGE("creating new acceptor");
auto newb_acceptor_ptr
= make_newb_acceptor<udp_newb_acceptor_t, udp_accept_policy_t>(sys, port);
dynamic_cast<udp_newb_acceptor_t*>(newb_acceptor_ptr.get())->responder
= helper_actor;
CAF_MESSAGE("contacting from 'old-style' broker");
auto test_broker = sys.middleman().spawn_broker(testing, host, port, main_actor);
main_actor->send(test_broker, send_atom::value, "hello world");
std::this_thread::sleep_for(std::chrono::seconds(1));
main_actor->receive(
[&](actor a) {
newb_actor = a;
}
);
CAF_MESSAGE("new newb was created");
main_actor->receive(
[](quit_atom) {
CAF_MESSAGE("check");
}
);
CAF_MESSAGE("sending message from newb");
main_actor->send(newb_actor, send_atom::value, actor_id{3}, actor_id{4},
"dlrow olleh");
main_actor->receive(
[](quit_atom) {
CAF_MESSAGE("check");
}
);
CAF_MESSAGE("shutting everything down");
newb_acceptor_ptr->stop();
anon_send(newb_actor, quit_atom::value);
anon_send(helper_actor, quit_atom::value);
anon_send(test_broker, quit_atom::value);
sys.await_all_actors_done();
CAF_MESSAGE("done");
}
CAF_TEST_FIXTURE_SCOPE_END();
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