Commit 05bae436 authored by Dominik Charousset's avatar Dominik Charousset

Streamline flag management in actors

parent b98cfd6e
......@@ -153,96 +153,25 @@ public:
static constexpr int is_terminated_flag = 0x0800; // local_actor
static constexpr int is_cleaned_up_flag = 0x1000; // monitorable_actor
inline void set_flag(bool enable_flag, int mask) {
inline void setf(int flag) {
auto x = flags();
flags(enable_flag ? x | mask : x & ~mask);
flags(x | flag);
}
inline bool get_flag(int mask) const {
return static_cast<bool>(flags() & mask);
}
inline bool has_timeout() const {
return get_flag(has_timeout_flag);
}
inline void has_timeout(bool value) {
set_flag(value, has_timeout_flag);
}
inline bool is_initialized() const {
return get_flag(is_initialized_flag);
}
inline void is_initialized(bool value) {
set_flag(value, is_initialized_flag);
}
inline bool is_blocking() const {
return get_flag(is_blocking_flag);
}
inline void is_blocking(bool value) {
set_flag(value, is_blocking_flag);
}
inline bool is_detached() const {
return get_flag(is_detached_flag);
}
inline void is_detached(bool value) {
set_flag(value, is_detached_flag);
}
inline bool is_priority_aware() const {
return get_flag(is_priority_aware_flag);
}
inline void is_priority_aware(bool value) {
set_flag(value, is_priority_aware_flag);
}
inline bool is_serializable() const {
return get_flag(is_serializable_flag);
}
inline void is_serializable(bool value) {
set_flag(value, is_serializable_flag);
}
inline bool is_migrated_from() const {
return get_flag(is_migrated_from_flag);
}
inline void is_migrated_from(bool value) {
set_flag(value, is_migrated_from_flag);
}
inline bool is_registered() const {
return get_flag(is_registered_flag);
}
void is_registered(bool value);
inline bool is_actor_decorator() const {
return static_cast<bool>(flags() & is_actor_decorator_mask);
}
inline bool is_terminated() const {
return get_flag(is_terminated_flag);
inline void unsetf(int flag) {
auto x = flags();
flags(x & ~flag);
}
inline void is_terminated(bool value) {
set_flag(value, is_terminated_flag);
inline bool getf(int flag) const {
return (flags() & flag) != 0;
}
inline bool is_cleaned_up() const {
return get_flag(is_cleaned_up_flag);
}
/// Sets `is_registered_flag` and calls `system().registry().inc_running()`.
void register_at_system();
inline void is_cleaned_up(bool value) {
set_flag(value, is_cleaned_up_flag);
}
/// Unsets `is_registered_flag` and calls `system().registry().dec_running()`.
void unregister_from_system();
virtual bool link_impl(linking_operation op, abstract_actor* other) = 0;
......
......@@ -45,7 +45,7 @@ public:
: Base(cfg, std::forward<Ts>(xs)...),
state(state_) {
if (detail::is_serializable<State>::value)
this->is_serializable(true);
this->setf(Base::is_serializable_flag);
}
~stateful_actor() {
......@@ -55,7 +55,7 @@ public:
/// Destroys the state of this actor (no further overriding allowed).
void on_exit() final {
CAF_LOG_TRACE("");
if (this->is_initialized())
if (this->getf(Base::is_initialized_flag))
state_.~State();
}
......
......@@ -68,7 +68,7 @@ public:
}
void initialize() override {
this->is_initialized(true);
this->setf(abstract_actor::is_initialized_flag);
auto bhvr = make_behavior();
CAF_LOG_DEBUG_IF(!bhvr, "make_behavior() did not return a behavior:"
<< CAF_ARG(this->has_behavior()));
......
......@@ -91,14 +91,18 @@ actor_system& abstract_actor::home_system() const noexcept {
return *(actor_control_block::from(this)->home_system);
}
void abstract_actor::is_registered(bool value) {
if (is_registered() == value)
void abstract_actor::register_at_system() {
if (getf(is_registered_flag))
return;
if (value)
home_system().registry().inc_running();
else
home_system().registry().dec_running();
set_flag(value, is_registered_flag);
setf(is_registered_flag);
home_system().registry().inc_running();
}
void abstract_actor::unregister_from_system() {
if (!getf(is_registered_flag))
return;
unsetf(is_registered_flag);
home_system().registry().dec_running();
}
namespace {
......
......@@ -53,7 +53,8 @@ void actor_companion::enqueue(strong_actor_ptr src, message_id mid,
}
void actor_companion::launch(execution_unit*, bool, bool hide) {
is_registered(!hide);
if (!hide)
register_at_system();
}
void actor_companion::on_exit() {
......
......@@ -74,8 +74,8 @@ void actor_ostream::redirect_all(actor_system& sys, std::string fn, int flags) {
}
void actor_ostream::init(abstract_actor* self) {
if (!self->get_flag(abstract_actor::has_used_aout_flag))
self->set_flag(true, abstract_actor::has_used_aout_flag);
if (!self->getf(abstract_actor::has_used_aout_flag))
self->setf(abstract_actor::has_used_aout_flag);
}
actor_ostream aout(local_actor* self) {
......
......@@ -124,7 +124,7 @@ void actor_pool::enqueue(mailbox_element_ptr what, execution_unit* eu) {
}
actor_pool::actor_pool(actor_config& cfg) : monitorable_actor(cfg) {
is_registered(true);
register_at_system();
}
void actor_pool::on_cleanup() {
......@@ -149,7 +149,7 @@ bool actor_pool::filter(upgrade_lock<detail::shared_spinlock>& guard,
unique_guard.unlock();
for (auto& w : workers)
anon_send(w, tmp);
is_registered(false);
unregister_from_system();
}
return true;
}
......@@ -210,7 +210,7 @@ void actor_pool::quit(execution_unit* host) {
// we can safely run our cleanup code here without holding
// workers_mtx_ because abstract_actor has its own lock
if (cleanup(planned_reason_, host))
is_registered(false);
unregister_from_system();
}
} // namespace caf
......@@ -76,8 +76,9 @@ const char* blocking_actor::name() const {
void blocking_actor::launch(execution_unit*, bool, bool hide) {
CAF_LOG_TRACE(CAF_ARG(hide));
CAF_ASSERT(is_blocking());
is_registered(!hide);
CAF_ASSERT(getf(is_blocking_flag));
if (!hide)
register_at_system();
home_system().inc_detached_threads();
std::thread([](strong_actor_ptr ptr) {
// actor lives in its own thread
......@@ -114,7 +115,8 @@ blocking_actor::receive_while(const bool& ref) {
}
void blocking_actor::await_all_other_actors_done() {
system().registry().await_running_count_equal(is_registered() ? 1 : 0);
system().registry().await_running_count_equal(getf(is_registered_flag) ? 1
: 0);
}
void blocking_actor::act() {
......
......@@ -32,7 +32,7 @@ event_based_actor::~event_based_actor() {
void event_based_actor::initialize() {
CAF_LOG_TRACE("subtype =" << logger::render_type_name(typeid(*this)).c_str());
is_initialized(true);
setf(is_initialized_flag);
auto bhvr = make_behavior();
CAF_LOG_DEBUG_IF(!bhvr, "make_behavior() did not return a behavior:"
<< CAF_ARG(has_behavior()));
......
......@@ -60,7 +60,7 @@ void local_actor::on_destroy() {
// alternatively, we would have to use a reference-counted,
// heap-allocated logger
CAF_SET_LOGGER_SYS(nullptr);
if (!is_cleaned_up()) {
if (!getf(is_cleaned_up_flag)) {
on_exit();
cleanup(exit_reason::unreachable, nullptr);
monitorable_actor::on_destroy();
......@@ -101,7 +101,7 @@ message_id local_actor::new_request_id(message_priority mp) {
}
mailbox_element_ptr local_actor::next_message() {
if (!is_priority_aware())
if (!getf(is_priority_aware_flag))
return mailbox_element_ptr{mailbox().try_pop()};
// we partition the mailbox into four segments in this case:
// <-------- ! was_skipped --------> | <-------- was_skipped -------->
......@@ -132,7 +132,7 @@ mailbox_element_ptr local_actor::next_message() {
}
bool local_actor::has_next_message() {
if (!is_priority_aware())
if (!getf(is_priority_aware_flag))
return mailbox_.can_fetch_more();
auto& mbox = mailbox();
auto& cache = mbox.cache();
......@@ -142,7 +142,7 @@ bool local_actor::has_next_message() {
void local_actor::push_to_cache(mailbox_element_ptr ptr) {
CAF_ASSERT(ptr != nullptr);
CAF_LOG_TRACE(CAF_ARG(*ptr));
if (!is_priority_aware() || !ptr->is_high_priority()) {
if (!getf(is_priority_aware_flag) || !ptr->is_high_priority()) {
mailbox().cache().insert(mailbox().cache().end(), ptr.release());
return;
}
......@@ -190,7 +190,7 @@ bool local_actor::cleanup(error&& fail_state, execution_unit* host) {
mailbox_.close(f);
}
// tell registry we're done
is_registered(false);
unregister_from_system();
monitorable_actor::cleanup(std::move(fail_state), host);
return true;
}
......
......@@ -42,7 +42,7 @@ void monitorable_actor::attach(attachable_ptr ptr) {
CAF_ASSERT(ptr);
error fail_state;
auto attached = exclusive_critical_section([&] {
if (is_terminated()) {
if (getf(is_terminated_flag)) {
fail_state = fail_state_;
return false;
}
......@@ -64,7 +64,7 @@ bool monitorable_actor::cleanup(error&& reason, execution_unit* host) {
CAF_LOG_TRACE(CAF_ARG(reason));
attachable_ptr head;
bool set_fail_state = exclusive_critical_section([&]() -> bool {
if (!is_cleaned_up()) {
if (!getf(is_cleaned_up_flag)) {
// local actors pass fail_state_ as first argument
if (&fail_state_ != &reason)
fail_state_ = std::move(reason);
......@@ -83,7 +83,7 @@ bool monitorable_actor::cleanup(error&& reason, execution_unit* host) {
for (attachable* i = head.get(); i != nullptr; i = i->next.get())
i->actor_exited(reason, host);
// tell printer to purge its state for us if we ever used aout()
if (get_flag(abstract_actor::has_used_aout_flag)) {
if (getf(abstract_actor::has_used_aout_flag)) {
auto pr = home_system().scheduler().printer();
pr->enqueue(make_mailbox_element(nullptr, message_id::make(), {},
delete_atom::value, id()),
......@@ -138,7 +138,7 @@ bool monitorable_actor::establish_link_impl(abstract_actor* x) {
bool send_exit_immediately = false;
auto tmp = default_attachable::make_link(address(), x->address());
auto success = exclusive_critical_section([&]() -> bool {
if (is_terminated()) {
if (getf(is_terminated_flag)) {
fail_state = fail_state_;
send_exit_immediately = true;
return false;
......@@ -165,7 +165,7 @@ bool monitorable_actor::establish_backlink_impl(abstract_actor* x) {
default_attachable::link};
auto tmp = default_attachable::make_link(address(), x->address());
auto success = exclusive_critical_section([&]() -> bool {
if (is_terminated()) {
if (getf(is_terminated_flag)) {
fail_state = fail_state_;
send_exit_immediately = true;
return false;
......
......@@ -58,7 +58,7 @@ result<message> drop(scheduled_actor*, message_view&) {
void scheduled_actor::default_error_handler(scheduled_actor* ptr, error& x) {
ptr->fail_state_ = std::move(x);
ptr->is_terminated(true);
ptr->setf(is_terminated_flag);
}
void scheduled_actor::default_down_handler(scheduled_actor* ptr, down_msg& x) {
......@@ -123,14 +123,14 @@ void scheduled_actor::enqueue(mailbox_element_ptr ptr,
CAF_PUSH_AID(id());
CAF_LOG_TRACE(CAF_ARG(*ptr));
CAF_ASSERT(ptr != nullptr);
CAF_ASSERT(!is_blocking());
CAF_ASSERT(!getf(is_blocking_flag));
auto mid = ptr->mid;
auto sender = ptr->sender;
switch (mailbox().enqueue(ptr.release())) {
case detail::enqueue_result::unblocked_reader: {
// add a reference count to this actor and re-schedule it
intrusive_ptr_add_ref(ctrl());
if (is_detached()) {
if (getf(is_detached_flag)) {
CAF_ASSERT(private_thread_ != nullptr);
private_thread_->resume();
} else {
......@@ -162,9 +162,10 @@ const char* scheduled_actor::name() const {
void scheduled_actor::launch(execution_unit* eu, bool lazy, bool hide) {
CAF_LOG_TRACE(CAF_ARG(lazy) << CAF_ARG(hide));
CAF_ASSERT(!is_blocking());
is_registered(!hide);
if (is_detached()) {
CAF_ASSERT(!getf(is_blocking_flag));
if (!hide)
register_at_system();
if (getf(is_detached_flag)) {
private_thread_ = new detail::private_thread(this);
private_thread_->start();
return;
......@@ -181,7 +182,7 @@ void scheduled_actor::launch(execution_unit* eu, bool lazy, bool hide) {
}
bool scheduled_actor::cleanup(error&& fail_state, execution_unit* host) {
if (is_detached()) {
if (getf(is_detached_flag)) {
CAF_ASSERT(private_thread_ != nullptr);
private_thread_->shutdown();
}
......@@ -264,17 +265,17 @@ proxy_registry* scheduled_actor::proxy_registry_ptr() {
void scheduled_actor::quit(error x) {
CAF_LOG_TRACE(CAF_ARG(x));
fail_state_ = std::move(x);
is_terminated(true);
setf(is_terminated_flag);
}
// -- timeout management -------------------------------------------------------
uint32_t scheduled_actor::request_timeout(const duration& d) {
if (!d.valid()) {
has_timeout(false);
unsetf(has_timeout_flag);
return 0;
}
has_timeout(true);
setf(has_timeout_flag);
auto result = ++timeout_id_;
auto msg = make_message(timeout_msg{++timeout_id_});
CAF_LOG_TRACE("send new timeout_msg, " << CAF_ARG(timeout_id_));
......@@ -288,13 +289,12 @@ uint32_t scheduled_actor::request_timeout(const duration& d) {
}
void scheduled_actor::reset_timeout(uint32_t timeout_id) {
if (is_active_timeout(timeout_id)) {
has_timeout(false);
}
if (is_active_timeout(timeout_id))
unsetf(has_timeout_flag);
}
bool scheduled_actor::is_active_timeout(uint32_t tid) const {
return has_timeout() && timeout_id_ == tid;
return getf(has_timeout_flag) && timeout_id_ == tid;
}
// -- message processing -------------------------------------------------------
......@@ -351,7 +351,7 @@ scheduled_actor::categorize(mailbox_element& x) {
// exit_reason::kill is always fatal
if (em.reason == exit_reason::kill) {
fail_state_ = std::move(em.reason);
is_terminated(true);
setf(is_terminated_flag);
} else {
exit_handler_(this, em);
}
......@@ -423,13 +423,13 @@ invoke_message_result scheduled_actor::consume(mailbox_element& x) {
case message_category::ordinary: {
detail::default_invoke_result_visitor visitor{this};
bool skipped = false;
auto had_timeout = has_timeout();
auto had_timeout = getf(has_timeout_flag);
if (had_timeout)
has_timeout(false);
unsetf(has_timeout_flag);
// restore timeout at scope exit if message was skipped
auto timeout_guard = detail::make_scope_guard([&] {
if (skipped && had_timeout)
has_timeout(true);
setf(has_timeout_flag);
});
auto call_default_handler = [&] {
auto sres = default_handler_(this, x);
......@@ -498,19 +498,20 @@ bool scheduled_actor::consume_from_cache() {
bool scheduled_actor::activate(execution_unit* ctx) {
CAF_LOG_TRACE("");
CAF_ASSERT(ctx != nullptr);
CAF_ASSERT(!is_blocking());
CAF_ASSERT(!getf(is_blocking_flag));
context(ctx);
if (is_initialized() && (!has_behavior() || is_terminated())) {
if (getf(is_initialized_flag)
&& (!has_behavior() || getf(is_terminated_flag))) {
CAF_LOG_DEBUG_IF(!has_behavior(),
"resume called on an actor without behavior");
CAF_LOG_DEBUG_IF(is_terminated(),
CAF_LOG_DEBUG_IF(getf(is_terminated_flag),
"resume called on a terminated actor");
return false;
}
# ifndef CAF_NO_EXCEPTIONS
try {
# endif // CAF_NO_EXCEPTIONS
if (!is_initialized()) {
if (!getf(is_initialized_flag)) {
initialize();
if (finalize()) {
CAF_LOG_DEBUG("actor_done() returned true right after make_behavior()");
......@@ -590,7 +591,7 @@ void scheduled_actor::do_become(behavior bhvr, bool discard_old) {
}
bool scheduled_actor::finalize() {
if (has_behavior() && !is_terminated())
if (has_behavior() && !getf(is_terminated_flag))
return false;
CAF_LOG_DEBUG("actor either has no behavior or has set an exit reason");
on_exit();
......
......@@ -31,7 +31,7 @@ namespace {
class impl : public blocking_actor {
public:
impl(actor_config& cfg) : blocking_actor(cfg) {
is_detached(true);
setf(is_detached_flag);
}
void act() override {
......@@ -52,7 +52,8 @@ scoped_actor::scoped_actor(actor_system& sys, bool hide) : context_(&sys) {
if (!hide)
prev_ = CAF_SET_AID(self_->id());
CAF_LOG_TRACE(CAF_ARG(hide));
ptr()->is_registered(!hide);
if (!hide)
ptr()->register_at_system();
}
scoped_actor::~scoped_actor() {
......@@ -60,9 +61,9 @@ scoped_actor::~scoped_actor() {
if (!self_)
return;
auto x = ptr();
if (x->is_registered())
if (x->getf(abstract_actor::is_registered_flag))
CAF_SET_AID(prev_);
if (!x->is_terminated())
if (!x->getf(abstract_actor::is_terminated_flag))
x->cleanup(exit_reason::normal, &context_);
}
......
......@@ -45,7 +45,7 @@ struct fixture {
auto ptr = actor_cast<abstract_actor*>(handle);
auto dptr = dynamic_cast<monitorable_actor*>(ptr);
CAF_REQUIRE(dptr != nullptr);
return dptr->is_terminated();
return dptr->getf(abstract_actor::is_terminated_flag);
}
fixture() : system(cfg) {
......
......@@ -67,7 +67,7 @@ struct fixture {
auto ptr = actor_cast<abstract_actor*>(handle);
auto dptr = dynamic_cast<monitorable_actor*>(ptr);
CAF_REQUIRE(dptr != nullptr);
return dptr->is_terminated();
return dptr->getf(abstract_actor::is_terminated_flag);
}
actor_system_config cfg;
......
......@@ -93,7 +93,7 @@ protected:
invoke_mailbox_element_impl(ctx, value_);
// only consume an activity token if actor did not produce them now
if (prev && activity_tokens_ && --(*activity_tokens_) == 0) {
if (this->parent()->is_terminated())
if (this->parent()->getf(abstract_actor::is_terminated_flag))
return false;
// tell broker it entered passive mode, this can result in
// producing, why we check the condition again afterwards
......
......@@ -50,7 +50,8 @@ void abstract_broker::launch(execution_unit* eu, bool is_lazy, bool is_hidden) {
CAF_ASSERT(eu != nullptr);
CAF_ASSERT(eu == &backend());
// add implicit reference count held by middleman/multiplexer
is_registered(!is_hidden);
if (!is_hidden)
register_at_system();
CAF_PUSH_AID(id());
CAF_LOG_TRACE("init and launch broker:" << CAF_ARG(id()));
if (is_lazy && mailbox().try_block())
......@@ -141,7 +142,7 @@ abstract_broker::add_tcp_scribe(network::native_socket fd) {
void abstract_broker::add_doorman(const intrusive_ptr<doorman>& ptr) {
doormen_.emplace(ptr->hdl(), ptr);
if (is_initialized())
if (getf(is_initialized_flag))
ptr->launch();
}
......@@ -218,7 +219,7 @@ const char* abstract_broker::name() const {
void abstract_broker::init_broker() {
CAF_LOG_TRACE("");
is_initialized(true);
setf(is_initialized_flag);
// launch backends now, because user-defined initialization
// might call functions like add_connection
for (auto& kvp : doormen_)
......
......@@ -96,7 +96,7 @@ strong_actor_ptr basp_broker_state::make_proxy(node_id nid, actor_id aid) {
// until the original instance terminates, thus preventing subtle
// bugs with attachables
auto bptr = static_cast<basp_broker*>(selfptr->get());
if (!bptr->is_terminated())
if (!bptr->getf(abstract_actor::is_terminated_flag))
bptr->state.proxies().erase(nid, res->id(), rsn);
});
});
......@@ -192,7 +192,7 @@ void basp_broker_state::proxy_announced(const node_id& nid, actor_id aid) {
auto bptr = static_cast<basp_broker*>(tmp->get());
// ... to make sure this is safe
if (bptr == mm->named_broker<basp_broker>(atom("BASP"))
&& !bptr->is_terminated())
&& !bptr->getf(abstract_actor::is_terminated_flag))
send_kill_proxy_instance(fail_state);
});
});
......
......@@ -288,9 +288,9 @@ void middleman::stop() {
for (auto& kvp : named_brokers_) {
auto& hdl = kvp.second;
auto ptr = static_cast<broker*>(actor_cast<abstract_actor*>(hdl));
if (!ptr->is_terminated()) {
if (!ptr->getf(abstract_actor::is_terminated_flag)) {
ptr->context(&backend());
ptr->is_terminated(true);
ptr->setf(abstract_actor::is_terminated_flag);
ptr->finalize();
}
}
......
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