Commit 3db0335c authored by Joseph Noir's avatar Joseph Noir

Add basic test for timeout with ordering policy

parent a0c40604
...@@ -32,6 +32,8 @@ ...@@ -32,6 +32,8 @@
// TODO: {receive_buffer => byte_buffer} (and use `byte` instead of `char`) // TODO: {receive_buffer => byte_buffer} (and use `byte` instead of `char`)
#include "caf/io/network/receive_buffer.hpp" #include "caf/io/network/receive_buffer.hpp"
#include "caf/detail/enum_to_string.hpp"
using namespace caf; using namespace caf;
using namespace caf::io; using namespace caf::io;
...@@ -48,6 +50,13 @@ struct protocol_policy_base; ...@@ -48,6 +50,13 @@ struct protocol_policy_base;
template <class T> template <class T>
struct protocol_policy; struct protocol_policy;
template <class T>
struct newb;
// -- atoms --------------------------------------------------------------------
using ordering_atom = atom_constant<atom("ordering")>;
// -- aliases ------------------------------------------------------------------ // -- aliases ------------------------------------------------------------------
using byte_buffer = network::receive_buffer; using byte_buffer = network::receive_buffer;
...@@ -60,7 +69,7 @@ struct basp_header { ...@@ -60,7 +69,7 @@ struct basp_header {
}; };
struct ordering_header { struct ordering_header {
int32_t seq_nr; uint32_t seq_nr;
}; };
// -- message types ------------------------------------------------------------ // -- message types ------------------------------------------------------------
...@@ -71,7 +80,7 @@ struct new_basp_message { ...@@ -71,7 +80,7 @@ struct new_basp_message {
size_t payload_size; size_t payload_size;
}; };
// -- new broker classes ------------------------------------------------------- // -- transport policy ---------------------------------------------------------
struct transport_policy { struct transport_policy {
virtual ~transport_policy() { virtual ~transport_policy() {
...@@ -88,11 +97,13 @@ struct transport_policy { ...@@ -88,11 +97,13 @@ struct transport_policy {
template <class T> template <class T>
expected<T> read_some(protocol_policy<T>& policy) { optional<T> read_some(newb<T>* parent, protocol_policy<T>& policy) {
auto err = read_some(); auto err = read_some();
if (err) if (err) {
return err; // Call something on parent?
return policy.read(receive_buffer.data(), receive_buffer.size()); return none;
}
return policy.read(parent, receive_buffer.data(), receive_buffer.size());
} }
virtual error read_some() { virtual error read_some() {
...@@ -105,6 +116,8 @@ struct transport_policy { ...@@ -105,6 +116,8 @@ struct transport_policy {
using transport_policy_ptr = std::unique_ptr<transport_policy>; using transport_policy_ptr = std::unique_ptr<transport_policy>;
// -- accept policy ------------------------------------------------------------
struct accept_policy { struct accept_policy {
virtual ~accept_policy() { virtual ~accept_policy() {
// nop // nop
...@@ -113,6 +126,8 @@ struct accept_policy { ...@@ -113,6 +126,8 @@ struct accept_policy {
virtual void init(newb_base&) = 0; virtual void init(newb_base&) = 0;
}; };
// -- protocol policies --------------------------------------------------------
struct protocol_policy_base { struct protocol_policy_base {
virtual ~protocol_policy_base() { virtual ~protocol_policy_base() {
// nop // nop
...@@ -129,11 +144,14 @@ struct protocol_policy : protocol_policy_base { ...@@ -129,11 +144,14 @@ struct protocol_policy : protocol_policy_base {
// nop // nop
} }
virtual expected<T> read(char* bytes, size_t count) = 0; virtual optional<T> read(newb<T>* parent, char* bytes, size_t count) = 0;
virtual optional<T> timeout(newb<T>* parent, caf::message& msg) = 0;
void write_header(byte_buffer&, size_t) override { void write_header(byte_buffer&, size_t) override {
// nop // nop
} }
}; };
template <class T> template <class T>
...@@ -143,8 +161,10 @@ struct basp_policy { ...@@ -143,8 +161,10 @@ struct basp_policy {
static constexpr size_t header_size = sizeof(basp_header); static constexpr size_t header_size = sizeof(basp_header);
static constexpr size_t offset = header_size; static constexpr size_t offset = header_size;
using type = new_basp_message; using type = new_basp_message;
using result_type = expected<new_basp_message>; using result_type = optional<new_basp_message>;
result_type read(char* bytes, size_t count) { //newb<type>* parent;
result_type read(newb<type>*, char* bytes, size_t count) {
new_basp_message msg; new_basp_message msg;
memcpy(&msg.header.from, bytes, sizeof(msg.header.from)); memcpy(&msg.header.from, bytes, sizeof(msg.header.from));
memcpy(&msg.header.to, bytes + sizeof(msg.header.from), sizeof(msg.header.to)); memcpy(&msg.header.to, bytes + sizeof(msg.header.from), sizeof(msg.header.to));
...@@ -152,7 +172,10 @@ struct basp_policy { ...@@ -152,7 +172,10 @@ struct basp_policy {
msg.payload_size = count - header_size; msg.payload_size = count - header_size;
return msg; return msg;
} }
scoped_execution_unit context;
result_type timeout(newb<type>*, caf::message&) {
return none;
}
}; };
template <class Next> template <class Next>
...@@ -161,13 +184,45 @@ struct ordering { ...@@ -161,13 +184,45 @@ struct ordering {
static constexpr size_t offset = Next::offset + header_size; static constexpr size_t offset = Next::offset + header_size;
using type = typename Next::type; using type = typename Next::type;
using result_type = typename Next::result_type; using result_type = typename Next::result_type;
uint32_t next_seq = 0;
Next next; Next next;
result_type read(char* bytes, size_t count) { std::unordered_map<uint32_t, std::vector<char>> pending;
result_type read(newb<type>* parent, char* bytes, size_t count) {
// TODO: What to do if we want to deliver multiple messages? For
// example when our buffer a missing message arrives and we
// can just deliver everything in our buffer?
uint32_t seq; uint32_t seq;
memcpy(&seq, bytes, sizeof(seq)); memcpy(&seq, bytes, sizeof(seq));
return next.read(bytes + header_size, count - header_size); if (seq != next_seq) {
CAF_MESSAGE("adding message to pending ");
// TODO: Only works if we have datagrams.
pending[seq] = std::vector<char>(bytes + header_size, bytes + count);
parent->set_timeout(std::chrono::seconds(2),
caf::make_message(ordering_atom::value, seq));
return none;
}
next_seq++;
return next.read(parent, bytes + header_size, count - header_size);
}
result_type timeout(newb<type>* parent, caf::message& msg) {
// TODO: Same as above.
auto matched = false;
result_type was_pending = none;
msg.apply([&](ordering_atom, uint32_t seq) {
matched = true;
if (pending.count(seq) > 0) {
CAF_MESSAGE("found pending message");
auto& buf = pending[seq];
was_pending = next.read(parent, buf.data(), buf.size());
}
});
if (matched)
return was_pending;
return next.timeout(parent, msg);
} }
scoped_execution_unit context;
}; };
template <class T> template <class T>
...@@ -177,15 +232,24 @@ struct protocol_policy_impl : protocol_policy<typename T::type> { ...@@ -177,15 +232,24 @@ struct protocol_policy_impl : protocol_policy<typename T::type> {
// nop // nop
} }
typename T::result_type read(char* bytes, size_t count) override { typename T::result_type read(newb<typename T::type>* parent, char* bytes,
return impl.read(bytes, count); size_t count) override {
return impl.read(parent, bytes, count);
} }
size_t offset() const noexcept override { size_t offset() const noexcept override {
return T::offset; return T::offset;
} }
typename T::result_type timeout(newb<typename T::type>* parent,
caf::message& msg) override {
return impl.timeout(parent, msg);
}
}; };
// -- new broker classes -------------------------------------------------------
/// Returned by funtion wr_buf
struct write_handle { struct write_handle {
protocol_policy_base* protocol; protocol_policy_base* protocol;
byte_buffer* buf; byte_buffer* buf;
...@@ -214,14 +278,15 @@ struct newb { ...@@ -214,14 +278,15 @@ struct newb {
return {protocol.get(), buf, header_offset}; return {protocol.get(), buf, header_offset};
} }
// Send
void flush() { void flush() {
} }
error read_event() { error read_event() {
auto maybe_msg = transport->read_some(*protocol); auto maybe_msg = transport->read_some(this, *protocol);
if (!maybe_msg) if (!maybe_msg)
return std::move(maybe_msg.error()); return make_error(sec::unexpected_message);
// TODO: create message on the stack and call message handler // TODO: create message on the stack and call message handler
handle(*maybe_msg); handle(*maybe_msg);
return none; return none;
...@@ -231,7 +296,31 @@ struct newb { ...@@ -231,7 +296,31 @@ struct newb {
// transport->write_some(); // transport->write_some();
} }
// 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>, caf::message msg) {
// TODO: Once this is an actor send yourself a delayed messge.
// And on receict call ...
set_timeout_impl(std::move(msg));
}
// Called on self when a timeout is received.
error timeout_event(caf::message& msg) {
auto maybe_msg = protocol->timeout(this, msg);
if (!maybe_msg)
return make_error(sec::unexpected_message);
handle(*maybe_msg);
return none;
}
// Allow protocol policies to enqueue a data for sending.
// Probably required for reliability.
// void direct_enqueue(char* bytes, size_t count);
virtual void handle(Message& msg) = 0; virtual void handle(Message& msg) = 0;
// TODO: Only here for some first tests.
virtual void set_timeout_impl(caf::message) = 0;
}; };
struct basp_newb : newb<new_basp_message> { struct basp_newb : newb<new_basp_message> {
...@@ -259,14 +348,22 @@ struct newb_acceptor { ...@@ -259,14 +348,22 @@ struct newb_acceptor {
}; };
*/ */
// client: sys.middleman().spawn_client<protocol_policy>(sock, std::move(transport_protocol_policy_impl), my_client); // client: sys.middleman().spawn_client<protocol_policy>(sock,
// server: sys.middleman().spawn_server<protocol_policy>(sock, std::move(accept_protocol_policy_impl), my_server); // std::move(transport_protocol_policy_impl), my_client);
// server: sys.middleman().spawn_server<protocol_policy>(sock,
// std::move(accept_protocol_policy_impl), my_server);
// -- test classes -------------------------------------------------------------
struct dummy_basp_newb : newb<new_basp_message> { struct dummy_basp_newb : newb<new_basp_message> {
void handle(new_basp_message& received_msg) override { void handle(new_basp_message& received_msg) override {
msg = received_msg; msg = received_msg;
} }
void set_timeout_impl(caf::message msg) override {
timeout_messages.emplace_back(std::move(msg));
}
std::vector<caf::message> timeout_messages;
new_basp_message msg; new_basp_message msg;
}; };
...@@ -284,10 +381,17 @@ struct fixture { ...@@ -284,10 +381,17 @@ struct fixture {
CAF_TEST_FIXTURE_SCOPE(protocol_policy_tests, fixture) CAF_TEST_FIXTURE_SCOPE(protocol_policy_tests, fixture)
CAF_TEST(ordering and basp) { CAF_TEST(ordering and basp read event) {
// Read process:
// - read event happens
// - read_event() called on newb
// - read_some(protocol&) called on transport policy
// - transport policies reads from network layer
// - calls read on protocol policy
// - the basp policy fills the expected result message with data
CAF_MESSAGE("create some values for our buffer"); CAF_MESSAGE("create some values for our buffer");
ordering_header ohdr{13}; ordering_header ohdr{0};
basp_header bhdr{41, 43}; basp_header bhdr{13, 42};
int payload = 1337; int payload = 1337;
CAF_MESSAGE("copy them into the buffer"); CAF_MESSAGE("copy them into the buffer");
auto& buf = self.transport->receive_buffer; auto& buf = self.transport->receive_buffer;
...@@ -303,7 +407,8 @@ CAF_TEST(ordering and basp) { ...@@ -303,7 +407,8 @@ CAF_TEST(ordering and basp) {
offset += sizeof(basp_header); offset += sizeof(basp_header);
memcpy(buf.data() + offset, &payload, sizeof(payload)); memcpy(buf.data() + offset, &payload, sizeof(payload));
CAF_MESSAGE("trigger a read event"); CAF_MESSAGE("trigger a read event");
self.read_event(); auto err = self.read_event();
CAF_REQUIRE(!err);
CAF_MESSAGE("check the basp header and payload"); CAF_MESSAGE("check the basp header and payload");
CAF_CHECK_EQUAL(self.msg.header.from, bhdr.from); CAF_CHECK_EQUAL(self.msg.header.from, bhdr.from);
CAF_CHECK_EQUAL(self.msg.header.to, bhdr.to); CAF_CHECK_EQUAL(self.msg.header.to, bhdr.to);
...@@ -313,4 +418,59 @@ CAF_TEST(ordering and basp) { ...@@ -313,4 +418,59 @@ CAF_TEST(ordering and basp) {
CAF_CHECK_EQUAL(return_payload, payload); CAF_CHECK_EQUAL(return_payload, payload);
} }
CAF_TEST(ordering and basp read event with timeout) {
// Read process:
// - read event happens
// - read_event() called on newb
// - read_some(protocol&) called on transport policy
// - transport policies reads from network layer
// - calls read on protocol policy
// - the ordering policy sets a timeout
// -
CAF_MESSAGE("create some values for our buffer");
// Should be an unexpected sequence number and lead to an error.
ordering_header ohdr{1};
basp_header bhdr{13, 42};
int payload = 1337;
CAF_MESSAGE("copy them into the buffer");
auto& buf = self.transport->receive_buffer;
// Make sure the buffer is big enough.
buf.resize(sizeof(ordering_header)
+ sizeof(basp_header)
+ sizeof(payload));
// Track an offset for writing.
size_t offset = 0;
memcpy(buf.data() + offset, &ohdr, sizeof(ordering_header));
offset += sizeof(ordering_header);
memcpy(buf.data() + offset, &bhdr, sizeof(basp_header));
offset += sizeof(basp_header);
memcpy(buf.data() + offset, &payload, sizeof(payload));
CAF_MESSAGE("trigger a read event, expecting an error");
auto err = self.read_event();
CAF_REQUIRE(err);
CAF_MESSAGE("check if we have a pending timeout now");
CAF_REQUIRE(!self.timeout_messages.empty());
auto& timeout_msg = self.timeout_messages.back();
auto read_message = false;
timeout_msg.apply([&](ordering_atom, uint32_t seq) {
if (seq == ohdr.seq_nr)
read_message = true;
});
CAF_REQUIRE(read_message);
CAF_MESSAGE("trigger timeout");
err = self.timeout_event(timeout_msg);
CAF_REQUIRE(!err);
CAF_MESSAGE("check delivered message");
CAF_CHECK_EQUAL(self.msg.header.from, bhdr.from);
CAF_CHECK_EQUAL(self.msg.header.to, bhdr.to);
CAF_CHECK_EQUAL(self.msg.payload_size, sizeof(payload));
int return_payload = 0;
memcpy(&return_payload, self.msg.payload, self.msg.payload_size);
CAF_CHECK_EQUAL(return_payload, payload);
}
CAF_TEST(ordering and basp write event) {
CAF_MESSAGE("Not implemented");
}
CAF_TEST_FIXTURE_SCOPE_END() 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