Commit b8bb11b5 authored by Joseph Noir's avatar Joseph Noir

Extend middleman and broker capabilities for udp

parent fb625c32
...@@ -106,10 +106,34 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee { ...@@ -106,10 +106,34 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
node_id id; node_id id;
// connected port // connected port
uint16_t remote_port; uint16_t remote_port;
// pending operations to be performed after handhsake completed // pending operations to be performed after handshake completed
optional<response_promise> callback; optional<response_promise> callback;
}; };
// stores meta information for dagram endpoints
struct endpoint_context {
// denotes what message we expect from the remote node next
basp::connection_state cstate;
// out current processed BSAP header
basp::header hdr;
// local sink
datagram_sink_handle sink;
// local source
datagram_source_handle source;
// network-agnositc node identifier
node_id id;
// ports
uint16_t local_port;
uint16_t remote_port;
// ordering information
uint32_t sequence_number_send = 0;
uint32_t sequence_number_receive = 0;
// TODO: local buffer for ordering
// pending operations to be performed after handshake completed
optional<response_promise> callback;
// TODO: reliability things
};
void set_context(connection_handle hdl); void set_context(connection_handle hdl);
// pointer to ourselves // pointer to ourselves
...@@ -119,7 +143,8 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee { ...@@ -119,7 +143,8 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
basp::instance instance; basp::instance instance;
// keeps context information for all open connections // keeps context information for all open connections
std::unordered_map<connection_handle, connection_context> ctx; std::unordered_map<connection_handle, connection_context> tcp_ctx;
std::unordered_map<datagram_sink_handle, endpoint_context> udp_ctx;
// points to the current context for callbacks such as `make_proxy` // points to the current context for callbacks such as `make_proxy`
connection_context* this_context = nullptr; connection_context* this_context = nullptr;
......
...@@ -149,14 +149,14 @@ void basp_broker_state::purge_state(const node_id& nid) { ...@@ -149,14 +149,14 @@ void basp_broker_state::purge_state(const node_id& nid) {
auto hdl = instance.tbl().lookup_direct(nid); auto hdl = instance.tbl().lookup_direct(nid);
if (hdl == invalid_connection_handle) if (hdl == invalid_connection_handle)
return; return;
auto i = ctx.find(hdl); auto i = tcp_ctx.find(hdl);
if (i != ctx.end()) { if (i != tcp_ctx.end()) {
auto& ref = i->second; auto& ref = i->second;
if (ref.callback) { if (ref.callback) {
CAF_LOG_DEBUG("connection closed during handshake"); CAF_LOG_DEBUG("connection closed during handshake");
ref.callback->deliver(sec::disconnect_during_handshake); ref.callback->deliver(sec::disconnect_during_handshake);
} }
ctx.erase(i); tcp_ctx.erase(i);
} }
proxies().erase(nid); proxies().erase(nid);
} }
...@@ -448,10 +448,10 @@ void basp_broker_state::learned_new_node_indirectly(const node_id& nid) { ...@@ -448,10 +448,10 @@ void basp_broker_state::learned_new_node_indirectly(const node_id& nid) {
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 = ctx.find(hdl); auto i = tcp_ctx.find(hdl);
if (i == ctx.end()) { if (i == tcp_ctx.end()) {
CAF_LOG_INFO("create new BASP context:" << CAF_ARG(hdl)); CAF_LOG_INFO("create new BASP context:" << CAF_ARG(hdl));
i = ctx.emplace(hdl, i = tcp_ctx.emplace(hdl,
connection_context{ connection_context{
basp::await_header, basp::await_header,
basp::header{basp::message_type::server_handshake, 0, basp::header{basp::message_type::server_handshake, 0,
...@@ -512,7 +512,7 @@ behavior basp_broker::make_behavior() { ...@@ -512,7 +512,7 @@ behavior basp_broker::make_behavior() {
ctx.callback->deliver(make_error(sec::disconnect_during_handshake)); ctx.callback->deliver(make_error(sec::disconnect_during_handshake));
} }
close(msg.handle); close(msg.handle);
state.ctx.erase(msg.handle); state.tcp_ctx.erase(msg.handle);
} else if (next != ctx.cstate) { } else if (next != ctx.cstate) {
auto rd_size = next == basp::await_payload auto rd_size = next == basp::await_payload
? ctx.hdr.payload_len ? ctx.hdr.payload_len
...@@ -618,13 +618,30 @@ behavior basp_broker::make_behavior() { ...@@ -618,13 +618,30 @@ behavior basp_broker::make_behavior() {
<< CAF_ARG(whom)); << CAF_ARG(whom));
} }
}, },
[=](publish_atom, datagram_source_handle hdl, uint16_t port,
const strong_actor_ptr& whom, std::set<std::string>& sigs) {
CAF_LOG_TRACE(CAF_ARG(hdl.id()) << CAF_ARG(whom) << CAF_ARG(port));
if (hdl.invalid()) {
CAF_LOG_WARNING("invalid handle");
return;
}
auto res = assign_datagram_source(hdl);
if (res) {
if (whom)
system().registry().put(whom->id(), whom);
state.instance.add_published_actor(port, whom, std::move(sigs));
} else {
CAF_LOG_DEBUG("failed to assign doorman from handle"
<< CAF_ARG(whom));
}
},
// received from middleman actor (delegated) // received from middleman actor (delegated)
[=](connect_atom, connection_handle hdl, uint16_t port) { [=](connect_atom, connection_handle hdl, uint16_t port) {
CAF_LOG_TRACE(CAF_ARG(hdl.id())); CAF_LOG_TRACE(CAF_ARG(hdl.id()));
auto rp = make_response_promise(); auto rp = make_response_promise();
auto res = assign_tcp_scribe(hdl); auto res = assign_tcp_scribe(hdl);
if (res) { if (res) {
auto& ctx = state.ctx[hdl]; auto& ctx = state.tcp_ctx[hdl];
ctx.hdl = hdl; ctx.hdl = hdl;
ctx.remote_port = port; ctx.remote_port = port;
ctx.cstate = basp::await_header; ctx.cstate = basp::await_header;
...@@ -636,6 +653,25 @@ behavior basp_broker::make_behavior() { ...@@ -636,6 +653,25 @@ behavior basp_broker::make_behavior() {
rp.deliver(std::move(res.error())); rp.deliver(std::move(res.error()));
} }
}, },
[=](connect_atom, datagram_sink_handle hdl, uint16_t port) {
CAF_LOG_TRACE(CAF_ARG(hdl.id()));
auto rp = make_response_promise();
auto res = assign_datagram_sink(hdl);
if (res) {
auto& ctx = state.udp_ctx[hdl];
ctx.sink = hdl;
ctx.remote_port = port;
ctx.cstate = basp::await_header;
ctx.callback = rp;
// await server handshake
// TODO: Is there special processing required?
// configure_read(hdl, receive_policy::exactly(basp::header_size));
} else {
CAF_LOG_DEBUG("failed to assign datagram sink from handle"
<< CAF_ARG(res));
rp.deliver(std::move(res.error()));
}
},
[=](delete_atom, const node_id& nid, actor_id aid) { [=](delete_atom, const node_id& nid, actor_id aid) {
CAF_LOG_TRACE(CAF_ARG(nid) << ", " << CAF_ARG(aid)); CAF_LOG_TRACE(CAF_ARG(nid) << ", " << CAF_ARG(aid));
state.proxies().erase(nid, aid); state.proxies().erase(nid, aid);
......
...@@ -41,6 +41,15 @@ namespace io { ...@@ -41,6 +41,15 @@ namespace io {
namespace { namespace {
// TODO: dispatching!
constexpr auto all_zeroes = "0.0.0.0";
const std::string tcp = "tcp";
const std::string udp = "udp";
} // namespace <anonymous>
namespace {
class middleman_actor_impl : public middleman_actor::base { class middleman_actor_impl : public middleman_actor::base {
public: public:
middleman_actor_impl(actor_config& cfg, actor default_broker) middleman_actor_impl(actor_config& cfg, actor default_broker)
...@@ -81,7 +90,8 @@ public: ...@@ -81,7 +90,8 @@ public:
using endpoint_data = std::tuple<node_id, strong_actor_ptr, mpi_set>; using endpoint_data = std::tuple<node_id, strong_actor_ptr, mpi_set>;
using endpoint = std::pair<std::string, uint16_t>; // using endpoint = std::pair<std::string, uint16_t>;
using endpoint = io::uri;
behavior_type make_behavior() override { behavior_type make_behavior() override {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
...@@ -89,70 +99,105 @@ public: ...@@ -89,70 +99,105 @@ public:
[=](publish_atom, strong_actor_ptr& whom, [=](publish_atom, strong_actor_ptr& whom,
mpi_set& sigs, uri& u, bool reuse) -> put_res { mpi_set& sigs, uri& u, bool reuse) -> put_res {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
std::string addr(u.host().first, u.host().second); return put(u, whom, sigs, reuse);
return put(u.port_as_int(), whom, sigs, addr.c_str(), reuse);
}, },
[=](open_atom, uri& u, bool reuse) -> put_res { [=](open_atom, uri& u, bool reuse) -> put_res {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
strong_actor_ptr whom; strong_actor_ptr whom;
mpi_set sigs; mpi_set sigs;
std::string addr(u.host().first, u.host().second); return put(u, whom, sigs, reuse);
return put(u.port_as_int(), whom, sigs, addr.c_str(), 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 hostname(u.host().first, u.host().second); std::string host(u.host().first, u.host().second);
auto port = u.port_as_int(); auto port = u.port_as_int();
endpoint key{std::move(hostname), port};
// respond immediately if endpoint is cached // respond immediately if endpoint is cached
auto x = cached(key); auto x = cached(u);
if (x) { if (x) {
CAF_LOG_DEBUG("found cached entry" << CAF_ARG(*x)); CAF_LOG_DEBUG("found cached entry" << CAF_ARG(*x));
rp.deliver(get<0>(*x), get<1>(*x), get<2>(*x)); rp.deliver(get<0>(*x), get<1>(*x), get<2>(*x));
return {}; return {};
} }
// attach this promise to a pending request if possible // attach this promise to a pending request if possible
auto rps = pending(key); auto rps = pending(u);
if (rps) { if (rps) {
CAF_LOG_DEBUG("attach to pending request"); CAF_LOG_DEBUG("attach to pending request");
rps->emplace_back(std::move(rp)); rps->emplace_back(std::move(rp));
return {}; return {};
} }
// connect to endpoint and initiate handhsake etc. if (std::distance(u.scheme().first, u.scheme().second) >= 3 &&
auto y = system().middleman().backend().new_tcp_scribe(key.first, port); equal(std::begin(tcp), std::end(tcp), u.scheme().first)) {
if (!y) { // connect to endpoint and initiate handhsake etc.
rp.deliver(std::move(y.error())); auto y = system().middleman().backend().new_tcp_scribe(host, port);
return {}; if (!y) {
} rp.deliver(std::move(y.error()));
auto hdl = *y; return {};
std::vector<response_promise> tmp{std::move(rp)}; }
pending_.emplace(key, std::move(tmp)); auto hdl = *y;
request(broker_, infinite, connect_atom::value, hdl, port).then( std::vector<response_promise> tmp{std::move(rp)};
[=](node_id& nid, strong_actor_ptr& addr, mpi_set& sigs) { pending_.emplace(u, std::move(tmp));
auto i = pending_.find(key); request(broker_, infinite, connect_atom::value, hdl, port).then(
if (i == pending_.end()) [=](node_id& nid, strong_actor_ptr& addr, mpi_set& sigs) {
return; auto i = pending_.find(u);
if (nid && addr) { if (i == pending_.end())
monitor(addr); return;
cached_.emplace(key, std::make_tuple(nid, addr, sigs)); if (nid && addr) {
monitor(addr);
cached_.emplace(u, std::make_tuple(nid, addr, sigs));
}
auto res = make_message(std::move(nid), std::move(addr),
std::move(sigs));
for (auto& promise : i->second)
promise.deliver(res);
pending_.erase(i);
},
[=](error& err) {
auto i = pending_.find(u);
if (i == pending_.end())
return;
for (auto& promise : i->second)
promise.deliver(err);
pending_.erase(i);
} }
auto res = make_message(std::move(nid), std::move(addr), );
std::move(sigs)); } else if (std::distance(u.scheme().first, u.scheme().second) >= 3 &&
for (auto& promise : i->second) equal(std::begin(udp), std::end(udp), u.scheme().first)) {
promise.deliver(res); // connect to endpoint and initiate handhsake etc.
pending_.erase(i); auto y = system().middleman().backend().new_datagram_sink(host, port);
}, if (!y) {
[=](error& err) { rp.deliver(std::move(y.error()));
auto i = pending_.find(key); return {};
if (i == pending_.end())
return;
for (auto& promise : i->second)
promise.deliver(err);
pending_.erase(i);
} }
); auto hdl = *y;
return {}; std::vector<response_promise> tmp{std::move(rp)};
pending_.emplace(u, std::move(tmp));
request(broker_, infinite, connect_atom::value, hdl, port).then(
[=](node_id& nid, strong_actor_ptr& addr, mpi_set& sigs) {
auto i = pending_.find(u);
if (i == pending_.end())
return;
if (nid && addr) {
monitor(addr);
cached_.emplace(u, std::make_tuple(nid, addr, sigs));
}
auto res = make_message(std::move(nid), std::move(addr),
std::move(sigs));
for (auto& promise : i->second)
promise.deliver(res);
pending_.erase(i);
},
[=](error& err) {
auto i = pending_.find(u);
if (i == pending_.end())
return;
for (auto& promise : i->second)
promise.deliver(err);
pending_.erase(i);
}
);
}
return {}; // sec::unsupported_protocol;
}, },
[=](unpublish_atom atm, actor_addr addr, uri& u) -> del_res { [=](unpublish_atom atm, actor_addr addr, uri& u) -> del_res {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
...@@ -183,24 +228,40 @@ public: ...@@ -183,24 +228,40 @@ public:
} }
private: private:
put_res put(uint16_t port, strong_actor_ptr& whom, put_res put(const uri& u, strong_actor_ptr& whom,
mpi_set& sigs, const char* in = nullptr, mpi_set& sigs, bool reuse_addr = false) {
bool reuse_addr = false) {
CAF_LOG_TRACE(CAF_ARG(port) << CAF_ARG(whom) << CAF_ARG(sigs) CAF_LOG_TRACE(CAF_ARG(port) << CAF_ARG(whom) << CAF_ARG(sigs)
<< CAF_ARG(in) << CAF_ARG(reuse_addr)); << CAF_ARG(in) << CAF_ARG(reuse_addr));
accept_handle hdl;
uint16_t actual_port; uint16_t actual_port;
// treat empty strings like nullptr // treat empty strings or 0.0.0.0 as nullptr
if (in != nullptr && in[0] == '\0') auto addr = std::string(u.host().first, u.host().second);
in = nullptr; const char* in = nullptr;
auto res = system().middleman().backend().new_tcp_doorman(port, in, if ((!addr.empty() && addr.front() != '\0') || addr == all_zeroes)
reuse_addr); in = addr.c_str();
if (!res) if (std::distance(u.scheme().first, u.scheme().second) >= 3 &&
return std::move(res.error()); equal(u.scheme().first, u.scheme().second, std::begin(tcp))) {
hdl = res->first; auto res = system().middleman().backend().new_tcp_doorman(u.port_as_int(),
actual_port = res->second; in, reuse_addr);
anon_send(broker_, publish_atom::value, hdl, actual_port, if (!res)
std::move(whom), std::move(sigs)); return std::move(res.error());
accept_handle hdl = res->first;
actual_port = res->second;
anon_send(broker_, publish_atom::value, hdl, actual_port,
std::move(whom), std::move(sigs));
} else if (std::distance(u.scheme().first, u.scheme().second) >= 3 &&
equal(u.scheme().first, u.scheme().second, std::begin(udp))) {
auto res
= system().middleman().backend().new_datagram_source(u.port_as_int(),
in, reuse_addr);
if (!res)
return std::move(res.error());
datagram_source_handle hdl = res->first;
actual_port = res->second;
anon_send(broker_, publish_atom::value, hdl, actual_port,
std::move(whom), std::move(sigs));
} else {
return sec::unsupported_protocol;
}
return actual_port; return actual_port;
} }
......
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