Commit c75024f8 authored by Joseph Noir's avatar Joseph Noir

Fix multiple message delivery

parent dbf7c528
......@@ -97,14 +97,11 @@ struct transport_policy {
return send_buffer;
}
template <class T>
optional<T> read_some(newb<T>* parent, protocol_policy<T>& policy) {
error read_some(newb<T>* parent, protocol_policy<T>& policy) {
auto err = read_some();
if (err) {
// Call something on parent?
return none;
}
if (err)
return err;
return policy.read(parent, receive_buffer.data(), receive_buffer.size());
}
......@@ -146,9 +143,9 @@ struct protocol_policy : protocol_policy_base {
// nop
}
virtual optional<T> read(newb<T>* parent, char* bytes, size_t count) = 0;
virtual error read(newb<T>* parent, char* bytes, size_t count) = 0;
virtual optional<T> timeout(newb<T>* parent, caf::message& msg) = 0;
virtual error timeout(newb<T>* parent, caf::message& msg) = 0;
/// TODO: Come up with something better than a write here?
/// Write header into buffer. Use push back to append only.
......@@ -158,93 +155,6 @@ struct protocol_policy : protocol_policy_base {
template <class T>
using protocol_policy_ptr = std::unique_ptr<protocol_policy<T>>;
/// @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<type>* parent;
result_type read(newb<message_type>*, char* bytes, size_t count) {
new_basp_message msg;
memcpy(&msg.header.from, bytes, sizeof(msg.header.from));
memcpy(&msg.header.to, bytes + sizeof(msg.header.from), sizeof(msg.header.to));
msg.payload = bytes + header_size;
msg.payload_size = count - header_size;
return msg;
}
result_type timeout(newb<message_type>*, caf::message&) {
return none;
}
size_t write_header(byte_buffer& buf, size_t offset, header_writer* hw) {
CAF_ASSERT(hw != nullptr);
(*hw)(buf);
return offset + header_size;
}
};
/// @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 next_seq_read = 0;
uint32_t next_seq_write = 0;
Next next;
std::unordered_map<uint32_t, std::vector<char>> pending;
result_type read(newb<message_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;
memcpy(&seq, bytes, sizeof(seq));
if (seq != next_seq_read) {
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_read++;
return next.read(parent, bytes + header_size, count - header_size);
}
result_type timeout(newb<message_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);
}
size_t write_header(byte_buffer& buf, size_t offset, header_writer* hw) {
std::array<char, sizeof (next_seq_write)> tmp;
memcpy(tmp.data(), &next_seq_write, sizeof(next_seq_write));
next_seq_write += 1;
for (auto& c : tmp)
buf.push_back(c);
return next.write_header(buf, offset + header_size, hw);
}
};
template <class T>
struct protocol_policy_impl : protocol_policy<typename T::message_type> {
T impl;
......@@ -252,7 +162,7 @@ struct protocol_policy_impl : protocol_policy<typename T::message_type> {
// nop
}
typename T::result_type read(newb<typename T::message_type>* parent, char* bytes,
error read(newb<typename T::message_type>* parent, char* bytes,
size_t count) override {
return impl.read(parent, bytes, count);
}
......@@ -261,7 +171,7 @@ struct protocol_policy_impl : protocol_policy<typename T::message_type> {
return T::offset;
}
typename T::result_type timeout(newb<typename T::message_type>* parent,
error timeout(newb<typename T::message_type>* parent,
caf::message& msg) override {
return impl.timeout(parent, msg);
}
......@@ -298,13 +208,6 @@ struct newb {
}
write_handle wr_buf(header_writer* hw) {
// Write the buffer fist. That allows variable sized headers and
// should be straight forward. The arguments can be consumed by
// the protocol policy layers to write their header such as the
// information required to write the BASP header.
// - get the write buffer from the transport layer
// - let the protocol policies write their headers
// - return a
// 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();
......@@ -318,12 +221,7 @@ struct newb {
}
error read_event() {
auto maybe_msg = transport->read_some(this, *protocol);
if (!maybe_msg)
return make_error(sec::unexpected_message);
// TODO: create message on the stack and call message handler
handle(*maybe_msg);
return none;
return transport->read_some(this, *protocol);
}
void write_event() {
......@@ -340,11 +238,7 @@ struct newb {
// 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;
return protocol->timeout(this, msg);
}
// Allow protocol policies to enqueue a data for sending.
......@@ -387,15 +281,123 @@ struct newb_acceptor {
// server: sys.middleman().spawn_server<protocol_policy>(sock,
// std::move(accept_protocol_policy_impl), my_server);
// -- 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<type>* parent;
error read(newb<message_type>* parent, char* bytes, size_t count) {
new_basp_message msg;
memcpy(&msg.header.from, bytes, sizeof(msg.header.from));
memcpy(&msg.header.to, bytes + sizeof(msg.header.from), sizeof(msg.header.to));
msg.payload = bytes + header_size;
msg.payload_size = count - header_size;
// Using the result for messages has some problems, so ...
parent->handle(msg);
return none;
}
error timeout(newb<message_type>*, caf::message&) {
return none;
}
size_t write_header(byte_buffer& buf, size_t offset, header_writer* hw) {
CAF_ASSERT(hw != nullptr);
(*hw)(buf);
return offset + header_size;
}
};
/// @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;
Next next;
std::unordered_map<uint32_t, std::vector<char>> pending;
error deliver_pending(newb<message_type>* parent) {
if (pending.empty())
return none;
while (pending.count(seq_read) > 0) {
auto& buf = pending[seq_read];
auto res = next.read(parent, buf.data(), buf.size());
pending.erase(seq_read);
if (res)
return res;
}
return none;
}
error read(newb<message_type>* parent, char* bytes, size_t count) {
uint32_t seq;
memcpy(&seq, bytes, sizeof(seq));
// TODO: Use the comparison function from BASP instance.
if (seq == seq_read) {
seq_read += 1;
auto res = next.read(parent, bytes + header_size, count - header_size);
if (res)
return res;
return deliver_pending(parent);
} else if (seq > seq_read) {
pending[seq] = std::vector<char>(bytes + header_size, bytes + count);
parent->set_timeout(std::chrono::seconds(2),
caf::make_message(ordering_atom::value, seq));
// TODO: Should this return an error?
return none;
}
// Is late, drop it. TODO: Should this return an error?
return none;
}
error timeout(newb<message_type>* parent, caf::message& msg) {
// TODO: Can't we just set the seq_read to the sequence number and call
// deliver messages?
error err = none;
msg.apply([&](ordering_atom, uint32_t seq) {
if (pending.count(seq) > 0) {
CAF_MESSAGE("found pending message");
auto& buf = pending[seq];
err = next.read(parent, buf.data(), buf.size());
seq_read = seq + 1;
}
});
if (err)
return err;
return deliver_pending(parent);
}
size_t write_header(byte_buffer& buf, size_t offset, header_writer* hw) {
std::array<char, sizeof (seq_write)> tmp;
memcpy(tmp.data(), &seq_write, sizeof(seq_write));
seq_write += 1;
for (auto& c : tmp)
buf.push_back(c);
return next.write_header(buf, offset + header_size, hw);
}
};
// -- test classes -------------------------------------------------------------
struct dummy_basp_newb : newb<new_basp_message> {
std::vector<caf::message> timeout_messages;
std::vector<new_basp_message> messages;
std::vector<std::pair<new_basp_message, std::vector<char>>> messages;
void handle(new_basp_message& received_msg) override {
messages.push_back(received_msg);
void handle(new_basp_message& msg) override {
std::vector<char> payload{msg.payload, msg.payload + msg.payload_size};
messages.emplace_back(msg, payload);
messages.back().first.payload = messages.back().second.data();
}
void set_timeout_impl(caf::message msg) override {
......@@ -403,19 +405,29 @@ struct dummy_basp_newb : newb<new_basp_message> {
}
};
struct fixture {
struct datagram_fixture {
dummy_basp_newb self;
fixture() {
datagram_fixture() {
self.transport.reset(new transport_policy);
self.protocol.reset(new protocol_policy_impl<ordering<basp_policy>>);
}
scoped_execution_unit context;
};
struct stream_fixture {
dummy_basp_newb self;
stream_fixture() {
self.transport.reset(new transport_policy);
self.protocol.reset(new protocol_policy_impl<basp_policy>);
}
scoped_execution_unit context;
};
} // namespace <anonymous>
CAF_TEST_FIXTURE_SCOPE(protocol_policy_tests, fixture)
CAF_TEST_FIXTURE_SCOPE(protocol_policy_tests, datagram_fixture)
CAF_TEST(ordering and basp read event) {
CAF_MESSAGE("create some values for our buffer");
......@@ -439,7 +451,7 @@ CAF_TEST(ordering and basp read event) {
auto err = self.read_event();
CAF_REQUIRE(!err);
CAF_MESSAGE("check the basp header and payload");
auto& msg = self.messages.front();
auto& msg = self.messages.front().first;
CAF_CHECK_EQUAL(msg.header.from, bhdr.from);
CAF_CHECK_EQUAL(msg.header.to, bhdr.to);
CAF_CHECK_EQUAL(msg.payload_size, sizeof(payload));
......@@ -468,9 +480,9 @@ CAF_TEST(ordering and basp read event with timeout) {
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");
CAF_MESSAGE("trigger a read event");
auto err = self.read_event();
CAF_REQUIRE(err);
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();
......@@ -484,7 +496,7 @@ CAF_TEST(ordering and basp read event with timeout) {
err = self.timeout_event(timeout_msg);
CAF_REQUIRE(!err);
CAF_MESSAGE("check delivered message");
auto& msg = self.messages.front();
auto& msg = self.messages.front().first;
CAF_CHECK_EQUAL(msg.header.from, bhdr.from);
CAF_CHECK_EQUAL(msg.header.to, bhdr.to);
CAF_CHECK_EQUAL(msg.payload_size, sizeof(payload));
......@@ -526,7 +538,7 @@ CAF_TEST(ordering and basp multiple messages) {
memcpy(buf.data() + offset, &payload_second, sizeof(payload_second));
CAF_MESSAGE("trigger a read event, expecting an error");
auto err = self.read_event();
CAF_REQUIRE(err);
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();
......@@ -544,21 +556,21 @@ CAF_TEST(ordering and basp multiple messages) {
memcpy(buf.data() + offset, &bhdr_first, sizeof(basp_header));
offset += sizeof(basp_header);
memcpy(buf.data() + offset, &payload_first, sizeof(payload_first));
CAF_MESSAGE("trigger a read event, expecting an error");
// This not clean.
CAF_MESSAGE("trigger a read event");
err = self.read_event();
CAF_REQUIRE(!err);
CAF_CHECK(self.messages.size() == 2);
CAF_MESSAGE("check handled messages");
// Message one
auto& msg = self.messages.front();
CAF_MESSAGE("check first message");
auto& msg = self.messages.front().first;
CAF_CHECK_EQUAL(msg.header.from, bhdr_first.from);
CAF_CHECK_EQUAL(msg.header.to, bhdr_first.to);
CAF_CHECK_EQUAL(msg.payload_size, sizeof(payload_first));
int return_payload = 0;
memcpy(&return_payload, msg.payload, msg.payload_size);
CAF_CHECK_EQUAL(return_payload, payload_first);
// Message two
msg = self.messages.back();
CAF_MESSAGE("check second message");
msg = self.messages.back().first;
CAF_CHECK_EQUAL(msg.header.from, bhdr_second.from);
CAF_CHECK_EQUAL(msg.header.to, bhdr_second.to);
CAF_CHECK_EQUAL(msg.payload_size, sizeof(payload_second));
......@@ -595,7 +607,7 @@ CAF_TEST(ordering and basp write buf) {
auto err = self.read_event();
CAF_REQUIRE(!err);
CAF_MESSAGE("check the basp header and payload");
auto& msg = self.messages.front();
auto& msg = self.messages.front().first;
CAF_CHECK_EQUAL(msg.header.from, bhdr.from);
CAF_CHECK_EQUAL(msg.header.to, bhdr.to);
CAF_CHECK_EQUAL(msg.payload_size, sizeof(payload));
......@@ -604,4 +616,30 @@ CAF_TEST(ordering and basp write buf) {
CAF_CHECK_EQUAL(return_payload, payload);
}
CAF_TEST(slicing and basp) {
// TODO: Come up with a way to implement this with policies. What is missing
// in the API? The problem here is that this is not just a write_header call
// when acquireing the write buffer. There is more to it.
// TOUGHTS: For receipt this is located somewhere between ordering /
// reliability and BASP. For sending it would need to split messages into
// multiple parts after the BASP header was written but before ordering or
// reliability add their headers.
CAF_MESSAGE("not implemented");
}
CAF_TEST_FIXTURE_SCOPE_END()
CAF_TEST_FIXTURE_SCOPE(more_protocol_policy_tests, stream_fixture)
CAF_TEST(basp and streaming) {
// TODO: As far as I can tell atm, we would need to reimplement the BASP
// policy as it works differently on datgrams and streams. Needs two reads
// for streaming: once for the header and once for the payload.
// IDEA: We could let the policy track itself if it expects payload. When it
// receives a message it dezerializes the header and tests if it still has a
// payload of the expected size. Or something like that.
// Easiest way is to implement BASP twice.
CAF_MESSAGE("not implemented");
}
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