Commit c5a3f545 authored by Joseph Noir's avatar Joseph Noir

Fix handshake response and align write usage

parent 4803624d
......@@ -354,12 +354,12 @@ public:
anon_send(actor_cast<actor>(peer_server), put_atom::value,
to_string(hdr.source_node), make_message(addrs));
// Write handshake as client in response.
if (tcp_based)
if (tcp_based) {
write_client_handshake(ctx, callee_.get_buffer(hdl), hdr.source_node);
flush(hdl);
}
callee_.learned_new_node(hdr.source_node);
callee_.finalize_handshake(hdr.source_node, aid, sigs);
flush(hdl);
// TODO: Do we always want to do this or only if `tcp_based`?
callee_.send_buffered_messages(ctx, hdr.source_node, hdl);
break;
}
......@@ -385,11 +385,23 @@ public:
// Handshakes were only exchanged if `hdl` is set.
auto lr = tbl_.lookup(hdr.source_node);
auto new_node = (this_node() != hdr.source_node && !lr.hdl);
auto pending = lr.known && !lr.hdl;
if (pending && hdr.source_node < this_node()) {
CAF_LOG_INFO("simultaneous handshake, let the other node act as "
"the server");
break;
}
if (!new_node) {
if (tcp_based) {
CAF_LOG_INFO("received second client handshake:"
<< CAF_ARG(hdr.source_node));
break;
} else {
if (lr.hdl && get_if<datagram_handle>(&(*lr.hdl)) == nullptr
&& get_if<datagram_handle>(&(*lr.hdl))->id() != hdl.id()) {
CAF_LOG_INFO("dropping repeated handshake on different handle");
break;
}
}
} else {
// Add this node to our contacts.
......@@ -405,7 +417,7 @@ public:
}
// Since udp is unreliable we answer, maybe our message was lost.
if (!tcp_based) {
uint16_t seq = (ep && ep->requires_ordering) ? ep->seq_outgoing++ : 0;
uint16_t seq = ep->requires_ordering ? ep->seq_outgoing++ : 0;
write_server_handshake(ctx, callee_.get_buffer(hdl), port, seq);
callee_.flush(hdl);
}
......@@ -413,11 +425,9 @@ public:
// `learned_new_node` expects there to be an entry in the routing table.
// TODO: Can we move this in the else block above with the changes to
// the routing table?
if (new_node) {
if (new_node)
callee_.learned_new_node(hdr.source_node);
// TODO: Only send buffered messaged for new nodes?
callee_.send_buffered_messages(ctx, hdr.source_node, hdl);
}
callee_.send_buffered_messages(ctx, hdr.source_node, hdl);
break;
}
case message_type::dispatch_message: {
......
......@@ -142,17 +142,16 @@ strong_actor_ptr basp_broker_state::make_proxy(node_id nid, actor_id aid) {
// TODO: Can it happen that things have changed here?
lr = instance.tbl().lookup(nid);
if (lr.hdl) {
auto& ctx = *this_context;
// Tell remote side we are monitoring this actor now.
instance.write_announce_proxy(self->context(),
get_buffer(nid),
get_buffer(*lr.hdl),
nid, aid,
ctx.requires_ordering ? ctx.seq_outgoing++ : 0);
instance.flush(*lr.hdl);
visit(seq_num_visitor{this}, *lr.hdl));
flush(*lr.hdl);
} else {
instance.write_announce_proxy(self->context(),
get_buffer(nid),
nid, aid,0);
nid, aid, 0);
}
mm->notify<hook::new_remote_actor>(res);
return res;
......@@ -212,12 +211,10 @@ void basp_broker_state::send_kill_proxy_instance(const node_id& nid,
get_buffer(hdl),
nid, aid, rsn,
visit(seq_num_visitor{this}, hdl));
instance.flush(hdl);
flush(hdl);
} else {
buffer_type buf;
instance.write_kill_proxy(self->context(), buf,
instance.write_kill_proxy(self->context(), get_buffer(nid),
nid, aid, rsn, 0);
pending_connectivity[nid].emplace_back(std::move(buf));
}
}
......@@ -407,11 +404,9 @@ void basp_broker_state::learned_new_node(const node_id& nid) {
// 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(hdl), hdr, &writer);
instance.flush(hdl);
flush(hdl);
} else {
buffer_type buf;
instance.write(self->context(), buf, hdr, &writer);
pending_connectivity[nid].emplace_back(std::move(buf));
instance.write(self->context(), get_buffer(nid), hdr, &writer);
}
}
......@@ -460,7 +455,7 @@ void basp_broker_state::establish_communication(const node_id& nid) {
visit(seq_num_visitor{this}, hdl)};
instance.write(self->context(), get_buffer(hdl),
hdr, &writer);
instance.flush(hdl);
flush(hdl);
};
auto item = to_string(nid);
if (enable_tcp)
......@@ -590,8 +585,9 @@ bool basp_broker_state::deliver_pending(execution_unit* ctx,
std::vector<char>* payload = nullptr;
auto i = ep.pending.begin();
// Force delivery of at least the first messages, if desired.
if (force)
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;
......@@ -670,9 +666,8 @@ basp_broker_state::get_buffer(connection_handle hdl) {
basp_broker_state::buffer_type&
basp_broker_state::get_buffer(node_id nid) {
auto res = instance.tbl().lookup(nid);
if (res.known && res.hdl) {
if (res.known && res.hdl)
return get_buffer(*res.hdl);
}
auto& msgs = pending_connectivity[nid];
msgs.emplace_back();
return msgs.back();
......@@ -694,8 +689,9 @@ void basp_broker_state::flush(endpoint_handle hdl) {
}
void basp_broker_state::flush(datagram_handle hdl) {
if (!cached_buffers.empty() && !cached_buffers.top().empty())
if (!cached_buffers.empty() && !cached_buffers.top().empty()) {
self->enqueue_datagram(hdl, pop_datagram_buffer(hdl));
}
self->flush(hdl);
}
......@@ -874,11 +870,10 @@ behavior basp_broker::make_behavior() {
hdr.sequence_number = visit(seq_num_visitor{&state}, hdl);
state.instance.write(context(), state.get_buffer(hdl),
hdr, &writer);
state.instance.flush(hdl);
state.flush(hdl);
} else {
std::vector<char> buf;
state.instance.write(context(), buf, hdr, &writer);
state.pending_connectivity[dest_node].emplace_back(buf);
state.instance.write(context(), state.get_buffer(dest_node), hdr,
&writer);
}
return delegated<message>();
},
......
......@@ -71,7 +71,6 @@ behavior datagram_connection_broker(broker* self, uint16_t port,
};
}
bool establish_stream_connection(stateful_actor<connection_helper_state>* self,
const actor& b, uint16_t port,
network::address_listing& addresses) {
......
......@@ -182,7 +182,7 @@ void instance::write(execution_unit* ctx, instance::endpoint_handle hdl,
CAF_LOG_TRACE(CAF_ARG(hdr));
CAF_ASSERT(hdr.payload_len == 0 || writer != nullptr);
write(ctx, callee_.get_buffer(hdl), hdr, writer);
flush(hdl);
callee_.flush(hdl);
}
void instance::add_published_actor(uint16_t port,
......@@ -269,7 +269,7 @@ bool instance::dispatch(execution_unit* ctx, const strong_actor_ptr& sender,
auto hdl = std::move(*res.hdl);
hdr.sequence_number = visit(seq_num_visitor{callee_}, hdl);
write(ctx, callee_.get_buffer(hdl), hdr, &writer);
flush(hdl);
callee_.flush(hdl);
notify<hook::message_sent>(sender, receiver->node(), receiver, mid, msg);
return true;
} else {
......
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