Commit c0e117cc authored by Dominik Charousset's avatar Dominik Charousset

Transmit named actors in BASP handshake

parent b45f558f
No preview for this file type
...@@ -83,7 +83,9 @@ public: ...@@ -83,7 +83,9 @@ public:
actor get_named(atom_value key) const; actor get_named(atom_value key) const;
std::vector<std::pair<atom_value, actor>> named_actors() const; using named_entries = std::unordered_map<atom_value, actor>;
named_entries named_actors() const;
static actor_registry* create_singleton(); static actor_registry* create_singleton();
...@@ -94,8 +96,6 @@ public: ...@@ -94,8 +96,6 @@ public:
void initialize(); void initialize();
private: private:
using named_entries = std::unordered_map<atom_value, actor>;
using entries = std::unordered_map<actor_id, value_type>; using entries = std::unordered_map<actor_id, value_type>;
actor_registry(); actor_registry();
......
...@@ -56,6 +56,8 @@ class mailbox_element; ...@@ -56,6 +56,8 @@ class mailbox_element;
class message_handler; class message_handler;
class uniform_type_info; class uniform_type_info;
class event_based_actor; class event_based_actor;
class binary_serializer;
class binary_deserializer;
class forwarding_actor_proxy; class forwarding_actor_proxy;
// structs // structs
......
...@@ -137,11 +137,8 @@ actor actor_registry::get_named(atom_value key) const { ...@@ -137,11 +137,8 @@ actor actor_registry::get_named(atom_value key) const {
return i->second; return i->second;
} }
std::vector<std::pair<atom_value, actor>> actor_registry::named_actors() const { auto actor_registry::named_actors() const -> named_entries {
std::vector<std::pair<atom_value, actor>> result; return named_entries_;
for (auto& kvp : named_entries_)
result.emplace_back(kvp);
return result;
} }
actor_registry* actor_registry::create_singleton() { actor_registry* actor_registry::create_singleton() {
......
...@@ -27,6 +27,7 @@ ...@@ -27,6 +27,7 @@
#include <unordered_map> #include <unordered_map>
#include <unordered_set> #include <unordered_set>
#include "caf/fwd.hpp"
#include "caf/node_id.hpp" #include "caf/node_id.hpp"
#include "caf/actor_addr.hpp" #include "caf/actor_addr.hpp"
#include "caf/serializer.hpp" #include "caf/serializer.hpp"
...@@ -164,7 +165,7 @@ namespace basp { ...@@ -164,7 +165,7 @@ namespace basp {
/// The current BASP version. Different BASP versions will not /// The current BASP version. Different BASP versions will not
/// be able to exchange messages. /// be able to exchange messages.
constexpr uint64_t version = 1; constexpr uint64_t version = 2;
/// Storage type for raw bytes. /// Storage type for raw bytes.
using buffer_type = std::vector<char>; using buffer_type = std::vector<char>;
...@@ -490,6 +491,9 @@ public: ...@@ -490,6 +491,9 @@ public:
/// written (e.g. used when establishing direct connections on-the-fly). /// written (e.g. used when establishing direct connections on-the-fly).
void write_server_handshake(buffer_type& buf, optional<uint16_t> port); void write_server_handshake(buffer_type& buf, optional<uint16_t> port);
/// Writes the client handshake to `buf`.
void write_client_handshake(buffer_type& buf);
/// Writes a `dispatch_error` to `buf`. /// Writes a `dispatch_error` to `buf`.
void write_dispatch_error(buffer_type& buf, void write_dispatch_error(buffer_type& buf,
const node_id& source_node, const node_id& source_node,
...@@ -508,11 +512,22 @@ public: ...@@ -508,11 +512,22 @@ public:
return this_node_; return this_node_;
} }
using named_actors_map = std::unordered_map<atom_value, actor>;
/// Returns the named actors for this node.
named_actors_map named_actors();
/// Returns the named actors for `nid`.
optional<const named_actors_map&> named_actors(const node_id& nid);
private: private:
void read_named_actors(binary_deserializer& source, const node_id& nid);
routing_table tbl_; routing_table tbl_;
published_actor_map published_actors_; published_actor_map published_actors_;
node_id this_node_; node_id this_node_;
callee& callee_; callee& callee_;
std::unordered_map<node_id, named_actors_map> named_actors_;
}; };
/// Checks whether given header contains a handshake. /// Checks whether given header contains a handshake.
......
...@@ -116,17 +116,13 @@ bool server_handshake_valid(const header& hdr) { ...@@ -116,17 +116,13 @@ bool server_handshake_valid(const header& hdr) {
&& ! valid(hdr.dest_node) && ! valid(hdr.dest_node)
&& zero(hdr.dest_actor) && zero(hdr.dest_actor)
&& ! zero(hdr.operation_data) && ! zero(hdr.operation_data)
&& ( (! zero(hdr.source_actor) && ! zero(hdr.payload_len)) && ! zero(hdr.payload_len);
|| (zero(hdr.source_actor) && zero(hdr.payload_len)));
} }
bool client_handshake_valid(const header& hdr) { bool client_handshake_valid(const header& hdr) {
return valid(hdr.source_node) return valid(hdr.source_node)
&& valid(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.payload_len)
&& zero(hdr.operation_data); && zero(hdr.operation_data);
} }
...@@ -146,7 +142,6 @@ bool announce_proxy_instance_valid(const header& hdr) { ...@@ -146,7 +142,6 @@ bool announce_proxy_instance_valid(const header& hdr) {
&& zero(hdr.operation_data); && zero(hdr.operation_data);
} }
bool kill_proxy_instance_valid(const header& hdr) { bool kill_proxy_instance_valid(const header& hdr) {
return valid(hdr.source_node) return valid(hdr.source_node)
&& valid(hdr.dest_node) && valid(hdr.dest_node)
...@@ -355,13 +350,16 @@ connection_state instance::handle(const new_data_msg& dm, header& hdr, ...@@ -355,13 +350,16 @@ connection_state instance::handle(const new_data_msg& dm, header& hdr,
// handle message to ourselves // handle message to ourselves
switch (hdr.operation) { switch (hdr.operation) {
case message_type::server_handshake: { case message_type::server_handshake: {
if (! payload_valid()) {
CAF_LOG_WARNING("received server handshake without payload");
return err();
}
actor_id aid = invalid_actor_id; actor_id aid = invalid_actor_id;
std::set<std::string> sigs; std::set<std::string> sigs;
if (payload_valid()) {
binary_deserializer bd{payload->data(), payload->size(), binary_deserializer bd{payload->data(), payload->size(),
&get_namespace()}; &get_namespace()};
// read first half of payload (ID and interface of published actor)
bd >> aid >> sigs; bd >> aid >> sigs;
}
// close self connection after handshake is done // close self connection after handshake is done
if (hdr.source_node == this_node_) { if (hdr.source_node == this_node_) {
CAF_LOG_INFO("close connection to self immediately"); CAF_LOG_INFO("close connection to self immediately");
...@@ -384,10 +382,9 @@ connection_state instance::handle(const new_data_msg& dm, header& hdr, ...@@ -384,10 +382,9 @@ connection_state instance::handle(const new_data_msg& dm, header& hdr,
CAF_LOG_ERROR("no route to host after server handshake"); CAF_LOG_ERROR("no route to host after server handshake");
return err(); return err();
} }
write(path->wr_buf, // read remainig payload
message_type::client_handshake, nullptr, 0, read_named_actors(bd, hdr.source_node);
this_node_, hdr.source_node, write_client_handshake(path->wr_buf);
invalid_actor_id, invalid_actor_id);
// tell client to create proxy etc. // tell client to create proxy etc.
callee_.finalize_handshake(hdr.source_node, aid, sigs); callee_.finalize_handshake(hdr.source_node, aid, sigs);
flush(*path); flush(*path);
...@@ -401,6 +398,11 @@ connection_state instance::handle(const new_data_msg& dm, header& hdr, ...@@ -401,6 +398,11 @@ connection_state instance::handle(const new_data_msg& dm, header& hdr,
} }
// add direct route to this node // add direct route to this node
tbl_.add_direct(dm.handle, hdr.source_node); tbl_.add_direct(dm.handle, hdr.source_node);
if (payload_valid()) {
binary_deserializer bd{payload->data(), payload->size(),
&get_namespace()};
read_named_actors(bd, hdr.source_node);
}
break; break;
case message_type::dispatch_message: { case message_type::dispatch_message: {
if (! payload_valid()) if (! payload_valid())
...@@ -584,12 +586,13 @@ void instance::write_server_handshake(buffer_type& out_buf, ...@@ -584,12 +586,13 @@ void instance::write_server_handshake(buffer_type& out_buf,
pa = &i->second; pa = &i->second;
} }
auto writer = make_callback([&](serializer& sink) { auto writer = make_callback([&](serializer& sink) {
if (! pa) { if (pa)
return; sink << pa->first.id() << pa->second;
} else
sink << pa->first.id() << static_cast<uint32_t>(pa->second.size()); sink << invalid_actor_id << std::set<std::string>{};
for (auto& sig : pa->second) auto na = named_actors();
sink << sig; for (auto& kvp : na)
sink << kvp.first << kvp.second;
}); });
header hdr{message_type::server_handshake, 0, version, header hdr{message_type::server_handshake, 0, version,
this_node_, invalid_node_id, this_node_, invalid_node_id,
...@@ -597,6 +600,19 @@ void instance::write_server_handshake(buffer_type& out_buf, ...@@ -597,6 +600,19 @@ void instance::write_server_handshake(buffer_type& out_buf,
write(out_buf, hdr, &writer); write(out_buf, hdr, &writer);
} }
void instance::write_client_handshake(buffer_type& buf) {
auto writer = make_callback([&](serializer& sink) {
auto na = named_actors();
for (auto& kvp : na)
sink << kvp.first << kvp.second;
});
write(buf,
message_type::client_handshake, nullptr, 0,
this_node_, invalid_node_id,
invalid_actor_id, invalid_actor_id,
&writer);
}
void instance::write_dispatch_error(buffer_type& buf, void instance::write_dispatch_error(buffer_type& buf,
const node_id& source_node, const node_id& source_node,
const node_id& dest_node, const node_id& dest_node,
...@@ -625,6 +641,30 @@ void instance::write_kill_proxy_instance(buffer_type& buf, ...@@ -625,6 +641,30 @@ void instance::write_kill_proxy_instance(buffer_type& buf,
write(buf, hdr); write(buf, hdr);
} }
auto instance::named_actors() -> named_actors_map {
return detail::singletons::get_actor_registry()->named_actors();
}
auto instance::named_actors(const node_id& nid)
-> optional<const named_actors_map&> {
auto i = named_actors_.find(nid);
if (i == named_actors_.end())
return none;
return i->second;
}
void instance::read_named_actors(binary_deserializer& bd, const node_id& nid) {
auto& storage = named_actors_[nid];
CAF_ASSERT(storage.empty());
while (! bd.at_end()) {
atom_value key;
actor value;
bd >> key >> value;
storage.emplace(key, std::move(value));
}
}
} // namespace basp } // namespace basp
} // namespace io } // namespace io
} // namespace caf } // namespace caf
...@@ -227,18 +227,15 @@ public: ...@@ -227,18 +227,15 @@ public:
{basp::message_type::client_handshake, 0, 0, {basp::message_type::client_handshake, 0, 0,
remote_node(i), this_node(), remote_node(i), this_node(),
invalid_actor_id, invalid_actor_id}); invalid_actor_id, invalid_actor_id});
if (published_actor_id != invalid_actor_id) auto na = registry()->named_actors();
m.expect(hdl, m.expect(hdl,
{basp::message_type::server_handshake, 0, basp::version, {basp::message_type::server_handshake, 0, basp::version,
this_node(), invalid_node_id, this_node(), invalid_node_id,
published_actor_id, invalid_actor_id}, published_actor_id, invalid_actor_id},
published_actor_id, published_actor_id,
published_actor_ifs); published_actor_ifs,
else atom("spawner"),
m.expect(hdl, na[atom("spawner")]);
{basp::message_type::server_handshake, 0, basp::version,
this_node(), invalid_node_id,
invalid_actor_id, invalid_actor_id});
// test whether basp instance correctly updates the // test whether basp instance correctly updates the
// routing table upon receiving client handshakes // routing table upon receiving client handshakes
auto path = tbl().lookup(remote_node(i)); auto path = tbl().lookup(remote_node(i));
...@@ -358,14 +355,14 @@ CAF_TEST(empty_server_handshake) { ...@@ -358,14 +355,14 @@ CAF_TEST(empty_server_handshake) {
buffer payload; buffer payload;
std::tie(hdr, payload) = from_buf(buf); std::tie(hdr, payload) = from_buf(buf);
basp::header expected{basp::message_type::server_handshake, basp::header expected{basp::message_type::server_handshake,
0, static_cast<uint32_t>(payload.size()),
basp::version, basp::version,
this_node(), invalid_node_id, this_node(), invalid_node_id,
invalid_actor_id, invalid_actor_id}; invalid_actor_id, invalid_actor_id};
CAF_CHECK(basp::valid(hdr)); CAF_CHECK(basp::valid(hdr));
CAF_CHECK(basp::is_handshake(hdr)); CAF_CHECK(basp::is_handshake(hdr));
CAF_CHECK(hdr == expected); CAF_CHECK(hdr == expected);
CAF_CHECK(payload.empty()); CAF_CHECK(! payload.empty());
} }
CAF_TEST(non_empty_server_handshake) { CAF_TEST(non_empty_server_handshake) {
...@@ -376,11 +373,13 @@ CAF_TEST(non_empty_server_handshake) { ...@@ -376,11 +373,13 @@ CAF_TEST(non_empty_server_handshake) {
{"caf::replies_to<@u16>::with<@u16>"}); {"caf::replies_to<@u16>::with<@u16>"});
instance().write_server_handshake(buf, 4242); instance().write_server_handshake(buf, 4242);
buffer expected_buf; buffer expected_buf;
auto na = registry()->named_actors();
basp::header expected{basp::message_type::server_handshake, 0, basp::version, basp::header expected{basp::message_type::server_handshake, 0, basp::version,
this_node(), invalid_node_id, this_node(), invalid_node_id,
self()->id(), invalid_actor_id}; self()->id(), invalid_actor_id};
to_buf(expected_buf, expected, nullptr, to_buf(expected_buf, expected, nullptr,
self()->id(), set<string>{"caf::replies_to<@u16>::with<@u16>"}); self()->id(), set<string>{"caf::replies_to<@u16>::with<@u16>"},
atom("spawner"), na[atom("spawner")]);
CAF_CHECK(hexstr(buf) == hexstr(expected_buf)); CAF_CHECK(hexstr(buf) == hexstr(expected_buf));
} }
...@@ -457,6 +456,7 @@ CAF_TEST(remote_actor_and_send) { ...@@ -457,6 +456,7 @@ CAF_TEST(remote_actor_and_send) {
CAF_CHECK(mpx()->pending_scribes().count(make_pair("localhost", 4242)) == 0); CAF_CHECK(mpx()->pending_scribes().count(make_pair("localhost", 4242)) == 0);
// build a fake server handshake containing the id of our first pseudo actor // build a fake server handshake containing the id of our first pseudo actor
CAF_TEST_VERBOSE("server handshake => client handshake + proxy announcement"); CAF_TEST_VERBOSE("server handshake => client handshake + proxy announcement");
auto na = registry()->named_actors();
mock(remote_hdl(0), mock(remote_hdl(0),
{basp::message_type::server_handshake, 0, basp::version, {basp::message_type::server_handshake, 0, basp::version,
remote_node(0), invalid_node_id, remote_node(0), invalid_node_id,
...@@ -465,8 +465,9 @@ CAF_TEST(remote_actor_and_send) { ...@@ -465,8 +465,9 @@ CAF_TEST(remote_actor_and_send) {
uint32_t{0}) uint32_t{0})
.expect(remote_hdl(0), .expect(remote_hdl(0),
{basp::message_type::client_handshake, 0, 0, {basp::message_type::client_handshake, 0, 0,
this_node(), remote_node(0), this_node(), invalid_node_id,
invalid_actor_id, invalid_actor_id}) invalid_actor_id, invalid_actor_id},
atom("spawner"), na[atom("spawner")])
.expect(remote_hdl(0), .expect(remote_hdl(0),
{basp::message_type::announce_proxy_instance, 0, 0, {basp::message_type::announce_proxy_instance, 0, 0,
this_node(), remote_node(0), this_node(), remote_node(0),
......
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