Commit 80f74146 authored by Dominik Charousset's avatar Dominik Charousset

Auto-delete unreachable actors, close #420

parent 47efe497
......@@ -40,8 +40,8 @@ using kickoff_atom = atom_constant<atom("kickoff")>;
// utility function to print an exit message with custom name
void print_on_exit(const actor& hdl, const std::string& name) {
hdl->attach_functor([=](abstract_actor* ptr, exit_reason reason) {
aout(ptr) << name << " exited with reason " << reason << endl;
hdl->attach_functor([=](exit_reason reason) {
cout << name << " exited with reason " << to_string(reason) << endl;
});
}
......
......@@ -75,8 +75,8 @@ public:
/// Executed if the actor finished execution with given `reason`.
/// The default implementation does nothing.
virtual void actor_exited(abstract_actor* self, exit_reason reason,
execution_unit* host);
/// @warning `host` can be `nullptr`
virtual void actor_exited(exit_reason reason, execution_unit* host);
/// Returns `true` if `what` selects this instance, otherwise `false`.
virtual bool matches(const token& what);
......
......@@ -38,17 +38,20 @@ public:
static constexpr size_t token_type = attachable::token::observer;
};
void actor_exited(abstract_actor* self, exit_reason reason,
execution_unit* host) override;
void actor_exited(exit_reason reason, execution_unit* host) override;
bool matches(const token& what) override;
inline static attachable_ptr make_monitor(actor_addr observer) {
return attachable_ptr{new default_attachable(std::move(observer), monitor)};
inline static attachable_ptr make_monitor(actor_addr observed,
actor_addr observer) {
return attachable_ptr{new default_attachable(std::move(observed),
std::move(observer), monitor)};
}
inline static attachable_ptr make_link(actor_addr observer) {
return attachable_ptr{new default_attachable(std::move(observer), link)};
inline static attachable_ptr make_link(actor_addr observed,
actor_addr observer) {
return attachable_ptr{new default_attachable(std::move(observed),
std::move(observer), link)};
}
class predicate {
......@@ -69,7 +72,8 @@ public:
};
private:
default_attachable(actor_addr observer, observe_type type);
default_attachable(actor_addr observed, actor_addr observer, observe_type ot);
actor_addr observed_;
actor_addr observer_;
observe_type type_;
};
......
......@@ -31,28 +31,15 @@ namespace detail {
template <class F,
int Args = tl_size<typename get_callable_trait<F>::arg_types>::value>
struct functor_attachable : attachable {
static_assert(Args == 2, "Only 0, 1 and 2 arguments for F are supported");
static_assert(Args == 1, "Only 0 or 1 arguments for F are supported");
F functor_;
functor_attachable(F arg) : functor_(std::move(arg)) {
// nop
}
void actor_exited(abstract_actor* self, exit_reason reason,
execution_unit*) override {
functor_(self, reason);
}
static constexpr size_t token_type = attachable::token::anonymous;
};
template <class F>
struct functor_attachable<F, 1> : attachable {
F functor_;
functor_attachable(F arg) : functor_(std::move(arg)) {
// nop
}
void actor_exited(abstract_actor*, exit_reason reason,
execution_unit*) override {
void actor_exited(exit_reason reason, execution_unit*) override {
functor_(reason);
}
static constexpr size_t token_type = attachable::token::anonymous;
};
template <class F>
......@@ -61,7 +48,7 @@ struct functor_attachable<F, 0> : attachable {
functor_attachable(F arg) : functor_(std::move(arg)) {
// nop
}
void actor_exited(abstract_actor*, exit_reason, execution_unit*) override {
void actor_exited(exit_reason, execution_unit*) override {
functor_();
}
};
......
......@@ -52,7 +52,10 @@ enum class exit_reason : uint8_t {
/// Indicates that an actor finishied execution because a connection
/// to a remote link was closed unexpectedly.
remote_link_unreachable = 0x21
remote_link_unreachable = 0x21,
/// Indicates that the actor was killed because it became unreachable.
unreachable = 0x40
};
/// Returns a string representation of given exit reason.
......
......@@ -121,6 +121,7 @@ private:
policy_.after_resume(this, job);
switch (res) {
case resumable::resume_later: {
// keep reference to this actor, as it remains in the "loop"
policy_.resume_job_later(this, job);
break;
}
......@@ -130,7 +131,8 @@ private:
break;
}
case resumable::awaiting_message: {
// resumable will be enqueued again later
// resumable will maybe be enqueued again later, deref it for now
intrusive_ptr_release(job);
break;
}
case resumable::shutdown_execution_unit: {
......
......@@ -109,8 +109,8 @@ actor actor_pool::make(execution_unit* eu, size_t num_workers,
auto res_addr = ptr->address();
for (size_t i = 0; i < num_workers; ++i) {
auto worker = fac();
worker->attach(default_attachable::make_monitor(res_addr));
ptr->workers_.push_back(worker);
worker->attach(default_attachable::make_monitor(worker.address(), res_addr));
ptr->workers_.push_back(std::move(worker));
}
return res;
}
......@@ -181,10 +181,10 @@ bool actor_pool::filter(upgrade_lock<detail::shared_spinlock>& guard,
}
if (msg.match_elements<sys_atom, put_atom, actor>()) {
auto& worker = msg.get_as<actor>(2);
if (worker == invalid_actor) {
if (worker == invalid_actor)
return true;
}
worker->attach(default_attachable::make_monitor(address()));
worker->attach(default_attachable::make_monitor(worker.address(),
address()));
upgrade_to_unique_lock<detail::shared_spinlock> unique_guard{guard};
workers_.push_back(worker);
return true;
......
......@@ -44,7 +44,7 @@ adapter::adapter(actor_addr decorated, message msg)
// bound actor has dependency on the decorated actor by default;
// if the decorated actor is already dead upon establishing the
// dependency, the actor is spawned dead
decorated_->attach(default_attachable::make_monitor(address()));
decorated_->attach(default_attachable::make_monitor(decorated_, address()));
}
void adapter::enqueue(mailbox_element_ptr what, execution_unit* host) {
......
......@@ -34,7 +34,7 @@ maybe<exit_reason> attachable::handle_exception(const std::exception_ptr&) {
return none;
}
void attachable::actor_exited(abstract_actor*, exit_reason, execution_unit*) {
void attachable::actor_exited(exit_reason, execution_unit*) {
// nop
}
......
......@@ -20,8 +20,8 @@
#include "caf/default_attachable.hpp"
#include "caf/message.hpp"
#include "caf/system_messages.hpp"
#include "caf/actor_cast.hpp"
#include "caf/system_messages.hpp"
namespace caf {
......@@ -34,13 +34,12 @@ message make(abstract_actor* self, exit_reason reason) {
} // namespace <anonymous>
void default_attachable::actor_exited(abstract_actor* self, exit_reason rsn,
execution_unit* host) {
CAF_ASSERT(self->address() != observer_);
void default_attachable::actor_exited(exit_reason rsn, execution_unit* host) {
CAF_ASSERT(observed_ != observer_);
auto factory = type_ == monitor ? &make<down_msg> : &make<exit_msg>;
auto ptr = actor_cast<abstract_actor_ptr>(observer_);
ptr->enqueue(self->address(), message_id{}.with_high_priority(),
factory(self, rsn), host);
ptr->enqueue(observed_, message_id{}.with_high_priority(),
factory(actor_cast<abstract_actor*>(observed_), rsn), host);
}
bool default_attachable::matches(const token& what) {
......@@ -51,8 +50,10 @@ bool default_attachable::matches(const token& what) {
return ot.observer == observer_ && ot.type == type_;
}
default_attachable::default_attachable(actor_addr observer, observe_type type)
: observer_(std::move(observer)),
default_attachable::default_attachable(actor_addr observed, actor_addr observer,
observe_type type)
: observed_(std::move(observed)),
observer_(std::move(observer)),
type_(type) {
// nop
}
......
......@@ -68,24 +68,20 @@ local_actor::local_actor(actor_system* sys, actor_id aid,
}
local_actor::~local_actor() {
if (! mailbox_.closed()) {
detail::sync_request_bouncer f{get_exit_reason()};
mailbox_.close(f);
}
if (planned_exit_reason() == exit_reason::not_exited)
cleanup(exit_reason::unreachable, nullptr);
}
void local_actor::monitor(const actor_addr& whom) {
if (whom == invalid_actor_addr) {
if (whom == invalid_actor_addr)
return;
}
auto ptr = actor_cast<abstract_actor_ptr>(whom);
ptr->attach(default_attachable::make_monitor(address()));
ptr->attach(default_attachable::make_monitor(whom, address()));
}
void local_actor::demonitor(const actor_addr& whom) {
if (whom == invalid_actor_addr) {
if (whom == invalid_actor_addr)
return;
}
auto ptr = actor_cast<abstract_actor_ptr>(whom);
default_attachable::observe_token tk{address(), default_attachable::monitor};
ptr->detach(tk);
......@@ -694,14 +690,14 @@ void local_actor::launch(execution_unit* eu, bool lazy, bool hide) {
return;
}
CAF_ASSERT(eu != nullptr);
// the scheduler keeps an implicit reference count for
// cooperatively scheduled that is released in finalize()
ref();
// do not schedule immediately when spawned with `lazy_init`
// mailbox could be set to blocked
if (lazy && mailbox().try_block()) {
return;
}
// scheduler has a reference count to the actor as long as
// it is waiting to get scheduled
ref();
eu->exec_later(this);
}
......@@ -727,7 +723,8 @@ void local_actor::enqueue(mailbox_element_ptr ptr, execution_unit* eu) {
auto sender = ptr->sender;
switch (mailbox().enqueue(ptr.release())) {
case detail::enqueue_result::unblocked_reader: {
// re-schedule actor
// add a reference count to this actor and re-schedule it
ref();
if (eu)
eu->exec_later(this);
else
......@@ -1087,8 +1084,10 @@ bool local_actor::finished() {
void local_actor::cleanup(exit_reason reason, execution_unit* host) {
CAF_LOG_TRACE(CAF_ARG(reason));
current_mailbox_element().reset();
detail::sync_request_bouncer f{reason};
mailbox_.close(f);
if (! mailbox_.closed()) {
detail::sync_request_bouncer f{reason};
mailbox_.close(f);
}
awaited_responses_.clear();
multiplexed_responses_.clear();
{ // lifetime scope of temporary
......
......@@ -48,7 +48,7 @@ void monitorable_actor::attach(attachable_ptr ptr) {
}
}
CAF_LOG_DEBUG("cannot attach functor to terminated actor: call immediately");
ptr->actor_exited(this, reason, nullptr);
ptr->actor_exited(reason, nullptr);
}
size_t monitorable_actor::detach(const attachable::token& what) {
......@@ -73,7 +73,7 @@ void monitorable_actor::cleanup(exit_reason reason, execution_unit* host) {
"cleanup" << CAF_ARG(id()) << CAF_ARG(reason));
// send exit messages
for (attachable* i = head.get(); i != nullptr; i = i->next.get())
i->actor_exited(this, reason, host);
i->actor_exited(reason, host);
}
monitorable_actor::monitorable_actor(actor_config& cfg)
......@@ -141,7 +141,7 @@ bool monitorable_actor::establish_link_impl(const actor_addr& other) {
} else if (ptr->establish_backlink(address())) {
// add link if not already linked to other
// (checked by establish_backlink)
auto tmp = default_attachable::make_link(other);
auto tmp = default_attachable::make_link(address(), other);
attach_impl(tmp);
return true;
}
......@@ -158,7 +158,7 @@ bool monitorable_actor::establish_backlink_impl(const actor_addr& other) {
reason = get_exit_reason();
if (reason == exit_reason::not_exited) {
if (detach_impl(tk, attachables_head_, true, true) == 0) {
auto tmp = default_attachable::make_link(other);
auto tmp = default_attachable::make_link(address(), other);
attach_impl(tmp);
return true;
}
......
......@@ -39,9 +39,9 @@ sequencer::sequencer(actor_addr f, actor_addr g, message_types_set msg_types)
// composed actor has dependency on constituent actors by default;
// if either constituent actor is already dead upon establishing
// the dependency, the actor is spawned dead
f_->attach(default_attachable::make_monitor(address()));
f_->attach(default_attachable::make_monitor(f_, address()));
if (g_ != f_)
g_->attach(default_attachable::make_monitor(address()));
g_->attach(default_attachable::make_monitor(g_, address()));
}
void sequencer::enqueue(mailbox_element_ptr what, execution_unit* context) {
......
......@@ -77,7 +77,7 @@ splitter::splitter(std::vector<actor_addr> workers, message_types_set msg_types)
// the dependency, the actor is spawned dead
auto addr = address();
for (auto& worker : workers_)
worker->attach(default_attachable::make_monitor(addr));
worker->attach(default_attachable::make_monitor(worker, addr));
}
void splitter::enqueue(mailbox_element_ptr what, execution_unit* context) {
......
......@@ -73,8 +73,10 @@ CAF_TEST(identity) {
CAF_CHECK(bound->node() == dbl->node());
CAF_CHECK(bound != dbl);
CAF_CHECK(bound->id() != dbl->id());
anon_send_exit(bound, exit_reason::kill);
//anon_send_exit(bound, exit_reason::kill);
anon_send_exit(dbl, exit_reason::kill);
// killing dbl triggers a down message to bound, which becomes unreachable
// when it no longer monitors dbl as a result and goes out of scope here
}
// bound actor spawned dead if decorated
......
......@@ -90,19 +90,6 @@ simple_cell(cell::stateful_pointer<cell_state> self) {
struct fixture {
actor_system system;
std::vector<actor_addr> testees;
template <class F>
typename infer_handle_from_fun<F>::type spawn(F fun) {
auto result = system.spawn(fun);
testees.emplace_back(result.address());
return result;
}
~fixture() {
for (auto& testee : testees)
anon_send_exit(testee, exit_reason::kill);
}
};
} // namespace <anonymous>
......@@ -121,7 +108,7 @@ CAF_TEST(empty_function_fiew) {
}
CAF_TEST(single_res_function_view) {
auto f = make_function_view(spawn(adder));
auto f = make_function_view(system.spawn(adder));
CAF_CHECK(f(3, 4) == 7);
CAF_CHECK(f != nullptr);
CAF_CHECK(nullptr != f);
......@@ -132,9 +119,9 @@ CAF_TEST(single_res_function_view) {
CAF_CHECK(g != nullptr);
CAF_CHECK(nullptr != g);
CAF_CHECK(g(10, 20) == 30);
g.assign(spawn(multiplier));
g.assign(system.spawn(multiplier));
CAF_CHECK(g(10, 20) == 200);
g.assign(spawn(divider));
g.assign(system.spawn(divider));
try {
g(1, 0);
CAF_ERROR("expected exception");
......@@ -143,17 +130,17 @@ CAF_TEST(single_res_function_view) {
CAF_CHECK(true);
}
CAF_CHECK(g == nullptr);
g.assign(spawn(divider));
g.assign(system.spawn(divider));
CAF_CHECK(g(4, 2) == 2);
}
CAF_TEST(tuple_res_function_view) {
auto f = make_function_view(spawn(simple_doubler));
auto f = make_function_view(system.spawn(simple_doubler));
CAF_CHECK(f(10) == std::make_tuple(10, 10));
}
CAF_TEST(cell_function_view) {
auto f = make_function_view(spawn(simple_cell));
auto f = make_function_view(system.spawn(simple_cell));
CAF_CHECK(f(get_atom::value) == 0);
f(put_atom::value, 1024);
CAF_CHECK(f(get_atom::value) == 1024);
......
......@@ -275,7 +275,6 @@ CAF_TEST(test_void_res) {
CAF_MESSAGE("received void res");
}
);
self->send_exit(buddy, exit_reason::kill);
}
CAF_TEST(pending_quit) {
......
......@@ -92,9 +92,10 @@ CAF_TEST(identity) {
CAF_CHECK(h->id() != g->id());
CAF_CHECK(h != g);
CAF_CHECK(h->message_types() == g->home_system().message_types(h));
anon_send_exit(h, exit_reason::kill);
anon_send_exit(f, exit_reason::kill);
anon_send_exit(g, exit_reason::kill);
// killing both f and g triggers down messages to h,
// ultimately clearing up held references and h will become unreachable
}
// spawned dead if `g` is already dead upon spawning
......
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