Unverified Commit 2cae44fc authored by Joseph Noir's avatar Joseph Noir Committed by GitHub

Merge pull request #730

Improve deterministic I/O DSL and port remote group test to DSL
parents 9b2c2867 214f101f
......@@ -17,7 +17,7 @@
******************************************************************************/
#define CAF_SUITE config_option
#include "caf/test/unit_test.hpp"
#include "caf/test/dsl.hpp"
#include "caf/config_option.hpp"
#include "caf/make_config_option.hpp"
......@@ -87,13 +87,6 @@ void check_integer_options() {
check_integer_options<T>(tk);
}
template <class T>
T unbox(optional<T> x) {
if (!x)
CAF_FAIL("no value to unbox");
return std::move(*x);
}
} // namespace <anonymous>
CAF_TEST(type_bool) {
......
......@@ -168,7 +168,7 @@ expected<uint16_t> middleman::publish_local_groups(uint16_t port,
}
};
};
auto gn = system().spawn<hidden>(group_nameserver);
auto gn = system().spawn<hidden + lazy_init>(group_nameserver);
auto result = publish(gn, port, in, reuse);
// link gn to our manager
if (result)
......@@ -241,15 +241,40 @@ expected<group> middleman::remote_group(const std::string& group_uri) {
}
expected<group> middleman::remote_group(const std::string& group_identifier,
const std::string& host, uint16_t port) {
const std::string& host,
uint16_t port) {
CAF_LOG_TRACE(CAF_ARG(group_identifier) << CAF_ARG(host) << CAF_ARG(port));
auto group_server = remote_actor(host, port);
if (!group_server)
return std::move(group_server.error());
scoped_actor self{system(), true};
self->send(*group_server, get_atom::value, group_identifier);
// Helper actor that first connects to the remote actor at `host:port` and
// then tries to get a valid group from that actor.
auto two_step_lookup = [=](event_based_actor* self,
middleman_actor mm) -> behavior {
return {
[=](get_atom) {
/// We won't receive a second message, so we drop our behavior here to
/// terminate the actor after both requests finish.
self->unbecome();
auto rp = self->make_response_promise();
self->request(mm, infinite, connect_atom::value, host, port).then(
[=](const node_id&, strong_actor_ptr& ptr,
const std::set<std::string>&) mutable {
auto hdl = actor_cast<actor>(ptr);
self->request(hdl, infinite, get_atom::value, group_identifier)
.then(
[=](group& result) mutable {
rp.deliver(std::move(result));
}
);
}
);
}
};
};
// Spawn the helper actor and wait for the result.
expected<group> result{sec::cannot_connect_to_node};
self->receive(
scoped_actor self{system(), true};
self->request(self->spawn<lazy_init>(two_step_lookup, actor_handle()),
infinite, get_atom::value)
.receive(
[&](group& grp) {
result = std::move(grp);
},
......@@ -357,8 +382,10 @@ void middleman::init(actor_system_config& cfg) {
// never detach actors when using the testing multiplexer
auto network_backend = get_or(cfg, "middleman.network-backend",
defaults::middleman::network_backend);
if (network_backend == atom("testing"))
cfg.set("middleman.attach-utility-actors", true);
if (network_backend == atom("testing")) {
cfg.set("middleman.attach-utility-actors", true)
.set("middleman.manual-multiplexing", true);
}
// add remote group module to config
struct remote_groups : group_module {
public:
......
......@@ -19,7 +19,7 @@
#include "caf/config.hpp"
#define CAF_SUITE io_dynamic_remote_group
#include "caf/test/unit_test.hpp"
#include "caf/test/io_dsl.hpp"
#include <vector>
#include <algorithm>
......@@ -31,140 +31,98 @@ using namespace caf;
namespace {
constexpr char local_host[] = "127.0.0.1";
class config : public caf::actor_system_config {
public:
config() {
load<caf::io::middleman>();
add_message_type<std::vector<actor>>("std::vector<actor>");
}
config& parse() {
actor_system_config::parse(caf::test::engine::argc(),
caf::test::engine::argv());
return *this;
}
};
struct fixture {
config server_side_cfg;
caf::actor_system server_side{server_side_cfg.parse()};
config client_side_cfg;
caf::actor_system client_side{client_side_cfg.parse()};
io::middleman& server_side_mm = server_side.middleman();
io::middleman& client_side_mm = client_side.middleman();
};
const uint16_t port = 8080;
const char* server = "mars";
behavior make_reflector_behavior(event_based_actor* self) {
const char* group_name = "foobar";
size_t received_messages = 0u;
behavior group_receiver(event_based_actor* self) {
self->set_default_handler(reflect_and_quit);
return {
[] {
// nop
[](ok_atom) {
++received_messages;
}
};
}
using spawn_atom = atom_constant<atom("Spawn")>;
using get_group_atom = atom_constant<atom("GetGroup")>;
struct await_reflector_reply_behavior {
event_based_actor* self;
int cnt;
int downs;
std::vector<actor> vec;
void operator()(const std::string& str, double val) {
CAF_CHECK_EQUAL(str, "Hello reflector!");
CAF_CHECK_EQUAL(val, 5.0);
if (++cnt == 7) {
for (const auto& actor : vec)
self->monitor(actor);
self->set_down_handler([=](down_msg&) {
if (++downs == 5)
self->quit();
});
}
// Our server is `mars` and our client is `earth`.
struct fixture : point_to_point_fixture<test_coordinator_fixture<config>> {
fixture() {
prepare_connection(mars, earth, server, port);
}
};
// `grp` may be either local or remote
void make_client_behavior(event_based_actor* self,
const actor& server, group grp) {
self->set_default_handler(skip);
self->spawn_in_group(grp, make_reflector_behavior);
self->spawn_in_group(grp, make_reflector_behavior);
self->request(server, infinite, spawn_atom::value, grp).then(
[=](const std::vector<actor>& vec) {
auto is_remote = [=](actor actor) {
return actor->node() != self->node();
};
CAF_CHECK(std::all_of(vec.begin(), vec.end(), is_remote));
self->send(grp, "Hello reflector!", 5.0);
self->become(await_reflector_reply_behavior{self, 0, 0, vec});
~fixture() {
for (auto& receiver : receivers)
anon_send_exit(receiver, exit_reason::user_shutdown);
}
);
}
behavior make_server_behavior(event_based_actor* self) {
return {
[=](get_group_atom) {
return self->system().groups().get_local("foobar");
},
[=](spawn_atom, group group) -> std::vector<actor> {
std::vector<actor> vec;
for (auto i = 0; i < 5; ++i) {
vec.push_back(self->spawn_in_group(group, make_reflector_behavior));
}
self->quit();
return vec;
void spawn_receivers(planet_type& planet, group grp, size_t count) {
for (size_t i = 0; i < count; ++i)
receivers.emplace_back(planet.sys.spawn_in_group(grp, group_receiver));
}
};
}
std::vector<actor> receivers;
};
} // namespace <anonymous>
CAF_TEST_FIXTURE_SCOPE(dynamic_remote_group_tests, fixture)
CAF_TEST_DISABLED(remote_group_conn) {
// server side
CAF_EXP_THROW(port, server_side_mm.publish_local_groups(0));
CAF_REQUIRE(port != 0);
// client side
CAF_CHECK(client_side_mm.remote_group("whatever", local_host, port));
CAF_TEST(publish_local_groups) {
loop_after_next_enqueue(mars);
CAF_CHECK_EQUAL(mars.sys.middleman().publish_local_groups(port), port);
}
CAF_TEST_DISABLED(server_side_group_comm) {
// server side
CAF_EXP_THROW(port,
server_side_mm.publish(server_side.spawn(make_server_behavior),
0, local_host));
CAF_REQUIRE(port != 0);
// client side
CAF_EXP_THROW(server, client_side_mm.remote_actor(local_host, port));
scoped_actor group_resolver(client_side, true);
group grp;
group_resolver->request(server, infinite, get_group_atom::value).receive(
[&](const group& x) {
grp = x;
},
[&](error& err) {
CAF_FAIL("error: " << client_side.render(err));
}
);
client_side.spawn(make_client_behavior, server, grp);
CAF_TEST(connecting to remote group) {
CAF_MESSAGE("publish local groups on mars");
loop_after_next_enqueue(mars);
CAF_CHECK_EQUAL(mars.sys.middleman().publish_local_groups(port), port);
CAF_MESSAGE("call remote_group on earth");
loop_after_next_enqueue(earth);
auto grp = unbox(earth.mm.remote_group(group_name, server, port));
CAF_CHECK(grp);
CAF_CHECK_EQUAL(grp->get()->identifier(), group_name);
}
CAF_TEST_DISABLED(client_side_group_comm) {
// server side
CAF_EXP_THROW(port,
server_side_mm.publish(server_side.spawn(make_server_behavior),
0, local_host));
CAF_REQUIRE(port != 0);
// client side
CAF_EXP_THROW(server, client_side_mm.remote_actor(local_host, port));
client_side.spawn(make_client_behavior, server,
client_side.groups().get_local("foobar"));
CAF_TEST_DISABLED(message transmission) {
CAF_MESSAGE("spawn 5 receivers on mars");
auto mars_grp = mars.sys.groups().get_local(group_name);
spawn_receivers(mars, mars_grp, 5u);
CAF_MESSAGE("publish local groups on mars");
loop_after_next_enqueue(mars);
CAF_CHECK_EQUAL(mars.sys.middleman().publish_local_groups(port), port);
CAF_MESSAGE("call remote_group on earth");
loop_after_next_enqueue(earth);
auto earth_grp = unbox(earth.mm.remote_group(group_name, server, port));
CAF_MESSAGE("spawn 5 more receivers on earth");
spawn_receivers(earth, earth_grp, 5u);
CAF_MESSAGE("send message on mars and expect 10 handled messages total");
{
received_messages = 0u;
scoped_actor self{mars.sys};
self->send(mars_grp, ok_atom::value);
exec_all();
CAF_CHECK_EQUAL(received_messages, 10u);
}
CAF_MESSAGE("send message on earth and again expect 10 handled messages");
{
received_messages = 0u;
scoped_actor self{earth.sys};
self->send(earth_grp, ok_atom::value);
exec_all();
CAF_CHECK_EQUAL(received_messages, 10u);
}
}
CAF_TEST_FIXTURE_SCOPE_END()
......@@ -580,7 +580,9 @@ struct test_coordinator_fixture {
explicit test_coordinator_fixture(Ts&&... xs)
: cfg(std::forward<Ts>(xs)...),
sys(cfg.parse(caf::test::engine::argc(), caf::test::engine::argv())
.set("scheduler.policy", caf::atom("testing"))),
.set("scheduler.policy", caf::atom("testing"))
.set("logger.inline-output", true)
.set("middleman.network-backend", caf::atom("testing"))),
self(sys, true),
sched(dynamic_cast<scheduler_type&>(sys.scheduler())),
credit_round_interval(cfg.streaming_credit_round_interval()),
......@@ -622,14 +624,6 @@ struct test_coordinator_fixture {
return f(res_hdl);
}
/// Unboxes an expected value or fails the test if it doesn't exist.
template <class T>
T unbox(caf::expected<T> x) {
if (!x)
FAIL(sys.render(x.error()));
return std::move(*x);
}
template <class T>
const T& peek() {
return sched.template peek<T>();
......@@ -643,6 +637,22 @@ struct test_coordinator_fixture {
}
};
/// Unboxes an expected value or fails the test if it doesn't exist.
template <class T>
T unbox(caf::expected<T> x) {
if (!x)
CAF_FAIL(to_string(x.error()));
return std::move(*x);
}
/// Unboxes an optional value or fails the test if it doesn't exist.
template <class T>
T unbox(caf::optional<T> x) {
if (!x)
CAF_FAIL("x == none");
return std::move(*x);
}
} // namespace <anonymous>
/// Expands to its argument.
......
......@@ -37,14 +37,12 @@ public:
exec_all_nodes_fun exec_all_nodes;
caf::io::middleman& mm;
caf::io::network::test_multiplexer& mpx;
caf::io::basp_broker* basp;
/// @param fun A function object for delegating to the parent's `exec_all`.
test_node_fixture(exec_all_nodes_fun fun)
: exec_all_nodes(std::move(fun)),
mm(this->sys.middleman()),
mpx(dynamic_cast<caf::io::network::test_multiplexer&>(mm.backend())),
basp(get_basp_broker()) {
mpx(dynamic_cast<caf::io::network::test_multiplexer&>(mm.backend())) {
// nop
}
......@@ -178,6 +176,7 @@ public:
// accept all pending connections, and running all broker and regular actor
// messages.
void exec_all() {
CAF_LOG_TRACE("");
exec_all_fixtures(std::begin(planets_), std::end(planets_));
}
......@@ -186,6 +185,10 @@ public:
return [&] { exec_all(); };
}
void loop_after_next_enqueue(PlanetType& planet) {
planet.sched.after_next_enqueue(exec_all_callback());
}
private:
int64_t hdl_id_ = 0;
std::vector<PlanetType*> planets_;
......@@ -209,7 +212,8 @@ public:
: super({&earth, &mars}),
earth(this->exec_all_callback()),
mars(this->exec_all_callback()) {
// nop
// Run initialization code.
this->exec_all();
}
};
......
......@@ -36,6 +36,7 @@ def read_lines(fp, ids):
indent = print_indented(line, indent)
def read_ids(ids_file):
if ids_file and len(ids_file) > 0:
if os.path.isfile(ids_file):
with open(ids_file) as fp:
return fp.read().splitlines()
......
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