Unverified Commit 0da72e8e authored by Dominik Charousset's avatar Dominik Charousset Committed by GitHub

Merge pull request #655

Improve reordering latency
parents 41fe4314 1d327ea3
......@@ -301,6 +301,7 @@ public:
bool middleman_enable_tcp;
bool middleman_enable_udp;
size_t middleman_cached_udp_buffers;
size_t middleman_max_pending_msgs;
// -- config parameters of the OpenCL module ---------------------------------
......
......@@ -139,6 +139,7 @@ actor_system_config::actor_system_config()
middleman_enable_tcp = true;
middleman_enable_udp = false;
middleman_cached_udp_buffers = 10;
middleman_max_pending_msgs = 10;
// fill our options vector for creating INI and CLI parsers
opt_group{options_, "scheduler"}
.add(scheduler_policy, "policy",
......@@ -208,6 +209,9 @@ actor_system_config::actor_system_config()
"enable communication via UDP (off by default)")
.add(middleman_cached_udp_buffers, "cached-udp-buffers",
"sets the max number of UDP send buffers that will be cached for reuse "
"(default: 10)")
.add(middleman_max_pending_msgs, "max-pending-messages",
"sets the max number of UDP pending messages due to ordering "
"(default: 10)");
opt_group(options_, "opencl")
.add(opencl_device_ids, "device-ids",
......
......@@ -36,8 +36,7 @@ namespace basp {
// stores meta information for active endpoints
struct endpoint_context {
using pending_map = std::unordered_map<uint16_t,
std::pair<basp::header,
using pending_map = std::map<sequence_type, std::pair<basp::header,
std::vector<char>>>;
// denotes what message we expect from the remote node next
basp::connection_state cstate;
......@@ -55,9 +54,12 @@ struct endpoint_context {
// protocols that do not implement ordering are ordered by CAF
bool requires_ordering;
// sequence numbers and a buffer to establish order
uint16_t seq_incoming;
uint16_t seq_outgoing;
sequence_type seq_incoming;
sequence_type seq_outgoing;
// pending messages due to ordering
pending_map pending;
// track if a timeout to deliver pending messages is set
bool did_set_timeout;
};
} // namespace basp
......
......@@ -112,16 +112,18 @@ public:
/// Adds a message with a future sequence number to the pending messages
/// of a given endpoint context.
virtual void add_pending(sequence_type seq, endpoint_context& ep,
header hdr, std::vector<char> payload) = 0;
virtual void add_pending(execution_unit* ctx, endpoint_context& ep,
sequence_type seq, header hdr,
std::vector<char> payload) = 0;
/// Delivers a pending incoming messages for an endpoint with
/// application layer ordering.
virtual bool deliver_pending(execution_unit* ctx,
endpoint_context& ep) = 0;
/// Delivers a pending incoming messages for an endpoint `ep` with
/// application layer ordering. Delivery of the next available packet can
/// be forced to simply skip undeliverd messages via the `force` flag.
virtual bool deliver_pending(execution_unit* ctx, endpoint_context& ep,
bool force) = 0;
/// Drop pending messages with sequence number `seq`.
virtual void drop_pending(sequence_type seq, endpoint_context& ep) = 0;
virtual void drop_pending(endpoint_context& ep, sequence_type seq) = 0;
/// Returns a reference to the current sent buffer, dispatching the call
/// based on the type contained in `hdl`.
......
......@@ -97,15 +97,16 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
uint16_t next_sequence_number(datagram_handle hdl) override;
// inherited from basp::instance::callee
void add_pending(uint16_t seq, basp::endpoint_context& ep, basp::header hdr,
void add_pending(execution_unit* ctx, basp::endpoint_context& ep,
uint16_t seq, basp::header hdr,
std::vector<char> payload) override;
// inherited from basp::instance::callee
bool deliver_pending(execution_unit* ctx,
basp::endpoint_context& ep) override;
bool deliver_pending(execution_unit* ctx, basp::endpoint_context& ep,
bool force) override;
// inherited from basp::instance::callee
void drop_pending(uint16_t seq, basp::endpoint_context& ep) override;
void drop_pending(basp::endpoint_context& ep, uint16_t seq) override;
// inherited from basp::instance::callee
buffer_type& get_buffer(endpoint_handle hdl) override;
......@@ -174,6 +175,12 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
const size_t max_buffers;
std::stack<buffer_type> cached_buffers;
// maximum queue size for pending messages of endpoints with ordering
const size_t max_pending_messages;
// timeout for delivery of pending messages of endpoints with ordering
const std::chrono::milliseconds pending_to = std::chrono::milliseconds(100);
// returns the node identifier of the underlying BASP instance
const node_id& this_node() const {
return instance.this_node();
......
......@@ -77,7 +77,8 @@ basp_broker_state::basp_broker_state(broker* selfptr)
static_cast<proxy_registry::backend&>(*this)),
self(selfptr),
instance(selfptr, *this),
max_buffers(self->system().config().middleman_cached_udp_buffers) {
max_buffers(self->system().config().middleman_cached_udp_buffers),
max_pending_messages(self->system().config().middleman_max_pending_msgs) {
CAF_ASSERT(this_node() != none);
}
......@@ -447,8 +448,8 @@ void basp_broker_state::set_context(connection_handle hdl) {
basp::header{basp::message_type::server_handshake, 0,
0, 0, none, none,
invalid_actor_id, invalid_actor_id},
hdl, none, 0, 0, none,
false, 0, 0, basp::endpoint_context::pending_map()
hdl, none, 0, 0, none, false, 0, 0,
basp::endpoint_context::pending_map(), false
}
).first;
}
......@@ -467,8 +468,8 @@ void basp_broker_state::set_context(datagram_handle hdl) {
basp::header{basp::message_type::server_handshake,
0, 0, 0, none, none,
invalid_actor_id, invalid_actor_id},
hdl, none, 0, 0, none,
true, 0, 0, basp::endpoint_context::pending_map()
hdl, none, 0, 0, none, true, 0, 0,
basp::endpoint_context::pending_map(), false
}
).first;
}
......@@ -533,37 +534,49 @@ basp_broker_state::next_sequence_number(datagram_handle hdl) {
return 0;
}
void basp_broker_state::add_pending(basp::sequence_type seq,
void basp_broker_state::add_pending(execution_unit* ctx,
basp::endpoint_context& ep,
basp::sequence_type seq,
basp::header hdr,
std::vector<char> payload) {
if (!ep.requires_ordering)
return;
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(20), pending_atom::value,
get<datagram_handle>(ep.hdl), seq);
if (ep.pending.size() >= max_pending_messages)
deliver_pending(ctx, ep, true);
else if (!ep.did_set_timeout)
self->delayed_send(self, pending_to, pending_atom::value,
get<datagram_handle>(ep.hdl));
}
bool basp_broker_state::deliver_pending(execution_unit* ctx,
basp::endpoint_context& ep) {
if (!ep.requires_ordering)
basp::endpoint_context& ep,
bool force) {
if (!ep.requires_ordering || ep.pending.empty())
return true;
std::vector<char>* payload = nullptr;
auto itr = ep.pending.find(ep.seq_incoming);
while (itr != ep.pending.end()) {
ep.hdr = std::move(itr->second.first);
payload = &itr->second.second;
auto i = ep.pending.begin();
// Force delivery of at least the first messages, if desired.
if (force)
ep.seq_incoming = i->first;
while (i != ep.pending.end() && i->first == ep.seq_incoming) {
ep.hdr = std::move(i->second.first);
payload = &i->second.second;
if (!instance.handle(ctx, get<datagram_handle>(ep.hdl),
ep.hdr, payload, false, ep, none))
return false;
ep.pending.erase(itr);
i = ep.pending.erase(i);
ep.seq_incoming += 1;
itr = ep.pending.find(ep.seq_incoming);
}
// Set a timeout if there are still pending messages.
if (!ep.pending.empty() && !ep.did_set_timeout)
self->delayed_send(self, pending_to, pending_atom::value,
get<datagram_handle>(ep.hdl));
return true;
}
void basp_broker_state::drop_pending(basp::sequence_type seq,
basp::endpoint_context& ep) {
void basp_broker_state::drop_pending(basp::endpoint_context& ep,
basp::sequence_type seq) {
if (!ep.requires_ordering)
return;
ep.pending.erase(seq);
......@@ -913,18 +926,19 @@ behavior basp_broker::make_behavior() {
delayed_send(this, std::chrono::milliseconds{interval},
tick_atom::value, interval);
},
[=](pending_atom, datagram_handle hdl, basp::sequence_type seq) {
[=](pending_atom, datagram_handle hdl) {
auto& ep = state.ctx_udp[hdl];
auto itr = ep.pending.find(seq);
if (itr != ep.pending.end()) {
ep.did_set_timeout = false;
if (ep.pending.empty())
return;
auto i = ep.pending.begin();
auto seq = i->first;
if (seq == ep.seq_incoming ||
basp::instance::is_greater(seq, ep.seq_incoming)) {
// skip missing messages
ep.seq_incoming = seq;
state.deliver_pending(context(), ep);
// Skip missing messages and force delivery.
state.deliver_pending(context(), ep, true);
} else {
state.drop_pending(seq, ep);
}
state.drop_pending(ep, seq);
}
}
};
......
......@@ -162,20 +162,20 @@ bool instance::handle(execution_unit* ctx, new_datagram_msg& dm,
return err();
}
}
// Handle FIFO ordering of datagrams
// Handle ordering of datagrams.
if (is_greater(ep.hdr.sequence_number, ep.seq_incoming)) {
// Message arrived "early", add to pending messages
// Add early messages to the pending message buffer.
auto s = ep.hdr.sequence_number;
callee_.add_pending(s, ep, std::move(ep.hdr), std::move(pl_buf));
callee_.add_pending(ctx, ep, s, std::move(ep.hdr), std::move(pl_buf));
return true;
} else if (ep.hdr.sequence_number != ep.seq_incoming) {
// Message arrived late, drop it!
// Drop messages that arrive late.
CAF_LOG_DEBUG("dropping message " << CAF_ARG(dm));
return true;
}
// Message arrived as expected
// This is the expected message.
ep.seq_incoming += 1;
// TODO: Add optional reliability here (send acks, ...)
// TODO: add optional reliability here
if (!is_handshake(ep.hdr) && !is_heartbeat(ep.hdr)
&& ep.hdr.dest_node != this_node_) {
CAF_LOG_DEBUG("forward message");
......@@ -208,8 +208,8 @@ bool instance::handle(execution_unit* ctx, new_datagram_msg& dm,
}
if (!handle(ctx, dm.handle, ep.hdr, payload, false, ep, ep.local_port))
return err();
// Look for pending messages
if (!callee_.deliver_pending(ctx, ep))
// See if the next message was delivered early and is already bufferd.
if (!callee_.deliver_pending(ctx, ep, false))
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