Commit aa3f2518 authored by Joseph Noir's avatar Joseph Noir

Buffer messages if connection is pending

parent 24073a6d
......@@ -131,6 +131,11 @@ public:
/// Returns a reference to the sent buffer.
virtual buffer_type& get_buffer(connection_handle hdl) = 0;
/// Returns a reference to a buffer to be sent to node with `nid`.
/// If communication with the node is esstablished, it picks the first
/// available handle, otherwise a buffer for a pending message is returned.
virtual buffer_type& get_buffer(node_id nid) = 0;
/// Returns the buffer accessed through a call to `get_buffer` when
/// passing a datagram handle and removes it from the callee.
virtual buffer_type pop_datagram_buffer(datagram_handle hdl) = 0;
......
......@@ -109,6 +109,9 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
// inherited from basp::instance::callee
buffer_type& get_buffer(connection_handle hdl) override;
// inherited from basp::instance::callee
buffer_type& get_buffer(node_id nid) override;
// inherited from basp::instance::callee
buffer_type pop_datagram_buffer(datagram_handle hdl) override;
......@@ -170,8 +173,12 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
// maximum queue size for pending messages of endpoints with ordering
const size_t max_pending_messages;
// buffer messages for nodes while connectivity is established
std::unordered_map<node_id, std::vector<buffer_type>> pending_connectivity;
// timeout for delivery of pending messages of endpoints with ordering
const std::chrono::milliseconds pending_to = std::chrono::milliseconds(100);
const std::chrono::milliseconds pending_timeout
= std::chrono::milliseconds(100);
// returns the node identifier of the underlying BASP instance
const node_id& this_node() const {
......
......@@ -201,9 +201,11 @@ void basp_broker_state::send_kill_proxy_instance(const node_id& nid,
visit(seq_num_visitor{this}, *c.hdl));
instance.flush(*c.hdl);
} else {
// TODO: Buffer message!
CAF_CRITICAL("ibasp_broker_state::send_kill_proxy_instance with buffering "
"not implemented!");
// TODO: Buffer message until communication is enstablished.
buffer_type buf;
instance.write_kill_proxy(self->context(), buf,
nid, aid, rsn, 0);
pending_connectivity[nid].emplace_back(std::move(buf));
}
}
......@@ -383,12 +385,13 @@ void basp_broker_state::learned_new_node(const node_id& nid) {
return;
}
auto c = std::move(*ec);
if (c.conn == basp::routing_table::connectivity::established) {
// send message to SpawnServ of remote node
basp::header hdr{basp::message_type::dispatch_message,
basp::header::named_receiver_flag,
0, 0, this_node(), nid, tmp.id(), invalid_actor_id,
visit(seq_num_visitor{this}, *c.hdl)};
0}; // sequence number only available with connectivity
if (c.conn == basp::routing_table::connectivity::established) {
hdr.sequence_number = visit(seq_num_visitor{this}, *c.hdl);
// writing std::numeric_limits<actor_id>::max() is a hack to get
// this send-to-named-actor feature working with older CAF releases
instance.write(self->context(), get_buffer(*c.hdl),
......@@ -396,8 +399,9 @@ void basp_broker_state::learned_new_node(const node_id& nid) {
instance.flush(*c.hdl);
} else {
// TODO: Implement this, can it happen?
CAF_CRITICAL("basp_broker_state::learned_new_node with buffering not "
"implemented!");
buffer_type buf;
instance.write(self->context(), buf, hdr, &writer);
pending_connectivity[nid].emplace_back(std::move(buf));
}
}
......@@ -510,7 +514,7 @@ void basp_broker_state::add_pending(execution_unit* ctx,
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,
self->delayed_send(self, pending_timeout, pending_atom::value,
get<datagram_handle>(ep.hdl));
}
......@@ -535,7 +539,7 @@ bool basp_broker_state::deliver_pending(execution_unit* ctx,
}
// 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,
self->delayed_send(self, pending_timeout, pending_atom::value,
get<datagram_handle>(ep.hdl));
return true;
}
......@@ -567,6 +571,17 @@ basp_broker_state::get_buffer(connection_handle hdl) {
return self->wr_buf(hdl);
}
basp_broker_state::buffer_type&
basp_broker_state::get_buffer(node_id nid) {
auto ec = instance.tbl().lookup(nid);
if (ec && ec->conn == basp::routing_table::connectivity::established && ec->hdl) {
return get_buffer(*(ec->hdl));
}
auto msgs = pending_connectivity[nid];
msgs.emplace_back();
return msgs.back();
}
basp_broker_state::buffer_type
basp_broker_state::pop_datagram_buffer(datagram_handle) {
std::vector<char> res;
......@@ -737,7 +752,6 @@ behavior basp_broker::make_behavior() {
return sec::no_route_to_receiving_node;
}
auto c = std::move(*ec);
if (c.conn == basp::routing_table::connectivity::established) {
if (system().node() == src->node())
system().registry().put(src->id(), src);
auto writer = make_callback([&](serializer& sink) -> error {
......@@ -747,14 +761,17 @@ behavior basp_broker::make_behavior() {
basp::header::named_receiver_flag,
0, cme->mid.integer_value(), state.this_node(),
dest_node, src->id(), invalid_actor_id,
visit(seq_num_visitor{&state}, *c.hdl)};
0};
if (c.conn == basp::routing_table::connectivity::established) {
hdr.sequence_number = visit(seq_num_visitor{&state}, *c.hdl);
state.instance.write(context(), state.get_buffer(*c.hdl),
hdr, &writer);
state.instance.flush(*c.hdl);
} else {
// TODO: Buffer the message in the basp broker.
CAF_CRITICAL("basp_broker forward_atom with buffering "
"not implemented!");
std::vector<char> buf;
state.instance.write(context(), buf, hdr, &writer);
state.pending_connectivity[dest_node].emplace_back(buf);
}
return delegated<message>();
},
......
......@@ -254,12 +254,11 @@ bool instance::dispatch(execution_unit* ctx, const strong_actor_ptr& sender,
CAF_ASSERT(receiver && system().node() != receiver->node());
auto ec = tbl_.lookup(receiver->node());
/// TODO: Let's assume that the handle is valid if the status is established.
if (!ec || ec->cs == routing_table::connectivity::failed) {
if (!ec || ec->conn == routing_table::connectivity::failed) {
notify<hook::message_sending_failed>(sender, receiver, mid, msg);
return false;
}
auto c = std::move(*ec);
if (c.cs == routing_table::connectivity::established) {
auto writer = make_callback([&](serializer& sink) -> error {
return sink(const_cast<std::vector<strong_actor_ptr>&>(forwarding_stack),
const_cast<message&>(msg));
......@@ -267,7 +266,9 @@ bool instance::dispatch(execution_unit* ctx, const strong_actor_ptr& sender,
header hdr{message_type::dispatch_message, 0, 0, mid.integer_value(),
sender ? sender->node() : this_node(), receiver->node(),
sender ? sender->id() : invalid_actor_id, receiver->id(),
visit(seq_num_visitor{callee_}, *c.hdl)};
0};
if (c.conn == routing_table::connectivity::established) {
hdr.sequence_number = visit(seq_num_visitor{callee_}, *c.hdl);
write(ctx, callee_.get_buffer(*c.hdl), hdr, &writer);
flush(*c.hdl);
notify<hook::message_sent>(sender, receiver->node(), receiver, mid, msg);
......@@ -275,8 +276,11 @@ bool instance::dispatch(execution_unit* ctx, const strong_actor_ptr& sender,
} else {
// lr.cs == routing_table::communication::pending
// TODO: Buffer the message in the basp broker.
CAF_CRITICAL("instance::disaptch with buffering not implemented!");
return false;
write(ctx, callee_.get_buffer(receiver->node()), hdr, &writer);
// TODO: should the hook really be called here, or should we delay this
// until communication is established?
notify<hook::message_sent>(sender, receiver->node(), receiver, mid, msg);
return true;
}
}
......
......@@ -85,7 +85,7 @@ void routing_table::add(const node_id& nid) {
bool routing_table::reachable(const node_id& dest) {
auto i = node_information_base_.find(dest);
if (i != node_information_base_.end())
return i->second.details.cs == connectivity::established;
return i->second.details.conn == connectivity::established;
return false;
}
......@@ -94,7 +94,7 @@ bool routing_table::status(const node_id& nid,
auto i = node_information_base_.find(nid);
if (i == node_information_base_.end())
return false;
i->second.details.cs = new_status;
i->second.details.conn = new_status;
return true;
}
......@@ -103,7 +103,7 @@ routing_table::status(const node_id& nid) {
auto i = node_information_base_.find(nid);
if (i == node_information_base_.end())
return none;
return i->second.details.cs;
return i->second.details.conn;
}
bool routing_table::forwarder(const node_id& nid,
......
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