Commit f5559262 authored by Dominik Charousset's avatar Dominik Charousset

Let actor system wait for actors on shutdown

parent a334ca5b
...@@ -30,6 +30,4 @@ int main() { ...@@ -30,6 +30,4 @@ int main() {
); );
}); });
} }
// wait until all other actors we've spawned are done
system.await_all_actors_done();
} }
...@@ -174,5 +174,4 @@ int main(int argc, char** argv) { ...@@ -174,5 +174,4 @@ int main(int argc, char** argv) {
<< endl; << endl;
} }
}); });
system.await_all_actors_done();
} }
...@@ -227,5 +227,4 @@ int main(int argc, char** argv) { ...@@ -227,5 +227,4 @@ int main(int argc, char** argv) {
<< res.helptext << endl; << res.helptext << endl;
return 1; return 1;
} }
system.await_all_actors_done();
} }
...@@ -105,5 +105,4 @@ int main(int argc, const char** argv) { ...@@ -105,5 +105,4 @@ int main(int argc, const char** argv) {
std::getline(std::cin, dummy); std::getline(std::cin, dummy);
// kill server // kill server
anon_send_exit(*server_actor, exit_reason::user_shutdown); anon_send_exit(*server_actor, exit_reason::user_shutdown);
system.await_all_actors_done();
} }
...@@ -41,6 +41,5 @@ int main() { ...@@ -41,6 +41,5 @@ int main() {
auto mirror_actor = system.spawn(mirror); auto mirror_actor = system.spawn(mirror);
// create another actor that calls 'hello_world(mirror_actor)'; // create another actor that calls 'hello_world(mirror_actor)';
system.spawn(hello_world, mirror_actor); system.spawn(hello_world, mirror_actor);
// wait until all other actors we have spawned are done // system will wait until both actors are done before leaving main
system.await_all_actors_done();
} }
...@@ -73,5 +73,4 @@ void dancing_kirby(event_based_actor* self) { ...@@ -73,5 +73,4 @@ void dancing_kirby(event_based_actor* self) {
int main() { int main() {
actor_system system; actor_system system;
system.spawn(dancing_kirby); system.spawn(dancing_kirby);
system.await_all_actors_done();
} }
...@@ -210,5 +210,4 @@ int main(int, char**) { ...@@ -210,5 +210,4 @@ int main(int, char**) {
"Nietzsche", "Descartes"}; "Nietzsche", "Descartes"};
for (size_t i = 0; i < 5; ++i) for (size_t i = 0; i < 5; ++i)
self->spawn<philosopher>(names[i], chopsticks[i], chopsticks[(i + 1) % 5]); self->spawn<philosopher>(names[i], chopsticks[i], chopsticks[(i + 1) % 5]);
system.await_all_actors_done();
} }
...@@ -83,5 +83,4 @@ int main() { ...@@ -83,5 +83,4 @@ int main() {
system.await_all_actors_done(); system.await_all_actors_done();
// test class-based impl // test class-based impl
system.spawn(tester, system.spawn<typed_calculator_class>()); system.spawn(tester, system.spawn<typed_calculator_class>());
system.await_all_actors_done();
} }
...@@ -296,5 +296,4 @@ int main(int argc, char** argv) { ...@@ -296,5 +296,4 @@ int main(int argc, char** argv) {
else { else {
client_repl(system, host, port); client_repl(system, host, port);
} }
system.await_all_actors_done();
} }
...@@ -156,5 +156,4 @@ int main(int argc, char** argv) { ...@@ -156,5 +156,4 @@ int main(int argc, char** argv) {
} }
// force actor to quit // force actor to quit
anon_send_exit(client_actor, exit_reason::user_shutdown); anon_send_exit(client_actor, exit_reason::user_shutdown);
system.await_all_actors_done();
} }
...@@ -366,6 +366,18 @@ public: ...@@ -366,6 +366,18 @@ public:
return spawn_in_groups<T>({grp}, std::forward<Ts>(xs)...); return spawn_in_groups<T>({grp}, std::forward<Ts>(xs)...);
} }
/// Returns whether this actor system calls `await_all_actors_done`
/// in its destructor before shutting down.
inline bool await_actors_before_shutdown() const {
return await_actors_before_shutdown_;
}
/// Configures whether this actor system calls `await_all_actors_done`
/// in its destructor before shutting down.
inline void await_actors_before_shutdown(bool x) {
await_actors_before_shutdown_ = x;
}
/// @cond PRIVATE /// @cond PRIVATE
inline atom_value backend_name() const { inline atom_value backend_name() const {
return backend_name_; return backend_name_;
...@@ -415,6 +427,7 @@ private: ...@@ -415,6 +427,7 @@ private:
scoped_execution_unit dummy_execution_unit_; scoped_execution_unit dummy_execution_unit_;
atom_value backend_name_; atom_value backend_name_;
riac::probe* probe_; riac::probe* probe_;
bool await_actors_before_shutdown_;
}; };
} // namespace caf } // namespace caf
......
...@@ -56,7 +56,8 @@ actor_system::actor_system(actor_system_config&& cfg) ...@@ -56,7 +56,8 @@ actor_system::actor_system(actor_system_config&& cfg)
registry_(*this), registry_(*this),
groups_(*this), groups_(*this),
middleman_(nullptr), middleman_(nullptr),
dummy_execution_unit_(this) { dummy_execution_unit_(this),
await_actors_before_shutdown_(true) {
CAF_SET_LOGGER_SYS(this); CAF_SET_LOGGER_SYS(this);
backend_name_ = cfg.middleman_network_backend; backend_name_ = cfg.middleman_network_backend;
for (auto& f : cfg.module_factories_) { for (auto& f : cfg.module_factories_) {
...@@ -134,6 +135,8 @@ actor_system::actor_system(actor_system_config&& cfg) ...@@ -134,6 +135,8 @@ actor_system::actor_system(actor_system_config&& cfg)
} }
actor_system::~actor_system() { actor_system::~actor_system() {
if (await_actors_before_shutdown_)
await_all_actors_done();
// stop modules in reverse order // stop modules in reverse order
registry_.stop(); registry_.stop();
for (auto i = modules_.rbegin(); i != modules_.rend(); ++i) for (auto i = modules_.rbegin(); i != modules_.rend(); ++i)
......
...@@ -103,10 +103,6 @@ struct fixture { ...@@ -103,10 +103,6 @@ struct fixture {
actor spawn(Ts&&... xs) { actor spawn(Ts&&... xs) {
return system.spawn<T, Os>(xs...); return system.spawn<T, Os>(xs...);
} }
~fixture() {
system.await_all_actors_done();
}
}; };
} // namespace <anonymous> } // namespace <anonymous>
......
...@@ -41,10 +41,6 @@ using foo_atom = atom_constant<atom("foo")>; ...@@ -41,10 +41,6 @@ using foo_atom = atom_constant<atom("foo")>;
struct fixture { struct fixture {
actor_system system; actor_system system;
~fixture() {
system.await_all_actors_done();
}
}; };
} // namespace <anonymous> } // namespace <anonymous>
......
...@@ -88,5 +88,4 @@ CAF_TEST(constructor_attach) { ...@@ -88,5 +88,4 @@ CAF_TEST(constructor_attach) {
}; };
actor_system system; actor_system system;
anon_send(system.spawn<spawner>(), die_atom::value); anon_send(system.spawn<spawner>(), die_atom::value);
system.await_all_actors_done();
} }
...@@ -330,7 +330,6 @@ struct fixture { ...@@ -330,7 +330,6 @@ struct fixture {
} }
~fixture() { ~fixture() {
system.await_all_actors_done();
system.~actor_system(); system.~actor_system();
// destructor of actor_system must make sure all // destructor of actor_system must make sure all
// destructors of all actors have been run // destructors of all actors have been run
......
...@@ -56,5 +56,4 @@ void testee(event_based_actor* self) { ...@@ -56,5 +56,4 @@ void testee(event_based_actor* self) {
CAF_TEST(test_local_group) { CAF_TEST(test_local_group) {
actor_system system; actor_system system;
system.spawn(testee); system.spawn(testee);
system.await_all_actors_done();
} }
...@@ -99,10 +99,6 @@ struct fixture { ...@@ -99,10 +99,6 @@ struct fixture {
for (size_t i = 0; i < 100; ++i) for (size_t i = 0; i < 100; ++i)
system.spawn<tester>(system.spawn<testee, Os>()); system.spawn<tester>(system.spawn<testee, Os>());
} }
~fixture() {
system.await_all_actors_done();
}
}; };
} // namespace <anonymous> } // namespace <anonymous>
......
...@@ -61,10 +61,6 @@ struct fixture { ...@@ -61,10 +61,6 @@ struct fixture {
self->send_exit(testee, exit_reason::user_shutdown); self->send_exit(testee, exit_reason::user_shutdown);
self->await_all_other_actors_done(); self->await_all_other_actors_done();
} }
~fixture() {
system.await_all_actors_done();
}
}; };
} // namespace <anonymous> } // namespace <anonymous>
......
...@@ -254,10 +254,6 @@ behavior server(event_based_actor* self) { ...@@ -254,10 +254,6 @@ behavior server(event_based_actor* self) {
struct fixture { struct fixture {
actor_system system; actor_system system;
~fixture() {
system.await_all_actors_done();
}
}; };
} // namespace <anonymous> } // namespace <anonymous>
......
...@@ -89,10 +89,6 @@ behavior ping2(event_based_actor* self, const actor& pong_actor) { ...@@ -89,10 +89,6 @@ behavior ping2(event_based_actor* self, const actor& pong_actor) {
struct fixture { struct fixture {
actor_system system; actor_system system;
~fixture() {
system.await_all_actors_done();
}
}; };
} // namespace <anonymous> } // namespace <anonymous>
......
...@@ -87,18 +87,15 @@ CAF_TEST(test_serial_reply) { ...@@ -87,18 +87,15 @@ CAF_TEST(test_serial_reply) {
} }
); );
}); });
{ // lifetime scope of self scoped_actor self{system};
scoped_actor self{system}; CAF_MESSAGE("ID of main: " << self->id());
CAF_MESSAGE("ID of main: " << self->id()); self->request(master, hi_atom::value).await(
self->request(master, hi_atom::value).await( [](ho_atom) {
[](ho_atom) { CAF_MESSAGE("received 'ho'");
CAF_MESSAGE("received 'ho'"); },
}, [&](const error& err) {
[&](const error& err) { CAF_TEST_ERROR("Error: " << self->system().render(err));
CAF_TEST_ERROR("Error: " << self->system().render(err)); }
} );
); self->send_exit(master, exit_reason::user_shutdown);
self->send_exit(master, exit_reason::user_shutdown);
}
system.await_all_actors_done();
} }
...@@ -186,10 +186,6 @@ struct fixture { ...@@ -186,10 +186,6 @@ struct fixture {
rs.str.assign(string(str.rbegin(), str.rend())); rs.str.assign(string(str.rbegin(), str.rend()));
msg = make_message(i32, te, str, rs); msg = make_message(i32, te, str, rs);
} }
~fixture() {
system.await_all_actors_done();
}
}; };
struct is_message { struct is_message {
......
...@@ -37,14 +37,11 @@ CAF_TEST(simple_reply_response) { ...@@ -37,14 +37,11 @@ CAF_TEST(simple_reply_response) {
} }
); );
}); });
{ scoped_actor self{system};
scoped_actor self{system}; self->send(s, ok_atom::value);
self->send(s, ok_atom::value); self->receive(
self->receive( others >> [&] {
others >> [&] { CAF_CHECK(self->current_message() == make_message(ok_atom::value));
CAF_CHECK(self->current_message() == make_message(ok_atom::value)); }
} );
);
}
system.await_all_actors_done();
} }
...@@ -52,10 +52,6 @@ timer::behavior_type timer_impl(timer::pointer self) { ...@@ -52,10 +52,6 @@ timer::behavior_type timer_impl(timer::pointer self) {
struct fixture { struct fixture {
actor_system system; actor_system system;
~fixture() {
system.await_all_actors_done();
}
}; };
} // namespace <anonymous> } // namespace <anonymous>
......
...@@ -116,10 +116,6 @@ struct fixture { ...@@ -116,10 +116,6 @@ struct fixture {
} }
); );
} }
~fixture() {
system.await_all_actors_done();
}
}; };
} // namespace <anonymous> } // namespace <anonymous>
......
...@@ -99,10 +99,6 @@ private: ...@@ -99,10 +99,6 @@ private:
struct fixture { struct fixture {
actor_system system; actor_system system;
~fixture() {
system.await_all_actors_done();
}
}; };
} // namespace <anonymous> } // namespace <anonymous>
......
...@@ -344,10 +344,6 @@ struct fixture { ...@@ -344,10 +344,6 @@ struct fixture {
.add_message_type<get_state_msg>("get_state_msg")) { .add_message_type<get_state_msg>("get_state_msg")) {
// nop // nop
} }
~fixture() {
system.await_all_actors_done();
}
}; };
} // namespace <anonymous> } // namespace <anonymous>
...@@ -363,17 +359,14 @@ CAF_TEST(typed_spawns) { ...@@ -363,17 +359,14 @@ CAF_TEST(typed_spawns) {
test_typed_spawn(system.spawn(typed_server1)); test_typed_spawn(system.spawn(typed_server1));
system.await_all_actors_done(); system.await_all_actors_done();
CAF_MESSAGE("finished test series with `typed_server1`"); CAF_MESSAGE("finished test series with `typed_server1`");
test_typed_spawn(system.spawn(typed_server2)); test_typed_spawn(system.spawn(typed_server2));
system.await_all_actors_done(); system.await_all_actors_done();
CAF_MESSAGE("finished test series with `typed_server2`"); CAF_MESSAGE("finished test series with `typed_server2`");
{ scoped_actor self{system};
scoped_actor self{system}; test_typed_spawn(self->spawn<typed_server3>("hi there", self));
test_typed_spawn(self->spawn<typed_server3>("hi there", self)); self->receive(on("hi there") >> [] {
self->receive(on("hi there") >> [] { CAF_MESSAGE("received \"hi there\"");
CAF_MESSAGE("received \"hi there\""); });
});
}
} }
CAF_TEST(event_testee_series) { CAF_TEST(event_testee_series) {
...@@ -558,6 +551,8 @@ CAF_TEST(dot_composition) { ...@@ -558,6 +551,8 @@ CAF_TEST(dot_composition) {
CAF_CHECK(res == (42 * 2.0) * (42 * 4.0)); CAF_CHECK(res == (42 * 2.0) * (42 * 4.0));
} }
); );
anon_send_exit(first, exit_reason::user_shutdown);
anon_send_exit(second, exit_reason::user_shutdown);
} }
CAF_TEST(currying) { CAF_TEST(currying) {
......
...@@ -151,7 +151,6 @@ public: ...@@ -151,7 +151,6 @@ public:
nid = invalid_node_id; nid = invalid_node_id;
for (auto& ptr : pseudo_remote_) for (auto& ptr : pseudo_remote_)
ptr.reset(); ptr.reset();
system.await_all_actors_done();
} }
// our "virtual communication backend" // our "virtual communication backend"
......
...@@ -171,7 +171,6 @@ void run_client(int argc, char** argv, uint16_t port) { ...@@ -171,7 +171,6 @@ void run_client(int argc, char** argv, uint16_t port) {
CAF_MESSAGE("spawn_client finished"); CAF_MESSAGE("spawn_client finished");
anon_send(p, kickoff_atom::value, *cl); anon_send(p, kickoff_atom::value, *cl);
CAF_MESSAGE("`kickoff_atom` has been send"); CAF_MESSAGE("`kickoff_atom` has been send");
system.await_all_actors_done();
} }
void run_server(int argc, char** argv) { void run_server(int argc, char** argv) {
......
...@@ -200,7 +200,6 @@ public: ...@@ -200,7 +200,6 @@ public:
// since we do not invoke any "I/O" from this point on that would // since we do not invoke any "I/O" from this point on that would
// trigger the exit message implicitly // trigger the exit message implicitly
mpx_->flush_runnables(); mpx_->flush_runnables();
system.await_all_actors_done();
} }
// helper class for a nice-and-easy "mock(...).expect(...)" syntax // helper class for a nice-and-easy "mock(...).expect(...)" syntax
......
...@@ -528,7 +528,6 @@ void launch_remote_side(int argc, char** argv, uint16_t group_port, ...@@ -528,7 +528,6 @@ void launch_remote_side(int argc, char** argv, uint16_t group_port,
CAF_CHECK_EQUAL(dm.reason, exit_reason::normal); CAF_CHECK_EQUAL(dm.reason, exit_reason::normal);
} }
); );
system.await_all_actors_done();
} }
void test_remote_actor(int argc, char** argv) { void test_remote_actor(int argc, char** argv) {
......
...@@ -103,7 +103,6 @@ void run_client(int argc, char** argv, uint16_t port) { ...@@ -103,7 +103,6 @@ void run_client(int argc, char** argv, uint16_t port) {
auto serv = system.middleman().remote_actor("localhost", port); auto serv = system.middleman().remote_actor("localhost", port);
CAF_REQUIRE(serv); CAF_REQUIRE(serv);
system.spawn(client, serv); system.spawn(client, serv);
system.await_all_actors_done();
} }
void run_server(int argc, char** argv) { void run_server(int argc, char** argv) {
...@@ -114,7 +113,6 @@ void run_server(int argc, char** argv) { ...@@ -114,7 +113,6 @@ void run_server(int argc, char** argv) {
auto port = *mport; auto port = *mport;
CAF_MESSAGE("published server at port " << port); CAF_MESSAGE("published server at port " << port);
std::thread child([=] { run_client(argc, argv, port); }); std::thread child([=] { run_client(argc, argv, port); });
system.await_all_actors_done();
child.join(); child.join();
} }
......
...@@ -189,7 +189,6 @@ void run_client(int argc, char** argv, uint16_t port) { ...@@ -189,7 +189,6 @@ void run_client(int argc, char** argv, uint16_t port) {
CAF_MESSAGE("spawn_client_typed finished"); CAF_MESSAGE("spawn_client_typed finished");
anon_send(p, kickoff_atom::value, *cl); anon_send(p, kickoff_atom::value, *cl);
CAF_MESSAGE("`kickoff_atom` has been send"); CAF_MESSAGE("`kickoff_atom` has been send");
system.await_all_actors_done();
} }
void run_server(int argc, char** argv) { void run_server(int argc, char** argv) {
......
...@@ -85,7 +85,6 @@ CAF_TEST(unpublishing) { ...@@ -85,7 +85,6 @@ CAF_TEST(unpublishing) {
CAF_MESSAGE("unpublish succeeded"); CAF_MESSAGE("unpublish succeeded");
} }
anon_send_exit(d, exit_reason::user_shutdown); anon_send_exit(d, exit_reason::user_shutdown);
system.await_all_actors_done();
} }
// check after dtor of system was called // check after dtor of system was called
CAF_CHECK_EQUAL(s_dtor_called.load(), 2); CAF_CHECK_EQUAL(s_dtor_called.load(), 2);
......
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