Commit fd1e13b3 authored by Dominik Charousset's avatar Dominik Charousset

Count nr. of actor instances in test_spawn

parent f20f86a2
#include <stack> #include <stack>
#include <atomic>
#include <chrono> #include <chrono>
#include <iostream> #include <iostream>
#include <functional> #include <functional>
...@@ -13,52 +14,75 @@ using namespace caf; ...@@ -13,52 +14,75 @@ using namespace caf;
namespace { namespace {
class event_testee : public sb_actor<event_testee> { std::atomic<long> s_max_actor_instances;
std::atomic<long> s_actor_instances;
void inc_actor_instances() {
long v1 = ++s_actor_instances;
long v2 = s_max_actor_instances.load();
while (v1 > v2) {
s_max_actor_instances.compare_exchange_weak(v2, v1);
}
}
friend class sb_actor<event_testee>; void dec_actor_instances() {
--s_actor_instances;
}
class event_testee : public sb_actor<event_testee> {
public:
event_testee();
~event_testee();
behavior wait4string; behavior wait4string;
behavior wait4float; behavior wait4float;
behavior wait4int; behavior wait4int;
behavior& init_state = wait4int; behavior& init_state = wait4int;
};
public: event_testee::event_testee() {
inc_actor_instances();
event_testee() { wait4string.assign(
wait4string = behavior{
[=](const std::string&) { [=](const std::string&) {
become(wait4int); become(wait4int);
}, },
[=](get_atom) { [=](get_atom) {
return "wait4string"; return "wait4string";
} }
}; );
wait4float = behavior{ wait4float.assign(
[=](float) { [=](float) {
become(wait4string); become(wait4string);
}, },
[=](get_atom) { [=](get_atom) {
return "wait4float"; return "wait4float";
} }
}; );
wait4int = behavior{ wait4int.assign(
[=](int) { [=](int) {
become(wait4float); become(wait4float);
}, },
[=](get_atom) { [=](get_atom) {
return "wait4int"; return "wait4int";
} }
}; );
} }
}; event_testee::~event_testee() {
dec_actor_instances();
}
// quits after 5 timeouts // quits after 5 timeouts
actor spawn_event_testee2(actor parent) { actor spawn_event_testee2(actor parent) {
struct impl : event_based_actor { struct impl : event_based_actor {
actor parent; actor parent;
impl(actor parent_actor) : parent(parent_actor) { } impl(actor parent_actor) : parent(parent_actor) {
inc_actor_instances();
}
~impl() {
dec_actor_instances();
}
behavior wait4timeout(int remaining) { behavior wait4timeout(int remaining) {
CAF_LOG_TRACE(CAF_ARG(remaining)); CAF_LOG_TRACE(CAF_ARG(remaining));
return { return {
...@@ -99,8 +123,13 @@ struct chopstick : public sb_actor<chopstick> { ...@@ -99,8 +123,13 @@ struct chopstick : public sb_actor<chopstick> {
behavior& init_state = available; behavior& init_state = available;
chopstick() { chopstick();
available = ( ~chopstick();
};
chopstick::chopstick() {
inc_actor_instances();
available.assign(
on(atom("take"), arg_match) >> [=](actor whom) -> atom_value { on(atom("take"), arg_match) >> [=](actor whom) -> atom_value {
become(taken_by(whom)); become(taken_by(whom));
return atom("taken"); return atom("taken");
...@@ -109,16 +138,46 @@ struct chopstick : public sb_actor<chopstick> { ...@@ -109,16 +138,46 @@ struct chopstick : public sb_actor<chopstick> {
quit(); quit();
} }
); );
} }
chopstick::~chopstick() {
dec_actor_instances();
}
class testee_actor : public blocking_actor {
public:
testee_actor();
~testee_actor();
void act() override;
private:
void wait4string();
void wait4float();
}; };
class testee_actor { testee_actor::testee_actor() {
inc_actor_instances();
}
testee_actor::~testee_actor() {
dec_actor_instances();
}
void testee_actor::act() {
receive_loop (
[&](int) {
wait4float();
},
[&](get_atom) {
return "wait4int";
}
);
}
void wait4string(blocking_actor* self) { void testee_actor::wait4string() {
bool string_received = false; bool string_received = false;
self->do_receive ( do_receive (
on<string>() >> [&] { [&](const string&) {
string_received = true; string_received = true;
}, },
[&](get_atom) { [&](get_atom) {
...@@ -126,12 +185,12 @@ class testee_actor { ...@@ -126,12 +185,12 @@ class testee_actor {
} }
) )
.until([&] { return string_received; }); .until([&] { return string_received; });
} }
void wait4float(blocking_actor* self) { void testee_actor::wait4float() {
bool float_received = false; bool float_received = false;
self->do_receive ( do_receive (
on<float>() >> [&] { [&](float) {
float_received = true; float_received = true;
}, },
[&](get_atom) { [&](get_atom) {
...@@ -139,31 +198,33 @@ class testee_actor { ...@@ -139,31 +198,33 @@ class testee_actor {
} }
) )
.until([&] { return float_received; }); .until([&] { return float_received; });
wait4string(self); wait4string();
} }
// self->receives one timeout and quits
class testee1 : public event_based_actor {
public: public:
testee1();
~testee1();
behavior make_behavior() override;
};
void operator()(blocking_actor* self) { testee1::testee1() {
self->receive_loop ( inc_actor_instances();
on<int>() >> [&] { }
wait4float(self);
},
[&](get_atom) {
return "wait4int";
}
);
}
}; testee1::~testee1() {
dec_actor_instances();
}
// self->receives one timeout and quits behavior testee1::make_behavior() {
void testee1(event_based_actor* self) {
CAF_LOGF_TRACE(""); CAF_LOGF_TRACE("");
self->become(after(chrono::milliseconds(10)) >> [=] { return {
after(chrono::milliseconds(10)) >> [=] {
CAF_LOGF_TRACE(""); CAF_LOGF_TRACE("");
self->unbecome(); unbecome();
}); }
};
} }
string behavior_test(scoped_actor& self, actor et) { string behavior_test(scoped_actor& self, actor et) {
...@@ -193,19 +254,49 @@ string behavior_test(scoped_actor& self, actor et) { ...@@ -193,19 +254,49 @@ string behavior_test(scoped_actor& self, actor et) {
return result; return result;
} }
behavior echo_actor(event_based_actor* self) { class echo_actor : public event_based_actor {
return ( public:
echo_actor();
~echo_actor();
behavior make_behavior() override;
};
echo_actor::echo_actor() {
inc_actor_instances();
}
echo_actor::~echo_actor() {
dec_actor_instances();
}
behavior echo_actor::make_behavior() {
return {
others() >> [=]() -> message { others() >> [=]() -> message {
self->quit(exit_reason::normal); quit(exit_reason::normal);
return self->last_dequeued(); return last_dequeued();
} }
); };
}
class simple_mirror : public event_based_actor {
public:
simple_mirror();
~simple_mirror();
behavior make_behavior() override;
};
simple_mirror::simple_mirror() {
inc_actor_instances();
} }
behavior simple_mirror(event_based_actor* self) { simple_mirror::~simple_mirror() {
dec_actor_instances();
}
behavior simple_mirror::make_behavior() {
return { return {
others() >> [=] { others() >> [=] {
return self->last_dequeued(); return last_dequeued();
} }
}; };
} }
...@@ -297,7 +388,7 @@ void test_spawn() { ...@@ -297,7 +388,7 @@ void test_spawn() {
CAF_CHECKPOINT(); CAF_CHECKPOINT();
CAF_PRINT("test mirror"); { CAF_PRINT("test mirror"); {
auto mirror = self->spawn<monitored>(simple_mirror); auto mirror = self->spawn<simple_mirror, monitored>();
self->send(mirror, "hello mirror"); self->send(mirror, "hello mirror");
self->receive ( self->receive (
on("hello mirror") >> CAF_CHECKPOINT_CB(), on("hello mirror") >> CAF_CHECKPOINT_CB(),
...@@ -318,7 +409,7 @@ void test_spawn() { ...@@ -318,7 +409,7 @@ void test_spawn() {
} }
CAF_PRINT("test detached mirror"); { CAF_PRINT("test detached mirror"); {
auto mirror = self->spawn<monitored+detached>(simple_mirror); auto mirror = self->spawn<simple_mirror, monitored+detached>();
self->send(mirror, "hello mirror"); self->send(mirror, "hello mirror");
self->receive ( self->receive (
on("hello mirror") >> CAF_CHECKPOINT_CB(), on("hello mirror") >> CAF_CHECKPOINT_CB(),
...@@ -339,7 +430,7 @@ void test_spawn() { ...@@ -339,7 +430,7 @@ void test_spawn() {
} }
CAF_PRINT("test priority aware mirror"); { CAF_PRINT("test priority aware mirror"); {
auto mirror = self->spawn<monitored+priority_aware>(simple_mirror); auto mirror = self->spawn<simple_mirror, monitored + priority_aware>();
CAF_CHECKPOINT(); CAF_CHECKPOINT();
self->send(mirror, "hello mirror"); self->send(mirror, "hello mirror");
self->receive ( self->receive (
...@@ -361,7 +452,7 @@ void test_spawn() { ...@@ -361,7 +452,7 @@ void test_spawn() {
} }
CAF_PRINT("test echo actor"); CAF_PRINT("test echo actor");
auto mecho = spawn(echo_actor); auto mecho = spawn<echo_actor>();
self->send(mecho, "hello echo"); self->send(mecho, "hello echo");
self->receive ( self->receive (
on("hello echo") >> [] { }, on("hello echo") >> [] { },
...@@ -380,7 +471,7 @@ void test_spawn() { ...@@ -380,7 +471,7 @@ void test_spawn() {
self->receive(after(chrono::milliseconds(1)) >> [] { }); self->receive(after(chrono::milliseconds(1)) >> [] { });
CAF_CHECKPOINT(); CAF_CHECKPOINT();
spawn(testee1); spawn<testee1>();
self->await_all_other_actors_done(); self->await_all_other_actors_done();
CAF_CHECKPOINT(); CAF_CHECKPOINT();
...@@ -450,21 +541,33 @@ void test_spawn() { ...@@ -450,21 +541,33 @@ void test_spawn() {
); );
CAF_CHECKPOINT(); CAF_CHECKPOINT();
struct inflater : public event_based_actor {
auto inflater = [](event_based_actor* s, const string& name, actor buddy) { public:
CAF_LOGF_TRACE(CAF_ARG(s) << ", " << CAF_ARG(name) inflater(string name, actor buddy)
<< ", " << CAF_TARG(buddy, to_string)); : m_name(std::move(name)),
s->become( m_buddy(std::move(buddy)) {
inc_actor_instances();
}
~inflater() {
dec_actor_instances();
}
behavior make_behavior() override {
return {
[=](int n, const string& str) { [=](int n, const string& str) {
s->send(buddy, n * 2, str + " from " + name); send(m_buddy, n * 2, str + " from " + m_name);
}, },
on(atom("done")) >> [=] { on(atom("done")) >> [=] {
s->quit(); quit();
} }
);
}; };
auto joe = spawn(inflater, "Joe", self); }
auto bob = spawn(inflater, "Bob", joe);
private:
string m_name;
actor m_buddy;
};
auto joe = spawn<inflater>("Joe", self);
auto bob = spawn<inflater>("Bob", joe);
self->send(bob, 1, "hello actor"); self->send(bob, 1, "hello actor");
self->receive ( self->receive (
on(4, "hello actor from Bob from Joe") >> CAF_CHECKPOINT_CB(), on(4, "hello actor from Bob from Joe") >> CAF_CHECKPOINT_CB(),
...@@ -476,28 +579,37 @@ void test_spawn() { ...@@ -476,28 +579,37 @@ void test_spawn() {
anon_send(bob, poison_pill); anon_send(bob, poison_pill);
self->await_all_other_actors_done(); self->await_all_other_actors_done();
function<actor (const string&, const actor&)> spawn_next; class kr34t0r : public event_based_actor {
// it's safe to capture spawn_next as reference here, because public:
// - it is guaranteeed to outlive kr34t0r by general scoping rules kr34t0r(string name, actor pal)
// - the lambda is always executed in the current actor's thread : m_name(std::move(name)),
// but using spawn_next in a message handler could m_pal(std::move(pal)) {
// still cause undefined behavior! inc_actor_instances();
auto kr34t0r = [&spawn_next](event_based_actor* s, const string& name, actor pal) { }
if (name == "Joe" && !pal) { ~kr34t0r() {
pal = spawn_next("Bob", s); dec_actor_instances();
} }
s->become ( behavior make_behavior() override {
if (m_name == "Joe" && m_pal == invalid_actor) {
m_pal = spawn<kr34t0r>("Bob", this);
}
return {
others() >> [=] { others() >> [=] {
// forward message and die // forward message and die
s->send(pal, s->last_dequeued()); send(m_pal, last_dequeued());
s->quit(); quit();
} }
);
}; };
spawn_next = [&kr34t0r](const string& name, const actor& pal) { }
return spawn(kr34t0r, name, pal); void on_exit() {
m_pal = invalid_actor; // break cycle
}
private:
string m_name;
actor m_pal;
}; };
auto joe_the_second = spawn(kr34t0r, "Joe", invalid_actor); auto joe_the_second = spawn<kr34t0r>("Joe", invalid_actor);
self->send(joe_the_second, atom("done")); self->send(joe_the_second, atom("done"));
self->await_all_other_actors_done(); self->await_all_other_actors_done();
...@@ -527,7 +639,7 @@ void test_spawn() { ...@@ -527,7 +639,7 @@ void test_spawn() {
self->await_all_other_actors_done(); self->await_all_other_actors_done();
CAF_CHECKPOINT(); CAF_CHECKPOINT();
auto res1 = behavior_test(self, spawn<blocking_api>(testee_actor{})); auto res1 = behavior_test(self, spawn<testee_actor, blocking_api>());
CAF_CHECK_EQUAL("wait4int", res1); CAF_CHECK_EQUAL("wait4int", res1);
CAF_CHECK_EQUAL(behavior_test(self, spawn<event_testee>()), "wait4int"); CAF_CHECK_EQUAL(behavior_test(self, spawn<event_testee>()), "wait4int");
self->await_all_other_actors_done(); self->await_all_other_actors_done();
...@@ -535,13 +647,24 @@ void test_spawn() { ...@@ -535,13 +647,24 @@ void test_spawn() {
// create some actors linked to one single actor // create some actors linked to one single actor
// and kill them all through killing the link // and kill them all through killing the link
auto legion = spawn([](event_based_actor* s) { class legion_actor : public event_based_actor {
CAF_PRINT("spawn 100 actors"); public:
legion_actor() {
inc_actor_instances();
for (int i = 0; i < 100; ++i) { for (int i = 0; i < 100; ++i) {
s->spawn<event_testee, linked>(); spawn<event_testee, linked>();
} }
s->become(others() >> CAF_UNEXPECTED_MSG_CB(s)); }
}); ~legion_actor() {
dec_actor_instances();
}
behavior make_behavior() override {
return {
others() >> CAF_UNEXPECTED_MSG_CB(this)
};
}
};
auto legion = spawn<legion_actor>();
self->send_exit(legion, exit_reason::user_shutdown); self->send_exit(legion, exit_reason::user_shutdown);
self->await_all_other_actors_done(); self->await_all_other_actors_done();
CAF_CHECKPOINT(); CAF_CHECKPOINT();
...@@ -619,15 +742,31 @@ void test_spawn() { ...@@ -619,15 +742,31 @@ void test_spawn() {
CAF_CHECKPOINT(); CAF_CHECKPOINT();
} }
void counting_actor(event_based_actor* self) { class counting_actor : public event_based_actor {
public:
counting_actor();
~counting_actor();
behavior make_behavior() override;
};
counting_actor::counting_actor() {
inc_actor_instances();
}
counting_actor::~counting_actor() {
dec_actor_instances();
}
behavior counting_actor::make_behavior() {
for (int i = 0; i < 100; ++i) { for (int i = 0; i < 100; ++i) {
self->send(self, atom("dummy")); send(this, atom("dummy"));
} }
CAF_CHECK_EQUAL(self->mailbox().count(), 100); CAF_CHECK_EQUAL(mailbox().count(), 100);
for (int i = 0; i < 100; ++i) { for (int i = 0; i < 100; ++i) {
self->send(self, atom("dummy")); send(this, atom("dummy"));
} }
CAF_CHECK_EQUAL(self->mailbox().count(), 200); CAF_CHECK_EQUAL(mailbox().count(), 200);
return {};
} }
// tests attach_functor() inside of an actor's constructor // tests attach_functor() inside of an actor's constructor
...@@ -769,8 +908,10 @@ void test_typed_testee() { ...@@ -769,8 +908,10 @@ void test_typed_testee() {
int main() { int main() {
CAF_TEST(test_spawn); CAF_TEST(test_spawn);
spawn(counting_actor); { // lifetime scope of temporary counting_actor handle
spawn<counting_actor>();
await_all_actors_done(); await_all_actors_done();
}
CAF_CHECKPOINT(); CAF_CHECKPOINT();
test_spawn(); test_spawn();
CAF_CHECKPOINT(); CAF_CHECKPOINT();
...@@ -794,5 +935,7 @@ int main() { ...@@ -794,5 +935,7 @@ int main() {
CAF_CHECKPOINT(); CAF_CHECKPOINT();
shutdown(); shutdown();
CAF_CHECKPOINT(); CAF_CHECKPOINT();
CAF_CHECK_EQUAL(s_actor_instances.load(), 0);
CAF_PRINT("max. nr. of actor instances: " << s_max_actor_instances.load());
return CAF_TEST_RESULT(); return CAF_TEST_RESULT();
} }
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