Commit e327c64e authored by Joseph Noir's avatar Joseph Noir

WIP Intergrating datagrams into the basp broker

parent 1fc1d5e5
......@@ -116,8 +116,11 @@ public:
/// Handles received data and returns a config for receiving the
/// next data or `none` if an error occured.
connection_state handle(execution_unit* ctx,
new_data_msg& dm, header& hdr, bool is_payload);
connection_state handle(execution_unit* ctx, new_data_msg& dm, header& hdr,
bool is_payload);
/// Handles a received datagram
bool handle(execution_unit* ctx, new_datagram_msg& dm, header& hdr);
/// Sends heartbeat messages to all valid nodes those are directly connected.
void handle_heartbeat(execution_unit* ctx);
......
......@@ -42,6 +42,7 @@ namespace caf {
namespace io {
struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
basp_broker_state(broker* self);
~basp_broker_state();
......@@ -94,31 +95,17 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
// nop
}
// stores meta information for open connections
struct connection_context {
// stores meta information for active endpoints
struct endpoint_context {
// denotes what message we expect from the remote node next
basp::connection_state cstate;
// our currently processed BASP header
basp::header hdr;
// the connection handle for I/O operations
connection_handle hdl;
// the handle for I/O operations
variant<connection_handle, dgram_scribe_handle> hdl;
// network-agnostic node identifier
node_id id;
// connected port
uint16_t remote_port;
// pending operations to be performed after handshake completed
optional<response_promise> callback;
};
// stores meta information for dgram endpoints
struct endpoint_context {
// our current processed BSAP header
basp::header hdr;
// local source
dgram_scribe_handle hdl;
// network-agnositc node identifier
node_id id;
// ports
// port
uint16_t remote_port;
// pending operations to be performed after handshake completed
optional<response_promise> callback;
......@@ -127,6 +114,15 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
};
void set_context(connection_handle hdl);
void set_context(dgram_scribe_handle hdl);
struct wr_buf_visitor {
wr_buf_visitor(broker* ptr) : ptr{ptr} { /* nop */ }
using result_type = std::vector<char>&;
result_type operator()(connection_handle hdl) { return ptr->wr_buf(hdl); }
result_type operator()(dgram_scribe_handle hdl) { return ptr->wr_buf(hdl); }
broker* ptr;
} wr_buf_of_hdl;
// pointer to ourselves
broker* self;
......@@ -135,12 +131,11 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
basp::instance instance;
// keeps context information for all open connections
std::unordered_map<connection_handle, connection_context> tcp_ctx;
std::unordered_map<connection_handle, endpoint_context> tcp_ctx;
std::unordered_map<dgram_scribe_handle, endpoint_context> udp_ctx;
// points to the current context for callbacks such as `make_proxy`
connection_context* this_context = nullptr;
endpoint_context* current_context = nullptr;
endpoint_context* this_context = nullptr;
// stores handles to spawn servers for other nodes; these servers
// are spawned whenever the broker learns a new node ID and try to
......
......@@ -71,7 +71,7 @@ public:
} // namespace io
} // namespace caf
namespace std{
namespace std {
template<>
struct hash<caf::io::connection_handle> {
......
......@@ -140,6 +140,19 @@ inspect(Inspector& f, dgram_scribe_closed_msg& x) {
return f(meta::type_name("dgram_scribe_closed_msg"), x.handle);
}
/// Signalizes that a {@link broker} dgram_acceptor has been closed.
struct dgram_acceptor_closed_msg {
/// Handle to the closed connection.
dgram_doorman_handle handle;
};
/// @relates connection_closed_msg
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f,
dgram_acceptor_closed_msg& x) {
return f(meta::type_name("dgram_acceptor_closed_msg"), x.handle);
}
/// Signalizes that a {@link broker} datagram source was closed.
struct dgram_doorman_closed_msg {
dgram_doorman_handle handle;
......@@ -155,17 +168,17 @@ inspect(Inspector& f, dgram_doorman_closed_msg& x) {
/// Signalizes newly arrived datagram for a {@link broker}.
struct new_endpoint_msg {
/// Handle to the related datagram endpoint.
dgram_doorman_handle handle;
dgram_doorman_handle source;
// Buffer containing the received data.
// std::vector<char> buf;
/// Handle to new dgram_scribe
dgram_scribe_handle endpoint;
dgram_scribe_handle handle;
};
/// @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.handle, x.endpoint);
return f(meta::type_name("new_endpoint_msg"), x.source, x.handle);
}
/// Signalizes that a datagram with a certain size has been sent.
......
......@@ -50,6 +50,7 @@ const char* basp_broker_state::name = "basp_broker";
basp_broker_state::basp_broker_state(broker* selfptr)
: basp::instance::callee(selfptr->system(),
static_cast<proxy_registry::backend&>(*this)),
wr_buf_of_hdl(selfptr),
self(selfptr),
instance(selfptr, *this) {
CAF_ASSERT(this_node() != none);
......@@ -105,7 +106,9 @@ strong_actor_ptr basp_broker_state::make_proxy(node_id nid, actor_id aid) {
<< CAF_ARG(nid) << CAF_ARG(aid));
// tell remote side we are monitoring this actor now
instance.write_announce_proxy(self->context(),
self->wr_buf(this_context->hdl), nid, aid);
apply_visitor(wr_buf_of_hdl,
this_context->hdl),
nid, aid);
instance.tbl().flush(*path);
mm->notify<hook::new_remote_actor>(res);
return res;
......@@ -451,18 +454,39 @@ void basp_broker_state::set_context(connection_handle hdl) {
auto i = tcp_ctx.find(hdl);
if (i == tcp_ctx.end()) {
CAF_LOG_INFO("create new BASP context:" << CAF_ARG(hdl));
i = tcp_ctx.emplace(hdl,
connection_context{
basp::await_header,
basp::header{basp::message_type::server_handshake, 0,
0, 0, none, none,
invalid_actor_id, invalid_actor_id},
hdl,
none,
0,
none}).first;
i = tcp_ctx.emplace(
hdl,
endpoint_context{
basp::await_header,
basp::header{basp::message_type::server_handshake,
0, 0, 0, none, none,
invalid_actor_id, invalid_actor_id},
hdl, none, 0, none
}
).first;
}
this_context = &i->second;
}
void basp_broker_state::set_context(dgram_scribe_handle hdl) {
CAF_LOG_TRACE(CAF_ARG(hdl));
auto i = udp_ctx.find(hdl);
if (i == udp_ctx.end()) {
CAF_LOG_INFO("create new BASP context:" << CAF_ARG(hdl));
i = udp_ctx.emplace(
hdl,
endpoint_context{
basp::await_header,
basp::header{basp::message_type::client_handshake,
0, 0, 0, none, none,
invalid_actor_id, invalid_actor_id},
hdl, none, 0, none
}
).first;
}
this_context = &i->second;
}
/******************************************************************************
......@@ -525,14 +549,21 @@ behavior basp_broker::make_behavior() {
[=](new_datagram_msg& msg) {
CAF_LOG_TRACE(CAF_ARG(msg.handle));
CAF_LOG_DEBUG("Received new_datagram_msg: " << CAF_ARG(msg));
static_cast<void>(msg);
// state.set_context(msg.handle);
// auto& ctx = *state.udp_context;
//auto next = state.instance.handle(context(), msg, ctx.hdr, ...);
// TODO: implement this
// look for existing context or create a new one
// handle messge
static_cast<void>(msg);
state.set_context(msg.handle);
auto& ctx = *state.this_context;
// TODO: ordering
// TODO: reliability?
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 {
// TODO: Is there configuration necessary?
// configure_datagram_size(...);
}
},
// received from proxy instances
[=](forward_atom, strong_actor_ptr& src,
......@@ -613,6 +644,27 @@ behavior basp_broker::make_behavior() {
auto port = local_port(msg.handle);
state.instance.remove_published_actor(port);
},
// received from underlying broker implementation
[=](const new_endpoint_msg& msg) {
CAF_LOG_TRACE(CAF_ARG(msg.handle));
auto& bi = state.instance;
bi.write_server_handshake(context(), wr_buf(msg.handle),
local_port(msg.source));
// flush(msg.handle);
// TODO: is this right?
},
// received from underlying broker implementation
[=](const dgram_scribe_closed_msg& msg) {
CAF_LOG_TRACE(CAF_ARG(msg.handle));
// TODO: dgram_scribe_closed_msg
static_cast<void>(msg);
},
// received from underlying broker implementation
[=](const dgram_acceptor_closed_msg& msg) {
CAF_LOG_TRACE(CAF_ARG(msg.handle));
// TODO: dgram_acceptor_closed_msg
static_cast<void>(msg);
},
// received from middleman actor
[=](publish_atom, accept_handle hdl, uint16_t port,
const strong_actor_ptr& whom, std::set<std::string>& sigs) {
......
......@@ -48,7 +48,7 @@ void dgram_doorman::io_failure(execution_unit* ctx, network::operation op) {
// size_t besize) {
bool dgram_doorman::new_endpoint(execution_unit* ctx,
dgram_scribe_handle endpoint) {
msg().endpoint = endpoint;
msg().handle = endpoint;
return invoke_mailbox_element(ctx);
}
......
......@@ -252,6 +252,13 @@ connection_state instance::handle(execution_unit* ctx,
return await_header;
}
bool instance::handle(execution_unit* ctx, new_datagram_msg& dm, header& hdr) {
static_cast<void>(ctx);
static_cast<void>(dm);
static_cast<void>(hdr);
return false;
};
void instance::handle_heartbeat(execution_unit* ctx) {
CAF_LOG_TRACE("");
for (auto& kvp: tbl_.direct_by_hdl_) {
......
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