Commit 1961b3fd authored by Joseph Noir's avatar Joseph Noir

Reworkign UDP handshake

parent e6a8e04f
...@@ -203,7 +203,8 @@ public: ...@@ -203,7 +203,8 @@ public:
/// Answer client handshake ... TODO /// Answer client handshake ... TODO
void write_udp_server_handshake(execution_unit* ctx, buffer_type& buf, void write_udp_server_handshake(execution_unit* ctx, buffer_type& buf,
const node_id& remote_side); const node_id& remote_side,
optional<uint16_t> port);
/// 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,
......
...@@ -154,7 +154,7 @@ public: ...@@ -154,7 +154,7 @@ public:
expected<ActorHandle> remote_actor(uri u) { expected<ActorHandle> remote_actor(uri u) {
CAF_LOG_TRACE(CAF_ARG(u)); CAF_LOG_TRACE(CAF_ARG(u));
detail::type_list<ActorHandle> tk; detail::type_list<ActorHandle> tk;
auto x = remote_actor(system().message_types(tk), u); auto x = remote_actor(system().message_types(tk), std::move(u));
if (!x) if (!x)
return x.error(); return x.error();
CAF_ASSERT(x && *x); CAF_ASSERT(x && *x);
......
...@@ -585,7 +585,6 @@ behavior basp_broker::make_behavior() { ...@@ -585,7 +585,6 @@ behavior basp_broker::make_behavior() {
// received from underlying broker implementation // received from underlying broker implementation
[=](new_datagram_msg& msg) { [=](new_datagram_msg& msg) {
CAF_LOG_TRACE(CAF_ARG(msg.handle)); CAF_LOG_TRACE(CAF_ARG(msg.handle));
CAF_LOG_DEBUG("Received new_datagram_msg: " << CAF_ARG(msg));
state.set_context(msg.handle); state.set_context(msg.handle);
auto& ctx = *state.this_context; auto& ctx = *state.this_context;
// TODO: ordering // TODO: ordering
...@@ -603,7 +602,7 @@ behavior basp_broker::make_behavior() { ...@@ -603,7 +602,7 @@ behavior basp_broker::make_behavior() {
// TODO: if callback available, handle that // TODO: if callback available, handle that
} }
// TODO: Is there configuration necessary? // TODO: Is there configuration necessary?
// configure_datagram_size(...); configure_datagram_size(msg.handle, 1500);
} }
}, },
// received from proxy instances // received from proxy instances
...@@ -682,24 +681,19 @@ behavior basp_broker::make_behavior() { ...@@ -682,24 +681,19 @@ behavior basp_broker::make_behavior() {
// received from underlying broker implementation // received from underlying broker implementation
[=](const acceptor_closed_msg& msg) { [=](const acceptor_closed_msg& msg) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
std::cerr << "[BB] Received acceptor closed msg" << std::endl;
auto port = local_port(msg.handle); auto port = local_port(msg.handle);
state.instance.remove_published_actor(port); state.instance.remove_published_actor(port);
}, },
// received from underlying broker implementation // received from underlying broker implementation
[=](const new_endpoint_msg& msg) { [=](const new_endpoint_msg& msg) {
CAF_LOG_TRACE(CAF_ARG(msg.handle)); CAF_LOG_TRACE(CAF_ARG(msg.handle));
std::cerr << "[BB] new endpoint msg received" << std::endl;
//auto& bi = state.instance;
//bi.write_server_handshake(context(), wr_buf(msg.handle),
// local_port(msg.source));
static_cast<void>(msg); static_cast<void>(msg);
// flush(msg.handle); // TODO: Do I need this or does the handshake procedure handle things?
// TODO: is this right?
}, },
/* /*
[=](const dgram_delegate_msg&) { [=](const dgram_delegate_msg&) {
CAF_LOG_TRACE(CAF_ARG(msg.handle)); CAF_LOG_TRACE(CAF_ARG(msg.handle));
// TODO: Remove msg type?
}, },
*/ */
// received from underlying broker implementation // received from underlying broker implementation
...@@ -717,7 +711,6 @@ behavior basp_broker::make_behavior() { ...@@ -717,7 +711,6 @@ behavior basp_broker::make_behavior() {
}, },
[=](const dgram_doorman_closed_msg& msg) { [=](const dgram_doorman_closed_msg& msg) {
CAF_LOG_TRACE(CAF_ARG(msg.handle)); CAF_LOG_TRACE(CAF_ARG(msg.handle));
std::cerr << "[BB] Received dgram_doorman_closed_msg" << std::endl;
auto port = local_port(msg.handle); auto port = local_port(msg.handle);
state.instance.remove_published_actor(port); state.instance.remove_published_actor(port);
}, },
...@@ -777,8 +770,6 @@ behavior basp_broker::make_behavior() { ...@@ -777,8 +770,6 @@ behavior basp_broker::make_behavior() {
[=](connect_atom, dgram_scribe_handle hdl, [=](connect_atom, dgram_scribe_handle hdl,
const std::string& host, uint16_t port) { const std::string& host, uint16_t port) {
CAF_LOG_TRACE(CAF_ARG(hdl.id())); CAF_LOG_TRACE(CAF_ARG(hdl.id()));
std::cerr << "[BB] \"connect\" for dgram to "
<< host << ":" << port << std::endl;
auto rp = make_response_promise(); auto rp = make_response_promise();
auto res = assign_dgram_scribe(hdl, host, port); auto res = assign_dgram_scribe(hdl, host, port);
if (res) { if (res) {
...@@ -787,8 +778,6 @@ behavior basp_broker::make_behavior() { ...@@ -787,8 +778,6 @@ behavior basp_broker::make_behavior() {
ctx.remote_port = port; ctx.remote_port = port;
ctx.callback = rp; ctx.callback = rp;
auto& bi = state.instance; auto& bi = state.instance;
// bi.write_server_handshake(context(), wr_buf(hdl), port);
// bi.write_client_handshake(context(), wr_buf(hdl), local_port(hdl));
bi.write_udp_client_handshake(context(), wr_buf(hdl)); bi.write_udp_client_handshake(context(), wr_buf(hdl));
flush(hdl); flush(hdl);
configure_datagram_size(hdl, 1500); configure_datagram_size(hdl, 1500);
......
...@@ -1254,6 +1254,7 @@ bool send_datagram(size_t& result, native_socket fd, void* buf, size_t buf_len, ...@@ -1254,6 +1254,7 @@ bool send_datagram(size_t& result, native_socket fd, void* buf, size_t buf_len,
auto sres = ::sendto(fd, reinterpret_cast<socket_send_ptr>(buf), buf_len, auto sres = ::sendto(fd, reinterpret_cast<socket_send_ptr>(buf), buf_len,
no_sigpipe_flag, reinterpret_cast<sockaddr*>(&sa), no_sigpipe_flag, reinterpret_cast<sockaddr*>(&sa),
sa_len); sa_len);
/*
{ {
std::string host; std::string host;
uint16_t port; uint16_t port;
...@@ -1261,6 +1262,7 @@ bool send_datagram(size_t& result, native_socket fd, void* buf, size_t buf_len, ...@@ -1261,6 +1262,7 @@ bool send_datagram(size_t& result, native_socket fd, void* buf, size_t buf_len,
std::cerr << "[SD] Sent " << sres << " bytes to " std::cerr << "[SD] Sent " << sres << " bytes to "
<< host << ":" << port << std::endl; << host << ":" << port << std::endl;
} }
*/
if (is_error(sres, true)) { if (is_error(sres, true)) {
CAF_LOG_ERROR("sendto returned" << CAF_ARG(sres)); CAF_LOG_ERROR("sendto returned" << CAF_ARG(sres));
return false; return false;
...@@ -1276,13 +1278,15 @@ bool receive_datagram(size_t& result, native_socket fd, void* buf, size_t len, ...@@ -1276,13 +1278,15 @@ bool receive_datagram(size_t& result, native_socket fd, void* buf, size_t len,
auto sres = ::recvfrom(fd, buf, len, 0, auto sres = ::recvfrom(fd, buf, len, 0,
reinterpret_cast<struct sockaddr *>(&sa), reinterpret_cast<struct sockaddr *>(&sa),
&sa_len); &sa_len);
/*
{ {
std::string host; std::string host;
uint16_t port; uint16_t port;
std::tie(host,port) = sender_from_sockaddr(sa, sa_len); std::tie(host,port) = sender_from_sockaddr(sa, sa_len);
std::cerr << "[SD] Received " << sres << " bytes from " std::cerr << "[RD] Received " << sres << " bytes from "
<< host << ":" << port << std::endl; << host << ":" << port << std::endl;
} }
*/
// TODO: Check if sres > len and do some error handling ... // TODO: Check if sres > len and do some error handling ...
if (is_error(sres, true)) { if (is_error(sres, true)) {
CAF_LOG_ERROR("recvfrom returned" << CAF_ARG(sres)); CAF_LOG_ERROR("recvfrom returned" << CAF_ARG(sres));
...@@ -1665,7 +1669,7 @@ void dgram_communicator::handle_event(operation op) { ...@@ -1665,7 +1669,7 @@ void dgram_communicator::handle_event(operation op) {
passivate(); passivate();
return; return;
} }
std::cerr << "[C] Received " << rb << " bytes" << std::endl; //std::cerr << "[C] Received " << rb << " bytes" << std::endl;
// currently handles the change from acceptor // currently handles the change from acceptor
// to communicator, TODO: Find a better solution // to communicator, TODO: Find a better solution
// Either keep sending to the same endpoint, // Either keep sending to the same endpoint,
...@@ -1675,7 +1679,9 @@ void dgram_communicator::handle_event(operation op) { ...@@ -1675,7 +1679,9 @@ void dgram_communicator::handle_event(operation op) {
std::tie(host_, port_) = sender_from_sockaddr(sockaddr_, sockaddr_len_); std::tie(host_, port_) = sender_from_sockaddr(sockaddr_, sockaddr_len_);
remote_endpoint_addr_ = std::move(sockaddr_); remote_endpoint_addr_ = std::move(sockaddr_);
remote_endpoint_addr_len_ = sockaddr_len_; remote_endpoint_addr_len_ = sockaddr_len_;
sockaddr_len_ = 0; sockaddr_len_ = 0;
//std::cerr << "[C] Adapted new endpoint: " << host_ << ":" << port_
// << std::endl;
} }
if (rb > 0) { if (rb > 0) {
auto res = reader_->consume(&backend(), rd_buf_.data(), rb); auto res = reader_->consume(&backend(), rd_buf_.data(), rb);
...@@ -1695,6 +1701,8 @@ void dgram_communicator::handle_event(operation op) { ...@@ -1695,6 +1701,8 @@ void dgram_communicator::handle_event(operation op) {
writer_->io_failure(&backend(), operation::write); writer_->io_failure(&backend(), operation::write);
backend().del(operation::write, fd(), this); backend().del(operation::write, fd(), this);
} else if (wb > 0) { } else if (wb > 0) {
//std::cerr << "[C] Sent " << wb << " bytes to "
// << host_ << ":" << port_ << std::endl;
CAF_ASSERT(wb == wr_buf_.size()); CAF_ASSERT(wb == wr_buf_.size());
if (ack_writes_) if (ack_writes_)
writer_->datagram_sent(&backend(), wb); writer_->datagram_sent(&backend(), wb);
...@@ -1736,9 +1744,10 @@ void dgram_communicator::prepare_next_read() { ...@@ -1736,9 +1744,10 @@ void dgram_communicator::prepare_next_read() {
void dgram_communicator::set_endpoint_addr(const sockaddr_storage& sa, void dgram_communicator::set_endpoint_addr(const sockaddr_storage& sa,
size_t len, size_t len,
bool is_tmp_endpoint) { bool is_tmp_endpoint) {
waiting_for_remote_endpoint = !is_tmp_endpoint; waiting_for_remote_endpoint = is_tmp_endpoint;
remote_endpoint_addr_ = sa; remote_endpoint_addr_ = sa;
remote_endpoint_addr_len_ = len; remote_endpoint_addr_len_ = len;
std::tie(host_, port_) = sender_from_sockaddr(sa, len);
} }
dgram_acceptor::dgram_acceptor(default_multiplexer& backend_ref, dgram_acceptor::dgram_acceptor(default_multiplexer& backend_ref,
...@@ -1787,16 +1796,14 @@ void dgram_acceptor::handle_event(operation op) { ...@@ -1787,16 +1796,14 @@ void dgram_acceptor::handle_event(operation op) {
// --> create a new dgram_scribe // --> create a new dgram_scribe
// TODO: some idenfification to prevent arbitrary // TODO: some idenfification to prevent arbitrary
// message from creating endpoints? // message from creating endpoints?
rd_buf_.clear();
rd_buf_.resize(dgram_size_);
size_t rb; size_t rb;
memset(&sockaddr_,0, sockaddr_len_);
if (!receive_datagram(rb, fd(), rd_buf_.data(), rd_buf_.size(), if (!receive_datagram(rb, fd(), rd_buf_.data(), rd_buf_.size(),
sockaddr_, sockaddr_len_)) { sockaddr_, sockaddr_len_)) {
mgr_->io_failure(&backend(), operation::read); mgr_->io_failure(&backend(), operation::read);
passivate(); passivate();
return; return;
} }
//std::cerr << "[A] Received message with " << rb << " bytes" << std::endl;
bytes_read_ = rb; bytes_read_ = rb;
if (rb > 0) { if (rb > 0) {
std::tie(host_,port_) = sender_from_sockaddr(sockaddr_, sockaddr_len_); std::tie(host_,port_) = sender_from_sockaddr(sockaddr_, sockaddr_len_);
......
...@@ -91,14 +91,15 @@ bool udp_server_handshake_valid(const header& hdr) { ...@@ -91,14 +91,15 @@ bool udp_server_handshake_valid(const header& hdr) {
return valid(hdr.source_node) return valid(hdr.source_node)
&& valid(hdr.dest_node) && valid(hdr.dest_node)
&& hdr.source_node != hdr.dest_node && hdr.source_node != hdr.dest_node
&& zero(hdr.source_actor) //&& zero(hdr.source_actor)
&& zero(hdr.dest_actor) && zero(hdr.dest_actor)
&& zero(hdr.operation_data); && !zero(hdr.operation_data);
} }
bool udp_client_handshake_valid(const header& hdr) { bool udp_client_handshake_valid(const header& hdr) {
return valid(hdr.source_node) return valid(hdr.source_node)
&& !valid(hdr.dest_node) && !valid(hdr.dest_node)
&& zero(hdr.source_actor)
&& zero(hdr.dest_actor) && zero(hdr.dest_actor)
&& !zero(hdr.operation_data); && !zero(hdr.operation_data);
} }
......
...@@ -17,6 +17,8 @@ ...@@ -17,6 +17,8 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#include <iterator>
#include "caf/io/basp/instance.hpp" #include "caf/io/basp/instance.hpp"
#include "caf/streambuf.hpp" #include "caf/streambuf.hpp"
...@@ -258,6 +260,8 @@ connection_state instance::handle(execution_unit* ctx, ...@@ -258,6 +260,8 @@ connection_state instance::handle(execution_unit* ctx,
} }
bool instance::handle(execution_unit* ctx, new_datagram_msg& dm, header& hdr) { bool instance::handle(execution_unit* ctx, new_datagram_msg& dm, header& hdr) {
using itr_t = std::vector<char>::iterator;
// Call in case of an error
auto err = [&]() -> bool { auto err = [&]() -> bool {
auto cb = make_callback([&](const node_id& nid) -> error { auto cb = make_callback([&](const node_id& nid) -> error {
callee_.purge_state(nid); callee_.purge_state(nid);
...@@ -266,299 +270,222 @@ bool instance::handle(execution_unit* ctx, new_datagram_msg& dm, header& hdr) { ...@@ -266,299 +270,222 @@ bool instance::handle(execution_unit* ctx, new_datagram_msg& dm, header& hdr) {
tbl_.erase(dm.handle, cb); tbl_.erase(dm.handle, cb);
return false; return false;
}; };
std::cerr << "New datagram received!" << std::endl; // Split message into hdr and payload
static_cast<void>(ctx); auto dgram_itr = dm.buf.begin();
static_cast<void>(dm); do {
static_cast<void>(hdr); // extract header
std::vector<char> hdr_buf{std::begin(dm.buf), std::vector<char> hdr_buf{std::move_iterator<itr_t>(dgram_itr),
std::begin(dm.buf) + basp::header_size}; std::move_iterator<itr_t>(dgram_itr +
std::vector<char> pl_buf{std::begin(dm.buf) + basp::header_size, basp::header_size)};
std::end(dm.buf)}; dgram_itr += basp::header_size;
binary_deserializer bd{ctx, hdr_buf}; // deserialize header
auto e = bd(hdr); binary_deserializer bd{ctx, hdr_buf};
if (e || !valid(hdr)) { auto e = bd(hdr);
CAF_LOG_WARNING("received invalid header:" << CAF_ARG(hdr)); if (e || !valid(hdr)) {
std::cerr << "Received invalid header!" << std::endl; CAF_LOG_WARNING("received invalid header:" << CAF_ARG(hdr));
return err(); std::cerr << "Received invalid header!" << std::endl;
}
CAF_LOG_DEBUG(CAF_ARG(hdr));
std::vector<char>* payload = nullptr;
if (hdr.payload_len > 0) {
payload = &pl_buf;
if (payload->size() != hdr.payload_len) {
CAF_LOG_WARNING("received invalid payload");
std::cerr << "Received invalid payload" << std::endl;
return err(); return err();
} }
std::cerr << "Received " << payload->size() << " bytes payload." CAF_LOG_DEBUG(CAF_ARG(hdr));
<< std::endl; // extract payload
} std::vector<char> pl_buf{std::move_iterator<itr_t>(dgram_itr),
// TODO: handle payload std::move_iterator<itr_t>(dgram_itr +
// needs forwarding? hdr.payload_len)};
if (!is_handshake(hdr) && !is_heartbeat(hdr) && hdr.dest_node != this_node_) { dgram_itr += hdr.payload_len;
std::cerr << "[I] Needs forwarding?" << std::endl; // deserialize handshake
/* std::vector<char>* payload = nullptr;
CAF_LOG_DEBUG("forward message"); if (hdr.payload_len > 0) {
auto path = lookup(hdr.dest_node); payload = &pl_buf;
if (path) { //if (payload->size() != hdr.payload_len) {
binary_serializer bs{ctx, path->wr_buf}; // CAF_LOG_WARNING("received invalid payload");
auto e = bs(hdr); // // TODO: This kind of should not happen ...
if (e) // std::cerr << "Received invalid payload with size " << payload->size()
return err(); // << " although header says " << hdr.payload_len << std::endl;
if (payload) // return err();
bs.apply_raw(payload->size(), payload->data()); //}
tbl_.flush(*path);
notify<hook::message_forwarded>(hdr, payload);
} else {
CAF_LOG_INFO("cannot forward message, no route to destination");
if (hdr.source_node != this_node_) {
// TODO: signalize error back to sending node
auto reverse_path = lookup(hdr.source_node);
if (!reverse_path) {
CAF_LOG_WARNING("cannot send error message: no route to source");
} else {
CAF_LOG_WARNING("not implemented yet: signalize forward failure");
}
} else {
CAF_LOG_WARNING("lost packet with probably spoofed source");
}
notify<hook::message_forwarding_failed>(hdr, payload);
} }
return await_header; // needs forwarding?
*/ if (!is_handshake(hdr) && !is_heartbeat(hdr) && hdr.dest_node != this_node_) {
} std::cerr << "[I] Needs forwarding?" << std::endl;
// function object for checking payload validity // TODO: anything to do here? (i.e., do we still need forwarding?)
auto payload_valid = [&]() -> bool { return err();
return payload != nullptr && payload->size() == hdr.payload_len;
};
// handle message to ourselves
switch (hdr.operation) {
case message_type::udp_server_handshake: {
std::cerr << "[I] Received UDP server handshake" << std::endl;
break;
} }
case message_type::udp_client_handshake: { // function object for checking payload validity
std::cerr << "[I] Received UDP client handshake" << std::endl; auto payload_valid = [&]() -> bool {
// udp scribes start with a client handshake when sending the first return payload != nullptr && payload->size() == hdr.payload_len;
// message, should be answered by a server handshake! };
actor_id aid = invalid_actor_id; // handle message ourselves
std::set<std::string> sigs; switch (hdr.operation) {
if (payload_valid()) { case message_type::udp_server_handshake: {
binary_deserializer bd{ctx, *payload}; std::cerr << "[I] Received UDP server handshake" << std::endl;
std::string remote_appid; actor_id aid = invalid_actor_id;
auto e = bd(remote_appid); std::set<std::string> sigs;
if (e) if (payload_valid()) {
binary_deserializer bd{ctx, *payload};
std::string remote_appid;
auto e = bd(remote_appid);
if (e)
return err();
if (remote_appid !=
callee_.system().config().middleman_app_identifier) {
CAF_LOG_ERROR("app identifier mismatch");
return err();
}
e = bd(aid, sigs);
if (e)
return err();
} else {
CAF_LOG_ERROR("fail to receive the app identifier");
return err(); return err();
if (remote_appid != callee_.system().config().middleman_app_identifier) { }
CAF_LOG_ERROR("app identifier mismatch"); // close self connection after handshake is done
std::cerr << "app identifier mismatch" << std::endl; if (hdr.source_node == this_node_) {
CAF_LOG_INFO("close connection to self immediately");
callee_.finalize_handshake(hdr.source_node, aid, sigs);
return err(); return err();
} }
e = bd(aid, sigs); // close this connection if we already have a direct connection
if (e) if (tbl_.lookup_hdl(hdr.source_node)) {
CAF_LOG_INFO("close connection since we already have a "
"direct connection: " << CAF_ARG(hdr.source_node));
callee_.finalize_handshake(hdr.source_node, aid, sigs);
return err(); return err();
} else { }
CAF_LOG_ERROR("fail to receive the app identifier"); // add direct route to this node and remove any indirect entry
std::cerr << "fail to receive the app identifier" << std::endl; CAF_LOG_INFO("new direct connection:" << CAF_ARG(hdr.source_node));
return err(); tbl_.add(dm.handle, hdr.source_node);
} // auto was_indirect = tbl_.erase_indirect(hdr.source_node);
// close self connection after handshake is done // write handshake as client in response
if (hdr.source_node == this_node_) { auto path = tbl_.lookup(hdr.source_node);
CAF_LOG_INFO("close connection to self immediately"); if (!path) {
std::cerr << "close connection to self immediately" << std::endl; CAF_LOG_ERROR("no route to host after server handshake");
callee_.finalize_handshake(hdr.source_node, aid, sigs); return err();
return err(); }
} write_client_handshake(ctx, path->wr_buf, hdr.source_node);
// close this connection if we already have a direct connection callee_.learned_new_node_directly(hdr.source_node);
if (tbl_.lookup_hdl(hdr.source_node)) { std::cerr << "[I] Before finalize handshake" << std::endl;
CAF_LOG_INFO("close connection since we already have a "
"direct connection: " << CAF_ARG(hdr.source_node));
std::cerr << "close connection since we already have a "
"direct connection: " << to_string(hdr.source_node)
<< std::endl;
callee_.finalize_handshake(hdr.source_node, aid, sigs); callee_.finalize_handshake(hdr.source_node, aid, sigs);
return err(); std::cerr << "[I] After finalize handshake" << std::endl;
} flush(*path);
// add direct route to this node and remove any indirect entry break;
CAF_LOG_INFO("new direct connection:" << CAF_ARG(hdr.source_node));
std::cerr << "new direct connection: " << to_string(hdr.source_node)
<< std::endl;
tbl_.add(dm.handle, hdr.source_node);
// auto was_indirect = tbl_.erase_indirect(hdr.source_node);
// write handshake as client in response
auto path = tbl_.lookup(hdr.source_node);
if (!path) {
CAF_LOG_ERROR("no route to host after server handshake");
return err();
} }
write_udp_server_handshake(ctx, path->wr_buf, hdr.source_node); case message_type::udp_client_handshake: {
callee_.learned_new_node_directly(hdr.source_node); std::cerr << "[I] Received UDP client handshake" << std::endl;
callee_.finalize_handshake(hdr.source_node, aid, sigs); if (tbl_.lookup_hdl(hdr.source_node)) {
flush(*path); CAF_LOG_INFO("received second client handshake:"
break; << CAF_ARG(hdr.source_node));
} break;
case message_type::server_handshake: { }
std::cerr << "[I] message_type::server_handshake" << std::endl; if (payload_valid()) {
/* binary_deserializer bd{ctx, *payload};
actor_id aid = invalid_actor_id; std::string remote_appid;
std::set<std::string> sigs; auto e = bd(remote_appid);
if (payload_valid()) { if (e)
binary_deserializer bd{ctx, *payload}; return err();
std::string remote_appid; if (remote_appid !=
auto e = bd(remote_appid); callee_.system().config().middleman_app_identifier) {
if (e) CAF_LOG_ERROR("app identifier mismatch");
return err(); return err();
if (remote_appid != callee_.system().config().middleman_app_identifier) { }
CAF_LOG_ERROR("app identifier mismatch"); } else {
CAF_LOG_ERROR("fail to receive the app identifier");
return err(); return err();
} }
e = bd(aid, sigs); // add direct route to this node and remove any indirect entry
if (e) CAF_LOG_INFO("new direct connection:" << CAF_ARG(hdr.source_node));
tbl_.add(dm.handle, hdr.source_node);
// write handshake as server in response
auto path = tbl_.lookup(hdr.source_node);
if (!path) {
CAF_LOG_ERROR("no route to host after server handshake");
return err(); return err();
} else { }
CAF_LOG_ERROR("fail to receive the app identifier"); write_udp_server_handshake(ctx, path->wr_buf, hdr.source_node, none);
return err(); // auto was_indirect = tbl_.erase_indirect(hdr.source_node);
} callee_.learned_new_node_directly(hdr.source_node);
// close self connection after handshake is done break;
if (hdr.source_node == this_node_) {
CAF_LOG_INFO("close connection to self immediately");
callee_.finalize_handshake(hdr.source_node, aid, sigs);
return err();
}
// close this connection if we already have a direct connection
if (tbl_.lookup_hdl(hdr.source_node)) {
CAF_LOG_INFO("close connection since we already have a "
"direct connection: " << CAF_ARG(hdr.source_node));
callee_.finalize_handshake(hdr.source_node, aid, sigs);
return err();
} }
// add direct route to this node and remove any indirect entry case message_type::server_handshake: {
CAF_LOG_INFO("new direct connection:" << CAF_ARG(hdr.source_node)); std::cerr << "[I] message_type::server_handshake" << std::endl;
tbl_.add(dm.handle, hdr.source_node); break;
// auto was_indirect = tbl_.erase_indirect(hdr.source_node);
// write handshake as client in response
auto path = tbl_.lookup(hdr.source_node);
if (!path) {
CAF_LOG_ERROR("no route to host after server handshake");
return err();
} }
write_client_handshake(ctx, path->wr_buf, hdr.source_node); case message_type::client_handshake: {
callee_.learned_new_node_directly(hdr.source_node); std::cerr << "[I] Received TCP client handshake" << std::endl;
callee_.finalize_handshake(hdr.source_node, aid, sigs);
flush(*path);
*/
break;
}
case message_type::client_handshake: {
std::cerr << "[I] message_type::client_handshake" << std::endl;
/*
if (tbl_.lookup_hdl(hdr.source_node)) {
CAF_LOG_INFO("received second client handshake:"
<< CAF_ARG(hdr.source_node));
break; break;
} }
if (payload_valid()) { case message_type::dispatch_message: {
std::cerr << "[I] Received dispatch message" << std::endl;
if (!payload_valid())
return err();
// in case the sender of this message was received via a third node,
// we assume that that node to offers a route to the original source
auto last_hop = tbl_.lookup_node(dm.handle);
// TODO: if we don't need this anymore, what does the rest do here?
//if (hdr.source_node != none
// && hdr.source_node != this_node_
// && last_hop != hdr.source_node
// && tbl_.lookup_hdl(hdr.source_node))
// && tbl_.add_indirect(last_hop, hdr.source_node))
// callee_.learned_new_node_indirectly(hdr.source_node);
binary_deserializer bd{ctx, *payload}; binary_deserializer bd{ctx, *payload};
std::string remote_appid; auto receiver_name = static_cast<atom_value>(0);
auto e = bd(remote_appid); std::vector<strong_actor_ptr> forwarding_stack;
message msg;
if (hdr.has(header::named_receiver_flag)) {
auto e = bd(receiver_name);
if (e)
return err();
}
auto e = bd(forwarding_stack, msg);
if (e) if (e)
return err(); return err();
if (remote_appid != callee_.system().config().middleman_app_identifier) { CAF_LOG_DEBUG(CAF_ARG(forwarding_stack) << CAF_ARG(msg));
CAF_LOG_ERROR("app identifier mismatch"); if (hdr.has(header::named_receiver_flag))
return err(); callee_.deliver(hdr.source_node, hdr.source_actor, receiver_name,
} message_id::from_integer_value(hdr.operation_data),
} else { forwarding_stack, msg);
CAF_LOG_ERROR("fail to receive the app identifier"); else
return err(); callee_.deliver(hdr.source_node, hdr.source_actor, hdr.dest_actor,
message_id::from_integer_value(hdr.operation_data),
forwarding_stack, msg);
//return err();
break;
} }
// add direct route to this node and remove any indirect entry case message_type::announce_proxy:
CAF_LOG_INFO("new direct connection:" << CAF_ARG(hdr.source_node)); std::cerr << "[I] Received announce proxy message" << std::endl;
tbl_.add(dm.handle, hdr.source_node); callee_.proxy_announced(hdr.source_node, hdr.dest_actor);
// auto was_indirect = tbl_.erase_indirect(hdr.source_node); break;
callee_.learned_new_node_directly(hdr.source_node); case message_type::kill_proxy: {
*/ std::cerr << "[I] message_type::kill_proxy" << std::endl;
break; if (!payload_valid())
} return err();
case message_type::dispatch_message: { binary_deserializer bd{ctx, *payload};
std::cerr << "[I] message_type::dispatch_message" << std::endl; error fail_state;
/* auto e = bd(fail_state);
if (!payload_valid())
return err();
// in case the sender of this message was received via a third node,
// we assume that that node to offers a route to the original source
auto last_hop = tbl_.lookup_node(dm.handle);
// TODO: if we don't need this anymore, what does the rest do here?
//if (hdr.source_node != none
// && hdr.source_node != this_node_
// && last_hop != hdr.source_node
// && tbl_.lookup_hdl(hdr.source_node))
// && tbl_.add_indirect(last_hop, hdr.source_node))
// callee_.learned_new_node_indirectly(hdr.source_node);
binary_deserializer bd{ctx, *payload};
auto receiver_name = static_cast<atom_value>(0);
std::vector<strong_actor_ptr> forwarding_stack;
message msg;
if (hdr.has(header::named_receiver_flag)) {
auto e = bd(receiver_name);
if (e) if (e)
return err(); return err();
callee_.kill_proxy(hdr.source_node, hdr.source_actor, fail_state);
break;
} }
auto e = bd(forwarding_stack, msg); case message_type::heartbeat: {
if (e) std::cerr << "[I] Received heartbeat message" << std::endl;
return err(); CAF_LOG_TRACE("received heartbeat: " << CAF_ARG(hdr.source_node));
CAF_LOG_DEBUG(CAF_ARG(forwarding_stack) << CAF_ARG(msg)); std::cerr << "[I] Received hearbeat: " << to_string(hdr.source_node)
if (hdr.has(header::named_receiver_flag)) << std::endl;
callee_.deliver(hdr.source_node, hdr.source_actor, receiver_name, callee_.handle_heartbeat(hdr.source_node);
message_id::from_integer_value(hdr.operation_data), //return true;
forwarding_stack, msg); break;
else }
callee_.deliver(hdr.source_node, hdr.source_actor, hdr.dest_actor, default:
message_id::from_integer_value(hdr.operation_data), CAF_LOG_ERROR("invalid operation");
forwarding_stack, msg); std::cerr << "[I] Invalid operation" << std::endl;
*/
return err();
break;
}
case message_type::announce_proxy:
std::cerr << "[I] message_type::announce_proxy" << std::endl;
/*
callee_.proxy_announced(hdr.source_node, hdr.dest_actor);
*/
break;
case message_type::kill_proxy: {
std::cerr << "[I] message_type::kill_proxy" << std::endl;
/*
if (!payload_valid())
return err();
binary_deserializer bd{ctx, *payload};
error fail_state;
auto e = bd(fail_state);
if (e)
return err(); return err();
callee_.kill_proxy(hdr.source_node, hdr.source_actor, fail_state);
*/
break;
}
case message_type::heartbeat: {
std::cerr << "[I] message_type::heartbeat" << std::endl;
/*
CAF_LOG_TRACE("received heartbeat: " << CAF_ARG(hdr.source_node));
std::cerr << "[I] Received hearbeat: " << to_string(hdr.source_node)
<< std::endl;
callee_.handle_heartbeat(hdr.source_node);
*/
return true;
} }
default: } while (dgram_itr != dm.buf.end());
CAF_LOG_ERROR("invalid operation");
std::cerr << "[I] Invalid operation" << std::endl;
return err();
}
// TODO: Is this reachable? // TODO: Is this reachable?
return err(); return true;
}; };
void instance::handle_heartbeat(execution_unit* ctx) { void instance::handle_heartbeat(execution_unit* ctx) {
...@@ -679,9 +606,11 @@ bool instance::dispatch(execution_unit* ctx, const strong_actor_ptr& sender, ...@@ -679,9 +606,11 @@ 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 << "In instance::write" << std::endl; //std::cerr << "[W] Writing " << buf.size() << " bytes in "
// << to_string(hdr.operation) << " message" << std::endl;
error err; error err;
if (pw) { if (pw) {
//std::cerr << "[W] Has payload writer" << std::endl;
auto pos = buf.size(); auto pos = buf.size();
// write payload first (skip first 72 bytes and write header later) // write payload first (skip first 72 bytes and write header later)
char placeholder[basp::header_size]; char placeholder[basp::header_size];
...@@ -715,11 +644,6 @@ void instance::write_server_handshake(execution_unit* ctx, ...@@ -715,11 +644,6 @@ void instance::write_server_handshake(execution_unit* ctx,
pa = &i->second; pa = &i->second;
} }
CAF_LOG_DEBUG_IF(!pa && port, "no actor published"); CAF_LOG_DEBUG_IF(!pa && port, "no actor published");
if (!pa && port) {
std::cerr << "No actor published." << std::endl;
} else {
std::cerr << "Found locally published actor." << std::endl;
}
auto writer = make_callback([&](serializer& sink) -> error { auto writer = make_callback([&](serializer& sink) -> error {
auto& ref = callee_.system().config().middleman_app_identifier; auto& ref = callee_.system().config().middleman_app_identifier;
auto e = sink(const_cast<std::string&>(ref)); auto e = sink(const_cast<std::string&>(ref));
...@@ -757,6 +681,13 @@ void instance::write_client_handshake(execution_unit* ctx, ...@@ -757,6 +681,13 @@ void instance::write_client_handshake(execution_unit* ctx,
void instance::write_udp_client_handshake(execution_unit* ctx, void instance::write_udp_client_handshake(execution_unit* ctx,
buffer_type& buf) { buffer_type& buf) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
auto writer = make_callback([&](serializer& sink) -> error {
auto& str = callee_.system().config().middleman_app_identifier;
return sink(const_cast<std::string&>(str));
});
header hdr{message_type::udp_client_handshake, 0, 0, version,
this_node_, none, invalid_actor_id, invalid_actor_id};
write(ctx, buf, hdr, &writer);
/* /*
CAF_LOG_TRACE(CAF_ARG(port)); CAF_LOG_TRACE(CAF_ARG(port));
using namespace detail; using namespace detail;
...@@ -772,7 +703,6 @@ void instance::write_udp_client_handshake(execution_unit* ctx, ...@@ -772,7 +703,6 @@ void instance::write_udp_client_handshake(execution_unit* ctx,
} else { } else {
std::cerr << "Found locally published actor." << std::endl; std::cerr << "Found locally published actor." << std::endl;
} }
*/
auto writer = make_callback([&](serializer& sink) -> error { auto writer = make_callback([&](serializer& sink) -> error {
auto& ref = callee_.system().config().middleman_app_identifier; auto& ref = callee_.system().config().middleman_app_identifier;
auto e = sink(const_cast<std::string&>(ref)); auto e = sink(const_cast<std::string&>(ref));
...@@ -790,11 +720,47 @@ void instance::write_udp_client_handshake(execution_unit* ctx, ...@@ -790,11 +720,47 @@ void instance::write_udp_client_handshake(execution_unit* ctx,
header hdr{message_type::udp_client_handshake, 0, 0, version, header hdr{message_type::udp_client_handshake, 0, 0, version,
this_node_, none, invalid_actor_id, invalid_actor_id}; this_node_, none, invalid_actor_id, invalid_actor_id};
write(ctx, buf, hdr, &writer); write(ctx, buf, hdr, &writer);
*/
} }
void instance::write_udp_server_handshake(execution_unit* ctx, void instance::write_udp_server_handshake(execution_unit* ctx,
buffer_type& buf, buffer_type& buf,
const node_id& remote_side) { const node_id& remote_side,
optional<uint16_t> port) {
CAF_LOG_TRACE(CAF_ARG(port));
using namespace detail;
published_actor* pa = nullptr;
if (port) {
auto i = published_actors_.find(*port);
if (i != published_actors_.end())
pa = &i->second;
}
CAF_LOG_DEBUG_IF(!pa && port, "no actor published");
if (!pa && port) {
std::cerr << "No actor published." << std::endl;
} else {
std::cerr << "Found locally published actor." << std::endl;
}
auto writer = make_callback([&](serializer& sink) -> error {
auto& ref = callee_.system().config().middleman_app_identifier;
auto e = sink(const_cast<std::string&>(ref));
if (e)
return e;
if (pa) {
auto i = pa->first ? pa->first->id() : invalid_actor_id;
return sink(i, pa->second);
} else {
auto aid = invalid_actor_id;
std::set<std::string> tmp;
return sink(aid, tmp);
}
});
header hdr{message_type::udp_server_handshake, 0, 0, version,
this_node_, remote_side,
pa && pa->first ? pa->first->id() : invalid_actor_id,
invalid_actor_id};
write(ctx, buf, hdr, &writer);
/*
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;
...@@ -803,6 +769,7 @@ void instance::write_udp_server_handshake(execution_unit* ctx, ...@@ -803,6 +769,7 @@ void instance::write_udp_server_handshake(execution_unit* ctx,
header hdr{message_type::udp_server_handshake, 0, 0, 0, header hdr{message_type::udp_server_handshake, 0, 0, 0,
this_node_, remote_side, invalid_actor_id, invalid_actor_id}; this_node_, remote_side, invalid_actor_id, invalid_actor_id};
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,
......
...@@ -33,7 +33,9 @@ const char* message_type_strings[] = { ...@@ -33,7 +33,9 @@ const char* message_type_strings[] = {
"dispatch_message", "dispatch_message",
"announce_proxy_instance", "announce_proxy_instance",
"kill_proxy_instance", "kill_proxy_instance",
"heartbeat" "heartbeat",
"udp_client_handshake",
"udp_server_handshake"
}; };
} // namespace <anonymous> } // namespace <anonymous>
......
...@@ -196,12 +196,15 @@ expected<strong_actor_ptr> middleman::remote_actor(std::set<std::string> ifs, ...@@ -196,12 +196,15 @@ expected<strong_actor_ptr> middleman::remote_actor(std::set<std::string> ifs,
uri u) { uri u) {
CAF_LOG_TRACE(CAF_ARG(ifs) << CAF_ARG(u)); CAF_LOG_TRACE(CAF_ARG(ifs) << CAF_ARG(u));
auto f = make_function_view(actor_handle()); auto f = make_function_view(actor_handle());
auto res = f(connect_atom::value, std::move(u)); auto res = f(connect_atom::value, u);
if (!res) if (!res)
return std::move(res.error()); return std::move(res.error());
strong_actor_ptr ptr = std::move(std::get<1>(*res)); strong_actor_ptr ptr = std::move(std::get<1>(*res));
if (!ptr) if (!ptr) {
std::cerr << "[remote_actor] no actor published on port " << u.port_as_int()
<< std::endl;
return make_error(sec::no_actor_published_at_port, u.port_as_int()); return make_error(sec::no_actor_published_at_port, u.port_as_int());
}
if (!system().assignable(std::get<2>(*res), ifs)) if (!system().assignable(std::get<2>(*res), ifs))
return make_error(sec::unexpected_actor_messaging_interface, std::move(ifs), return make_error(sec::unexpected_actor_messaging_interface, std::move(ifs),
std::move(std::get<2>(*res))); std::move(std::get<2>(*res)));
......
...@@ -108,7 +108,7 @@ public: ...@@ -108,7 +108,7 @@ public:
mpi_set sigs; mpi_set sigs;
return put(u, whom, sigs, reuse); return put(u, whom, sigs, reuse);
}, },
[=](connect_atom, uri& u) -> get_res { [=](connect_atom, uri u) -> get_res {
CAF_LOG_TRACE(CAF_ARG(u)); CAF_LOG_TRACE(CAF_ARG(u));
auto rp = make_response_promise(); auto rp = make_response_promise();
std::string host(u.host().first, u.host().second); std::string host(u.host().first, u.host().second);
......
...@@ -122,12 +122,12 @@ CAF_TEST(test_datagram_remote_actor) { ...@@ -122,12 +122,12 @@ CAF_TEST(test_datagram_remote_actor) {
auto res3 = io::uri::make(string(uri_no_port) + ":" + to_string(*res2)); auto res3 = io::uri::make(string(uri_no_port) + ":" + to_string(*res2));
CAF_CHECK(res3); CAF_CHECK(res3);
CAF_MESSAGE("Published pong on: " + to_string(*res3) + "."); CAF_MESSAGE("Published pong on: " + to_string(*res3) + ".");
/*
CAF_SET_LOGGER_SYS(&server_side); CAF_SET_LOGGER_SYS(&server_side);
//CAF_MESSAGE("Local call to remote actor should acquire the actor."); CAF_MESSAGE("Local call to remote actor should acquire the actor.");
//auto res4 = server_side_mm.remote_actor(*res3); auto res4 = server_side_mm.remote_actor(*res3);
//CAF_CHECK(res4); CAF_CHECK(res4);
*/
CAF_MESSAGE("Checking from different actor system next."); CAF_MESSAGE("Checking from different actor system next.");
auto res5 = client_side_mm.remote_actor(*res3); auto res5 = client_side_mm.remote_actor(*res3);
CAF_CHECK(res5); CAF_CHECK(res5);
......
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