Commit b5c049b6 authored by Joseph Noir's avatar Joseph Noir

Add autoconn test for UDP and fix related bugs

parent 2eed3d57
...@@ -46,8 +46,7 @@ struct connection_helper_state { ...@@ -46,8 +46,7 @@ struct connection_helper_state {
static const char* name; static const char* name;
}; };
behavior datagram_connection_broker(broker* self, behavior datagram_connection_broker(broker* self, uint16_t port,
uint16_t port,
network::address_listing addresses, network::address_listing addresses,
actor system_broker, actor system_broker,
basp::instance* instance); basp::instance* instance);
......
...@@ -632,19 +632,19 @@ void basp_broker_state::send_buffered_messages(execution_unit* ctx, ...@@ -632,19 +632,19 @@ void basp_broker_state::send_buffered_messages(execution_unit* ctx,
datagram_handle hdl) { datagram_handle hdl) {
if (pending_connectivity.count(nid) > 0) { if (pending_connectivity.count(nid) > 0) {
for (auto& msg : pending_connectivity[nid]) { for (auto& msg : pending_connectivity[nid]) {
// TODO: Validate that this actually works.
auto seq_num = next_sequence_number(hdl); auto seq_num = next_sequence_number(hdl);
auto seq_size = sizeof(basp::sequence_type); auto seq_size = sizeof(basp::sequence_type);
auto offset = basp::header_size - seq_size; auto offset = basp::header_size - seq_size;
auto& buf = get_buffer(hdl); auto& buf = msg;
stream_serializer<charbuf> out{ctx, buf.data() + offset, seq_size}; stream_serializer<charbuf> out{ctx, buf.data() + offset, seq_size};
auto err = out(seq_num); auto err = out(seq_num);
if (err) if (err)
CAF_LOG_ERROR(CAF_ARG(err)); CAF_LOG_ERROR(CAF_ARG(err));
buf.insert(buf.end(), msg.begin(), msg.end()); self->enqueue_datagram(hdl, std::move(buf));
self->flush(hdl);
} }
pending_connectivity[nid].clear();
} }
flush(hdl);
} }
basp_broker_state::buffer_type& basp_broker_state::buffer_type&
...@@ -720,13 +720,22 @@ behavior basp_broker::make_behavior() { ...@@ -720,13 +720,22 @@ behavior basp_broker::make_behavior() {
state.enable_udp = system().config().middleman_enable_udp; state.enable_udp = system().config().middleman_enable_udp;
if (system().config().middleman_enable_automatic_connections) { if (system().config().middleman_enable_automatic_connections) {
CAF_LOG_INFO("enable automatic connections"); CAF_LOG_INFO("enable automatic connections");
auto addrs = network::interfaces::list_addresses(false);
for (auto& p : addrs) {
auto& vec = p.second;
// Remove link local addresses.
vec.erase(std::remove_if(std::begin(vec), std::end(vec),
[](const std::string& str) {
return str.find("fe80") == 0;
}),
vec.end());
}
// Open a random port and store a record for our peers how to // Open a random port and store a record for our peers how to
// connect to this broker directly in the configuration server. // connect to this broker directly in the configuration server.
if (state.enable_tcp) { if (state.enable_tcp) {
auto res = add_tcp_doorman(uint16_t{0}); auto res = add_tcp_doorman(uint16_t{0});
if (res) { if (res) {
auto port = res->second; auto port = res->second;
auto addrs = network::interfaces::list_addresses(false);
state.instance.tbl().local_addresses(network::protocol::tcp, state.instance.tbl().local_addresses(network::protocol::tcp,
{port, addrs}); {port, addrs});
auto config_server = system().registry().get(atom("ConfigServ")); auto config_server = system().registry().get(atom("ConfigServ"));
...@@ -739,7 +748,6 @@ behavior basp_broker::make_behavior() { ...@@ -739,7 +748,6 @@ behavior basp_broker::make_behavior() {
auto res = add_udp_datagram_servant(uint16_t{0}); auto res = add_udp_datagram_servant(uint16_t{0});
if (res) { if (res) {
auto port = res->second; auto port = res->second;
auto addrs = network::interfaces::list_addresses(false);
state.instance.tbl().local_addresses(network::protocol::udp, state.instance.tbl().local_addresses(network::protocol::udp,
{port, addrs}); {port, addrs});
auto config_server = system().registry().get(atom("ConfigServ")); auto config_server = system().registry().get(atom("ConfigServ"));
......
...@@ -45,13 +45,17 @@ behavior datagram_connection_broker(broker* self, uint16_t port, ...@@ -45,13 +45,17 @@ behavior datagram_connection_broker(broker* self, uint16_t port,
if (eptr) { if (eptr) {
auto hdl = (*eptr)->hdl(); auto hdl = (*eptr)->hdl();
self->add_datagram_servant(std::move(*eptr)); self->add_datagram_servant(std::move(*eptr));
instance->write_client_handshake(self->context(), std::vector<char> buf;
self->wr_buf(hdl), instance->write_client_handshake(self->context(), buf,
none, this_node, none, this_node,
app_id); app_id);
self->enqueue_datagram(hdl, std::move(buf));
self->flush(hdl);
} }
} }
} }
// We are not interested in attempts that do not work.
self->set_default_handler(drop);
return { return {
[=](new_datagram_msg& msg) { [=](new_datagram_msg& msg) {
auto hdl = msg.handle; auto hdl = msg.handle;
...@@ -61,7 +65,7 @@ behavior datagram_connection_broker(broker* self, uint16_t port, ...@@ -61,7 +65,7 @@ behavior datagram_connection_broker(broker* self, uint16_t port,
after(autoconnect_timeout) >> [=]() { after(autoconnect_timeout) >> [=]() {
CAF_LOG_TRACE(CAF_ARG("")); CAF_LOG_TRACE(CAF_ARG(""));
// Nothing heard in about 10 minutes... just call it a day, then. // Nothing heard in about 10 minutes... just call it a day, then.
CAF_LOG_INFO("aborted direct connection attempt after 10min"); CAF_LOG_INFO("aborted direct connection attempt after 10 min");
self->quit(exit_reason::user_shutdown); self->quit(exit_reason::user_shutdown);
} }
}; };
...@@ -76,8 +80,7 @@ bool establish_stream_connection(stateful_actor<connection_helper_state>* self, ...@@ -76,8 +80,7 @@ bool establish_stream_connection(stateful_actor<connection_helper_state>* self,
for (auto& addr : kvp.second) { for (auto& addr : kvp.second) {
auto hdl = mx.new_tcp_scribe(addr, port); auto hdl = mx.new_tcp_scribe(addr, port);
if (hdl) { if (hdl) {
// Gotcha! Send scribe to our BASP broker // Gotcha! Send scribe to our BASP broker to initiate handshake etc.
// to initiate handshake etc.
CAF_LOG_INFO("connected directly:" << CAF_ARG(addr)); CAF_LOG_INFO("connected directly:" << CAF_ARG(addr));
self->send(b, connect_atom::value, *hdl, port); self->send(b, connect_atom::value, *hdl, port);
return true; return true;
......
This diff is collapsed.
...@@ -68,6 +68,17 @@ public: ...@@ -68,6 +68,17 @@ public:
return *res; return *res;
} }
/// Convenience function for calling `mm.publish` and requiring a valid
/// result.
template <class Handle>
uint16_t publish_udp(Handle whom, uint16_t port, const char* in = nullptr,
bool reuse = false) {
this->sched.inline_next_enqueue();
auto res = mm.publish_udp(whom, port, in, reuse);
CAF_REQUIRE(res);
return *res;
}
/// Convenience function for calling `mm.remote_actor` and requiring a valid /// Convenience function for calling `mm.remote_actor` and requiring a valid
/// result. /// result.
template <class Handle = caf::actor> template <class Handle = caf::actor>
...@@ -79,6 +90,17 @@ public: ...@@ -79,6 +90,17 @@ public:
return *res; return *res;
} }
/// Convenience function for calling `mm.remote_actor` and requiring a valid
/// result.
template <class Handle = caf::actor>
Handle remote_actor_udp(std::string host, uint16_t port) {
this->sched.inline_next_enqueue();
this->sched.after_next_enqueue(exec_all_nodes);
auto res = mm.remote_actor_udp<Handle>(std::move(host), port);
CAF_REQUIRE(res);
return *res;
}
private: private:
caf::io::basp_broker* get_basp_broker() { caf::io::basp_broker* get_basp_broker() {
auto hdl = mm.named_broker<caf::io::basp_broker>(caf::atom("BASP")); auto hdl = mm.named_broker<caf::io::basp_broker>(caf::atom("BASP"));
...@@ -126,6 +148,8 @@ public: ...@@ -126,6 +148,8 @@ public:
using accept_handle = caf::io::accept_handle; using accept_handle = caf::io::accept_handle;
using datagram_handle = caf::io::datagram_handle;
fake_network_fixture_base(planets_vector xs) : planets_(std::move(xs)) { fake_network_fixture_base(planets_vector xs) : planets_(std::move(xs)) {
// nop // nop
} }
...@@ -140,6 +164,11 @@ public: ...@@ -140,6 +164,11 @@ public:
return connection_handle::from_int(++hdl_id_); return connection_handle::from_int(++hdl_id_);
} }
/// Returns a unique datagram handle.
datagram_handle next_datagram_handle() {
return datagram_handle::from_int(++hdl_id_);
}
/// Prepare a connection from `client` (calls `remote_actor`) to `server` /// Prepare a connection from `client` (calls `remote_actor`) to `server`
/// (calls `publish`). /// (calls `publish`).
/// @returns randomly picked connection handles for the server and the client. /// @returns randomly picked connection handles for the server and the client.
...@@ -164,6 +193,31 @@ public: ...@@ -164,6 +193,31 @@ public:
next_accept_handle()); next_accept_handle());
} }
/// Prepares comminication between `client` (calls `remote_actor_udp`) and
/// `server` (calls `publish_udp`).
/// @returns randomly picked datagram handles for the server and the client.
std::pair<datagram_handle, datagram_handle>
prepare_endpoints(PlanetType& server, PlanetType& client,
std::string host, uint16_t port,
datagram_handle server_endpoint) {
auto server_hdl = next_datagram_handle();
auto client_hdl = next_datagram_handle();
server.mpx.prepare_endpoints(server_endpoint, server_hdl, client.mpx,
std::move(host), port, client_hdl);
return std::make_pair(server_hdl, client_hdl);
}
/// Prepares comminication between `client` (calls `remote_actor_udp`) and
/// `server` (calls `publish_udp`).
/// @returns randomly picked datagram handles for the server and the client.
std::pair<datagram_handle, datagram_handle>
prepare_endpoints(PlanetType& server, PlanetType& client,
std::string host, uint16_t port) {
return prepare_endpoints(server, client, std::move(host), port,
next_datagram_handle());
}
// Convenience function for transmitting all "network" traffic (no new // Convenience function for transmitting all "network" traffic (no new
// connections are accepted). // connections are accepted).
void network_traffic() { void network_traffic() {
......
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