Commit e40b6668 authored by Joseph Noir's avatar Joseph Noir

Extract protocol policies into headers

parent 94a379c7
...@@ -61,6 +61,7 @@ add(measurements one_raw_tcp) ...@@ -61,6 +61,7 @@ add(measurements one_raw_tcp)
add(measurements one_raw_udp) add(measurements one_raw_udp)
add(measurements one_basp_tcp) add(measurements one_basp_tcp)
add(measurements one_basp_udp) add(measurements one_basp_udp)
add(measurements four_udp)
# basic I/O with brokers # basic I/O with brokers
add(broker simple_broker) add(broker simple_broker)
......
#include "caf/io/network/newb.hpp"
#include "caf/logger.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/detail/call_cfun.hpp"
#include "caf/policy/newb_basp.hpp"
#include "caf/policy/newb_ordering.hpp"
#include "caf/policy/newb_udp.hpp"
using namespace caf;
using caf::io::network::default_multiplexer;
using caf::io::network::invalid_native_socket;
using caf::io::network::make_client_newb;
using caf::io::network::make_server_newb;
using caf::io::network::native_socket;
using caf::policy::accept_udp;
using caf::policy::udp_protocol;
using caf::policy::udp_transport;
namespace {
using ordering_atom = atom_constant<atom("ordering")>;
constexpr size_t chunk_size = 1024; //8192; //128; //1024;
struct dummy_transport : public io::network::transport_policy {
dummy_transport() {
// nop
}
inline error read_some(io::network::event_handler*) override {
return none;
}
inline bool should_deliver() override {
return true;
}
void prepare_next_read(io::network::event_handler*) override {
received_bytes = 0;
receive_buffer.resize(maximum);
}
inline void configure_read(io::receive_policy::config) override {
// nop
}
inline error write_some(io::network::event_handler* parent) override {
written += send_sizes.front();
send_sizes.pop_front();
auto remaining = send_buffer.size() - written;
count += 1;
if (remaining == 0)
prepare_next_write(parent);
return none;
}
void prepare_next_write(io::network::event_handler*) override {
written = 0;
send_buffer.clear();
send_sizes.clear();
if (offline_buffer.empty()) {
writing = false;
} else {
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() override {
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 {
if (!offline_buffer.empty() && !writing) {
writing = true;
prepare_next_write(parent);
}
}
expected<io::network::native_socket>
connect(const std::string&, uint16_t,
optional<io::network::protocol::network> = none) override {
return invalid_native_socket;
}
// 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;
};
struct raw_newb : public io::network::newb<caf::policy::new_basp_message> {
using message_type = caf::policy::new_basp_message;
raw_newb(caf::actor_config& cfg, default_multiplexer& dm,
native_socket sockfd)
: newb<message_type>(cfg, dm, sockfd) {
// nop
CAF_LOG_TRACE("");
}
void handle(message_type&) override {
CAF_PUSH_AID_FROM_PTR(this);
CAF_LOG_TRACE("");
}
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);
}
};
}
};
class config : public actor_system_config {
public:
int iterations = 10;
config() {
opt_group{custom_options_, "global"}
.add(iterations, "iterations,i", "set iterations");
}
};
void caf_main(actor_system& sys, const config& cfg) {
using clock = std::chrono::system_clock;
using resolution = std::chrono::milliseconds;
auto n = io::network::make_newb<raw_newb>(sys, invalid_native_socket);
auto ptr = caf::actor_cast<caf::abstract_actor*>(n);
auto& ref = dynamic_cast<raw_newb&>(*ptr);
ref.transport.reset(new dummy_transport);
ref.protocol.reset(new udp_protocol<policy::ordering<caf::policy::datagram_basp>>(&ref));
auto start = clock::now();
for (int i = 0; i < cfg.iterations; ++i) {
auto hw = caf::make_callback([&](io::network::byte_buffer& buf) -> error {
binary_serializer bs(sys, buf);
bs(policy::basp_header{0, actor_id{}, actor_id{}});
return none;
});
auto whdl = ref.wr_buf(&hw);
CAF_ASSERT(whdl.buf != nullptr);
CAF_ASSERT(whdl.protocol != nullptr);
binary_serializer bs(sys, *whdl.buf);
auto start = whdl.buf->size();
whdl.buf->resize(start + chunk_size);
std::fill(whdl.buf->begin() + start, whdl.buf->end(), 'a');
}
auto end = clock::now();
auto ticks = std::chrono::duration_cast<resolution>(end - start).count();
std::cout << cfg.iterations << ", " << ticks << std::endl;
}
} // namespace anonymous
CAF_MAIN(io::middleman);
...@@ -5,6 +5,7 @@ ...@@ -5,6 +5,7 @@
#include "caf/binary_deserializer.hpp" #include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp" #include "caf/binary_serializer.hpp"
#include "caf/detail/call_cfun.hpp" #include "caf/detail/call_cfun.hpp"
#include "caf/policy/newb_basp.hpp"
#include "caf/policy/newb_tcp.hpp" #include "caf/policy/newb_tcp.hpp"
using namespace caf; using namespace caf;
...@@ -28,102 +29,8 @@ using responder_atom = atom_constant<atom("responder")>; ...@@ -28,102 +29,8 @@ using responder_atom = atom_constant<atom("responder")>;
constexpr size_t chunk_size = 1024; //128; //8192; //1024; constexpr size_t chunk_size = 1024; //128; //8192; //1024;
struct basp_header { struct basp_newb : public io::network::newb<policy::new_basp_message> {
uint32_t payload_len; using message_type = policy::new_basp_message;
actor_id from;
actor_id to;
};
constexpr size_t basp_header_len = sizeof(uint32_t) + sizeof(actor_id) * 2;
template <class Inspector>
typename Inspector::result_type inspect(Inspector& fun, basp_header& hdr) {
return fun(meta::type_name("tcp_basp_header"),
hdr.payload_len, hdr.from, hdr.to);
}
struct new_basp_message {
basp_header header;
char* payload;
size_t payload_len;
};
template <class Inspector>
typename Inspector::result_type inspect(Inspector& fun,
new_basp_message& msg) {
return fun(meta::type_name("tcp_new_basp_message"), msg.header,
msg.payload_len);
}
struct basp {
static constexpr size_t header_size = basp_header_len;
using message_type = new_basp_message;
using result_type = optional<message_type>;
io::network::newb<message_type>* parent;
message_type msg;
bool expecting_header = true;
basp(io::network::newb<message_type>* parent) : parent(parent) {
// TODO: this is dangerous ...
// Maybe we need an init function that is called with `start()`?
parent->configure_read(io::receive_policy::exactly(basp_header_len));
}
error read_header(char* bytes, size_t count) {
if (count < basp_header_len)
return sec::unexpected_message;
binary_deserializer bd{&parent->backend(), bytes, count};
bd(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(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 offset, size_t plen) {
stream_serializer<charbuf> out{&parent->backend(),
buf.data() + hstart + offset,
sizeof(uint32_t)};
auto len = static_cast<uint32_t>(plen);
out(len);
}
};
struct basp_newb : public io::network::newb<new_basp_message> {
using message_type = new_basp_message;
basp_newb(caf::actor_config& cfg, default_multiplexer& dm, basp_newb(caf::actor_config& cfg, default_multiplexer& dm,
native_socket sockfd) native_socket sockfd)
...@@ -162,7 +69,7 @@ struct basp_newb : public io::network::newb<new_basp_message> { ...@@ -162,7 +69,7 @@ struct basp_newb : public io::network::newb<new_basp_message> {
delayed_send(this, interval, send_atom::value, char((c + 1) % 256)); delayed_send(this, interval, send_atom::value, char((c + 1) % 256));
auto hw = caf::make_callback([&](io::network::byte_buffer& buf) -> error { auto hw = caf::make_callback([&](io::network::byte_buffer& buf) -> error {
binary_serializer bs(&backend(), buf); binary_serializer bs(&backend(), buf);
bs(basp_header{0, id(), actor_id{}}); bs(policy::basp_header{0, id(), actor_id{}});
return none; return none;
}); });
auto whdl = wr_buf(&hw); auto whdl = wr_buf(&hw);
...@@ -199,6 +106,7 @@ struct basp_newb : public io::network::newb<new_basp_message> { ...@@ -199,6 +106,7 @@ struct basp_newb : public io::network::newb<new_basp_message> {
for (auto& t : data) { for (auto& t : data) {
auto expected = (1000000 / get<0>(t).count()); auto expected = (1000000 / get<0>(t).count());
aggregate[expected].push_back(get<1>(t)); aggregate[expected].push_back(get<1>(t));
std::cerr << expected << ", " << get<1>(t) << ", " << get<2>(t) << std::endl;
} }
for (auto& p : aggregate) { for (auto& p : aggregate) {
std::cerr << p.first; std::cerr << p.first;
...@@ -251,7 +159,7 @@ struct tcp_acceptor ...@@ -251,7 +159,7 @@ struct tcp_acceptor
ref.transport = std::move(pol); ref.transport = std::move(pol);
ref.protocol.reset(new ProtocolPolicy(&ref)); ref.protocol.reset(new ProtocolPolicy(&ref));
ref.responder = responder; ref.responder = responder;
ref.configure_read(io::receive_policy::exactly(basp_header_len)); ref.configure_read(io::receive_policy::exactly(policy::basp_header_len));
anon_send(responder, n); anon_send(responder, n);
return n; return n;
} }
...@@ -274,7 +182,7 @@ public: ...@@ -274,7 +182,7 @@ public:
}; };
void caf_main(actor_system& sys, const config& cfg) { void caf_main(actor_system& sys, const config& cfg) {
using acceptor_t = tcp_acceptor<tcp_protocol<basp>>; using acceptor_t = tcp_acceptor<tcp_protocol<policy::stream_basp>>;
const char* host = cfg.host.c_str(); const char* host = cfg.host.c_str();
const uint16_t port = cfg.port; const uint16_t port = cfg.port;
scoped_actor self{sys}; scoped_actor self{sys};
...@@ -332,7 +240,7 @@ void caf_main(actor_system& sys, const config& cfg) { ...@@ -332,7 +240,7 @@ void caf_main(actor_system& sys, const config& cfg) {
} else { } else {
std::cout << "creating new client" << std::endl; std::cout << "creating new client" << std::endl;
auto client = make_client_newb<basp_newb, tcp_transport, auto client = make_client_newb<basp_newb, tcp_transport,
tcp_protocol<basp>>(sys, host, port); tcp_protocol<policy::stream_basp>>(sys, host, port);
self->send(client, responder_atom::value, helper); self->send(client, responder_atom::value, helper);
self->send(client, send_atom::value, char(0)); self->send(client, send_atom::value, char(0));
self->send(client, interval_atom::value); self->send(client, interval_atom::value);
......
...@@ -6,17 +6,17 @@ ...@@ -6,17 +6,17 @@
#include "caf/binary_serializer.hpp" #include "caf/binary_serializer.hpp"
#include "caf/detail/call_cfun.hpp" #include "caf/detail/call_cfun.hpp"
#include "caf/policy/newb_udp.hpp" #include "caf/policy/newb_udp.hpp"
#include "caf/policy/newb_basp.hpp"
#include "caf/policy/newb_ordering.hpp"
using namespace caf; using namespace caf;
using namespace caf::policy;
using caf::io::network::default_multiplexer; using caf::io::network::default_multiplexer;
using caf::io::network::invalid_native_socket; using caf::io::network::invalid_native_socket;
using caf::io::network::make_client_newb; using caf::io::network::make_client_newb;
using caf::io::network::make_server_newb; using caf::io::network::make_server_newb;
using caf::io::network::native_socket; using caf::io::network::native_socket;
using caf::policy::accept_udp;
using caf::policy::udp_protocol;
using caf::policy::udp_transport;
namespace { namespace {
...@@ -30,190 +30,6 @@ using handshake_atom = atom_constant<atom("handshake")>; ...@@ -30,190 +30,6 @@ using handshake_atom = atom_constant<atom("handshake")>;
constexpr size_t chunk_size = 8192; //8192; //128; //1024; constexpr size_t chunk_size = 8192; //8192; //128; //1024;
struct basp_header {
uint32_t payload_len;
actor_id from;
actor_id to;
};
template <class Inspector>
typename Inspector::result_type inspect(Inspector& fun, basp_header& hdr) {
return fun(meta::type_name("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 ordering_header {
sequence_type seq;
};
template <class Inspector>
typename Inspector::result_type inspect(Inspector& fun, ordering_header& hdr) {
return fun(meta::type_name("ordering_header"), hdr.seq);
}
constexpr size_t udp_ordering_header_len = sizeof(sequence_type);
struct new_basp_message {
basp_header header;
char* payload;
size_t payload_len;
};
template <class Inspector>
typename Inspector::result_type inspect(Inspector& fun,
new_basp_message& msg) {
return fun(meta::type_name("new_basp_message"), msg.header,
msg.payload_len);
}
struct basp {
static constexpr size_t header_size = udp_basp_header_len;
using message_type = new_basp_message;
using result_type = optional<message_type>;
io::network::newb<message_type>* parent;
message_type msg;
basp(io::network::newb<message_type>* parent) : parent(parent) {
// nop
}
error read(char* bytes, size_t count) {
// Read header.
if (count < udp_basp_header_len) {
CAF_LOG_DEBUG("not enought bytes for basp header");
return sec::unexpected_message;
}
binary_deserializer bd{&parent->backend(), bytes, count};
bd(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_ERROR("not enough bytes remaining to fit payload");
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 offset, size_t plen) {
stream_serializer<charbuf> out{&parent->backend(),
buf.data() + hstart + offset,
sizeof(uint32_t)};
auto len = static_cast<uint32_t>(plen);
out(len);
}
};
template <class Next>
struct 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;
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;
ordering_header hdr;
binary_deserializer bd(&parent->backend(), bytes, count);
bd(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(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 offset, size_t plen) {
next.prepare_for_sending(buf, hstart, offset + header_size, plen);
}
};
struct raw_newb : public io::network::newb<new_basp_message> { struct raw_newb : public io::network::newb<new_basp_message> {
using message_type = new_basp_message; using message_type = new_basp_message;
...@@ -406,7 +222,8 @@ public: ...@@ -406,7 +222,8 @@ public:
}; };
void caf_main(actor_system& sys, const config& cfg) { void caf_main(actor_system& sys, const config& cfg) {
using acceptor_t = udp_acceptor<udp_protocol<ordering<basp>>>; using policy_t = udp_protocol<ordering<datagram_basp>>;
using acceptor_t = udp_acceptor<policy_t>;
const char* host = cfg.host.c_str(); const char* host = cfg.host.c_str();
const uint16_t port = cfg.port; const uint16_t port = cfg.port;
scoped_actor self{sys}; scoped_actor self{sys};
...@@ -463,8 +280,8 @@ void caf_main(actor_system& sys, const config& cfg) { ...@@ -463,8 +280,8 @@ void caf_main(actor_system& sys, const config& cfg) {
await_done("done"); await_done("done");
} else { } else {
std::cout << "creating new client" << std::endl; std::cout << "creating new client" << std::endl;
auto client = make_client_newb<raw_newb, udp_transport, auto client = make_client_newb<raw_newb, udp_transport, policy_t>(sys, host,
udp_protocol<ordering<basp>>>(sys, host, port); port);
self->send(client, responder_atom::value, helper); self->send(client, responder_atom::value, helper);
self->send(client, handshake_atom::value); self->send(client, handshake_atom::value);
await_done("let's start"); await_done("let's start");
......
...@@ -5,6 +5,7 @@ ...@@ -5,6 +5,7 @@
#include "caf/binary_deserializer.hpp" #include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp" #include "caf/binary_serializer.hpp"
#include "caf/detail/call_cfun.hpp" #include "caf/detail/call_cfun.hpp"
#include "caf/policy/newb_raw.hpp"
#include "caf/policy/newb_tcp.hpp" #include "caf/policy/newb_tcp.hpp"
using namespace caf; using namespace caf;
...@@ -28,49 +29,8 @@ using responder_atom = atom_constant<atom("responder")>; ...@@ -28,49 +29,8 @@ using responder_atom = atom_constant<atom("responder")>;
constexpr size_t chunk_size = 8192; //128; //8192; //1024; constexpr size_t chunk_size = 8192; //128; //8192; //1024;
struct new_data { struct raw_newb : public io::network::newb<policy::raw_data_message> {
char* payload; using message_type = policy::raw_data_message;
size_t payload_len;
};
template <class Inspector>
typename Inspector::result_type inspect(Inspector& fun, new_data& data) {
return fun(meta::type_name("new_data"), data.payload_len);
}
struct raw_tcp {
using message_type = new_data;
using result_type = optional<message_type>;
io::network::newb<message_type>* parent;
message_type msg;
raw_tcp(io::network::newb<message_type>* parent) : parent(parent) {
// nop
}
error read(char* bytes, size_t count) {
msg.payload = bytes;
msg.payload_len = count;
parent->handle(msg);
return none;
}
error timeout(atom_value, uint32_t) {
return none;
}
size_t write_header(io::network::byte_buffer&,
io::network::header_writer*) {
return 0;
}
void prepare_for_sending(io::network::byte_buffer&, size_t, size_t, size_t) {
// nop
}
};
struct raw_newb : public io::network::newb<new_data> {
using message_type = new_data;
raw_newb(caf::actor_config& cfg, default_multiplexer& dm, raw_newb(caf::actor_config& cfg, default_multiplexer& dm,
native_socket sockfd) native_socket sockfd)
...@@ -221,7 +181,7 @@ struct state { ...@@ -221,7 +181,7 @@ struct state {
}; };
void caf_main(actor_system& sys, const config& cfg) { void caf_main(actor_system& sys, const config& cfg) {
using acceptor_t = tcp_acceptor<tcp_protocol<raw_tcp>>; using acceptor_t = tcp_acceptor<tcp_protocol<policy::raw>>;
const char* host = cfg.host.c_str(); const char* host = cfg.host.c_str();
const uint16_t port = cfg.port; const uint16_t port = cfg.port;
scoped_actor self{sys}; scoped_actor self{sys};
...@@ -281,7 +241,7 @@ void caf_main(actor_system& sys, const config& cfg) { ...@@ -281,7 +241,7 @@ void caf_main(actor_system& sys, const config& cfg) {
} else { } else {
std::cout << "creating new client" << std::endl; std::cout << "creating new client" << std::endl;
auto client = make_client_newb<raw_newb, tcp_transport, auto client = make_client_newb<raw_newb, tcp_transport,
tcp_protocol<raw_tcp>>(sys, host, port); tcp_protocol<policy::raw>>(sys, host, port);
self->send(client, responder_atom::value, helper); self->send(client, responder_atom::value, helper);
self->send(client, send_atom::value, char(0)); self->send(client, send_atom::value, char(0));
self->send(client, interval_atom::value); self->send(client, interval_atom::value);
......
...@@ -5,6 +5,7 @@ ...@@ -5,6 +5,7 @@
#include "caf/binary_deserializer.hpp" #include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp" #include "caf/binary_serializer.hpp"
#include "caf/detail/call_cfun.hpp" #include "caf/detail/call_cfun.hpp"
#include "caf/policy/newb_raw.hpp"
#include "caf/policy/newb_udp.hpp" #include "caf/policy/newb_udp.hpp"
using namespace caf; using namespace caf;
...@@ -30,49 +31,8 @@ using handshake_atom = atom_constant<atom("handshake")>; ...@@ -30,49 +31,8 @@ using handshake_atom = atom_constant<atom("handshake")>;
constexpr size_t chunk_size = 8192; //128; //1024; constexpr size_t chunk_size = 8192; //128; //1024;
struct new_data { struct raw_newb : public io::network::newb<policy::raw_data_message> {
char* payload; using message_type = policy::raw_data_message;
size_t payload_len;
};
template <class Inspector>
typename Inspector::result_type inspect(Inspector& fun, new_data& data) {
return fun(meta::type_name("new_data"), data.payload_len);
}
struct raw_udp {
using message_type = new_data;
using result_type = optional<message_type>;
io::network::newb<message_type>* parent;
message_type msg;
raw_udp(io::network::newb<message_type>* parent) : parent(parent) {
// nop
}
error read(char* bytes, size_t count) {
msg.payload = bytes;
msg.payload_len = count;
parent->handle(msg);
return none;
}
error timeout(atom_value, uint32_t) {
return none;
}
size_t write_header(io::network::byte_buffer&,
io::network::header_writer*) {
return 0;
}
void prepare_for_sending(io::network::byte_buffer&, size_t, size_t, size_t) {
// nop
}
};
struct raw_newb : public io::network::newb<new_data> {
using message_type = new_data;
raw_newb(caf::actor_config& cfg, default_multiplexer& dm, raw_newb(caf::actor_config& cfg, default_multiplexer& dm,
native_socket sockfd) native_socket sockfd)
...@@ -247,7 +207,7 @@ public: ...@@ -247,7 +207,7 @@ public:
}; };
void caf_main(actor_system& sys, const config& cfg) { void caf_main(actor_system& sys, const config& cfg) {
using acceptor_t = udp_acceptor<udp_protocol<raw_udp>>; using acceptor_t = udp_acceptor<udp_protocol<policy::raw>>;
const char* host = cfg.host.c_str(); const char* host = cfg.host.c_str();
const uint16_t port = cfg.port; const uint16_t port = cfg.port;
scoped_actor self{sys}; scoped_actor self{sys};
...@@ -305,7 +265,7 @@ void caf_main(actor_system& sys, const config& cfg) { ...@@ -305,7 +265,7 @@ void caf_main(actor_system& sys, const config& cfg) {
} else { } else {
std::cout << "creating new client" << std::endl; std::cout << "creating new client" << std::endl;
auto client = make_client_newb<raw_newb, udp_transport, auto client = make_client_newb<raw_newb, udp_transport,
udp_protocol<raw_udp>>(sys, host, port); udp_protocol<policy::raw>>(sys, host, port);
self->send(client, responder_atom::value, helper); self->send(client, responder_atom::value, helper);
self->send(client, handshake_atom::value); self->send(client, handshake_atom::value);
await_done("let's start"); await_done("let's start");
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#include "caf/actor.hpp"
#include "caf/io/network/newb.hpp"
namespace caf {
namespace policy {
struct basp_header {
uint32_t payload_len;
actor_id from;
actor_id to;
};
template <class Inspector>
typename Inspector::result_type inspect(Inspector& fun, basp_header& hdr) {
return fun(meta::type_name("basp_header"), hdr.payload_len,
hdr.from, hdr.to);
}
constexpr size_t basp_header_len = sizeof(uint32_t) + sizeof(actor_id) * 2;
struct new_basp_message {
basp_header header;
char* payload;
size_t payload_len;
};
template <class Inspector>
typename Inspector::result_type inspect(Inspector& fun,
new_basp_message& msg) {
return fun(meta::type_name("new_basp_message"), msg.header,
msg.payload_len);
}
struct datagram_basp {
static constexpr size_t header_size = basp_header_len;
using message_type = new_basp_message;
using result_type = optional<message_type>;
io::network::newb<message_type>* parent;
message_type msg;
datagram_basp(io::network::newb<message_type>* parent) : parent(parent) {
// nop
}
error read(char* bytes, size_t count) {
// Read header.
if (count < basp_header_len) {
CAF_LOG_DEBUG("not enought bytes for basp header");
return sec::unexpected_message;
}
binary_deserializer bd{&parent->backend(), bytes, count};
bd(msg.header);
size_t payload_len = static_cast<size_t>(msg.header.payload_len);
// Read payload.
auto remaining = count - basp_header_len;
// TODO: Could be `!=` ?
if (remaining < payload_len) {
CAF_LOG_ERROR("not enough bytes remaining to fit payload");
return sec::unexpected_message;
}
msg.payload = bytes + 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 offset, size_t plen) {
stream_serializer<charbuf> out{&parent->backend(),
buf.data() + hstart + offset,
sizeof(uint32_t)};
auto len = static_cast<uint32_t>(plen);
out(len);
}
};
struct stream_basp {
static constexpr size_t header_size = basp_header_len;
using message_type = new_basp_message;
using result_type = optional<message_type>;
io::network::newb<message_type>* parent;
message_type msg;
bool expecting_header = true;
stream_basp(io::network::newb<message_type>* parent) : parent(parent) {
// TODO: this is dangerous ...
// Maybe we need an init function that is called with `start()`?
parent->configure_read(io::receive_policy::exactly(basp_header_len));
}
error read_header(char* bytes, size_t count) {
if (count < basp_header_len)
return sec::unexpected_message;
binary_deserializer bd{&parent->backend(), bytes, count};
bd(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(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 offset, size_t plen) {
stream_serializer<charbuf> out{&parent->backend(),
buf.data() + hstart + offset,
sizeof(uint32_t)};
auto len = static_cast<uint32_t>(plen);
out(len);
}
};
} // namespace policy
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#include <cstdint>
#include <unordered_map>
#include <chrono>
#include "caf/error.hpp"
#include "caf/io/network/newb.hpp"
namespace caf {
namespace policy {
using sequence_type = uint16_t;
using ordering_atom = atom_constant<atom("ordering")>;
struct ordering_header {
sequence_type seq;
};
constexpr size_t ordering_header_len = sizeof(sequence_type);
template <class Inspector>
typename Inspector::result_type inspect(Inspector& fun, ordering_header& hdr) {
return fun(meta::type_name("ordering_header"), hdr.seq);
}
template <class Next>
struct ordering {
static constexpr size_t header_size = 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;
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;
ordering_header hdr;
binary_deserializer bd(&parent->backend(), bytes, count);
bd(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(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 offset, size_t plen) {
next.prepare_for_sending(buf, hstart, offset + header_size, plen);
}
};
} // namespace policy
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#include "caf/actor.hpp"
#include "caf/io/network/newb.hpp"
namespace caf {
namespace policy {
struct raw_data_message {
char* payload;
size_t payload_len;
};
template <class Inspector>
typename Inspector::result_type inspect(Inspector& fun, raw_data_message& data) {
return fun(meta::type_name("raw_data_message"), data.payload_len);
}
struct raw {
using message_type = raw_data_message;
using result_type = optional<message_type>;
io::network::newb<message_type>* parent;
message_type msg;
raw(io::network::newb<message_type>* parent) : parent(parent) {
// nop
}
error read(char* bytes, size_t count) {
msg.payload = bytes;
msg.payload_len = count;
parent->handle(msg);
return none;
}
error timeout(atom_value, uint32_t) {
return none;
}
size_t write_header(io::network::byte_buffer&,
io::network::header_writer*) {
return 0;
}
void prepare_for_sending(io::network::byte_buffer&, size_t, size_t, size_t) {
// nop
}
};
} // namespace policy
} // namespace caf
...@@ -119,6 +119,7 @@ error udp_transport::write_some(io::network::event_handler* parent) { ...@@ -119,6 +119,7 @@ error udp_transport::write_some(io::network::event_handler* parent) {
CAF_LOG_ERROR("sendto returned" << CAF_ARG(sres)); CAF_LOG_ERROR("sendto returned" << CAF_ARG(sres));
return sec::runtime_error; return sec::runtime_error;
} }
// TODO: This only works if we always write send_sizes.front()
send_sizes.pop_front(); send_sizes.pop_front();
written += (sres > 0) ? static_cast<size_t>(sres) : 0; written += (sres > 0) ? static_cast<size_t>(sres) : 0;
auto remaining = send_buffer.size() - written; auto remaining = send_buffer.size() - written;
......
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