Commit fe9b3ed2 authored by Joseph Noir's avatar Joseph Noir

Add sequence numbers for ordering to basp messages

Work for messages in right order, next up: add out of order delivery
to test multiplexer and ensure buffering and delviery for unordered
messages.
parent 6374186c
...@@ -104,6 +104,12 @@ public: ...@@ -104,6 +104,12 @@ public:
return namespace_.system(); return namespace_.system();
} }
// return the next outgoing sequence number for a connection
virtual uint16_t next_sequence_number(connection_handle hdl) = 0;
// return the next outgoing sequence number for an endpoint
virtual uint16_t next_sequence_number(dgram_scribe_handle hdl) = 0;
protected: protected:
proxy_registry namespace_; proxy_registry namespace_;
}; };
...@@ -197,25 +203,30 @@ public: ...@@ -197,25 +203,30 @@ public:
/// actor published at `port` to `buf`. If `port == none` or /// actor published at `port` to `buf`. If `port == none` or
/// if no actor is published at this port then a standard handshake is /// if no actor is published at this port then a standard handshake is
/// written (e.g. used when establishing direct connections on-the-fly). /// written (e.g. used when establishing direct connections on-the-fly).
void write_server_handshake(execution_unit* ctx, void write_server_handshake(execution_unit* ctx, buffer_type& buf,
buffer_type& buf, optional<uint16_t> port); optional<uint16_t> port,
uint16_t sequence_number = 0);
/// Writes the client handshake to `buf`. /// Writes the client handshake to `buf`.
void write_client_handshake(execution_unit* ctx, void write_client_handshake(execution_unit* ctx, buffer_type& buf,
buffer_type& buf, const node_id& remote_side); const node_id& remote_side,
uint16_t sequence_number = 0);
/// Writes an `announce_proxy` to `buf`. /// Writes an `announce_proxy` to `buf`.
void write_announce_proxy(execution_unit* ctx, buffer_type& buf, void write_announce_proxy(execution_unit* ctx, buffer_type& buf,
const node_id& dest_node, actor_id aid); const node_id& dest_node, actor_id aid,
uint16_t sequence_number = 0);
/// Writes a `kill_proxy` to `buf`. /// Writes a `kill_proxy` to `buf`.
void write_kill_proxy(execution_unit* ctx, buffer_type& buf, void write_kill_proxy(execution_unit* ctx, buffer_type& buf,
const node_id& dest_node, actor_id aid, const node_id& dest_node, actor_id aid,
const error& fail_state); const error& fail_state,
uint16_t sequence_number = 0);
/// Writes a `heartbeat` to `buf`. /// Writes a `heartbeat` to `buf`.
void write_heartbeat(execution_unit* ctx, void write_heartbeat(execution_unit* ctx,
buffer_type& buf, const node_id& remote_side); buffer_type& buf, const node_id& remote_side,
uint16_t sequence_number = 0);
inline const node_id& this_node() const { inline const node_id& this_node() const {
return this_node_; return this_node_;
...@@ -234,9 +245,7 @@ public: ...@@ -234,9 +245,7 @@ public:
template <class Handle> template <class Handle>
bool handle_msg(execution_unit* ctx, const Handle& hdl, header& hdr, bool handle_msg(execution_unit* ctx, const Handle& hdl, header& hdr,
std::vector<char>* payload, bool tcp_based, std::vector<char>* payload, bool tcp_based,
optional<uint16_t> port) { optional<endpoint_context&> ep, optional<uint16_t> port) {
// std::cerr << "[MSG] From " << hdl.id() << " (" << to_string(hdr.operation)
// << ")" << std::endl;
auto payload_valid = [&]() -> bool { auto payload_valid = [&]() -> bool {
return payload != nullptr && payload->size() == hdr.payload_len; return payload != nullptr && payload->size() == hdr.payload_len;
}; };
...@@ -286,7 +295,8 @@ public: ...@@ -286,7 +295,8 @@ public:
return false; return false;
} }
if (tcp_based) if (tcp_based)
write_client_handshake(ctx, apply_visitor(wr_buf_, path->hdl), hdr.source_node); write_client_handshake(ctx, apply_visitor(wr_buf_, path->hdl),
hdr.source_node);
callee_.learned_new_node_directly(hdr.source_node); callee_.learned_new_node_directly(hdr.source_node);
callee_.finalize_handshake(hdr.source_node, aid, sigs); callee_.finalize_handshake(hdr.source_node, aid, sigs);
flush(*path); flush(*path);
...@@ -319,7 +329,8 @@ public: ...@@ -319,7 +329,8 @@ public:
tbl_.add(hdl, hdr.source_node); tbl_.add(hdl, hdr.source_node);
} }
if (!tcp_based) { if (!tcp_based) {
write_server_handshake(ctx, wr_buf_.ptr->wr_buf(hdl), port); auto seq = (ep && ep->requires_ordering) ? ep->seq_outgoing++ : 0;
write_server_handshake(ctx, wr_buf_.ptr->wr_buf(hdl), port, seq);
wr_buf_.ptr->flush(hdl); wr_buf_.ptr->flush(hdl);
} }
if (!is_known_node) { if (!is_known_node) {
...@@ -390,12 +401,22 @@ public: ...@@ -390,12 +401,22 @@ public:
} }
private: private:
struct sequence_number_visitor {
using result_type = uint16_t;
sequence_number_visitor(instance::callee& c) : cal{c} { }
template <class T>
result_type operator()(const T& hdl) {
return cal.next_sequence_number(hdl);
}
instance::callee& cal;
};
routing_table tbl_; routing_table tbl_;
published_actor_map published_actors_; published_actor_map published_actors_;
node_id this_node_; node_id this_node_;
callee& callee_; callee& callee_;
flush_visitor flush_; flush_visitor flush_;
wr_buf_visitor wr_buf_; wr_buf_visitor wr_buf_;
sequence_number_visitor seq_num_;
}; };
/// @} /// @}
......
...@@ -98,6 +98,13 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee { ...@@ -98,6 +98,13 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
// nop // nop
} }
// inherited from basp::instance::listener
uint16_t next_sequence_number(connection_handle hdl) override;
uint16_t next_sequence_number(dgram_scribe_handle hdl) override;
void purge(connection_handle hdl);
void purge(dgram_scribe_handle hdl);
void set_context(connection_handle hdl); void set_context(connection_handle hdl);
void set_context(dgram_scribe_handle hdl); void set_context(dgram_scribe_handle hdl);
...@@ -105,12 +112,24 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee { ...@@ -105,12 +112,24 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
struct purge_visitor { struct purge_visitor {
using result_type = void; using result_type = void;
purge_visitor(basp_broker_state* ptr) : state{ptr} { } purge_visitor(basp_broker_state* ptr) : state{ptr} { }
result_type operator()(const connection_handle& hdl); template <class T>
result_type operator()(const dgram_scribe_handle& hdl); result_type operator()(const T& hdl) {
return state->purge(hdl);
}
basp_broker_state* state;
};
struct sequence_number_visitor {
using result_type = uint16_t;
sequence_number_visitor(basp_broker_state* ptr) : state{ptr} { }
template <class T>
result_type operator()(const T& hdl) {
return state->next_sequence_number(hdl);
}
basp_broker_state* state; basp_broker_state* state;
}; };
purge_visitor purge_state_vis;
wr_buf_visitor wr_buf_vis; wr_buf_visitor wr_buf_vis;
purge_visitor purge_state_vis;
sequence_number_visitor seq_num_vis;
// pointer to ourselves // pointer to ourselves
broker* self; broker* self;
......
...@@ -48,6 +48,9 @@ struct endpoint_context { ...@@ -48,6 +48,9 @@ struct endpoint_context {
uint16_t remote_port; uint16_t remote_port;
// pending operations to be performed after handshake completed // pending operations to be performed after handshake completed
optional<response_promise> callback; optional<response_promise> callback;
// TODO: introduce some call to ask scribe for such info
// Should be configurable by policies in the future
bool requires_ordering;
uint16_t seq_incoming; uint16_t seq_incoming;
uint16_t seq_outgoing; uint16_t seq_outgoing;
std::unordered_map<uint16_t, std::pair<basp::header,std::vector<char>>> pending; std::unordered_map<uint16_t, std::pair<basp::header,std::vector<char>>> pending;
......
...@@ -50,8 +50,9 @@ const char* basp_broker_state::name = "basp_broker"; ...@@ -50,8 +50,9 @@ const char* basp_broker_state::name = "basp_broker";
basp_broker_state::basp_broker_state(broker* selfptr) basp_broker_state::basp_broker_state(broker* selfptr)
: basp::instance::callee(selfptr->system(), : basp::instance::callee(selfptr->system(),
static_cast<proxy_registry::backend&>(*this)), static_cast<proxy_registry::backend&>(*this)),
purge_state_vis(this),
wr_buf_vis(selfptr), wr_buf_vis(selfptr),
purge_state_vis(this),
seq_num_vis(this),
self(selfptr), self(selfptr),
instance(selfptr, *this) { instance(selfptr, *this) {
CAF_ASSERT(this_node() != none); CAF_ASSERT(this_node() != none);
...@@ -108,10 +109,12 @@ strong_actor_ptr basp_broker_state::make_proxy(node_id nid, actor_id aid) { ...@@ -108,10 +109,12 @@ strong_actor_ptr basp_broker_state::make_proxy(node_id nid, actor_id aid) {
CAF_LOG_INFO("successfully created proxy instance, " CAF_LOG_INFO("successfully created proxy instance, "
"write announce_proxy_instance:" "write announce_proxy_instance:"
<< CAF_ARG(nid) << CAF_ARG(aid)); << CAF_ARG(nid) << CAF_ARG(aid));
auto& ctx = *this_context;
// tell remote side we are monitoring this actor now // tell remote side we are monitoring this actor now
instance.write_announce_proxy(self->context(), instance.write_announce_proxy(self->context(),
apply_visitor(wr_buf_vis, this_context->hdl), apply_visitor(wr_buf_vis, ctx.hdl),
nid, aid); nid, aid,
ctx.requires_ordering ? ctx.seq_outgoing++ : 0);
instance.tbl().flush(*path); instance.tbl().flush(*path);
mm->notify<hook::new_remote_actor>(res); mm->notify<hook::new_remote_actor>(res);
return res; return res;
...@@ -175,7 +178,8 @@ void basp_broker_state::proxy_announced(const node_id& nid, actor_id aid) { ...@@ -175,7 +178,8 @@ void basp_broker_state::proxy_announced(const node_id& nid, actor_id aid) {
} }
instance.write_kill_proxy(self->context(), instance.write_kill_proxy(self->context(),
apply_visitor(wr_buf_vis, path->hdl), apply_visitor(wr_buf_vis, path->hdl),
nid, aid, rsn); nid, aid, rsn,
apply_visitor(seq_num_vis, path->hdl));
instance.tbl().flush(*path); instance.tbl().flush(*path);
}; };
auto ptr = actor_cast<strong_actor_ptr>(entry); auto ptr = actor_cast<strong_actor_ptr>(entry);
...@@ -353,7 +357,8 @@ void basp_broker_state::learned_new_node(const node_id& nid) { ...@@ -353,7 +357,8 @@ void basp_broker_state::learned_new_node(const node_id& nid) {
// send message to SpawnServ of remote node // send message to SpawnServ of remote node
basp::header hdr{basp::message_type::dispatch_message, basp::header hdr{basp::message_type::dispatch_message,
basp::header::named_receiver_flag, basp::header::named_receiver_flag,
0, 0, this_node(), nid, tmp.id(), invalid_actor_id}; 0, 0, this_node(), nid, tmp.id(), invalid_actor_id,
apply_visitor(seq_num_vis, path->hdl)};
// writing std::numeric_limits<actor_id>::max() is a hack to get // writing std::numeric_limits<actor_id>::max() is a hack to get
// this send-to-named-actor feature working with older CAF releases // this send-to-named-actor feature working with older CAF releases
instance.write(self->context(), apply_visitor(wr_buf_vis, path->hdl), hdr, instance.write(self->context(), apply_visitor(wr_buf_vis, path->hdl), hdr,
...@@ -444,12 +449,37 @@ void basp_broker_state::learned_new_node_indirectly(const node_id& nid) { ...@@ -444,12 +449,37 @@ void basp_broker_state::learned_new_node_indirectly(const node_id& nid) {
}); });
basp::header hdr{basp::message_type::dispatch_message, basp::header hdr{basp::message_type::dispatch_message,
basp::header::named_receiver_flag, basp::header::named_receiver_flag,
0, 0, this_node(), nid, tmp.id(), invalid_actor_id}; 0, 0, this_node(), nid, tmp.id(), invalid_actor_id,
apply_visitor(state.seq_num_vis, path->hdl)};
instance.write(self->context(), path->wr_buf, hdr, &writer); instance.write(self->context(), path->wr_buf, hdr, &writer);
instance.flush(*path); instance.flush(*path);
} }
*/ */
void basp_broker_state::purge(connection_handle h) {
auto i = tcp_ctx.find(h);
if (i != tcp_ctx.end()) {
auto& ref = i->second;
if (ref.callback) {
CAF_LOG_DEBUG("connection closed during handshake");
ref.callback->deliver(sec::disconnect_during_handshake);
}
tcp_ctx.erase(i);
}
}
void basp_broker_state::purge(dgram_scribe_handle h) {
auto i = udp_ctx.find(h);
if (i != udp_ctx.end()) {
auto& ref = i->second;
if (ref.callback) {
CAF_LOG_DEBUG("connection closed during handshake");
ref.callback->deliver(sec::disconnect_during_handshake);
}
udp_ctx.erase(i);
}
}
void basp_broker_state::set_context(connection_handle hdl) { void basp_broker_state::set_context(connection_handle hdl) {
CAF_LOG_TRACE(CAF_ARG(hdl)); CAF_LOG_TRACE(CAF_ARG(hdl));
auto i = tcp_ctx.find(hdl); auto i = tcp_ctx.find(hdl);
...@@ -463,7 +493,7 @@ void basp_broker_state::set_context(connection_handle hdl) { ...@@ -463,7 +493,7 @@ void basp_broker_state::set_context(connection_handle hdl) {
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, none, hdl, none, 0, none,
0, 0 false, 0, 0, {}
} }
).first; ).first;
} }
...@@ -484,37 +514,22 @@ void basp_broker_state::set_context(dgram_scribe_handle hdl) { ...@@ -484,37 +514,22 @@ void basp_broker_state::set_context(dgram_scribe_handle hdl) {
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, none, hdl, none, 0, none,
0, 0 true, 0, 0, {}
} }
).first; ).first;
} }
this_context = &i->second; this_context = &i->second;
} }
basp_broker_state::purge_visitor::result_type uint16_t basp_broker_state::next_sequence_number(connection_handle) {
basp_broker_state::purge_visitor::operator()(const connection_handle& h) { return 0;
auto i = state->tcp_ctx.find(h);
if (i != state->tcp_ctx.end()) {
auto& ref = i->second;
if (ref.callback) {
CAF_LOG_DEBUG("connection closed during handshake");
ref.callback->deliver(sec::disconnect_during_handshake);
}
state->tcp_ctx.erase(i);
}
} }
basp_broker_state::purge_visitor::result_type uint16_t basp_broker_state::next_sequence_number(dgram_scribe_handle hdl) {
basp_broker_state::purge_visitor::operator()(const dgram_scribe_handle& h) { auto i = udp_ctx.find(hdl);
auto i = state->udp_ctx.find(h); if (i != udp_ctx.end() && i->second.requires_ordering)
if (i != state->udp_ctx.end()) { return i->second.seq_outgoing++;
auto& ref = i->second; return 0;
if (ref.callback) {
CAF_LOG_DEBUG("connection closed during handshake");
ref.callback->deliver(sec::disconnect_during_handshake);
}
state->udp_ctx.erase(i);
}
} }
...@@ -607,8 +622,7 @@ behavior basp_broker::make_behavior() { ...@@ -607,8 +622,7 @@ behavior basp_broker::make_behavior() {
} }
if (src && system().node() == src->node()) if (src && system().node() == src->node())
system().registry().put(src->id(), src); system().registry().put(src->id(), src);
if (!state.instance.dispatch(context(), src, fwd_stack, if (!state.instance.dispatch(context(), src, fwd_stack, dest, mid, msg)
dest, mid, msg)
&& mid.is_request()) { && mid.is_request()) {
detail::sync_request_bouncer srb{exit_reason::remote_link_unreachable}; detail::sync_request_bouncer srb{exit_reason::remote_link_unreachable};
srb(src, mid); srb(src, mid);
...@@ -640,7 +654,8 @@ behavior basp_broker::make_behavior() { ...@@ -640,7 +654,8 @@ behavior basp_broker::make_behavior() {
basp::header hdr{basp::message_type::dispatch_message, basp::header hdr{basp::message_type::dispatch_message,
basp::header::named_receiver_flag, basp::header::named_receiver_flag,
0, cme->mid.integer_value(), state.this_node(), 0, cme->mid.integer_value(), state.this_node(),
dest_node, src->id(), invalid_actor_id}; dest_node, src->id(), invalid_actor_id,
apply_visitor(state.seq_num_vis, path->hdl)};
state.instance.write(context(), apply_visitor(wr_buf_, path->hdl), hdr, state.instance.write(context(), apply_visitor(wr_buf_, path->hdl), hdr,
&writer); &writer);
state.instance.flush(*path); state.instance.flush(*path);
...@@ -771,10 +786,17 @@ behavior basp_broker::make_behavior() { ...@@ -771,10 +786,17 @@ behavior basp_broker::make_behavior() {
ctx.hdl = hdl; ctx.hdl = hdl;
ctx.remote_port = port; ctx.remote_port = port;
ctx.callback = rp; ctx.callback = rp;
// assuming all UDP based protocol need ordering ...
// TODO: get this information from the policy
ctx.requires_ordering = true;
ctx.seq_incoming = 0;
ctx.seq_outgoing = 0;
auto& bi = state.instance; auto& bi = state.instance;
bi.write_client_handshake(context(), wr_buf(hdl), none); bi.write_client_handshake(context(), wr_buf(hdl), none,
ctx.seq_outgoing++);
flush(hdl); flush(hdl);
configure_datagram_size(hdl, 1500); configure_datagram_size(hdl, 1500);
// TODO: Set timeout for expected answer
} else { } else {
CAF_LOG_DEBUG("failed to assign datagram sink from handle" CAF_LOG_DEBUG("failed to assign datagram sink from handle"
<< CAF_ARG(res)); << CAF_ARG(res));
......
...@@ -21,6 +21,7 @@ ...@@ -21,6 +21,7 @@
#include "caf/io/basp/instance.hpp" #include "caf/io/basp/instance.hpp"
#include "caf/variant.hpp"
#include "caf/streambuf.hpp" #include "caf/streambuf.hpp"
#include "caf/binary_serializer.hpp" #include "caf/binary_serializer.hpp"
#include "caf/binary_deserializer.hpp" #include "caf/binary_deserializer.hpp"
...@@ -46,13 +47,13 @@ instance::instance(abstract_broker* parent, callee& lstnr) ...@@ -46,13 +47,13 @@ instance::instance(abstract_broker* parent, callee& lstnr)
this_node_(parent->system().node()), this_node_(parent->system().node()),
callee_(lstnr), callee_(lstnr),
flush_(parent), flush_(parent),
wr_buf_(parent) { wr_buf_(parent),
seq_num_(lstnr) {
CAF_ASSERT(this_node_ != none); CAF_ASSERT(this_node_ != none);
} }
connection_state instance::handle(execution_unit* ctx, connection_state instance::handle(execution_unit* ctx, new_data_msg& dm,
new_data_msg& dm, header& hdr, header& hdr, bool is_payload) {
bool is_payload) {
CAF_LOG_TRACE(CAF_ARG(dm) << CAF_ARG(is_payload)); CAF_LOG_TRACE(CAF_ARG(dm) << CAF_ARG(is_payload));
// function object providing cleanup code on errors // function object providing cleanup code on errors
auto err = [&]() -> connection_state { auto err = [&]() -> connection_state {
...@@ -113,7 +114,7 @@ connection_state instance::handle(execution_unit* ctx, ...@@ -113,7 +114,7 @@ connection_state instance::handle(execution_unit* ctx,
} }
return await_header; return await_header;
} }
if (!handle_msg(ctx, dm.handle, hdr, payload, true, none)) if (!handle_msg(ctx, dm.handle, hdr, payload, true, none, none))
return err(); return err();
return await_header; return await_header;
} }
...@@ -141,7 +142,7 @@ bool instance::handle(execution_unit* ctx, new_datagram_msg& dm, ...@@ -141,7 +142,7 @@ bool instance::handle(execution_unit* ctx, new_datagram_msg& dm,
auto e = bd(ep.hdr); auto e = bd(ep.hdr);
if (e || !valid(ep.hdr)) { if (e || !valid(ep.hdr)) {
CAF_LOG_WARNING("received invalid header:" << CAF_ARG(ep.hdr)); CAF_LOG_WARNING("received invalid header:" << CAF_ARG(ep.hdr));
std::cerr << "Received invalid header!" << std::endl; std::cerr << "[!!] invalid header!" << std::endl;
return err(); return err();
} }
CAF_LOG_DEBUG(CAF_ARG(ep.hdr)); CAF_LOG_DEBUG(CAF_ARG(ep.hdr));
...@@ -154,10 +155,35 @@ bool instance::handle(execution_unit* ctx, new_datagram_msg& dm, ...@@ -154,10 +155,35 @@ bool instance::handle(execution_unit* ctx, new_datagram_msg& dm,
} }
} }
// TODO: Ordering // 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;
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)));
return true;
}
ep.seq_incoming += 1;
// TODO: Reliability // TODO: Reliability
if (!handle_msg(ctx, dm.handle, ep.hdr, payload, false, none)) if (!handle_msg(ctx, dm.handle, ep.hdr, payload, false, ep, none))
return err(); return err();
auto itr = ep.pending.find(ep.seq_incoming);
while (itr != ep.pending.end()) {
ep.hdr = std::move(itr->second.first);
pl_buf = std::move(itr->second.second);
payload = &pl_buf;
if (!handle_msg(ctx, get<dgram_scribe_handle>(ep.hdl),
ep.hdr, payload, false, ep, none))
err();
ep.pending.erase(itr);
ep.seq_incoming += 1;
itr = ep.pending.find(ep.seq_incoming);
}
return true; return true;
}; };
...@@ -175,19 +201,31 @@ bool instance::handle(execution_unit* ctx, new_endpoint_msg& em, ...@@ -175,19 +201,31 @@ bool instance::handle(execution_unit* ctx, new_endpoint_msg& em,
return false; return false;
}; };
// extract payload // extract payload
std::vector<char> pl_buf{std::move_iterator<itr_t>(std::begin(em.buf) + std::vector<char> pl_buf{std::move_iterator<itr_t>(std::begin(em.buf)
basp::header_size), + basp::header_size),
std::move_iterator<itr_t>(std::end(em.buf))}; std::move_iterator<itr_t>(std::end(em.buf))};
// resize header // resize header
em.buf.resize(basp::header_size); em.buf.resize(basp::header_size);
// extract header // extract header
binary_deserializer bd{ctx, em.buf}; binary_deserializer bd{ctx, em.buf};
auto e = bd(ep.hdr); auto e = bd(ep.hdr);
// client handshake for UDP should be sequence number 0
if (e || !valid(ep.hdr)) { if (e || !valid(ep.hdr)) {
CAF_LOG_WARNING("received invalid header:" << CAF_ARG(ep.hdr)); CAF_LOG_WARNING("received invalid header:" << CAF_ARG(ep.hdr));
std::cerr << "Received invalid header!" << std::endl; std::cerr << "[<<] invalid header!" << std::endl;
return err(); return err();
} }
if (ep.hdr.sequence_number != ep.seq_incoming) {
CAF_LOG_WARNING("Handshake with unexected sequence number: "
<< CAF_ARG(ep.hdr.sequence_number));
std::cerr << "[<<] Unexpected sequence number '" << ep.hdr.sequence_number
<< "'in client handshake" << std::endl;
ep.seq_incoming = ep.hdr.sequence_number + 1;
} else {
std::cerr << "[<<] '" << to_string(ep.hdr.operation) << "' with seq '"
<< ep.hdr.sequence_number << "'" << std::endl;
ep.seq_incoming += 1;
}
CAF_LOG_DEBUG(CAF_ARG(ep.hdr)); CAF_LOG_DEBUG(CAF_ARG(ep.hdr));
std::vector<char>* payload = nullptr; std::vector<char>* payload = nullptr;
if (ep.hdr.payload_len > 0) { if (ep.hdr.payload_len > 0) {
...@@ -197,8 +235,9 @@ bool instance::handle(execution_unit* ctx, new_endpoint_msg& em, ...@@ -197,8 +235,9 @@ bool instance::handle(execution_unit* ctx, new_endpoint_msg& em,
return err(); return err();
} }
} }
if (!handle_msg(ctx, em.handle, ep.hdr, payload, false, em.port)) if (!handle_msg(ctx, em.handle, ep.hdr, payload, false, ep, em.port))
return err(); return err();
// TODO: Check for pending messages ...
return true; return true;
} }
...@@ -206,7 +245,8 @@ void instance::handle_heartbeat(execution_unit* ctx) { ...@@ -206,7 +245,8 @@ void instance::handle_heartbeat(execution_unit* ctx) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
for (auto& kvp: tbl_.direct_by_hdl_) { for (auto& kvp: tbl_.direct_by_hdl_) {
CAF_LOG_TRACE(CAF_ARG(kvp.first) << CAF_ARG(kvp.second)); CAF_LOG_TRACE(CAF_ARG(kvp.first) << CAF_ARG(kvp.second));
write_heartbeat(ctx, apply_visitor(wr_buf_, kvp.first), kvp.second); auto seq = apply_visitor(seq_num_, kvp.first);
write_heartbeat(ctx, apply_visitor(wr_buf_, kvp.first), kvp.second, seq);
apply_visitor(flush_, kvp.first); apply_visitor(flush_, kvp.first);
} }
} }
...@@ -309,7 +349,8 @@ bool instance::dispatch(execution_unit* ctx, const strong_actor_ptr& sender, ...@@ -309,7 +349,8 @@ bool instance::dispatch(execution_unit* ctx, const strong_actor_ptr& sender,
}); });
header hdr{message_type::dispatch_message, 0, 0, mid.integer_value(), header hdr{message_type::dispatch_message, 0, 0, mid.integer_value(),
sender ? sender->node() : this_node(), receiver->node(), sender ? sender->node() : this_node(), receiver->node(),
sender ? sender->id() : invalid_actor_id, receiver->id()}; sender ? sender->id() : invalid_actor_id, receiver->id(),
apply_visitor(seq_num_, path->hdl)};
write(ctx, apply_visitor(wr_buf_, path->hdl), hdr, &writer); write(ctx, apply_visitor(wr_buf_, path->hdl), hdr, &writer);
flush(*path); flush(*path);
notify<hook::message_sent>(sender, path->next_hop, receiver, mid, msg); notify<hook::message_sent>(sender, path->next_hop, receiver, mid, msg);
...@@ -319,8 +360,9 @@ bool instance::dispatch(execution_unit* ctx, const strong_actor_ptr& sender, ...@@ -319,8 +360,9 @@ bool instance::dispatch(execution_unit* ctx, const strong_actor_ptr& sender,
void instance::write(execution_unit* ctx, buffer_type& buf, void instance::write(execution_unit* ctx, buffer_type& buf,
header& hdr, payload_writer* pw) { header& hdr, payload_writer* pw) {
CAF_LOG_TRACE(CAF_ARG(hdr)); CAF_LOG_TRACE(CAF_ARG(hdr));
//std::cerr << "[W] Writing " << buf.size() << " bytes in " std::cerr << "[>>] '" << to_string(hdr.operation)
// << to_string(hdr.operation) << " message" << std::endl; << "' with seq '" << hdr.sequence_number << "'"
<< std::endl;
error err; error err;
if (pw) { if (pw) {
auto pos = buf.size(); auto pos = buf.size();
...@@ -334,8 +376,6 @@ void instance::write(execution_unit* ctx, buffer_type& buf, ...@@ -334,8 +376,6 @@ void instance::write(execution_unit* ctx, buffer_type& buf,
hdr.payload_len = static_cast<uint32_t>(plen); hdr.payload_len = static_cast<uint32_t>(plen);
stream_serializer<charbuf> out{ctx, buf.data() + pos, basp::header_size}; stream_serializer<charbuf> out{ctx, buf.data() + pos, basp::header_size};
err = out(hdr); err = out(hdr);
// std::cerr << "Wrote " << to_string(hdr.operation) << " with " << basp::header_size
// << " + " << hdr.payload_len << " bytes" << std::endl;
} else { } else {
binary_serializer bs{ctx, buf}; binary_serializer bs{ctx, buf};
err = bs(hdr); err = bs(hdr);
...@@ -347,7 +387,8 @@ void instance::write(execution_unit* ctx, buffer_type& buf, ...@@ -347,7 +387,8 @@ void instance::write(execution_unit* ctx, buffer_type& buf,
} }
void instance::write_server_handshake(execution_unit* ctx, buffer_type& buf, void instance::write_server_handshake(execution_unit* ctx, buffer_type& buf,
optional<uint16_t> port) { optional<uint16_t> port,
uint16_t sequence_number) {
CAF_LOG_TRACE(CAF_ARG(port)); CAF_LOG_TRACE(CAF_ARG(port));
using namespace detail; using namespace detail;
published_actor* pa = nullptr; published_actor* pa = nullptr;
...@@ -374,50 +415,54 @@ void instance::write_server_handshake(execution_unit* ctx, buffer_type& buf, ...@@ -374,50 +415,54 @@ void instance::write_server_handshake(execution_unit* ctx, buffer_type& buf,
header hdr{message_type::server_handshake, 0, 0, version, header hdr{message_type::server_handshake, 0, 0, version,
this_node_, none, this_node_, none,
pa && pa->first ? pa->first->id() : invalid_actor_id, pa && pa->first ? pa->first->id() : invalid_actor_id,
invalid_actor_id}; invalid_actor_id, sequence_number};
// std::cerr << "Writing server handshake, published actor "
// << (pa ? "found" : "unknwon") << std::endl;
write(ctx, buf, hdr, &writer); write(ctx, buf, hdr, &writer);
} }
void instance::write_client_handshake(execution_unit* ctx, buffer_type& buf, void instance::write_client_handshake(execution_unit* ctx, buffer_type& buf,
const node_id& remote_side) { const node_id& remote_side,
uint16_t sequence_number) {
CAF_LOG_TRACE(CAF_ARG(remote_side)); CAF_LOG_TRACE(CAF_ARG(remote_side));
auto writer = make_callback([&](serializer& sink) -> error { auto writer = make_callback([&](serializer& sink) -> error {
auto& str = callee_.system().config().middleman_app_identifier; auto& str = callee_.system().config().middleman_app_identifier;
return sink(const_cast<std::string&>(str)); return sink(const_cast<std::string&>(str));
}); });
header hdr{message_type::client_handshake, 0, 0, 0, header hdr{message_type::client_handshake, 0, 0, 0,
this_node_, remote_side, invalid_actor_id, invalid_actor_id}; this_node_, remote_side, invalid_actor_id, invalid_actor_id,
sequence_number};
write(ctx, buf, hdr, &writer); write(ctx, buf, hdr, &writer);
} }
void instance::write_announce_proxy(execution_unit* ctx, buffer_type& buf, void instance::write_announce_proxy(execution_unit* ctx, buffer_type& buf,
const node_id& dest_node, actor_id aid) { const node_id& dest_node, actor_id aid,
uint16_t sequence_number) {
CAF_LOG_TRACE(CAF_ARG(dest_node) << CAF_ARG(aid)); CAF_LOG_TRACE(CAF_ARG(dest_node) << CAF_ARG(aid));
header hdr{message_type::announce_proxy, 0, 0, 0, header hdr{message_type::announce_proxy, 0, 0, 0,
this_node_, dest_node, invalid_actor_id, aid}; this_node_, dest_node, invalid_actor_id, aid,
sequence_number};
write(ctx, buf, hdr); write(ctx, buf, hdr);
} }
void instance::write_kill_proxy(execution_unit* ctx, buffer_type& buf, void instance::write_kill_proxy(execution_unit* ctx, buffer_type& buf,
const node_id& dest_node, actor_id aid, const node_id& dest_node, actor_id aid,
const error& rsn) { const error& rsn, uint16_t sequence_number) {
CAF_LOG_TRACE(CAF_ARG(dest_node) << CAF_ARG(aid) << CAF_ARG(rsn)); CAF_LOG_TRACE(CAF_ARG(dest_node) << CAF_ARG(aid) << CAF_ARG(rsn));
auto writer = make_callback([&](serializer& sink) -> error { auto writer = make_callback([&](serializer& sink) -> error {
return sink(const_cast<error&>(rsn)); return sink(const_cast<error&>(rsn));
}); });
header hdr{message_type::kill_proxy, 0, 0, 0, header hdr{message_type::kill_proxy, 0, 0, 0,
this_node_, dest_node, aid, invalid_actor_id}; this_node_, dest_node, aid, invalid_actor_id,
sequence_number};
write(ctx, buf, hdr, &writer); write(ctx, buf, hdr, &writer);
} }
void instance::write_heartbeat(execution_unit* ctx, void instance::write_heartbeat(execution_unit* ctx, buffer_type& buf,
buffer_type& buf, const node_id& remote_side,
const node_id& remote_side) { uint16_t sequence_number) {
CAF_LOG_TRACE(CAF_ARG(remote_side)); CAF_LOG_TRACE(CAF_ARG(remote_side));
header hdr{message_type::heartbeat, 0, 0, 0, header hdr{message_type::heartbeat, 0, 0, 0,
this_node_, remote_side, invalid_actor_id, invalid_actor_id}; this_node_, remote_side, invalid_actor_id, invalid_actor_id,
sequence_number};
write(ctx, buf, hdr); write(ctx, buf, hdr);
} }
......
...@@ -1047,7 +1047,8 @@ CAF_TEST(client_handshake_and_dispatch_udp) { ...@@ -1047,7 +1047,8 @@ CAF_TEST(client_handshake_and_dispatch_udp) {
// send a message via `dispatch` from node 0 // send a message via `dispatch` from node 0
mock(jupiter().connection, mock(jupiter().connection,
{basp::message_type::dispatch_message, 0, 0, 0, {basp::message_type::dispatch_message, 0, 0, 0,
jupiter().id, this_node(), jupiter().dummy_actor->id(), self()->id()}, jupiter().id, this_node(), jupiter().dummy_actor->id(), self()->id(),
1},
std::vector<actor_addr>{}, std::vector<actor_addr>{},
make_message(1, 2, 3)) make_message(1, 2, 3))
.expect(jupiter().connection, .expect(jupiter().connection,
...@@ -1167,7 +1168,7 @@ CAF_TEST(remote_actor_and_send_udp) { ...@@ -1167,7 +1168,7 @@ CAF_TEST(remote_actor_and_send_udp) {
mock(jupiter().connection, mock(jupiter().connection,
{basp::message_type::dispatch_message, 0, 0, 0, {basp::message_type::dispatch_message, 0, 0, 0,
jupiter().id, this_node(), jupiter().id, this_node(),
jupiter().dummy_actor->id(), self()->id()}, jupiter().dummy_actor->id(), self()->id(), 1},
std::vector<actor_id>{}, std::vector<actor_id>{},
make_message("hi there!")); make_message("hi there!"));
self()->receive( self()->receive(
...@@ -1204,7 +1205,7 @@ CAF_TEST(actor_serialize_and_deserialize_udp) { ...@@ -1204,7 +1205,7 @@ CAF_TEST(actor_serialize_and_deserialize_udp) {
mock(jupiter().connection, mock(jupiter().connection,
{basp::message_type::dispatch_message, 0, 0, 0, {basp::message_type::dispatch_message, 0, 0, 0,
prx->node(), this_node(), prx->node(), this_node(),
prx->id(), testee->id()}, prx->id(), testee->id(), 1},
std::vector<actor_id>{}, std::vector<actor_id>{},
msg); msg);
// testee must've responded (process forwarded message in BASP broker) // testee must've responded (process forwarded message in BASP broker)
......
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