Commit af4f3410 authored by Dominik Charousset's avatar Dominik Charousset

Streamline composed_actor and bound_actor

parent 6395e33e
...@@ -43,9 +43,6 @@ public: ...@@ -43,9 +43,6 @@ public:
void enqueue(mailbox_element_ptr what, execution_unit* host) override; void enqueue(mailbox_element_ptr what, execution_unit* host) override;
private: private:
void handle_system_message(const message& msg, execution_unit* host);
static bool is_system_message(const message& msg);
actor_addr decorated_; actor_addr decorated_;
message merger_; message merger_;
}; };
......
...@@ -47,9 +47,6 @@ public: ...@@ -47,9 +47,6 @@ public:
message_types_set message_types() const override; message_types_set message_types() const override;
private: private:
void handle_system_message(const message& msg, execution_unit* host);
static bool is_system_message(const message& msg);
actor_addr f_; actor_addr f_;
actor_addr g_; actor_addr g_;
message_types_set msg_types_; message_types_set msg_types_;
......
...@@ -357,10 +357,7 @@ public: ...@@ -357,10 +357,7 @@ public:
attach(attachable_ptr{new functor_attachable(std::move(f))}); attach(attachable_ptr{new functor_attachable(std::move(f))});
} }
/// Returns an implementation-dependent name for logging purposes, which const char* name() const override;
/// is only valid as long as the actor is running. The default
/// implementation simply returns "actor".
virtual const char* name() const;
/// Serializes the state of this actor to `sink`. This function is /// Serializes the state of this actor to `sink`. This function is
/// only called if this actor has set the `is_serializable` flag. /// only called if this actor has set the `is_serializable` flag.
......
...@@ -32,6 +32,7 @@ ...@@ -32,6 +32,7 @@
#include <condition_variable> #include <condition_variable>
#include "caf/abstract_actor.hpp" #include "caf/abstract_actor.hpp"
#include "caf/mailbox_element.hpp"
#include "caf/detail/type_traits.hpp" #include "caf/detail/type_traits.hpp"
#include "caf/detail/functor_attachable.hpp" #include "caf/detail/functor_attachable.hpp"
...@@ -41,6 +42,11 @@ namespace caf { ...@@ -41,6 +42,11 @@ namespace caf {
/// Base class for all actor implementations. /// Base class for all actor implementations.
class monitorable_actor : public abstract_actor { class monitorable_actor : public abstract_actor {
public: public:
/// Returns an implementation-dependent name for logging purposes, which
/// is only valid as long as the actor is running. The default
/// implementation simply returns "actor".
virtual const char* name() const;
void attach(attachable_ptr ptr) override; void attach(attachable_ptr ptr) override;
size_t detach(const attachable::token& what) override; size_t detach(const attachable::token& what) override;
...@@ -82,7 +88,7 @@ protected: ...@@ -82,7 +88,7 @@ protected:
bool remove_backlink_impl(const actor_addr& other); bool remove_backlink_impl(const actor_addr& other);
// Tries to run a custom exception handler for `eptr`. // tries to run a custom exception handler for `eptr`
maybe<exit_reason> handle(const std::exception_ptr& eptr); maybe<exit_reason> handle(const std::exception_ptr& eptr);
inline void attach_impl(attachable_ptr& ptr) { inline void attach_impl(attachable_ptr& ptr) {
...@@ -95,6 +101,20 @@ protected: ...@@ -95,6 +101,20 @@ protected:
bool stop_on_first_hit = false, bool stop_on_first_hit = false,
bool dry_run = false); bool dry_run = false);
bool handle_system_message(mailbox_element& node, execution_unit* context,
bool trap_exit);
template <class F>
bool handle_system_message(mailbox_element& node, execution_unit* context,
bool trap_exit, F& down_msg_handler) {
auto& msg = node.msg;
if (msg.size() == 1 && msg.match_element<down_msg>(0)) {
down_msg_handler(msg.get_as<down_msg>(0));
return true;
}
return handle_system_message(node, context, trap_exit);
}
// initially set to exit_reason::not_exited // initially set to exit_reason::not_exited
std::atomic<exit_reason> exit_reason_; std::atomic<exit_reason> exit_reason_;
......
...@@ -41,6 +41,8 @@ enum class sec : uint8_t { ...@@ -41,6 +41,8 @@ enum class sec : uint8_t {
state_not_serializable, state_not_serializable,
/// An actor received an invalid key for `('sys', 'get', key)` messages. /// An actor received an invalid key for `('sys', 'get', key)` messages.
invalid_sys_key, invalid_sys_key,
/// An actor received an unsupported system message.
unsupported_sys_message,
/// A remote node disconnected during CAF handshake. /// A remote node disconnected during CAF handshake.
disconnect_during_handshake, disconnect_during_handshake,
/// Tried to forward a message via BASP to an invalid actor handle. /// Tried to forward a message via BASP to an invalid actor handle.
......
...@@ -87,8 +87,7 @@ actor operator*(actor f, actor g) { ...@@ -87,8 +87,7 @@ actor operator*(actor f, actor g) {
return invalid_actor; return invalid_actor;
auto ptr = make_counted<composed_actor>(f.address(), auto ptr = make_counted<composed_actor>(f.address(),
g.address(), g.address(),
g->home_system().message_types( std::set<std::string>{});
actor{}));
return actor_cast<actor>(std::move(ptr)); return actor_cast<actor>(std::move(ptr));
} }
......
...@@ -19,11 +19,12 @@ ...@@ -19,11 +19,12 @@
#include "caf/bound_actor.hpp" #include "caf/bound_actor.hpp"
#include "caf/sec.hpp"
#include "caf/actor_cast.hpp" #include "caf/actor_cast.hpp"
#include "caf/actor_system.hpp" #include "caf/actor_system.hpp"
#include "caf/default_attachable.hpp"
#include "caf/mailbox_element.hpp" #include "caf/mailbox_element.hpp"
#include "caf/system_messages.hpp" #include "caf/system_messages.hpp"
#include "caf/default_attachable.hpp"
#include "caf/detail/disposer.hpp" #include "caf/detail/disposer.hpp"
#include "caf/detail/merged_tuple.hpp" #include "caf/detail/merged_tuple.hpp"
...@@ -32,13 +33,13 @@ ...@@ -32,13 +33,13 @@
namespace caf { namespace caf {
bound_actor::bound_actor(actor_addr decorated, message msg) bound_actor::bound_actor(actor_addr decorated, message msg)
: monitorable_actor{ &decorated->home_system(), : monitorable_actor(&decorated->home_system(),
decorated->home_system().next_actor_id(), decorated->home_system().next_actor_id(),
decorated->node(), decorated->node(),
is_abstract_actor_flag is_abstract_actor_flag
| is_actor_bind_decorator_flag }, | is_actor_bind_decorator_flag),
decorated_{ std::move(decorated) }, decorated_(std::move(decorated)),
merger_{ std::move(msg) } { merger_(std::move(msg)) {
// bound actor has dependency on the decorated actor by default; // bound actor has dependency on the decorated actor by default;
// if the decorated actor is already dead upon establishing the // if the decorated actor is already dead upon establishing the
// dependency, the actor is spawned dead // dependency, the actor is spawned dead
...@@ -62,53 +63,22 @@ void bound_actor::enqueue(mailbox_element_ptr what, ...@@ -62,53 +63,22 @@ void bound_actor::enqueue(mailbox_element_ptr what,
} }
return; return;
} }
if (is_system_message(what->msg)) { auto down_handler = [&](const down_msg& dm) {
// handle and consume the system message; if (dm.source == decorated_)
// the only effect that MAY result from handling a system message cleanup(dm.reason, host);
// is to exit the actor if it hasn't exited already; };
// `handle_system_message()` is thread-safe, and if the actor // handle and consume the system message;
// has already exited upon the invocation, nothing is done // the only effect that MAY result from handling a system message
handle_system_message(what->msg, host); // is to exit the actor if it hasn't exited already;
} else { // `handle_system_message()` is thread-safe, and if the actor
// process and forward the non-system message // has already exited upon the invocation, nothing is done
auto& msg = what->msg; if (handle_system_message(*what, host, false, down_handler))
message tmp{ detail::merged_tuple::make(merger_, std::move(msg)) };
msg.swap(tmp);
decorated_->enqueue(std::move(what), host);
}
}
void bound_actor::handle_system_message(const message& msg,
execution_unit* host) {
// `monitorable_actor::cleanup()` is thread-safe, and if the actor
// has already exited upon the invocation, nothing is done;
// handles only `down_msg` from the decorated actor and `exit_msg` from anyone
if (msg.size() != 1)
return; // neither case
if (msg.match_element<down_msg>(0)) {
auto& dm = msg.get_as<down_msg>(0);
CAF_ASSERT(dm.reason != exit_reason::not_exited);
if (dm.source == decorated_) {
// decorated actor has exited, so exits
// the bound actor with the same reason
monitorable_actor::cleanup(dm.reason, host);
}
return; return;
} // process and forward non-system messages
if (msg.match_element<exit_msg>(0)) { auto& msg = what->msg;
auto& em = msg.get_as<exit_msg>(0); message tmp{detail::merged_tuple::make(merger_, std::move(msg))};
CAF_ASSERT(em.reason != exit_reason::not_exited); msg.swap(tmp);
// exit message received, so exits with the same reason decorated_->enqueue(std::move(what), host);
monitorable_actor::cleanup(em.reason, host);
return;
}
// drop other cases
}
bool bound_actor::is_system_message(const message& msg) {
return (msg.size() == 1 && (msg.match_element<exit_msg>(0)
|| msg.match_element<down_msg>(0)))
|| (msg.size() > 1 && msg.match_element<sys_atom>(0));
} }
} // namespace caf } // namespace caf
...@@ -29,14 +29,13 @@ namespace caf { ...@@ -29,14 +29,13 @@ namespace caf {
composed_actor::composed_actor(actor_addr f, actor_addr g, composed_actor::composed_actor(actor_addr f, actor_addr g,
message_types_set msg_types) message_types_set msg_types)
: monitorable_actor{ &g->home_system(), : monitorable_actor(&g->home_system(),
g->home_system().next_actor_id(), g->home_system().next_actor_id(),
g->node(), g->node(),
is_abstract_actor_flag is_abstract_actor_flag | is_actor_dot_decorator_flag),
| is_actor_dot_decorator_flag }, f_(std::move(f)),
f_{ std::move(f) }, g_(std::move(g)),
g_{ std::move(g) }, msg_types_(std::move(msg_types)) {
msg_types_{ std::move(msg_types) } {
// composed actor has dependency on constituent actors by default; // composed actor has dependency on constituent actors by default;
// if either constituent actor is already dead upon establishing // if either constituent actor is already dead upon establishing
// the dependency, the actor is spawned dead // the dependency, the actor is spawned dead
...@@ -46,7 +45,7 @@ composed_actor::composed_actor(actor_addr f, actor_addr g, ...@@ -46,7 +45,7 @@ composed_actor::composed_actor(actor_addr f, actor_addr g,
} }
void composed_actor::enqueue(mailbox_element_ptr what, void composed_actor::enqueue(mailbox_element_ptr what,
execution_unit* host) { execution_unit* context) {
if (! what) if (! what)
return; // not even an empty message return; // not even an empty message
auto reason = exit_reason_.load(); auto reason = exit_reason_.load();
...@@ -62,57 +61,27 @@ void composed_actor::enqueue(mailbox_element_ptr what, ...@@ -62,57 +61,27 @@ void composed_actor::enqueue(mailbox_element_ptr what,
} }
return; return;
} }
if (is_system_message(what->msg)) { auto down_msg_handler = [&](const down_msg& dm) {
// handle and consume the system message; // quit if either `f` or `g` are no longer available
// the only effect that MAY result from handling a system message if (dm.source == f_ || dm.source == g_)
// is to exit the actor if it hasn't exited already; monitorable_actor::cleanup(dm.reason, context);
// `handle_system_message()` is thread-safe, and if the actor };
// has already exited upon the invocation, nothing is done // handle and consume the system message;
handle_system_message(what->msg, host); // the only effect that MAY result from handling a system message
} else { // is to exit the actor if it hasn't exited already;
// process and forward the non-system message; // `handle_system_message()` is thread-safe, and if the actor
// store `f` as the next stage in the forwarding chain // has already exited upon the invocation, nothing is done
what->stages.push_back(f_); if (handle_system_message(*what, context, false, down_msg_handler))
// forward modified message to `g` return;
g_->enqueue(std::move(what), host); // process and forward the non-system message;
} // store `f` as the next stage in the forwarding chain
what->stages.push_back(f_);
// forward modified message to `g`
g_->enqueue(std::move(what), context);
} }
composed_actor::message_types_set composed_actor::message_types() const { composed_actor::message_types_set composed_actor::message_types() const {
return msg_types_; return msg_types_;
} }
void composed_actor::handle_system_message(const message& msg,
execution_unit* host) {
// `monitorable_actor::cleanup()` is thread-safe, and if the actor
// has already exited upon the invocation, nothing is done;
// handles only `down_msg` from constituent actors and `exit_msg` from anyone
if (msg.size() != 1)
return; // neither case
if (msg.match_element<down_msg>(0)) {
auto& dm = msg.get_as<down_msg>(0);
CAF_ASSERT(dm.reason != exit_reason::not_exited);
if (dm.source == f_ || dm.source == g_) {
// one of the constituent actors has exited, so
// exits the composed actor with the same reason
monitorable_actor::cleanup(dm.reason, host);
}
return;
}
if (msg.match_element<exit_msg>(0)) {
auto& em = msg.get_as<exit_msg>(0);
CAF_ASSERT(em.reason != exit_reason::not_exited);
// exit message received; exits with the same reason
monitorable_actor::cleanup(em.reason, host);
return;
}
// drop other cases
}
bool composed_actor::is_system_message(const message& msg) {
return (msg.size() == 1 && (msg.match_element<exit_msg>(0)
|| msg.match_element<down_msg>(0)))
|| (msg.size() > 1 && msg.match_element<sys_atom>(0));
}
} // namespace caf } // namespace caf
...@@ -19,6 +19,7 @@ ...@@ -19,6 +19,7 @@
#include "caf/monitorable_actor.hpp" #include "caf/monitorable_actor.hpp"
#include "caf/sec.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/actor_cast.hpp" #include "caf/actor_cast.hpp"
#include "caf/actor_system.hpp" #include "caf/actor_system.hpp"
...@@ -27,6 +28,10 @@ ...@@ -27,6 +28,10 @@
namespace caf { namespace caf {
const char* monitorable_actor::name() const {
return "monitorable_actor";
}
void monitorable_actor::attach(attachable_ptr ptr) { void monitorable_actor::attach(attachable_ptr ptr) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
if (ptr == nullptr) if (ptr == nullptr)
...@@ -210,4 +215,44 @@ size_t monitorable_actor::detach_impl(const attachable::token& what, ...@@ -210,4 +215,44 @@ size_t monitorable_actor::detach_impl(const attachable::token& what,
return detach_impl(what, ptr->next, stop_on_hit, dry_run); return detach_impl(what, ptr->next, stop_on_hit, dry_run);
} }
bool monitorable_actor::handle_system_message(mailbox_element& node,
execution_unit* context,
bool trap_exit) {
auto& msg = node.msg;
if (! trap_exit && msg.size() == 1 && msg.match_element<exit_msg>(0)) {
auto& em = msg.get_as<exit_msg>(0);
CAF_ASSERT(em.reason != exit_reason::not_exited);
if (em.reason != exit_reason::normal)
cleanup(em.reason, context);
return true;
} else if (msg.size() > 1 && msg.match_element<sys_atom>(0)) {
if (! node.sender)
return true;
error err;
mailbox_element_ptr res;
msg.apply({
[&](sys_atom, get_atom, std::string& what) {
CAF_LOG_TRACE(CAF_ARG(what));
if (what != "info") {
err = sec::invalid_sys_key;
return;
}
res = mailbox_element::make_joint(address(), node.mid.response_id(), {},
ok_atom::value, std::move(what),
address(), name());
},
others >> [&] {
err = sec::unsupported_sys_message;
}
});
if (err && node.mid.is_request())
res = mailbox_element::make_joint(address(), node.mid.response_id(),
{}, std::move(err));
if (res)
node.sender->enqueue(std::move(res), context);
return true;
}
return false;
}
} // namespace caf } // namespace caf
...@@ -57,7 +57,7 @@ struct fixture { ...@@ -57,7 +57,7 @@ struct fixture {
} }
actor_system system; actor_system system;
scoped_actor self{ system, true }; scoped_actor self{system, true};
}; };
} // namespace <anonymous> } // namespace <anonymous>
...@@ -103,8 +103,8 @@ CAF_TEST(lifetime_2) { ...@@ -103,8 +103,8 @@ CAF_TEST(lifetime_2) {
CAF_TEST(lifetime_3) { CAF_TEST(lifetime_3) {
auto dbl = system.spawn(dbl_bhvr); auto dbl = system.spawn(dbl_bhvr);
auto bound = dbl.bind(1); auto bound = dbl.bind(1);
anon_send(bound, down_msg{ self->address(), anon_send(bound, down_msg{self->address(),
exit_reason::kill }); exit_reason::kill});
CAF_CHECK(! exited(bound)); CAF_CHECK(! exited(bound));
self->monitor(bound); self->monitor(bound);
auto em_sender = system.spawn(dbl_bhvr); auto em_sender = system.spawn(dbl_bhvr);
...@@ -177,7 +177,7 @@ CAF_TEST(full_currying) { ...@@ -177,7 +177,7 @@ CAF_TEST(full_currying) {
[](int v) { [](int v) {
CAF_CHECK(v == 2); CAF_CHECK(v == 2);
}, },
[](error err) { [](error) {
CAF_CHECK(false); CAF_CHECK(false);
} }
); );
......
...@@ -72,12 +72,12 @@ struct fixture { ...@@ -72,12 +72,12 @@ struct fixture {
} }
actor_system system; actor_system system;
scoped_actor self{ system, true }; scoped_actor self{system, true};
}; };
} // namespace <anonymous> } // namespace <anonymous>
CAF_TEST_FIXTURE_SCOPE(bound_actor_tests, fixture) CAF_TEST_FIXTURE_SCOPE(composed_actor_tests, fixture)
CAF_TEST(identity) { CAF_TEST(identity) {
actor_system system_of_g; actor_system system_of_g;
...@@ -85,15 +85,14 @@ CAF_TEST(identity) { ...@@ -85,15 +85,14 @@ CAF_TEST(identity) {
auto g = system_of_g.spawn(first_stage_impl); auto g = system_of_g.spawn(first_stage_impl);
auto f = system_of_f.spawn(second_stage_impl); auto f = system_of_f.spawn(second_stage_impl);
CAF_CHECK(system_of_g.registry().running() == 1); CAF_CHECK(system_of_g.registry().running() == 1);
auto composed = f * g; auto h = f * g;
CAF_CHECK(system_of_g.registry().running() == 1); CAF_CHECK(system_of_g.registry().running() == 1);
CAF_CHECK(&composed->home_system() == &g->home_system()); CAF_CHECK(&h->home_system() == &g->home_system());
CAF_CHECK(composed->node() == g->node()); CAF_CHECK(h->node() == g->node());
CAF_CHECK(composed->id() != g->id()); CAF_CHECK(h->id() != g->id());
CAF_CHECK(composed != g); CAF_CHECK(h != g);
CAF_CHECK(composed->message_types() CAF_CHECK(h->message_types() == g->home_system().message_types(h));
== g->home_system().message_types(composed)); anon_send_exit(h, exit_reason::kill);
anon_send_exit(composed, exit_reason::kill);
anon_send_exit(f, exit_reason::kill); anon_send_exit(f, exit_reason::kill);
anon_send_exit(g, exit_reason::kill); anon_send_exit(g, exit_reason::kill);
} }
...@@ -105,8 +104,8 @@ CAF_TEST(lifetime_1a) { ...@@ -105,8 +104,8 @@ CAF_TEST(lifetime_1a) {
self->monitor(g); self->monitor(g);
anon_send_exit(g, exit_reason::kill); anon_send_exit(g, exit_reason::kill);
wait_until_exited(); wait_until_exited();
auto fg = f * g; auto h = f * g;
CAF_CHECK(exited(fg)); CAF_CHECK(exited(h));
anon_send_exit(f, exit_reason::kill); anon_send_exit(f, exit_reason::kill);
} }
...@@ -117,8 +116,8 @@ CAF_TEST(lifetime_1b) { ...@@ -117,8 +116,8 @@ CAF_TEST(lifetime_1b) {
self->monitor(f); self->monitor(f);
anon_send_exit(f, exit_reason::kill); anon_send_exit(f, exit_reason::kill);
wait_until_exited(); wait_until_exited();
auto fg = f * g; auto h = f * g;
CAF_CHECK(exited(fg)); CAF_CHECK(exited(h));
anon_send_exit(g, exit_reason::kill); anon_send_exit(g, exit_reason::kill);
} }
...@@ -126,8 +125,8 @@ CAF_TEST(lifetime_1b) { ...@@ -126,8 +125,8 @@ CAF_TEST(lifetime_1b) {
CAF_TEST(lifetime_2a) { CAF_TEST(lifetime_2a) {
auto g = system.spawn(dbl_bhvr); auto g = system.spawn(dbl_bhvr);
auto f = system.spawn(dbl_bhvr); auto f = system.spawn(dbl_bhvr);
auto fg = f * g; auto h = f * g;
self->monitor(fg); self->monitor(h);
anon_send(g, message{}); anon_send(g, message{});
wait_until_exited(); wait_until_exited();
anon_send_exit(f, exit_reason::kill); anon_send_exit(f, exit_reason::kill);
...@@ -137,8 +136,8 @@ CAF_TEST(lifetime_2a) { ...@@ -137,8 +136,8 @@ CAF_TEST(lifetime_2a) {
CAF_TEST(lifetime_2b) { CAF_TEST(lifetime_2b) {
auto g = system.spawn(dbl_bhvr); auto g = system.spawn(dbl_bhvr);
auto f = system.spawn(dbl_bhvr); auto f = system.spawn(dbl_bhvr);
auto fg = f * g; auto h = f * g;
self->monitor(fg); self->monitor(h);
anon_send(f, message{}); anon_send(f, message{});
wait_until_exited(); wait_until_exited();
anon_send_exit(g, exit_reason::kill); anon_send_exit(g, exit_reason::kill);
...@@ -150,13 +149,12 @@ CAF_TEST(lifetime_2b) { ...@@ -150,13 +149,12 @@ CAF_TEST(lifetime_2b) {
CAF_TEST(lifetime_3) { CAF_TEST(lifetime_3) {
auto g = system.spawn(dbl_bhvr); auto g = system.spawn(dbl_bhvr);
auto f = system.spawn(dbl_bhvr); auto f = system.spawn(dbl_bhvr);
auto fg = f * g; auto h = f * g;
self->monitor(fg); self->monitor(h);
anon_send(fg, down_msg{ self->address(), anon_send(h, down_msg{self->address(), exit_reason::kill});
exit_reason::kill }); CAF_CHECK(! exited(h));
CAF_CHECK(! exited(fg));
auto em_sender = system.spawn(dbl_bhvr); auto em_sender = system.spawn(dbl_bhvr);
em_sender->link_to(fg.address()); em_sender->link_to(h.address());
anon_send_exit(em_sender, exit_reason::kill); anon_send_exit(em_sender, exit_reason::kill);
wait_until_exited(); wait_until_exited();
self->request(f, 1).receive( self->request(f, 1).receive(
...@@ -182,10 +180,10 @@ CAF_TEST(lifetime_3) { ...@@ -182,10 +180,10 @@ CAF_TEST(lifetime_3) {
CAF_TEST(request_response_promise) { CAF_TEST(request_response_promise) {
auto g = system.spawn(dbl_bhvr); auto g = system.spawn(dbl_bhvr);
auto f = system.spawn(dbl_bhvr); auto f = system.spawn(dbl_bhvr);
auto fg = f * g; auto h = f * g;
anon_send_exit(fg, exit_reason::kill); anon_send_exit(h, exit_reason::kill);
CAF_CHECK(exited(fg)); CAF_CHECK(exited(h));
self->request(fg, 1).receive( self->request(h, 1).receive(
[](int) { [](int) {
CAF_CHECK(false); CAF_CHECK(false);
}, },
......
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