Commit e747d972 authored by Joseph Noir's avatar Joseph Noir

Implement network code for newb acceptor

parent 481bdfa6
...@@ -18,6 +18,8 @@ ...@@ -18,6 +18,8 @@
#define CAF_SUITE protocol_policy #define CAF_SUITE protocol_policy
#include "caf/config.hpp"
#include <cstdint> #include <cstdint>
#include <cstring> #include <cstring>
#include <tuple> #include <tuple>
...@@ -26,21 +28,66 @@ ...@@ -26,21 +28,66 @@
#include "caf/callback.hpp" #include "caf/callback.hpp"
#include "caf/config.hpp" #include "caf/config.hpp"
#include "caf/detail/call_cfun.hpp"
#include "caf/detail/enum_to_string.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/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/event_handler.hpp"
#include "caf/io/network/interfaces.hpp"
#include "caf/io/network/native_socket.hpp" #include "caf/io/network/native_socket.hpp"
#include "caf/mixin/behavior_changer.hpp" #include "caf/mixin/behavior_changer.hpp"
#include "caf/mixin/requester.hpp" #include "caf/mixin/requester.hpp"
#include "caf/mixin/sender.hpp" #include "caf/mixin/sender.hpp"
#include "caf/scheduled_actor.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 network::native_socket;
using network::invalid_native_socket;
using network::default_multiplexer; using network::default_multiplexer;
using network::last_socket_error_as_string;
namespace { namespace {
...@@ -56,6 +103,33 @@ struct protocol_policy; ...@@ -56,6 +103,33 @@ struct protocol_policy;
template <class T> template <class T>
struct newb; 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 <anonymous>
namespace caf { namespace caf {
...@@ -75,6 +149,7 @@ namespace { ...@@ -75,6 +149,7 @@ namespace {
using expect_atom = atom_constant<atom("expect")>; using expect_atom = atom_constant<atom("expect")>;
using ordering_atom = atom_constant<atom("ordering")>; using ordering_atom = atom_constant<atom("ordering")>;
using send_atom = atom_constant<atom("send")>; using send_atom = atom_constant<atom("send")>;
using shutdown_atom = atom_constant<atom("shutdown")>;
// -- aliases ------------------------------------------------------------------ // -- aliases ------------------------------------------------------------------
...@@ -129,26 +204,45 @@ struct transport_policy { ...@@ -129,26 +204,45 @@ struct transport_policy {
// nop // nop
} }
virtual error write_some(network::native_socket) { virtual error write_some(network::event_handler*) {
return none; 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
}
byte_buffer& wr_buf() { byte_buffer& wr_buf() {
return send_buffer; return offline_buffer;
} }
template <class T> template <class T>
error read_some(protocol_policy<T>& policy) { error read_some(network::event_handler* parent, protocol_policy<T>& policy) {
auto err = read_some(); auto res = read_some(parent);
if (err) if (!res && should_deliver()) {
return err; res = policy.read(receive_buffer.data(), receive_buffer.size());
return policy.read(receive_buffer.data(), receive_buffer.size()); prepare_next_read(parent);
} }
return res;
virtual error read_some() {
return none;
} }
byte_buffer offline_buffer;
byte_buffer receive_buffer; byte_buffer receive_buffer;
byte_buffer send_buffer; byte_buffer send_buffer;
}; };
...@@ -161,22 +255,11 @@ struct accept_policy { ...@@ -161,22 +255,11 @@ struct accept_policy {
virtual ~accept_policy() { virtual ~accept_policy() {
// nop // nop
} }
virtual std::pair<native_socket, transport_policy_ptr> accept() = 0;
virtual void init(network::event_handler&) = 0;
};
struct accept_policy_impl : accept_policy { virtual std::pair<native_socket, transport_policy_ptr>
std::pair<native_socket, transport_policy_ptr> accept() override { accept(network::event_handler*) = 0;
// 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.
native_socket sock{13337};
transport_policy_ptr ptr{new transport_policy};
return {sock, std::move(ptr)};
}
void init(network::event_handler& eh) override { virtual void init(network::event_handler&) = 0;
eh.handle_event(network::operation::read);
}
}; };
// -- protocol policies -------------------------------------------------------- // -- protocol policies --------------------------------------------------------
...@@ -186,9 +269,13 @@ struct protocol_policy_base { ...@@ -186,9 +269,13 @@ struct protocol_policy_base {
// nop // nop
} }
// virtual void write_header(byte_buffer& buf, size_t offset) = 0; /// TODO: Come up with something better than a write here?
/// Write header into buffer. Use push back to append only.
virtual void write_header(byte_buffer&, header_writer*) = 0;
virtual size_t offset() const noexcept = 0; //virtual size_t offset() const noexcept = 0;
virtual void prepare_for_sending(byte_buffer&, size_t, size_t) = 0;
}; };
template <class T> template <class T>
...@@ -201,10 +288,6 @@ struct protocol_policy : protocol_policy_base { ...@@ -201,10 +288,6 @@ struct protocol_policy : protocol_policy_base {
virtual error read(char* bytes, size_t count) = 0; virtual error read(char* bytes, size_t count) = 0;
virtual error timeout(atom_value, uint32_t) = 0; virtual error timeout(atom_value, uint32_t) = 0;
/// TODO: Come up with something better than a write here?
/// Write header into buffer. Use push back to append only.
virtual size_t write_header(byte_buffer&, header_writer*) = 0;
}; };
template <class T> template <class T>
...@@ -222,17 +305,23 @@ struct protocol_policy_impl : protocol_policy<typename T::message_type> { ...@@ -222,17 +305,23 @@ struct protocol_policy_impl : protocol_policy<typename T::message_type> {
return impl.read(bytes, count); return impl.read(bytes, count);
} }
/*
size_t offset() const noexcept override { size_t offset() const noexcept override {
return T::offset; return T::offset;
} }
*/
error timeout(atom_value atm, uint32_t id) override { error timeout(atom_value atm, uint32_t id) override {
return impl.timeout(atm, id); return impl.timeout(atm, id);
} }
size_t write_header(byte_buffer& buf, header_writer* hw) override { void write_header(byte_buffer& buf, header_writer* hw) override {
return impl.write_header(buf, hw); return impl.write_header(buf, hw);
} }
void prepare_for_sending(byte_buffer&, size_t, size_t) override {
return;
}
}; };
// -- new broker classes ------------------------------------------------------- // -- new broker classes -------------------------------------------------------
...@@ -242,14 +331,14 @@ struct protocol_policy_impl : protocol_policy<typename T::message_type> { ...@@ -242,14 +331,14 @@ struct protocol_policy_impl : protocol_policy<typename T::message_type> {
struct write_handle { struct write_handle {
protocol_policy_base* protocol; protocol_policy_base* protocol;
byte_buffer* buf; byte_buffer* buf;
size_t header_offset; size_t header_start;
size_t header_len;
/*
~write_handle() { ~write_handle() {
// TODO: maybe trigger transport policy for ... what again?
// Can we calculate added bytes for datagram things? // 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);
} }
*/
}; };
template <class Message> template <class Message>
...@@ -368,9 +457,10 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template ...@@ -368,9 +457,10 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template
// TODO: We somehow need to tell the transport policy how much we've // TODO: We somehow need to tell the transport policy how much we've
// written to enable it to split the buffer into datagrams. // written to enable it to split the buffer into datagrams.
auto& buf = transport->wr_buf(); auto& buf = transport->wr_buf();
CAF_REQUIRE(buf.empty()); auto hstart = buf.size();
auto header_offset = protocol->write_header(buf, hw); protocol->write_header(buf, hw);
return {protocol.get(), &buf, header_offset}; auto hlen = buf.size() - hstart;
return {protocol.get(), &buf, hstart, hlen};
} }
// Send // Send
...@@ -379,7 +469,7 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template ...@@ -379,7 +469,7 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template
} }
error read_event() { error read_event() {
return transport->read_some(*protocol); return transport->read_some(this, *protocol);
} }
void write_event() { void write_event() {
...@@ -473,6 +563,7 @@ struct newb_acceptor : public network::event_handler { ...@@ -473,6 +563,7 @@ struct newb_acceptor : public network::event_handler {
void removed_from_loop(network::operation) override { void removed_from_loop(network::operation) override {
CAF_MESSAGE("remove from loop not implemented in newb acceptor"); CAF_MESSAGE("remove from loop not implemented in newb acceptor");
close_read_channel();
// TODO: implement // TODO: implement
} }
...@@ -482,7 +573,7 @@ struct newb_acceptor : public network::event_handler { ...@@ -482,7 +573,7 @@ struct newb_acceptor : public network::event_handler {
CAF_MESSAGE("read event on newb acceptor"); CAF_MESSAGE("read event on newb acceptor");
native_socket sock; native_socket sock;
transport_policy_ptr transport; transport_policy_ptr transport;
std::tie(sock, transport) = acceptor->accept();; std::tie(sock, transport) = acceptor->accept(this);;
auto en = create_newb(sock, std::move(transport)); auto en = create_newb(sock, std::move(transport));
if (!en) if (!en)
return std::move(en.error()); return std::move(en.error());
...@@ -491,35 +582,15 @@ struct newb_acceptor : public network::event_handler { ...@@ -491,35 +582,15 @@ struct newb_acceptor : public network::event_handler {
return none; return none;
} }
void activate() {
event_handler::activate();
}
virtual expected<actor> create_newb(native_socket sock, transport_policy_ptr pol) = 0; virtual expected<actor> create_newb(native_socket sock, transport_policy_ptr pol) = 0;
std::unique_ptr<accept_policy> acceptor; std::unique_ptr<accept_policy> acceptor;
}; };
/*
behavior my_broker(newb<new_data_msg>* self) {
// nop ...
}
template <class AcceptPolicy, class ProtocolPolicy>
struct newb_acceptor {
std::unique_ptr<AcceptPolicy> acceptor;
// read = accept
error read_event() {
auto [sock, trans_pol] = acceptor.accept();
auto worker = sys.middleman.spawn_client<ProtocolPolicy>(
sock, std::move(trans_pol), fork_behavior);
acceptor.init(worker);
}
};
*/
// client: sys.middleman().spawn_client<protocol_policy>(sock,
// std::move(transport_protocol_policy_impl), my_client);
// server: sys.middleman().spawn_server<protocol_policy>(sock,
// std::move(accept_protocol_policy_impl), my_server);
// -- policies ----------------------------------------------------------------- // -- policies -----------------------------------------------------------------
/// @relates protocol_policy /// @relates protocol_policy
...@@ -553,10 +624,10 @@ struct basp_policy { ...@@ -553,10 +624,10 @@ struct basp_policy {
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;
} }
}; };
...@@ -626,16 +697,104 @@ struct ordering { ...@@ -626,16 +697,104 @@ struct ordering {
return next.timeout(atm, id); return next.timeout(atm, id);
} }
size_t 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(ordering_header{seq_write}); bs(ordering_header{seq_write});
seq_write += 1; seq_write += 1;
return header_size + next.write_header(buf, hw); next.write_header(buf, hw);
return;
} }
}; };
// -- test classes ------------------------------------------------------------- // -- 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();
}
// -- create newbs -------------------------------------------------------------
template <class Newb> template <class Newb>
actor make_newb(actor_system& sys, native_socket sockfd) { actor make_newb(actor_system& sys, native_socket sockfd) {
...@@ -648,15 +807,315 @@ actor make_newb(actor_system& sys, native_socket sockfd) { ...@@ -648,15 +807,315 @@ actor make_newb(actor_system& sys, native_socket sockfd) {
// TODO: I feel like this should include the ProtocolPolicy somehow. // TODO: I feel like this should include the ProtocolPolicy somehow.
template <class NewbAcceptor, class AcceptPolicy> template <class NewbAcceptor, class AcceptPolicy>
std::unique_ptr<newb_acceptor> make_newb_acceptor(actor_system& sys, std::unique_ptr<newb_acceptor> make_newb_acceptor(actor_system& sys,
native_socket sockfd) { uint16_t port,
const char* addr = nullptr,
bool reuse_addr = false) {
auto sockfd = 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()); auto& mpx = dynamic_cast<default_multiplexer&>(sys.middleman().backend());
std::unique_ptr<newb_acceptor> ptr{new NewbAcceptor(mpx, sockfd)}; std::unique_ptr<newb_acceptor> ptr{new NewbAcceptor(mpx, *sockfd)};
//std::unique_ptr<network::event_handler> ptr{new NewbAcceptor(mpx, sockfd)};
//static_cast<NewbAcceptor*>(ptr)->acceptor.reset(new AcceptPolicy);
ptr->acceptor.reset(new AcceptPolicy); ptr->acceptor.reset(new AcceptPolicy);
ptr->activate();
return ptr; return ptr;
} }
// -- tcp impls ----------------------------------------------------------------
struct tcp_basp_header {
uint32_t payload_len;
actor_id from;
actor_id to;
};
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_size;
};
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_size);
}
struct tcp_basp {
static constexpr size_t header_size = sizeof(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 < header_size) {
CAF_MESSAGE("data left in packet to small to contain the basp header");
return sec::unexpected_message;
}
binary_deserializer bd(&parent->backend(), bytes, count);
bd(msg.header);
expecting_header = false;
return none;
}
error read_payload(char* bytes, size_t count) {
if (count < msg.header.payload_len) {
return sec::unexpected_message;
}
msg.payload = bytes;
msg.payload_size = msg.header.payload_len;
parent->handle(msg);
expecting_header = true;
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(tcp_basp_header::payload_len)};
auto len = static_cast<uint32_t>(plen);
out(len);
}
};
struct tcp_transport_policy : public transport_policy {
error write_some(network::event_handler* parent) override {
const void* buf = send_buffer.data();
auto len = send_buffer.size();
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;
//result = (sres > 0) ? static_cast<size_t>(sres) : 0;
return none;
}
byte_buffer& wr_buf() {
return send_buffer;
}
error read_some(network::event_handler* parent) override {
size_t len = 1024;
receive_buffer.resize(len);
void* buf = receive_buffer.data();
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;
}
// auto result = (sres > 0) ? static_cast<size_t>(sres) : 0;
return none;
}
bool should_deliver() override {
return collected >= read_threshold;
}
void prepare_next_read(network::event_handler*) override {
collected = 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 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);
}
}
void configure_read(receive_policy::config config) override {
rd_flag = config.first;
maximum = config.second;
}
// 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);
}
/*
size_t offset() const noexcept override {
return T::offset;
}
*/
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&) override {
// TODO:
// - parse the payload of the message
// - send it somewhere
// - drop bytes from the buffer?
}
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, int payload) {
auto hw = caf::make_callback([&](byte_buffer& buf) -> error {
binary_serializer bs(&backend(), buf);
bs(basp_header{sender, receiver});
return none;
});
CAF_MESSAGE("get a write buffer");
auto whdl = wr_buf(&hw);
CAF_CHECK(whdl.buf != nullptr);
CAF_CHECK(whdl.protocol != nullptr);
CAF_MESSAGE("write the payload");
binary_serializer bs(&backend(), *whdl.buf);
bs(payload);
}
};
}
actor responder;
};
struct tcp_accept_policy : public accept_policy {
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(network::event_handler&) {
}
};
template <class ProtocolPolicy>
struct tcp_basp_acceptor : newb_acceptor {
tcp_basp_acceptor(default_multiplexer& dm, native_socket sockfd)
: newb_acceptor(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;
anon_send(responder, n);
return n;
}
actor responder;
};
// -- test classes -------------------------------------------------------------
struct dummy_basp_newb : newb<new_basp_message> { 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;
...@@ -728,6 +1187,20 @@ struct dummy_basp_newb : newb<new_basp_message> { ...@@ -728,6 +1187,20 @@ struct dummy_basp_newb : newb<new_basp_message> {
} }
}; };
struct accept_policy_impl : accept_policy {
std::pair<native_socket, transport_policy_ptr>
accept(network::event_handler*) override {
// 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.
transport_policy_ptr ptr{new transport_policy};
return {invalid_native_socket, std::move(ptr)};
}
void init(network::event_handler& eh) override {
eh.handle_event(network::operation::read);
}
};
template <class ProtocolPolicy> template <class ProtocolPolicy>
struct dummy_basp_newb_acceptor : newb_acceptor { struct dummy_basp_newb_acceptor : newb_acceptor {
using message_tuple_t = std::tuple<ordering_header, basp_header, int>; using message_tuple_t = std::tuple<ordering_header, basp_header, int>;
...@@ -769,6 +1242,52 @@ public: ...@@ -769,6 +1242,52 @@ public:
} }
}; };
class io_config : public actor_system_config {
public:
io_config() {
load<io::middleman>();
}
};
struct fixture {
using protocol_policy_t = tcp_protocol_policy<tcp_basp>;
using accept_policy_t = tcp_accept_policy;
using newb_acceptor_t = tcp_basp_acceptor<protocol_policy_t>;
using transport_policy_t = tcp_transport_policy;
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
}
}
template <class T = caf::scheduled_actor, class Handle = caf::actor>
T& deref(const Handle& hdl) {
auto ptr = caf::actor_cast<caf::abstract_actor*>(hdl);
CAF_REQUIRE(ptr != nullptr);
return dynamic_cast<T&>(*ptr);
}
};
struct dm_fixture { 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>;
...@@ -783,12 +1302,16 @@ struct dm_fixture { ...@@ -783,12 +1302,16 @@ struct dm_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.
self = make_newb<dummy_basp_newb>(sys, network::invalid_native_socket); self = make_newb<dummy_basp_newb>(sys, network::invalid_native_socket);
auto& ref = deref<newb<new_basp_message>>(self); auto& ref = deref<newb<new_basp_message>>(self);
ref.transport.reset(new transport_policy); ref.transport.reset(new transport_policy);
ref.protocol.reset(new protocol_policy_impl<ordering<basp_policy>>(&ref)); ref.protocol.reset(new protocol_policy_impl<ordering<basp_policy>>(&ref));
na = make_newb_acceptor<acceptor_t, accept_policy_impl>(sys, // Create acceptor.
network::invalid_native_socket); auto& mpx = dynamic_cast<default_multiplexer&>(sys.middleman().backend());
std::unique_ptr<newb_acceptor> ptr{new acceptor_t(mpx, invalid_native_socket)};
ptr->acceptor.reset(new accept_policy_impl);
na = std::move(ptr);
} }
// -- supporting ------------------------------------------------------------- // -- supporting -------------------------------------------------------------
...@@ -969,3 +1492,50 @@ CAF_TEST(ordering and basp acceptor) { ...@@ -969,3 +1492,50 @@ 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(accept test) {
actor newb_actor;
auto tester = [](broker* self, connection_handle hdl) -> behavior {
return {
[=](send_atom) {
byte_buffer buf;
self->write(hdl, buf.size(), buf.data());
},
[=](shutdown_atom) {
self->quit();
}
};
};
auto helper = sys.spawn([&](event_based_actor* self) -> behavior {
self->set_default_handler(print_and_drop);
return {
[&](int i) {
CAF_MESSAGE("Got int message " << i);
},
[&](actor a) {
CAF_MESSAGE("Got new newb handle");
newb_actor = a;
}
};
});
exec_all();
CAF_MESSAGE("creating new acceptor");
auto ptr = make_newb_acceptor<newb_acceptor_t, accept_policy_t>(sys, port);
dynamic_cast<newb_acceptor_t*>(ptr.get())->responder = helper;
exec_all();
CAF_MESSAGE("connecting from 'old' broker");
auto eb = sys.middleman().spawn_client(tester, host, port);
CAF_CHECK(eb);
auto e = std::move(*eb);
exec_all();
anon_send_exit(e, exit_reason::user_shutdown);
anon_send_exit(helper, exit_reason::user_shutdown);
ptr->passivate();
exec_all();
// Not a good solution but the newbs currently don't shut down cleanly.
sys.await_actors_before_shutdown(false);
}
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