Commit 1d93cf63 authored by Joseph Noir's avatar Joseph Noir

Add test for autoconnect and fix related bugs

parent a287b288
...@@ -337,7 +337,8 @@ public: ...@@ -337,7 +337,8 @@ public:
return false; return false;
} }
// Close this connection if we already established communication. // Close this connection if we already established communication.
if (tbl_.lookup(hdr.source_node).hdl) { auto lr = tbl_.lookup(hdr.source_node);
if (lr.hdl) {
CAF_LOG_INFO("close connection since we already have a " CAF_LOG_INFO("close connection since we already have a "
"connection: " << CAF_ARG(hdr.source_node)); "connection: " << CAF_ARG(hdr.source_node));
callee_.finalize_handshake(hdr.source_node, aid, sigs); callee_.finalize_handshake(hdr.source_node, aid, sigs);
...@@ -345,7 +346,10 @@ public: ...@@ -345,7 +346,10 @@ public:
} }
// Add this node to our contacts. // Add this node to our contacts.
CAF_LOG_INFO("new endpoint:" << CAF_ARG(hdr.source_node)); CAF_LOG_INFO("new endpoint:" << CAF_ARG(hdr.source_node));
tbl_.add(hdr.source_node, hdl); if (lr.known)
tbl_.handle(hdr.source_node, hdl);
else
tbl_.add(hdr.source_node, hdl);
auto config_server = system().registry().get(atom("ConfigServ")); auto config_server = system().registry().get(atom("ConfigServ"));
anon_send(actor_cast<actor>(config_server), put_atom::value, anon_send(actor_cast<actor>(config_server), put_atom::value,
to_string(hdr.source_node), make_message(addrs)); to_string(hdr.source_node), make_message(addrs));
......
...@@ -97,6 +97,11 @@ strong_actor_ptr basp_broker_state::make_proxy(node_id nid, actor_id aid) { ...@@ -97,6 +97,11 @@ strong_actor_ptr basp_broker_state::make_proxy(node_id nid, actor_id aid) {
// payload received from a remote node; if a remote node A sends // payload received from a remote node; if a remote node A sends
// us a handle to a third node B, then we assume that A can tell us // us a handle to a third node B, then we assume that A can tell us
// how to contact B. // how to contact B.
// TODO: This should probably happen somewhere else, but is usually only
// performed in `finalize_handshake` which is only called on receipt
// of server handshakes.
if (this_context->id == none)
this_context->id = instance.tbl().lookup(this_context->hdl);
auto lr = instance.tbl().lookup(nid); auto lr = instance.tbl().lookup(nid);
if (nid != this_context->id && !lr.known) { if (nid != this_context->id && !lr.known) {
instance.tbl().add(nid, this_context->id); instance.tbl().add(nid, this_context->id);
...@@ -414,7 +419,6 @@ void basp_broker_state::establish_communication(const node_id& nid) { ...@@ -414,7 +419,6 @@ void basp_broker_state::establish_communication(const node_id& nid) {
// TODO: Split this by functionality, address query & connecting? // TODO: Split this by functionality, address query & connecting?
CAF_ASSERT(this_context != nullptr); CAF_ASSERT(this_context != nullptr);
CAF_LOG_TRACE(CAF_ARG(nid)); CAF_LOG_TRACE(CAF_ARG(nid));
learned_new_node(nid);
if (!enable_automatic_connections) if (!enable_automatic_connections)
return; return;
// this member function gets only called once, after adding a new // this member function gets only called once, after adding a new
...@@ -452,7 +456,7 @@ void basp_broker_state::establish_communication(const node_id& nid) { ...@@ -452,7 +456,7 @@ void basp_broker_state::establish_communication(const node_id& nid) {
}); });
basp::header hdr{basp::message_type::dispatch_message, basp::header hdr{basp::message_type::dispatch_message,
basp::header::named_receiver_flag, basp::header::named_receiver_flag,
0, 0, this_node(), nid, tmp.id(), invalid_actor_id, 0, 0, this_node(), *origin, tmp.id(), invalid_actor_id,
visit(seq_num_visitor{this}, hdl)}; visit(seq_num_visitor{this}, hdl)};
instance.write(self->context(), get_buffer(hdl), instance.write(self->context(), get_buffer(hdl),
hdr, &writer); hdr, &writer);
...@@ -669,7 +673,7 @@ basp_broker_state::get_buffer(node_id nid) { ...@@ -669,7 +673,7 @@ basp_broker_state::get_buffer(node_id nid) {
if (res.known && res.hdl) { if (res.known && res.hdl) {
return get_buffer(*res.hdl); return get_buffer(*res.hdl);
} }
auto msgs = pending_connectivity[nid]; auto& msgs = pending_connectivity[nid];
msgs.emplace_back(); msgs.emplace_back();
return msgs.back(); return msgs.back();
} }
......
...@@ -108,6 +108,7 @@ bool routing_table::handle(const node_id& nid, ...@@ -108,6 +108,7 @@ bool routing_table::handle(const node_id& nid,
if (i == node_information_base_.end()) if (i == node_information_base_.end())
return false; return false;
i->second.hdl = hdl; i->second.hdl = hdl;
nid_by_hdl_.emplace(hdl, nid);
return true; return true;
} }
......
...@@ -25,6 +25,8 @@ ...@@ -25,6 +25,8 @@
#include <thread> #include <thread>
#include <vector> #include <vector>
#include "caf/test/io_dsl.hpp"
#include "caf/all.hpp" #include "caf/all.hpp"
#include "caf/io/all.hpp" #include "caf/io/all.hpp"
...@@ -39,6 +41,9 @@ using std::string; ...@@ -39,6 +41,9 @@ using std::string;
using ping_atom = atom_constant<atom("ping")>; using ping_atom = atom_constant<atom("ping")>;
using pong_atom = atom_constant<atom("pong")>; using pong_atom = atom_constant<atom("pong")>;
using test_one_atom = atom_constant<atom("test_one")>;
using done_atom = atom_constant<atom("shutdown")>;
/* /*
This test checks whether automatic connections work as expected This test checks whether automatic connections work as expected
...@@ -65,223 +70,190 @@ using pong_atom = atom_constant<atom("pong")>; ...@@ -65,223 +70,190 @@ using pong_atom = atom_constant<atom("pong")>;
*/ */
/* namespace {
std::thread run_prog(const char* arg, uint16_t port, bool use_asio) {
return detail::run_sub_unit_test(invalid_actor, constexpr uint16_t port_earth = 12340;
test::engine::path(), constexpr uint16_t port_mars = 12341;
test::engine::max_runtime(), constexpr uint16_t port_jupiter = 12342;
CAF_XSTR(CAF_SUITE),
use_asio, class config : public actor_system_config {
{"--port=" + std::to_string(port), arg}); public:
} config(bool use_tcp = true) {
load<caf::io::middleman, io::network::test_multiplexer>();
set("scheduler.policy", caf::atom("testing"));
set("middleman.detach-utility-actors", false);
set("middleman.enable-automatic-connections", true);
set("middleman.enable-tcp", use_tcp);
set("middleman.enable-udp", !use_tcp);
}
};
class simple_config : public actor_system_config {
public:
simple_config(bool use_tcp = true) {
load<caf::io::middleman>();
set("middleman.enable-automatic-connections", true);
set("middleman.enable-tcp", use_tcp);
set("middleman.enable-udp", !use_tcp);
}
};
class fixture {
public:
fixture(bool use_tcp = true)
: cfg_earth(use_tcp),
cfg_mars(use_tcp),
cfg_jupiter(use_tcp),
earth(cfg_earth),
mars(cfg_mars),
jupiter(cfg_jupiter) {
CAF_MESSAGE("Earth : " << to_string(earth.node()));
CAF_MESSAGE("Mars : " << to_string(mars.node()));
CAF_MESSAGE("Jupiter: " << to_string(jupiter.node()));
}
simple_config cfg_earth;
simple_config cfg_mars;
simple_config cfg_jupiter;
actor_system earth{cfg_earth};
actor_system mars{cfg_mars};
actor_system jupiter{cfg_jupiter};
};
// we run the same code on all three nodes, a simple ping-pong client class fixture_udp : public fixture {
struct testee_state { public:
std::set<actor> buddies; fixture_udp() : fixture(false) {
uint16_t port = 0; // nop
const char* name = "testee"; }
}; };
behavior testee(stateful_actor<testee_state>* self) { behavior actor_jupiter(event_based_actor* self, actor mars) {
return { return {
[self](ping_atom, actor buddy, bool please_broadcast) -> message { [=](test_one_atom) {
if (please_broadcast) CAF_MESSAGE("sending message from Jupiter to Mars");
for (auto& x : self->state.buddies) self->send(mars, test_one_atom::value, self);
if (x != buddy)
send_as(buddy, x, ping_atom::value, buddy, false);
self->state.buddies.emplace(std::move(buddy));
return make_message(pong_atom::value, self);
}, },
[self](pong_atom, actor buddy) { [=](done_atom) {
self->state.buddies.emplace(std::move(buddy)); CAF_MESSAGE("Jupiter received message from Earth, shutting down");
}, self->quit();
[self](put_atom, uint16_t new_port) {
self->state.port = new_port;
},
[self](get_atom) {
return self->state.port;
} }
}; };
} }
void run_earth(bool use_asio, bool as_server, uint16_t pub_port) { behavior actor_mars(event_based_actor* self, actor earth) {
scoped_actor self{system}; return {
struct captain : hook { [=](done_atom) {
public: CAF_MESSAGE("Mars received message from Earth, shutting down");
captain(actor parent) : parent_(std::move(parent)) { self->quit();
// nop },
[=](test_one_atom, actor jupiter) {
CAF_MESSAGE("sending message from Mars to Earth");
self->send(earth, test_one_atom::value, jupiter, self);
} }
};
}
void new_connection_established_cb(const node_id& node) override { behavior actor_earth(event_based_actor* self) {
anon_send(parent_, put_atom::value, node); return {
call_next<hook::new_connection_established>(node); [=](test_one_atom, actor jupiter, actor mars) {
CAF_MESSAGE("message from Jupiter reached Earth, "
"replying and shutting down");
self->send(mars, done_atom::value);
self->send(jupiter, done_atom::value);
self->quit();
} }
};
}
void new_remote_actor_cb(const actor_addr& addr) override { } // namespace <anonymous>
anon_send(parent_, put_atom::value, addr);
call_next<hook::new_remote_actor>(addr);
}
void connection_lost_cb(const node_id& dest) override { CAF_TEST_FIXTURE_SCOPE(autoconn_tcp_simple_test, fixture)
anon_send(parent_, delete_atom::value, dest);
}
private: CAF_TEST(build_triangle_simple_tcp) {
actor parent_; CAF_MESSAGE("setting up Earth");
}; auto on_earth = earth.spawn(actor_earth);
middleman::instance()->add_hook<captain>(self); auto earth_port = earth.middleman().publish(on_earth, 0);
auto aut = system.spawn(testee); CAF_REQUIRE(earth_port);
auto port = publish(aut, pub_port); CAF_MESSAGE("Earth reachable via " << *earth_port);
CAF_MESSAGE("published testee at port " << port); CAF_MESSAGE("setting up Mars");
std::thread mars_process; auto from_earth = mars.middleman().remote_actor("localhost", *earth_port);
std::thread jupiter_process; CAF_REQUIRE(from_earth);
// launch process for Mars auto on_mars = mars.spawn(actor_mars, *from_earth);
if (!as_server) { auto mars_port = mars.middleman().publish(on_mars, 0);
CAF_MESSAGE("launch process for Mars"); CAF_REQUIRE(mars_port);
mars_process = run_prog("--mars", port, use_asio); CAF_MESSAGE("Mars reachable via " << *mars_port);
} CAF_MESSAGE("setting up Jupiter");
CAF_MESSAGE("wait for Mars to connect"); auto from_mars = jupiter.middleman().remote_actor("localhost", *mars_port);
node_id mars; CAF_REQUIRE(from_mars);
self->receive( auto on_jupiter = jupiter.spawn(actor_jupiter, *from_mars);
[&](put_atom, const node_id& nid) { CAF_MESSAGE("forwarding an actor from Jupiter to Earth via Mars");
mars = nid; anon_send(on_jupiter, test_one_atom::value);
CAF_MESSAGE(CAF_ARG(mars)); jupiter.await_all_actors_done();
} mars.await_all_actors_done();
); earth.await_all_actors_done();
actor_addr mars_addr;
uint16_t mars_port;
self->receive_while([&] { return mars_addr == invalid_actor_addr; })(
[&](put_atom, const actor_addr& addr) {
auto hdl = actor_cast<actor>(addr);
self->request(hdl, sys_atom::value, get_atom::value, "info").then(
[&](ok_atom, const string&, const actor_addr&, const string& name) {
if (name != "testee")
return;
mars_addr = addr;
CAF_MESSAGE(CAF_ARG(mars_addr));
self->request(actor_cast<actor>(mars_addr), get_atom::value).then(
[&](uint16_t mp) {
CAF_MESSAGE("mars published its actor at port " << mp);
mars_port = mp;
}
);
}
);
}
);
// launch process for Jupiter
if (!as_server) {
CAF_MESSAGE("launch process for Jupiter");
jupiter_process = run_prog("--jupiter", mars_port, use_asio);
}
CAF_MESSAGE("wait for Jupiter to connect");
self->receive(
[](put_atom, const node_id& jupiter) {
CAF_MESSAGE(CAF_ARG(jupiter));
}
);
actor_addr jupiter_addr;
self->receive_while([&] { return jupiter_addr == invalid_actor_addr; })(
[&](put_atom, const actor_addr& addr) {
auto hdl = actor_cast<actor>(addr);
self->request(hdl, sys_atom::value, get_atom::value, "info").then(
[&](ok_atom, const string&, const actor_addr&, const string& name) {
if (name != "testee")
return;
jupiter_addr = addr;
CAF_MESSAGE(CAF_ARG(jupiter_addr));
}
);
}
);
CAF_MESSAGE("shutdown Mars");
anon_send_exit(mars_addr, exit_reason::kill);
if (mars_process.joinable())
mars_process.join();
self->receive(
[&](delete_atom, const node_id& nid) {
CAF_CHECK(nid == mars);
}
);
CAF_MESSAGE("check whether we still can talk to Jupiter");
self->send(aut, ping_atom::value, self, true);
std::set<actor_addr> found;
int i = 0;
self->receive_for(i, 2)(
[&](pong_atom, const actor&) {
found.emplace(self->current_sender());
}
);
std::set<actor_addr> expected{aut.address(), jupiter_addr};
CAF_CHECK(found == expected);
CAF_MESSAGE("shutdown Jupiter");
anon_send_exit(jupiter_addr, exit_reason::kill);
if (jupiter_process.joinable())
jupiter_process.join();
anon_send_exit(aut, exit_reason::kill);
} }
void run_mars(uint16_t port_to_earth, uint16_t pub_port) { CAF_TEST_FIXTURE_SCOPE_END()
auto aut = system.spawn(testee);
auto port = publish(aut, pub_port);
anon_send(aut, put_atom::value, port);
CAF_MESSAGE("published testee at port " << port);
auto earth = remote_actor("localhost", port_to_earth);
send_as(aut, earth, ping_atom::value, aut, false);
}
void run_jupiter(uint16_t port_to_mars) { CAF_TEST_FIXTURE_SCOPE(autoconn_tcp_test, belt_fixture_t<config>)
auto aut = system.spawn(testee);
auto mars = remote_actor("localhost", port_to_mars); CAF_TEST(build_triangle_tcp) {
send_as(aut, mars, ping_atom::value, aut, true); CAF_MESSAGE("Earth : " << to_string(earth.sys.node()));
} CAF_MESSAGE("Mars : " << to_string(mars.sys.node()));
*/ CAF_MESSAGE("Jupiter: " << to_string(jupiter.sys.node()));
CAF_MESSAGE("setting up Earth");
auto on_earth = earth.sys.spawn(actor_earth);
// scoped_actor on_earth{earth.sys};
CAF_MESSAGE("run initialization code");
exec_all();
CAF_MESSAGE("prepare connection");
prepare_connection(earth, mars, "earth", port_earth);
CAF_MESSAGE("publish dummy on earth");
// earth.publish(actor_cast<actor>(on_earth), port_earth);
earth.publish(on_earth, port_earth);
CAF_MESSAGE("setting up Mars");
auto from_earth = mars.remote_actor("earth", port_earth);
CAF_REQUIRE(from_earth);
auto on_mars = mars.sys.spawn(actor_mars, from_earth);
CAF_MESSAGE("run initialization code");
exec_all();
CAF_MESSAGE("prepare connection");
prepare_connection(mars, jupiter, "mars", port_mars);
CAF_MESSAGE("publish dummy on earth");
mars.publish(on_mars, port_mars);
CAF_TEST(triangle_setup) { CAF_MESSAGE("setting up Jupiter");
// this unit test is temporarily disabled until problems auto from_mars = jupiter.remote_actor("mars", port_mars);
// with OBS are sorted out or new actor_system API is in place CAF_REQUIRE(from_mars);
auto on_jupiter = jupiter.sys.spawn(actor_jupiter, from_mars);
// This handle will be created by the test multiplexer for the automatically
// opened socket when automatic connections are enabled.
auto hdl_jupiter = accept_handle::from_int(std::numeric_limits<int64_t>::max());
// Perpare autmomatic connection between Jupiter and Earth,
prepare_connection(jupiter, earth, "jupiter", port_jupiter, hdl_jupiter);
// Add the address information for this test to the config server on Mars.
auto mars_config_server = mars.sys.registry().get(atom("ConfigServ"));
network::address_listing interfaces{
{network::protocol::ipv4, std::vector<std::string>{"jupiter"}}
};
basp::routing_table::address_map addrs{
{network::protocol::tcp, {port_jupiter, interfaces}}
};
anon_send(actor_cast<actor>(mars_config_server), put_atom::value,
to_string(jupiter.sys.node()), make_message(addrs));
CAF_MESSAGE("forwarding an actor from Jupiter to Earth via Mars.");
anon_send(on_jupiter, test_one_atom::value);
exec_all();
} }
/* CAF_TEST(break_triangle_tcp) {
CAF_TEST(triangle_setup) { // TODO: Implement the same test as above, but kill the intermediate node
uint16_t port = 0; // that helped establish the connection.
uint16_t publish_port = 0;
auto argv = test::engine::argv();
auto argc = test::engine::argc();
auto r = message_builder(argv, argv + argc).extract_opts({
{"port,p", "port of remote side (when running mars or jupiter)", port},
{"mars", "run mars"},
{"jupiter", "run jupiter"},
{"use-asio", "use ASIO network backend (if available)"},
{"server,s", "run in server mode (don't run clients)", publish_port}
});
// check arguments
bool is_mars = r.opts.count("mars") > 0;
bool is_jupiter = r.opts.count("jupiter") > 0;
bool has_port = r.opts.count("port") > 0;
if (((is_mars || is_jupiter) && !has_port) || (is_mars && is_jupiter)) {
CAF_ERROR("need a port when running Mars or Jupiter and cannot "
"both at the same time");
return;
}
// enable automatic connections
anon_send(whereis(atom("ConfigServ")), put_atom::value,
"middleman.enable-automatic-connections", make_message(true));
auto use_asio = r.opts.count("use-asio") > 0;
# ifdef CAF_USE_ASIO
if (use_asio) {
CAF_MESSAGE("enable ASIO backend");
set_middleman<network::asio_multiplexer>();
}
# endif // CAF_USE_ASIO
auto as_server = r.opts.count("server") > 0;
if (is_mars)
run_mars(port, publish_port);
else if (is_jupiter)
run_jupiter(port);
else
run_earth(use_asio, as_server, publish_port);
await_all_actors_done();
shutdown();
} }
*/
CAF_TEST_FIXTURE_SCOPE_END()
...@@ -52,8 +52,7 @@ constexpr uint64_t no_operation_data = 0; ...@@ -52,8 +52,7 @@ constexpr uint64_t no_operation_data = 0;
constexpr auto basp_atom = caf::atom("BASP"); constexpr auto basp_atom = caf::atom("BASP");
constexpr auto spawn_serv_atom = caf::atom("SpawnServ"); constexpr auto spawn_serv_atom = caf::atom("SpawnServ");
// TODO: Will probably be removed when the new default autoconnect works. constexpr auto config_serv_atom = caf::atom("ConfigServ");
// constexpr auto config_serv_atom = caf::atom("ConfigServ");
} // namespace <anonymous> } // namespace <anonymous>
...@@ -278,7 +277,8 @@ public: ...@@ -278,7 +277,8 @@ public:
void connect_node(node& n, void connect_node(node& n,
optional<accept_handle> ax = none, optional<accept_handle> ax = none,
actor_id published_actor_id = invalid_actor_id, actor_id published_actor_id = invalid_actor_id,
const set<string>& published_actor_ifs = std::set<std::string>{}) { const set<string>& published_actor_ifs = std::set<std::string>{},
const basp::routing_table::address_map& local_addrs_map = {}) {
auto src = ax ? *ax : ahdl_; auto src = ax ? *ax : ahdl_;
CAF_MESSAGE("connect remote node " << n.name CAF_MESSAGE("connect remote node " << n.name
<< ", connection ID = " << n.connection.id() << ", connection ID = " << n.connection.id()
...@@ -299,7 +299,7 @@ public: ...@@ -299,7 +299,7 @@ public:
published_actor_id, invalid_actor_id, std::string{}, published_actor_id, invalid_actor_id, std::string{},
published_actor_id, published_actor_id,
published_actor_ifs, published_actor_ifs,
basp::routing_table::address_map{}) local_addrs_map)
// upon receiving our client handshake, BASP will check // upon receiving our client handshake, BASP will check
// whether there is a SpawnServ actor on this node // whether there is a SpawnServ actor on this node
.receive(hdl, .receive(hdl,
...@@ -456,12 +456,16 @@ public: ...@@ -456,12 +456,16 @@ public:
scheduler_type& sched; scheduler_type& sched;
middleman_actor mma; middleman_actor mma;
actor_system_config phobos_cfg;
actor_system phobos;
scoped_actor on_phobos;
autoconn_enabled_fixture() autoconn_enabled_fixture()
: fixture(true), : fixture(true),
sched(dynamic_cast<scheduler_type&>(sys.scheduler())), sched(dynamic_cast<scheduler_type&>(sys.scheduler())),
mma(sys.middleman().actor_handle()) { mma(sys.middleman().actor_handle()),
phobos(phobos_cfg.load<caf::io::middleman>()),
on_phobos(phobos) {
// nop // nop
} }
...@@ -710,9 +714,62 @@ CAF_TEST(actor_serialize_and_deserialize) { ...@@ -710,9 +714,62 @@ CAF_TEST(actor_serialize_and_deserialize) {
std::vector<actor_id>{}, msg); std::vector<actor_id>{}, msg);
} }
// TODO: Will be reactivated and adjusted when new conenction feature works. CAF_TEST_FIXTURE_SCOPE_END()
CAF_TEST_FIXTURE_SCOPE(basp_tests_with_autoconn, autoconn_enabled_fixture)
CAF_TEST(address_handshake) {
// Test whether basp instance correctly sends a server handshake
// when there's no actor published and automatic connections are enabled.
buffer buf;
instance().write_server_handshake(mpx(), buf, none);
auto addrs = instance().tbl().local_addresses();
CAF_CHECK(!addrs.empty());
CAF_CHECK(addrs.count(network::protocol::tcp) > 0 &&
!addrs[network::protocol::tcp].second.empty());
CAF_CHECK(addrs.count(network::protocol::udp) == 0);
buffer expected_buf;
basp::header expected{basp::message_type::server_handshake, 0, 0,
basp::version, this_node(), none,
invalid_actor_id, invalid_actor_id};
to_buf(expected_buf, expected, nullptr, std::string{},
invalid_actor_id, set<string>{}, addrs);
CAF_CHECK_EQUAL(hexstr(buf), hexstr(expected_buf));
}
CAF_TEST(read_address_after_handshake) {
mpx()->provide_scribe("jupiter", 8080, jupiter().connection);
CAF_CHECK(mpx()->has_pending_scribe("jupiter", 8080));
CAF_MESSAGE("self: " << to_string(self()->address()));
auto ax = accept_handle::from_int(4242);
mpx()->provide_acceptor(4242, ax);
publish(self(), 4242);
mpx()->flush_runnables(); // process publish message in basp_broker
CAF_MESSAGE("connect to mars");
auto& addrs = instance().tbl().local_addresses();
connect_node(mars(), ax, self()->id(), std::set<string>{}, addrs);
CAF_MESSAGE("Look for mars address information in our config server");
auto config_server = sys.registry().get(config_serv_atom);
self()->send(actor_cast<actor>(config_server), get_atom::value,
to_string(mars().id));
sched.run();
mpx()->flush_runnables(); // process get request and send answer
self()->receive(
[&](const std::string& item, message& msg) {
// Check that we got an entry under the name of our peer.
CAF_REQUIRE_EQUAL(item, to_string(mars().id));
msg.apply(
[&](basp::routing_table::address_map& addrs) {
// The addresses of our dummy node, thus empty.
CAF_CHECK(addrs.empty());
}
);
}
);
}
/* /*
CAF_TEST(indirect_connections) { CAF_TEST(build_connections) {
// this node receives a message from jupiter via mars and responds via mars // this node receives a message from jupiter via mars and responds via mars
// and any ad-hoc automatic connection requests are ignored // and any ad-hoc automatic connection requests are ignored
CAF_MESSAGE("self: " << to_string(self()->address())); CAF_MESSAGE("self: " << to_string(self()->address()));
...@@ -763,37 +820,92 @@ CAF_TEST(indirect_connections) { ...@@ -763,37 +820,92 @@ CAF_TEST(indirect_connections) {
} }
*/ */
CAF_TEST_FIXTURE_SCOPE_END()
CAF_TEST_FIXTURE_SCOPE(basp_tests_with_autoconn, autoconn_enabled_fixture)
// TODO: Will probably be removed when the new default autoconnect works.
/*
CAF_TEST(automatic_connection) { CAF_TEST(automatic_connection) {
// this tells our BASP broker to enable the automatic connection feature
//anon_send(aut(), ok_atom::value,
// "middleman.enable-automatic-connections", make_message(true));
//mpx()->exec_runnable(); // process publish message in basp_broker
// jupiter [remote hdl 0] -> mars [remote hdl 1] -> earth [this_node] // jupiter [remote hdl 0] -> mars [remote hdl 1] -> earth [this_node]
// (this node receives a message from jupiter via mars and responds via mars, // (this node receives a message from jupiter via mars and responds via mars,
// but then also establishes a connection to jupiter directly) // but then also establishes a connection to jupiter directly)
/*
CAF_MESSAGE("! This test case changes the node id and dummy actor of jupiter\n"
" to a different actor system to trigger the creation of a\n"
" proxy on receipt of the dummy actor.");
// Create a new system to have an actor with an id from a remote system
// to test the automatic connection setup.
actor_system_config fake_cfg;
fake_cfg.load<caf::io::middleman>();
actor_system fake_sys(fake_cfg);
scoped_actor another_fake_actor(fake_sys);
new (&jupiter().dummy_actor) scoped_actor(fake_sys);
jupiter().id = fake_sys.node();
CAF_MESSAGE("Jupiter: " << to_string(jupiter().id));
*/
/*
auto check_node_in_tbl = [&](node& n) { auto check_node_in_tbl = [&](node& n) {
io::id_visitor id_vis; io::id_visitor id_vis;
auto ehdl = tbl().lookup(n.id); auto lr = tbl().lookup(n.id);
CAF_REQUIRE(ehdl); CAF_REQUIRE(lr.hdl);
CAF_CHECK_EQUAL(visit(id_vis, *ehdl), n.connection.id()); CAF_CHECK_EQUAL(visit(id_vis, *lr.hdl), n.connection.id());
}; };
mpx()->provide_scribe("jupiter", 8080, jupiter().connection); mpx()->provide_scribe("jupiter", 8080, jupiter().connection);
CAF_CHECK(mpx()->has_pending_scribe("jupiter", 8080)); CAF_CHECK(mpx()->has_pending_scribe("jupiter", 8080));
CAF_MESSAGE("self: " << to_string(self()->address())); CAF_MESSAGE("Earth actor : " << to_string(self()));
CAF_MESSAGE("Jupiter actor: " << to_string(jupiter().dummy_actor));
CAF_MESSAGE("Mars actor : " << to_string(mars().dummy_actor));
auto ax = accept_handle::from_int(4242); auto ax = accept_handle::from_int(4242);
mpx()->provide_acceptor(4242, ax); mpx()->provide_acceptor(4242, ax);
publish(self(), 4242); publish(self(), 4242);
mpx()->flush_runnables(); // process publish message in basp_broker mpx()->flush_runnables(); // process publish message in basp_broker
CAF_MESSAGE("connect to mars"); CAF_MESSAGE("connect to mars");
connect_node(mars(), ax, self()->id()); auto& addrs = instance().tbl().local_addresses();
connect_node(mars(), ax, self()->id(), std::set<string>{}, addrs);
//CAF_CHECK_EQUAL(tbl().lookup_direct(mars().id).id(), mars().connection.id()); //CAF_CHECK_EQUAL(tbl().lookup_direct(mars().id).id(), mars().connection.id());
check_node_in_tbl(mars()); check_node_in_tbl(mars());
// TODO:
// - Send Jupiter's actor from mars to Earth.
// - Expect the ConfigServ msg at Mars.
// - Reply with the connection info for Mars.
// - Handle the direct handshake.
// - Sounds about right!
// Our dummy actors all belong to the same system,
// here is a fake actor from Jupiter.
// TODO: Should really be an actor ...
CAF_MESSAGE("Mars shares an actor located at Jupiter with Earth.");
// TODO: This fails. Apparently it is not the same thing, sending the actor
// or forwardings an actor received earlier. Not sure how to solve this, yet.
mock(mars().connection,
{basp::message_type::dispatch_message, 0, 0, 0,
mars().id, this_node(),
mars().dummy_actor->id(), self()->id()},
std::vector<actor_id>{},
make_message("Here is jupiter's actor!", on_phobos));
// CAF_MESSAGE("Receive the msg from Mars");
// self()->receive(
// [&](std::string& text, actor sender) {
// CAF_MESSAGE("Received '" << text << "' from '"
// << to_string(sender) << "'.");
// CAF_MESSAGE("Self is '" << to_string(self()) << "'.");
// }
// );
CAF_MESSAGE("Earth announces a proxy for the sender from Mars.");
mock()
.receive(mars().connection,
basp::message_type::announce_proxy,
no_flags, no_payload, no_operation_data,
this_node(), mars().id,
invalid_actor_id, mars().dummy_actor->id());
sched.run();
mpx()->flush_runnables();
CAF_MESSAGE("Earth requests connection info for Jupiter");
mock()
.receive(mars().connection,
basp::message_type::dispatch_message,
basp::header::named_receiver_flag, any_vals, no_operation_data,
this_node(), mars().id, any_vals, invalid_actor_id,
config_serv_atom,
std::vector<actor_id>{},
make_message(get_atom::value, to_string(jupiter().id)));
*/
/*
CAF_MESSAGE("simulate that an actor from jupiter " CAF_MESSAGE("simulate that an actor from jupiter "
"sends a message to us via mars"); "sends a message to us via mars");
mock(mars().connection, mock(mars().connection,
...@@ -823,7 +935,7 @@ CAF_TEST(automatic_connection) { ...@@ -823,7 +935,7 @@ CAF_TEST(automatic_connection) {
no_operation_data, this_node(), jupiter().id, no_operation_data, this_node(), jupiter().id,
invalid_actor_id, jupiter().dummy_actor->id()); invalid_actor_id, jupiter().dummy_actor->id());
CAF_CHECK_EQUAL(mpx()->output_buffer(mars().connection).size(), 0u); CAF_CHECK_EQUAL(mpx()->output_buffer(mars().connection).size(), 0u);
CAF_CHECK_EQUAL(tbl().lookup(jupiter().id), mars().id); CAF_CHECK_EQUAL(*tbl().lookup(jupiter().id).hdl, mars().id);
CAF_CHECK_EQUAL(tbl().lookup(mars().id), none); CAF_CHECK_EQUAL(tbl().lookup(mars().id), none);
auto connection_helper_actor = sys.latest_actor_id(); auto connection_helper_actor = sys.latest_actor_id();
CAF_CHECK_EQUAL(mpx()->output_buffer(mars().connection).size(), 0u); CAF_CHECK_EQUAL(mpx()->output_buffer(mars().connection).size(), 0u);
...@@ -833,7 +945,7 @@ CAF_TEST(automatic_connection) { ...@@ -833,7 +945,7 @@ CAF_TEST(automatic_connection) {
res[network::protocol::ipv4].emplace_back("jupiter"); res[network::protocol::ipv4].emplace_back("jupiter");
mock(mars().connection, mock(mars().connection,
{basp::message_type::dispatch_message, 0, 0, 0, {basp::message_type::dispatch_message, 0, 0, 0,
this_node(), this_node(), this_node(), jupiter(),
invalid_actor_id, connection_helper_actor}, invalid_actor_id, connection_helper_actor},
std::vector<actor_id>{}, std::vector<actor_id>{},
make_message("basp.default-connectivity-tcp", make_message("basp.default-connectivity-tcp",
...@@ -857,8 +969,8 @@ CAF_TEST(automatic_connection) { ...@@ -857,8 +969,8 @@ CAF_TEST(automatic_connection) {
basp::message_type::client_handshake, no_flags, 1u, basp::message_type::client_handshake, no_flags, 1u,
no_operation_data, this_node(), jupiter().id, no_operation_data, this_node(), jupiter().id,
invalid_actor_id, invalid_actor_id, std::string{}); invalid_actor_id, invalid_actor_id, std::string{});
CAF_CHECK_EQUAL(tbl().lookup_indirect(jupiter().id), none); CAF_CHECK_EQUAL(tbl().lookup(jupiter().id).hdl, none);
CAF_CHECK_EQUAL(tbl().lookup_indirect(mars().id), none); CAF_CHECK_EQUAL(tbl().lookup(mars().id).hdl, none);
check_node_in_tbl(jupiter()); check_node_in_tbl(jupiter());
check_node_in_tbl(mars()); check_node_in_tbl(mars());
CAF_MESSAGE("receive message from jupiter"); CAF_MESSAGE("receive message from jupiter");
...@@ -878,7 +990,7 @@ CAF_TEST(automatic_connection) { ...@@ -878,7 +990,7 @@ CAF_TEST(automatic_connection) {
std::vector<actor_id>{}, std::vector<actor_id>{},
make_message("hello from earth!")); make_message("hello from earth!"));
CAF_CHECK_EQUAL(mpx()->output_buffer(mars().connection).size(), 0u); CAF_CHECK_EQUAL(mpx()->output_buffer(mars().connection).size(), 0u);
*/
} }
*/
CAF_TEST_FIXTURE_SCOPE_END() CAF_TEST_FIXTURE_SCOPE_END()
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