Commit 1d327ea3 authored by Joseph Noir's avatar Joseph Noir

Improve reordering latency

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