Commit 2fbfa8f2 authored by Joseph Noir's avatar Joseph Noir

Fix pending message delivery and overflow handling

parent 970b1090
......@@ -18,12 +18,13 @@
#pragma once
#include <cstdint>
#include <unordered_map>
#include <chrono>
#include <cstdint>
#include <map>
#include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/defaults.hpp"
#include "caf/error.hpp"
#include "caf/io/network/newb.hpp"
......@@ -33,6 +34,31 @@ namespace policy {
using sequence_type = uint16_t;
using ordering_atom = atom_constant<atom("ordering")>;
} // namespace policy
} // namespace caf
namespace {
bool is_greater(caf::policy::sequence_type lhs, caf::policy::sequence_type rhs,
caf::policy::sequence_type max_distance
= std::numeric_limits<caf::policy::sequence_type>::max() / 2) {
// distance between lhs and rhs is smaller than max_distance.
return ((lhs > rhs) && (lhs - rhs <= max_distance)) ||
((lhs < rhs) && (rhs - lhs > max_distance));
}
struct sequence_comperator {
bool operator()(const caf::policy::sequence_type& lhs,
const caf::policy::sequence_type& rhs) const {
return is_greater(rhs, lhs);
}
};
} // namespace anonymous
namespace caf {
namespace policy {
struct ordering_header {
sequence_type seq;
};
......@@ -51,14 +77,19 @@ struct ordering {
using result_type = typename Next::result_type;
sequence_type seq_read = 0;
sequence_type seq_write = 0;
size_t max_pending_messages = 10;
size_t max_pending_messages;
bool use_timeouts;
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),
std::map<sequence_type, std::vector<char>, sequence_comperator> pending;
ordering(io::network::newb<message_type>* parent, bool use_timeouts = true)
: max_pending_messages(get_or(parent->config(),
"middleman.max-pending-messages",
caf::defaults::middleman::max_pending_messages)),
use_timeouts(use_timeouts),
parent(parent),
next(parent) {
// nop
}
......@@ -70,6 +101,7 @@ struct ordering {
auto& buf = pending[seq_read];
auto res = next.read(buf.data(), buf.size());
pending.erase(seq_read);
seq_read += 1;
// TODO: Cancel timeout.
if (res)
return res;
......@@ -79,7 +111,8 @@ struct ordering {
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 (use_timeouts)
parent->set_timeout(pending_to, ordering_atom::value, seq);
if (pending.size() > max_pending_messages) {
seq_read = pending.begin()->first;
return deliver_pending();
......@@ -93,14 +126,13 @@ struct ordering {
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) {
} else if (is_greater(hdr.seq, seq_read)) {
add_pending(bytes, count, hdr.seq);
return none;
}
......@@ -108,6 +140,8 @@ struct ordering {
}
error timeout(atom_value atm, uint32_t id) {
std::cout << "got timeout message" << std::endl;
std::abort();
if (atm == ordering_atom::value) {
error err = none;
sequence_type seq = static_cast<sequence_type>(id);
......
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