Commit 0daae04f authored by Joseph Noir's avatar Joseph Noir

Update UDP handshake process

parent 38892b14
......@@ -124,6 +124,9 @@ public:
/// Handles a received datagram
bool handle(execution_unit* ctx, new_datagram_msg& dm, header& hdr);
/// Handles a new UDP endpoint msg
bool handle(execution_unit* ctx, new_endpoint_msg& em, header& hdr);
/// Sends heartbeat messages to all valid nodes those are directly connected.
void handle_heartbeat(execution_unit* ctx);
......
......@@ -34,7 +34,7 @@ namespace io {
using dgram_doorman_base = broker_servant<network::dgram_acceptor_manager,
dgram_doorman_handle,
new_endpoint_msg>;
dgram_dummy_msg>;
/// Manages reading from a datagram source
/// @ingroup Broker
......@@ -57,9 +57,7 @@ public:
using dgram_doorman_base::new_endpoint;
bool new_endpoint(execution_unit* ctx, dgram_scribe_handle endpoint);
bool delegate_msg(execution_unit* ctx, dgram_scribe_handle endpoint,
bool new_endpoint(execution_unit* ctx, dgram_scribe_handle endpoint,
const void* buf, size_t num_bytes);
// needs to be launched explicitly
......
......@@ -165,20 +165,24 @@ inspect(Inspector& f, dgram_doorman_closed_msg& x) {
return f(meta::type_name("dgram_doorman_closed_msg"), x.handle);
}
/// Signalizes newly arrived datagram for a {@link broker}.
/// Signalizes newly disvocerd remote endpoint
/// and the responsible scribe to a {@link broker}.
struct new_endpoint_msg {
/// Handle to the related datagram endpoint.
// Handle to the related datagram endpoint.
dgram_doorman_handle source;
// Buffer containing the received data.
// std::vector<char> buf;
/// Handle to new dgram_scribe
std::vector<char> buf;
// Handle to new dgram_scribe
dgram_scribe_handle handle;
// Port of the addressed actor
uint16_t port;
};
/// @relates new_endpoint_msg
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, new_endpoint_msg& x) {
return f(meta::type_name("new_endpoint_msg"), x.source, x.handle);
return f(meta::type_name("new_endpoint_msg"), x.source, x.buf,
x.handle, x.port);
}
/// Signalizes that a datagram with a certain size has been sent.
......@@ -188,8 +192,6 @@ struct new_datagram_msg {
// Buffer containing received data.
std::vector<char> buf;
// Port of the doorman that accepted the handshake
// TODO: get rid of this because it is only sometime used
optional<uint16_t> port;
};
/// @relates new_datagram_msg
......@@ -236,21 +238,17 @@ inspect(Inspector& f, dgram_doorman_passivated_msg& x) {
return f(meta::type_name("dgram_doorman_passivated_msg"), x.handle);
}
/// delegates a handshake to a dgram scribe
struct dgram_delegate_msg {
// Sender handle
dgram_doorman_handle source;
// Responsible scribe
dgram_scribe_handle handle;
// Buffer containing received data.
std::vector<char> buf;
/// Dummy message for the dgram_doorman implementation
struct dgram_dummy_msg {
dgram_doorman_handle handle;
std::vector<char> buffer;
};
/// @relates dgram_delegate_msg
/// @relates dgram_dummy_msg
template <class Inspector>
typename Inspector::result_type
inspect(Inspector& f, dgram_delegate_msg& x) {
return f(meta::type_name("dgram_delegate_msg"), x.handle);
inspect(Inspector& f, dgram_dummy_msg& x) {
return f(meta::type_name("dgram_dummy_msg"), x.handle, x.buffer);
}
} // namespace io
......
......@@ -671,17 +671,21 @@ behavior basp_broker::make_behavior() {
state.instance.remove_published_actor(port);
},
// received from underlying broker implementation
[=](const new_endpoint_msg& msg) {
[=](new_endpoint_msg& msg) {
CAF_LOG_TRACE(CAF_ARG(msg.handle));
static_cast<void>(msg);
// TODO: Do I need this or does the handshake procedure handle things?
},
/*
[=](const dgram_delegate_msg&) {
CAF_LOG_TRACE(CAF_ARG(msg.handle));
// TODO: Remove msg type?
state.set_context(msg.handle);
auto& ctx = *state.this_context;
auto is_open = state.instance.handle(context(), msg, ctx.hdr);
if (!is_open) {
if (ctx.callback) {
CAF_LOG_WARNING("failed to handshake with remote node"
<< CAF_ARG(msg.handle));
ctx.callback->deliver(make_error(sec::disconnect_during_handshake));
}
} else {
configure_datagram_size(msg.handle, 1500);
}
},
*/
// received from underlying broker implementation
[=](const dgram_scribe_closed_msg& msg) {
CAF_LOG_TRACE(CAF_ARG(msg.handle));
......
......@@ -1008,35 +1008,30 @@ default_multiplexer::add_dgram_doorman(abstract_broker* self,
// further activities for the broker
return false;
auto& dm = acceptor_.backend();
auto endpoint_info = acceptor_.last_sender();
//auto endpoint_info = acceptor_.last_sender();
auto fd = new_dgram_scribe_impl(acceptor_.host(),
acceptor_.port());
/*
{
std::string host;
uint16_t port;
std::tie(host,port) = sender_from_sockaddr(endpoint_info.first,
endpoint_info.second);
// std::cerr << "[DDM] New scribe with " << fd << " for " << host << ":" << port
// << "(or " << acceptor_.host() << ":" << acceptor_.port() << ")"
// << std::endl;
std::cerr << "[DDM] New scribe with " << fd << " for " << host
<< ":" << port << "(or " << acceptor_.host() << ":"
<< acceptor_.port() << ")" << std::endl;
}
*/
if (!fd) {
CAF_LOG_ERROR(CAF_ARG(fd.error()));
return false;
// return std::move(fd.error());
}
// auto endpoint_info = acceptor_.last_sender();
auto endpoint_info = acceptor_.last_sender();
auto hdl = dm.add_dgram_scribe(parent(), *fd,
endpoint_info.first, endpoint_info.second,
false);
// TODO: needs a better design
// new_endpoint(...) registeres the new handle
// delegate_message assigns the new handle
// responsibility for the received msg.
auto res = dgram_doorman::new_endpoint(&dm, hdl);
if (!res)
return false;
return delegate_msg(&dm, hdl, buf, num_bytes);
return dgram_doorman::new_endpoint(&dm, hdl, buf, num_bytes);
}
void stop_reading() override {
CAF_LOG_TRACE("");
......
......@@ -44,29 +44,21 @@ void dgram_doorman::io_failure(execution_unit* ctx, network::operation op) {
detach(ctx, true);
}
// bool dgram_doorman::new_endpoint(execution_unit* ctx, const void* buf,
// size_t besize) {
bool dgram_doorman::new_endpoint(execution_unit* ctx,
dgram_scribe_handle endpoint) {
msg().handle = endpoint;
return invoke_mailbox_element(ctx);
}
bool dgram_doorman::delegate_msg(execution_unit* ctx,
dgram_scribe_handle endpoint,
const void*, size_t num_bytes) {
CAF_LOG_TRACE(CAF_ARG(endpoint) << CAF_ARG(num_bytes));
if (detached())
return false;
using delegate_t = new_datagram_msg; // dgram_delegate_msg;
using tmp_t = mailbox_element_vals<delegate_t>;
using endpoint_t = new_endpoint_msg;
using tmp_t = mailbox_element_vals<endpoint_t>;
auto guard = parent_;
auto& buf = rd_buf();
CAF_ASSERT(buf.size() >= num_bytes);
buf.resize(num_bytes);
tmp_t tmp{strong_actor_ptr{}, message_id::make(),
mailbox_element::forwarding_stack{},
delegate_t{endpoint, std::move(buf),
endpoint_t{hdl(), std::move(buf), endpoint,
parent()->local_port(hdl())}};
invoke_mailbox_element_impl(ctx,tmp);
return true;
......
......@@ -163,12 +163,52 @@ bool instance::handle(execution_unit* ctx, new_datagram_msg& dm, header& hdr) {
// TODO: anything to do here? (i.e., do we still need forwarding?)
return err();
}
if (!handle_msg(ctx, dm.handle, hdr, payload, false, dm.port))
if (!handle_msg(ctx, dm.handle, hdr, payload, false, none))
return err();
} while (dgram_itr != dm.buf.end());
return true;
};
bool instance::handle(execution_unit* ctx, new_endpoint_msg& em, header& hdr) {
using itr_t = std::vector<char>::iterator;
// function object providing cleanup code on errors
auto err = [&]() -> bool {
auto cb = make_callback([&](const node_id& nid) -> error {
callee_.purge_state(nid);
return none;
});
tbl_.erase(em.handle, cb);
return false;
};
// extract payload
std::vector<char> pl_buf{std::move_iterator<itr_t>(std::begin(em.buf) +
basp::header_size),
std::move_iterator<itr_t>(std::end(em.buf))};
// resize header
em.buf.resize(basp::header_size);
// extract header
binary_deserializer bd{ctx, em.buf};
auto e = bd(hdr);
if (e || !valid(hdr)) {
CAF_LOG_WARNING("received invalid header:" << CAF_ARG(hdr));
std::cerr << "Received invalid header!" << std::endl;
return err();
}
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");
return err();
}
}
if (!handle_msg(ctx, em.handle, hdr, payload, false, em.port))
return err();
return true;
}
void instance::handle_heartbeat(execution_unit* ctx) {
CAF_LOG_TRACE("");
for (auto& kvp: tbl_.direct_by_hdl_) {
......
......@@ -200,11 +200,8 @@ expected<strong_actor_ptr> middleman::remote_actor(std::set<std::string> ifs,
if (!res)
return std::move(res.error());
strong_actor_ptr ptr = std::move(std::get<1>(*res));
if (!ptr) {
std::cerr << "[remote_actor] no actor published on port " << u.port_as_int()
<< std::endl;
if (!ptr)
return make_error(sec::no_actor_published_at_port, u.port_as_int());
}
if (!system().assignable(std::get<2>(*res), ifs))
return make_error(sec::unexpected_actor_messaging_interface, std::move(ifs),
std::move(std::get<2>(*res)));
......
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