Commit 46f5a120 authored by Joseph Noir's avatar Joseph Noir

Set timeouts for pending messages

parent 500d672c
......@@ -155,6 +155,9 @@ using migrate_atom = atom_constant<atom("migrate")>;
/// Used for triggering periodic operations.
using tick_atom = atom_constant<atom("tick")>;
/// Used for pending out of order messages
using pending_atom = atom_constant<atom("pending")>;
/// Used as config parameter for the `logger`.
using trace_log_lvl_atom = atom_constant<atom("TRACE")>;
......
......@@ -110,6 +110,15 @@ public:
// return the next outgoing sequence number for an endpoint
virtual uint16_t next_sequence_number(dgram_scribe_handle hdl) = 0;
/// Add message with a future sequence number to the pending messages
/// of a given endpoint context
virtual void add_pending(uint16_t seq, endpoint_context& ep,
basp::header hdr, std::vector<char> payload) = 0;
/// Deliver pending incoming messages for an endpoint with
/// application layer ordering
virtual bool deliver_pending(execution_unit* ctx, endpoint_context& ep) = 0;
protected:
proxy_registry namespace_;
};
......@@ -134,10 +143,6 @@ public:
/// Handles a new UDP endpoint msg
bool handle(execution_unit* ctx, new_endpoint_msg& em, endpoint_context& ep);
/// Deliver pending incoming messages for an endpoint with
/// application layer ordering
bool deliver_pending(execution_unit* ctx, endpoint_context& ep);
/// Sends heartbeat messages to all valid nodes those are directly connected.
void handle_heartbeat(execution_unit* ctx);
......
......@@ -100,8 +100,17 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
// inherited from basp::instance::listener
uint16_t next_sequence_number(connection_handle hdl) override;
// inherited from basp::instance::listener
uint16_t next_sequence_number(dgram_scribe_handle hdl) override;
// inherited from basp::instance::listener
void add_pending(uint16_t seq, endpoint_context& ep, basp::header hdr,
std::vector<char> payload) override;
// inherited from basp::instance::listener
bool deliver_pending(execution_unit* ctx, endpoint_context& ep) override;
void purge(connection_handle hdl);
void purge(dgram_scribe_handle hdl);
......
......@@ -532,6 +532,36 @@ uint16_t basp_broker_state::next_sequence_number(dgram_scribe_handle hdl) {
return 0;
}
void basp_broker_state::add_pending(uint16_t seq, endpoint_context& ep,
basp::header hdr, std::vector<char> payload) {
ep.pending.emplace(seq, std::make_pair(std::move(hdr), std::move(payload)));
// TODO: choose reasonable default timeout, make configurable
self->delayed_send(self, std::chrono::milliseconds(500), pending_atom::value,
get<dgram_scribe_handle>(ep.hdl), seq);
}
bool basp_broker_state::deliver_pending(execution_unit* ctx,
endpoint_context& ep) {
if (!ep.requires_ordering)
return true;
std::vector<char>* payload = nullptr;
auto itr = ep.pending.find(ep.seq_incoming);
while (itr != ep.pending.end()) {
//std::cerr << "[++] '" << to_string(itr->second.first.operation)
// << "' with seq '" << itr->second.first.sequence_number
// << "' (was pending)" << std::endl;
ep.hdr = std::move(itr->second.first);
payload = &itr->second.second;
if (!instance.handle_msg(ctx, get<dgram_scribe_handle>(ep.hdl),
ep.hdr, payload, false, ep, none))
return false;
ep.pending.erase(itr);
ep.seq_incoming += 1;
itr = ep.pending.find(ep.seq_incoming);
}
return true;
}
/******************************************************************************
* basp_broker *
......@@ -844,6 +874,14 @@ behavior basp_broker::make_behavior() {
state.instance.handle_heartbeat(context());
delayed_send(this, std::chrono::milliseconds{interval},
tick_atom::value, interval);
},
[=](pending_atom, dgram_scribe_handle hdl, uint16_t seq) {
auto& ep = state.udp_ctx[hdl];
auto itr = ep.pending.find(seq);
if (itr != ep.pending.end()) {
ep.seq_incoming = seq;
state.deliver_pending(context(), ep);
}
}
};
}
......
......@@ -52,27 +52,6 @@ instance::instance(abstract_broker* parent, callee& lstnr)
CAF_ASSERT(this_node_ != none);
}
bool instance::deliver_pending(execution_unit* ctx, endpoint_context& ep) {
if (!ep.requires_ordering)
return true;
std::vector<char>* payload = nullptr;
auto itr = ep.pending.find(ep.seq_incoming);
while (itr != ep.pending.end()) {
//std::cerr << "[++] '" << to_string(itr->second.first.operation)
// << "' with seq '" << itr->second.first.sequence_number
// << "' (was pending)" << std::endl;
ep.hdr = std::move(itr->second.first);
payload = &itr->second.second;
if (!handle_msg(ctx, get<dgram_scribe_handle>(ep.hdl),
ep.hdr, payload, false, ep, none))
return false;
ep.pending.erase(itr);
ep.seq_incoming += 1;
itr = ep.pending.find(ep.seq_incoming);
}
return true;
}
connection_state instance::handle(execution_unit* ctx, new_data_msg& dm,
header& hdr, bool is_payload) {
CAF_LOG_TRACE(CAF_ARG(dm) << CAF_ARG(is_payload));
......@@ -175,25 +154,33 @@ bool instance::handle(execution_unit* ctx, new_datagram_msg& dm,
return err();
}
}
// TODO: Ordering
//std::cerr << "[<<] '" << to_string(ep.hdr.operation)
// << "' with seq '" << ep.hdr.sequence_number << "'" << std::endl;
if (ep.hdr.sequence_number != ep.seq_incoming) {
//std::cerr << "[!!] '" << to_string(ep.hdr.operation) << "' with seq '"
// << ep.hdr.sequence_number << "' (!= " << ep.seq_incoming
// << ")" << std::endl;
// Handle FIFO ordering of datagrams
/*
std::cerr << "[<<] '" << to_string(ep.hdr.operation)
<< "' with seq '" << ep.hdr.sequence_number << "'" << std::endl;
if (ep.hdr.sequence_numer != ep.seq_incoming) {
std::cerr << "[!!] '" << to_string(ep.hdr.operation) << "' with seq '"
<< ep.hdr.sequence_number << "' (!= " << ep.seq_incoming
<< ")" << std::endl;
}
*/
if (ep.hdr.sequence_number > ep.seq_incoming) {
// Message arrived "early", add to pending messages
auto s = ep.hdr.sequence_number;
auto h = std::move(ep.hdr);
auto b = std::move(pl_buf);
ep.pending.emplace(s, std::make_pair(std::move(h), std::move(b)));
callee_.add_pending(s, ep, std::move(ep.hdr), std::move(pl_buf));
return true;
} else if (ep.hdr.sequence_number < ep.seq_incoming) {
// Message arrived late, drop it!
CAF_LOG_DEBUG("dropping msg " << CAF_ARG(dm));
return true;
}
// Message arrived as expected
ep.seq_incoming += 1;
// TODO: Reliability
// TODO: Add optional reliability here (send acks, ...)
if (!handle_msg(ctx, dm.handle, ep.hdr, payload, false, ep, none))
return err();
if (!deliver_pending(ctx, ep))
// Look for pending messages
if (!callee_.deliver_pending(ctx, ep))
return err();
return true;
};
......@@ -248,7 +235,7 @@ bool instance::handle(execution_unit* ctx, new_endpoint_msg& em,
}
if (!handle_msg(ctx, em.handle, ep.hdr, payload, false, ep, em.port))
return err();
if (!deliver_pending(ctx, ep))
if (!callee_.deliver_pending(ctx, ep))
return err();
return true;
}
......
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