Commit 406f32df authored by Dominik Charousset's avatar Dominik Charousset

Remove attachable-related code from abstract_actor

parent 34908176
...@@ -55,6 +55,7 @@ set (LIBCAF_CORE_SRCS ...@@ -55,6 +55,7 @@ set (LIBCAF_CORE_SRCS
src/group.cpp src/group.cpp
src/group_manager.cpp src/group_manager.cpp
src/match_case.cpp src/match_case.cpp
src/monitorable_actor.cpp
src/local_actor.cpp src/local_actor.cpp
src/logger.cpp src/logger.cpp
src/mailbox_element.cpp src/mailbox_element.cpp
......
...@@ -59,7 +59,7 @@ class abstract_actor : public abstract_channel { ...@@ -59,7 +59,7 @@ class abstract_actor : public abstract_channel {
public: public:
/// Attaches `ptr` to this actor. The actor will call `ptr->detach(...)` on /// Attaches `ptr` to this actor. The actor will call `ptr->detach(...)` on
/// exit, or immediately if it already finished execution. /// exit, or immediately if it already finished execution.
void attach(attachable_ptr ptr); virtual void attach(attachable_ptr ptr) = 0;
/// Convenience function that attaches the functor `f` to this actor. The /// Convenience function that attaches the functor `f` to this actor. The
/// actor executes `f()` on exit or immediatley if it is not running. /// actor executes `f()` on exit or immediatley if it is not running.
...@@ -72,7 +72,7 @@ public: ...@@ -72,7 +72,7 @@ public:
actor_addr address() const; actor_addr address() const;
/// Detaches the first attached object that matches `what`. /// Detaches the first attached object that matches `what`.
size_t detach(const attachable::token& what); virtual size_t detach(const attachable::token& what) = 0;
/// Links this actor to `whom`. /// Links this actor to `whom`.
inline void link_to(const actor_addr& whom) { inline void link_to(const actor_addr& whom) {
...@@ -117,44 +117,19 @@ public: ...@@ -117,44 +117,19 @@ public:
/// an empty set if this actor is untyped. /// an empty set if this actor is untyped.
virtual std::set<std::string> message_types() const; virtual std::set<std::string> message_types() const;
/// Returns the execution unit currently used by this actor. /// Returns the system that created this actor (or proxy).
/// @warning not thread safe actor_system& home_system() {
inline execution_unit* context() const { CAF_ASSERT(home_system_ != nullptr);
return context_; return *home_system_;
}
/// Sets the execution unit for this actor.
inline void context(execution_unit* x) {
context_ = x;
}
/// Returns the hosting actor system.
inline actor_system& system() const {
CAF_ASSERT(context_);
return context_->system();
}
/// @cond PRIVATE
// Returns `exit_reason_ != exit_reason::not_exited`.
inline bool exited() const {
return exit_reason_ != exit_reason::not_exited;
} }
/// @endcond
protected: protected:
/// Creates a new actor instance.
abstract_actor(execution_unit* ptr, int flags);
/// Creates a new actor instance. /// Creates a new actor instance.
explicit abstract_actor(actor_config& cfg); explicit abstract_actor(actor_config& cfg);
/// Creates a new actor instance. /// Creates a new actor instance.
abstract_actor(actor_id aid, node_id nid); abstract_actor(actor_id aid, node_id nid);
/// Called by the runtime system to perform cleanup actions for this actor.
/// Subtypes should always call this member function when overriding it.
virtual void cleanup(exit_reason reason);
/**************************************************************************** /****************************************************************************
* here be dragons: end of public interface * * here be dragons: end of public interface *
****************************************************************************/ ****************************************************************************/
...@@ -162,11 +137,6 @@ protected: ...@@ -162,11 +137,6 @@ protected:
public: public:
/// @cond PRIVATE /// @cond PRIVATE
actor_system& home_system() {
CAF_ASSERT(home_system_ != nullptr);
return *home_system_;
}
enum linking_operation { enum linking_operation {
establish_link_op, establish_link_op,
establish_backlink_op, establish_backlink_op,
...@@ -256,52 +226,16 @@ public: ...@@ -256,52 +226,16 @@ public:
void is_registered(bool value); void is_registered(bool value);
// Tries to run a custom exception handler for `eptr`.
maybe<exit_reason> handle(const std::exception_ptr& eptr);
protected: protected:
virtual bool link_impl(linking_operation op, const actor_addr& other); virtual bool link_impl(linking_operation op, const actor_addr& other) = 0;
bool establish_link_impl(const actor_addr& other);
bool remove_link_impl(const actor_addr& other);
bool establish_backlink_impl(const actor_addr& other);
bool remove_backlink_impl(const actor_addr& other);
inline void attach_impl(attachable_ptr& ptr) {
ptr->next.swap(attachables_head_);
attachables_head_.swap(ptr);
}
static size_t detach_impl(const attachable::token& what,
attachable_ptr& ptr,
bool stop_on_first_hit = false,
bool dry_run = false);
// identifies the execution unit this actor is currently executed by
execution_unit* context_;
// cannot be changed after construction // cannot be changed after construction
const actor_id id_; const actor_id id_;
// initially set to exit_reason::not_exited // points to the actor system that created this actor
std::atomic<exit_reason> exit_reason_;
// guards access to exit_reason_, attachables_, links_,
// and enqueue operations if actor is thread-mapped
mutable std::mutex mtx_;
// only used in blocking and thread-mapped actors
mutable std::condition_variable cv_;
// attached functors that are executed on cleanup (monitors, links, etc)
attachable_ptr attachables_head_;
actor_system* home_system_; actor_system* home_system_;
/// @endcond /// @endcond
}; };
std::string to_string(abstract_actor::linking_operation op); std::string to_string(abstract_actor::linking_operation op);
......
...@@ -25,9 +25,9 @@ ...@@ -25,9 +25,9 @@
#include "caf/locks.hpp" #include "caf/locks.hpp"
#include "caf/actor.hpp" #include "caf/actor.hpp"
#include "caf/abstract_actor.hpp"
#include "caf/execution_unit.hpp" #include "caf/execution_unit.hpp"
#include "caf/mailbox_element.hpp" #include "caf/mailbox_element.hpp"
#include "caf/monitorable_actor.hpp"
#include "caf/detail/split_join.hpp" #include "caf/detail/split_join.hpp"
#include "caf/detail/shared_spinlock.hpp" #include "caf/detail/shared_spinlock.hpp"
...@@ -54,7 +54,7 @@ namespace caf { ...@@ -54,7 +54,7 @@ namespace caf {
/// during the enqueue operation. Any user-defined policy thus has to dispatch /// during the enqueue operation. Any user-defined policy thus has to dispatch
/// messages with as little overhead as possible, because the dispatching /// messages with as little overhead as possible, because the dispatching
/// runs in the context of the sender. /// runs in the context of the sender.
class actor_pool : public abstract_actor { class actor_pool : public monitorable_actor {
public: public:
using uplock = upgrade_lock<detail::shared_spinlock>; using uplock = upgrade_lock<detail::shared_spinlock>;
using actor_vec = std::vector<actor>; using actor_vec = std::vector<actor>;
...@@ -103,20 +103,19 @@ public: ...@@ -103,20 +103,19 @@ public:
void enqueue(mailbox_element_ptr what, execution_unit* host) override; void enqueue(mailbox_element_ptr what, execution_unit* host) override;
actor_pool(execution_unit* host); actor_pool(actor_config& cfg);
private: private:
bool filter(upgrade_lock<detail::shared_spinlock>&, const actor_addr& sender, bool filter(upgrade_lock<detail::shared_spinlock>&, const actor_addr& sender,
message_id mid, const message& content, execution_unit* host); message_id mid, const message& content, execution_unit* host);
// call without workers_mtx_ held // call without workers_mtx_ held
void quit(); void quit(execution_unit* host);
detail::shared_spinlock workers_mtx_; detail::shared_spinlock workers_mtx_;
std::vector<actor> workers_; std::vector<actor> workers_;
policy policy_; policy policy_;
exit_reason planned_reason_; exit_reason planned_reason_;
actor_system& system_;
}; };
} // namespace caf } // namespace caf
......
...@@ -24,6 +24,7 @@ ...@@ -24,6 +24,7 @@
#include <cstdint> #include <cstdint>
#include "caf/abstract_actor.hpp" #include "caf/abstract_actor.hpp"
#include "caf/monitorable_actor.hpp"
#include "caf/detail/shared_spinlock.hpp" #include "caf/detail/shared_spinlock.hpp"
...@@ -37,7 +38,7 @@ using actor_proxy_ptr = intrusive_ptr<actor_proxy>; ...@@ -37,7 +38,7 @@ using actor_proxy_ptr = intrusive_ptr<actor_proxy>;
/// Represents an actor running on a remote machine, /// Represents an actor running on a remote machine,
/// or different hardware, or in a separate process. /// or different hardware, or in a separate process.
class actor_proxy : public abstract_actor { class actor_proxy : public monitorable_actor {
public: public:
/// An anchor points to a proxy instance without sharing /// An anchor points to a proxy instance without sharing
/// ownership to it, i.e., models a weak ptr. /// ownership to it, i.e., models a weak ptr.
......
...@@ -75,7 +75,8 @@ public: ...@@ -75,7 +75,8 @@ public:
/// Executed if the actor finished execution with given `reason`. /// Executed if the actor finished execution with given `reason`.
/// The default implementation does nothing. /// The default implementation does nothing.
virtual void actor_exited(abstract_actor* self, exit_reason reason); virtual void actor_exited(abstract_actor* self, exit_reason reason,
execution_unit* host);
/// Returns `true` if `what` selects this instance, otherwise `false`. /// Returns `true` if `what` selects this instance, otherwise `false`.
virtual bool matches(const token& what); virtual bool matches(const token& what);
......
...@@ -38,7 +38,8 @@ public: ...@@ -38,7 +38,8 @@ public:
static constexpr size_t token_type = attachable::token::observer; static constexpr size_t token_type = attachable::token::observer;
}; };
void actor_exited(abstract_actor* self, exit_reason reason) override; void actor_exited(abstract_actor* self, exit_reason reason,
execution_unit* host) override;
bool matches(const token& what) override; bool matches(const token& what) override;
......
...@@ -36,7 +36,8 @@ struct functor_attachable : attachable { ...@@ -36,7 +36,8 @@ struct functor_attachable : attachable {
functor_attachable(F arg) : functor_(std::move(arg)) { functor_attachable(F arg) : functor_(std::move(arg)) {
// nop // nop
} }
void actor_exited(abstract_actor* self, exit_reason reason) override { void actor_exited(abstract_actor* self, exit_reason reason,
execution_unit*) override {
functor_(self, reason); functor_(self, reason);
} }
static constexpr size_t token_type = attachable::token::anonymous; static constexpr size_t token_type = attachable::token::anonymous;
...@@ -48,7 +49,8 @@ struct functor_attachable<F, 1> : attachable { ...@@ -48,7 +49,8 @@ struct functor_attachable<F, 1> : attachable {
functor_attachable(F arg) : functor_(std::move(arg)) { functor_attachable(F arg) : functor_(std::move(arg)) {
// nop // nop
} }
void actor_exited(abstract_actor*, exit_reason reason) override { void actor_exited(abstract_actor*, exit_reason reason,
execution_unit*) override {
functor_(reason); functor_(reason);
} }
}; };
...@@ -59,7 +61,7 @@ struct functor_attachable<F, 0> : attachable { ...@@ -59,7 +61,7 @@ struct functor_attachable<F, 0> : attachable {
functor_attachable(F arg) : functor_(std::move(arg)) { functor_attachable(F arg) : functor_(std::move(arg)) {
// nop // nop
} }
void actor_exited(abstract_actor*, exit_reason) override { void actor_exited(abstract_actor*, exit_reason, execution_unit*) override {
functor_(); functor_();
} }
}; };
......
...@@ -52,6 +52,7 @@ ...@@ -52,6 +52,7 @@
#include "caf/response_promise.hpp" #include "caf/response_promise.hpp"
#include "caf/message_priority.hpp" #include "caf/message_priority.hpp"
#include "caf/check_typed_input.hpp" #include "caf/check_typed_input.hpp"
#include "caf/monitorable_actor.hpp"
#include "caf/invoke_message_result.hpp" #include "caf/invoke_message_result.hpp"
#include "caf/detail/disposer.hpp" #include "caf/detail/disposer.hpp"
...@@ -64,7 +65,7 @@ namespace caf { ...@@ -64,7 +65,7 @@ namespace caf {
/// Base class for actors running on this node, either /// Base class for actors running on this node, either
/// living in an own thread or cooperatively scheduled. /// living in an own thread or cooperatively scheduled.
class local_actor : public abstract_actor, public resumable { class local_actor : public monitorable_actor, public resumable {
public: public:
using mailbox_type = detail::single_reader_queue<mailbox_element, using mailbox_type = detail::single_reader_queue<mailbox_element,
detail::disposer>; detail::disposer>;
...@@ -237,6 +238,23 @@ public: ...@@ -237,6 +238,23 @@ public:
* miscellaneous actor operations * * miscellaneous actor operations *
****************************************************************************/ ****************************************************************************/
/// Returns the execution unit currently used by this actor.
/// @warning not thread safe
inline execution_unit* context() const {
return context_;
}
/// Sets the execution unit for this actor.
inline void context(execution_unit* x) {
context_ = x;
}
/// Returns the hosting actor system.
inline actor_system& system() const {
CAF_ASSERT(context_);
return context_->system();
}
/// Causes this actor to subscribe to the group `what`. /// Causes this actor to subscribe to the group `what`.
/// The group will be unsubscribed if the actor finishes execution. /// The group will be unsubscribed if the actor finishes execution.
void join(const group& what); void join(const group& what);
...@@ -505,7 +523,7 @@ public: ...@@ -505,7 +523,7 @@ public:
// valid behavior left or has set a planned exit reason // valid behavior left or has set a planned exit reason
bool finished(); bool finished();
void cleanup(exit_reason reason) override; void cleanup(exit_reason reason, execution_unit* host) override;
// an actor can have multiple pending timeouts, but only // an actor can have multiple pending timeouts, but only
// the latest one is active (i.e. the pending_timeouts_.back()) // the latest one is active (i.e. the pending_timeouts_.back())
...@@ -572,6 +590,9 @@ protected: ...@@ -572,6 +590,9 @@ protected:
// used only in thread-mapped actors // used only in thread-mapped actors
void await_data(); void await_data();
// identifies the execution unit this actor is currently executed by
execution_unit* context_;
// identifies the ID of the last sent synchronous request // identifies the ID of the last sent synchronous request
message_id last_request_id_; message_id last_request_id_;
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_MONITORABLE_ACTOR_HPP
#define CAF_MONITORABLE_ACTOR_HPP
#include <set>
#include <mutex>
#include <atomic>
#include <memory>
#include <string>
#include <vector>
#include <cstdint>
#include <exception>
#include <type_traits>
#include <condition_variable>
#include "caf/abstract_actor.hpp"
#include "caf/detail/type_traits.hpp"
#include "caf/detail/functor_attachable.hpp"
namespace caf {
/// Base class for all actor implementations.
class monitorable_actor : public abstract_actor {
public:
void attach(attachable_ptr ptr) override;
size_t detach(const attachable::token& what) override;
/// @cond PRIVATE
// Returns `exit_reason_ != exit_reason::not_exited`.
inline bool exited() const {
return exit_reason_ != exit_reason::not_exited;
}
/// @endcond
protected:
/// Creates a new actor instance.
explicit monitorable_actor(actor_config& cfg);
/// Creates a new actor instance.
monitorable_actor(actor_id aid, node_id nid);
/// Called by the runtime system to perform cleanup actions for this actor.
/// Subtypes should always call this member function when overriding it.
virtual void cleanup(exit_reason reason, execution_unit* host);
/****************************************************************************
* here be dragons: end of public interface *
****************************************************************************/
bool link_impl(linking_operation op, const actor_addr& other) override;
bool establish_link_impl(const actor_addr& other);
bool remove_link_impl(const actor_addr& other);
bool establish_backlink_impl(const actor_addr& other);
bool remove_backlink_impl(const actor_addr& other);
// Tries to run a custom exception handler for `eptr`.
maybe<exit_reason> handle(const std::exception_ptr& eptr);
inline void attach_impl(attachable_ptr& ptr) {
ptr->next.swap(attachables_head_);
attachables_head_.swap(ptr);
}
static size_t detach_impl(const attachable::token& what,
attachable_ptr& ptr,
bool stop_on_first_hit = false,
bool dry_run = false);
// initially set to exit_reason::not_exited
std::atomic<exit_reason> exit_reason_;
// guards access to exit_reason_, attachables_, links_,
// and enqueue operations if actor is thread-mapped
mutable std::mutex mtx_;
// only used in blocking and thread-mapped actors
mutable std::condition_variable cv_;
// attached functors that are executed on cleanup (monitors, links, etc)
attachable_ptr attachables_head_;
/// @endcond
};
std::string to_string(abstract_actor::linking_operation op);
} // namespace caf
#endif // CAF_MONITORABLE_ACTOR_HPP
...@@ -38,7 +38,7 @@ public: ...@@ -38,7 +38,7 @@ public:
response_promise& operator=(response_promise&&) = default; response_promise& operator=(response_promise&&) = default;
response_promise& operator=(const response_promise&) = default; response_promise& operator=(const response_promise&) = default;
response_promise(const actor_addr& from, const actor_addr& to, response_promise(local_actor* self, const actor_addr& to,
const message_id& response_id); const message_id& response_id);
/// Queries whether this promise is still valid, i.e., no response /// Queries whether this promise is still valid, i.e., no response
...@@ -62,7 +62,7 @@ public: ...@@ -62,7 +62,7 @@ public:
private: private:
void deliver_impl(message response_message) const; void deliver_impl(message response_message) const;
actor_addr from_; local_actor* self_;
actor_addr to_; actor_addr to_;
message_id id_; message_id id_;
}; };
......
...@@ -41,194 +41,28 @@ ...@@ -41,194 +41,28 @@
namespace caf { namespace caf {
namespace {
using guard_type = std::unique_lock<std::mutex>;
} // namespace <anonymous>
// exit_reason_ is guaranteed to be set to 0, i.e., exit_reason::not_exited, // exit_reason_ is guaranteed to be set to 0, i.e., exit_reason::not_exited,
// by std::atomic<> constructor // by std::atomic<> constructor
abstract_actor::abstract_actor(execution_unit* ptr, int flags)
: abstract_channel(flags | abstract_channel::is_abstract_actor_flag,
ptr->system().node()),
context_(ptr),
id_(ptr->system().next_actor_id()),
exit_reason_(exit_reason::not_exited),
home_system_(&ptr->system()) {
// nop
}
abstract_actor::abstract_actor(actor_config& cfg) abstract_actor::abstract_actor(actor_config& cfg)
: abstract_actor(cfg.host, cfg.flags) { : abstract_channel(cfg.flags | abstract_channel::is_abstract_actor_flag,
cfg.host->system().node()),
id_(cfg.host->system().next_actor_id()),
home_system_(&cfg.host->system()) {
// nop // nop
} }
abstract_actor::abstract_actor(actor_id aid, node_id nid) abstract_actor::abstract_actor(actor_id aid, node_id nid)
: abstract_channel(abstract_channel::is_abstract_actor_flag, : abstract_channel(abstract_channel::is_abstract_actor_flag,
std::move(nid)), std::move(nid)),
context_(nullptr), id_(aid) {
id_(aid),
exit_reason_(exit_reason::not_exited) {
// nop // nop
} }
void abstract_actor::attach(attachable_ptr ptr) {
CAF_LOG_TRACE("");
if (ptr == nullptr) {
return;
}
caf::exit_reason reason;
{ // lifetime scope of guard
guard_type guard{mtx_};
reason = exit_reason_;
if (reason == exit_reason::not_exited) {
attach_impl(ptr);
return;
}
}
CAF_LOG_DEBUG("cannot attach functor to terminated actor: call immediately");
ptr->actor_exited(this, reason);
}
size_t abstract_actor::detach_impl(const attachable::token& what,
attachable_ptr& ptr,
bool stop_on_hit,
bool dry_run) {
CAF_LOG_TRACE("");
if (! ptr) {
CAF_LOG_DEBUG("invalid ptr");
return 0;
}
if (ptr->matches(what)) {
if (! dry_run) {
CAF_LOG_DEBUG("removed element");
attachable_ptr next;
next.swap(ptr->next);
ptr.swap(next);
}
return stop_on_hit ? 1 : 1 + detach_impl(what, ptr, stop_on_hit, dry_run);
}
return detach_impl(what, ptr->next, stop_on_hit, dry_run);
}
size_t abstract_actor::detach(const attachable::token& what) {
CAF_LOG_TRACE("");
guard_type guard{mtx_};
return detach_impl(what, attachables_head_);
}
bool abstract_actor::link_impl(linking_operation op, const actor_addr& other) {
CAF_LOG_TRACE(CAF_ARG(op) << CAF_ARG(other));
switch (op) {
case establish_link_op:
return establish_link_impl(other);
case establish_backlink_op:
return establish_backlink_impl(other);
case remove_link_op:
return remove_link_impl(other);
default:
CAF_ASSERT(op == remove_backlink_op);
return remove_backlink_impl(other);
}
}
bool abstract_actor::establish_link_impl(const actor_addr& other) {
CAF_LOG_TRACE(CAF_ARG(other));
if (other && other != this) {
guard_type guard{mtx_};
auto ptr = actor_cast<abstract_actor_ptr>(other);
// send exit message if already exited
if (exited()) {
ptr->enqueue(address(), invalid_message_id,
make_message(exit_msg{address(), exit_reason_}), context_);
} 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);
attach_impl(tmp);
return true;
}
}
return false;
}
bool abstract_actor::establish_backlink_impl(const actor_addr& other) {
CAF_LOG_TRACE(CAF_ARG(other));
auto reason = exit_reason::not_exited;
default_attachable::observe_token tk{other, default_attachable::link};
if (other && other != this) {
guard_type guard{mtx_};
reason = exit_reason_;
if (reason == exit_reason::not_exited) {
if (detach_impl(tk, attachables_head_, true, true) == 0) {
auto tmp = default_attachable::make_link(other);
attach_impl(tmp);
return true;
}
}
}
// send exit message without lock
if (reason != exit_reason::not_exited) {
auto ptr = actor_cast<abstract_actor_ptr>(other);
ptr->enqueue(address(), invalid_message_id,
make_message(exit_msg{address(), exit_reason_}), context_);
}
return false;
}
bool abstract_actor::remove_link_impl(const actor_addr& other) {
CAF_LOG_TRACE(CAF_ARG(other));
if (other == invalid_actor_addr || other == this)
return false;
default_attachable::observe_token tk{other, default_attachable::link};
guard_type guard{mtx_};
// remove_backlink returns true if this actor is linked to other
auto ptr = actor_cast<abstract_actor_ptr>(other);
if (detach_impl(tk, attachables_head_, true) > 0) {
// tell remote side to remove link as well
ptr->remove_backlink(address());
return true;
}
return false;
}
bool abstract_actor::remove_backlink_impl(const actor_addr& other) {
CAF_LOG_TRACE(CAF_ARG(other));
default_attachable::observe_token tk{other, default_attachable::link};
if (other && other != this) {
guard_type guard{mtx_};
return detach_impl(tk, attachables_head_, true) > 0;
}
return false;
}
actor_addr abstract_actor::address() const { actor_addr abstract_actor::address() const {
return actor_addr{const_cast<abstract_actor*>(this)}; return actor_addr{const_cast<abstract_actor*>(this)};
} }
void abstract_actor::cleanup(exit_reason reason) {
CAF_LOG_TRACE(CAF_ARG(reason));
CAF_ASSERT(reason != exit_reason::not_exited);
// move everyhting out of the critical section before processing it
attachable_ptr head;
{ // lifetime scope of guard
guard_type guard{mtx_};
if (exit_reason_ != exit_reason::not_exited) {
// already exited
return;
}
exit_reason_ = reason;
attachables_head_.swap(head);
}
CAF_LOG_INFO_IF(node() == system().node(),
"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);
}
std::set<std::string> abstract_actor::message_types() const { std::set<std::string> abstract_actor::message_types() const {
// defaults to untyped // defaults to untyped
return std::set<std::string>{}; return std::set<std::string>{};
...@@ -244,21 +78,6 @@ void abstract_actor::is_registered(bool value) { ...@@ -244,21 +78,6 @@ void abstract_actor::is_registered(bool value) {
set_flag(value, is_registered_flag); set_flag(value, is_registered_flag);
} }
maybe<exit_reason> abstract_actor::handle(const std::exception_ptr& eptr) {
guard_type guard{mtx_};
for (auto i = attachables_head_.get(); i != nullptr; i = i->next.get()) {
try {
auto result = i->handle_exception(eptr);
if (result)
return *result;
}
catch (...) {
// ignore exceptions
}
}
return none;
}
std::string to_string(abstract_actor::linking_operation op) { std::string to_string(abstract_actor::linking_operation op) {
switch (op) { switch (op) {
case abstract_actor::establish_link_op: case abstract_actor::establish_link_op:
......
...@@ -28,7 +28,7 @@ void actor_companion::disconnect(exit_reason rsn) { ...@@ -28,7 +28,7 @@ void actor_companion::disconnect(exit_reason rsn) {
std::lock_guard<lock_type> guard(lock_); std::lock_guard<lock_type> guard(lock_);
on_enqueue_.swap(tmp); on_enqueue_.swap(tmp);
} }
cleanup(rsn); cleanup(rsn, context());
} }
void actor_companion::on_enqueue(enqueue_handler handler) { void actor_companion::on_enqueue(enqueue_handler handler) {
......
...@@ -29,7 +29,7 @@ namespace caf { ...@@ -29,7 +29,7 @@ namespace caf {
actor_ostream::actor_ostream(abstract_actor* self) actor_ostream::actor_ostream(abstract_actor* self)
: self_(self), : self_(self),
printer_(self->system().scheduler().printer()) { printer_(self->home_system().scheduler().printer()) {
// nop // nop
} }
...@@ -47,7 +47,7 @@ void actor_ostream::redirect(abstract_actor* self, std::string fn, int flags) { ...@@ -47,7 +47,7 @@ void actor_ostream::redirect(abstract_actor* self, std::string fn, int flags) {
if (! self) if (! self)
return; return;
intrusive_ptr<abstract_actor> ptr{self}; intrusive_ptr<abstract_actor> ptr{self};
send_as(actor_cast<actor>(ptr), self->system().scheduler().printer(), send_as(actor_cast<actor>(ptr), self->home_system().scheduler().printer(),
redirect_atom::value, self->address(), std::move(fn), flags); redirect_atom::value, self->address(), std::move(fn), flags);
} }
......
...@@ -96,7 +96,8 @@ actor_pool::~actor_pool() { ...@@ -96,7 +96,8 @@ actor_pool::~actor_pool() {
actor actor_pool::make(execution_unit* eu, policy pol) { actor actor_pool::make(execution_unit* eu, policy pol) {
CAF_ASSERT(eu); CAF_ASSERT(eu);
intrusive_ptr<actor_pool> ptr; intrusive_ptr<actor_pool> ptr;
ptr = make_counted<actor_pool>(eu); actor_config cfg{eu};
ptr = make_counted<actor_pool>(cfg);
ptr->policy_ = std::move(pol); ptr->policy_ = std::move(pol);
return actor_cast<actor>(ptr); return actor_cast<actor>(ptr);
} }
...@@ -117,25 +118,22 @@ actor actor_pool::make(execution_unit* eu, size_t num_workers, ...@@ -117,25 +118,22 @@ actor actor_pool::make(execution_unit* eu, size_t num_workers,
void actor_pool::enqueue(const actor_addr& sender, message_id mid, void actor_pool::enqueue(const actor_addr& sender, message_id mid,
message content, execution_unit* eu) { message content, execution_unit* eu) {
upgrade_lock<detail::shared_spinlock> guard{workers_mtx_}; upgrade_lock<detail::shared_spinlock> guard{workers_mtx_};
if (filter(guard, sender, mid, content, eu)) { if (filter(guard, sender, mid, content, eu))
return; return;
}
auto ptr = mailbox_element::make(sender, mid, std::move(content)); auto ptr = mailbox_element::make(sender, mid, std::move(content));
policy_(system_, guard, workers_, ptr, eu); policy_(home_system(), guard, workers_, ptr, eu);
} }
void actor_pool::enqueue(mailbox_element_ptr what, execution_unit* eu) { void actor_pool::enqueue(mailbox_element_ptr what, execution_unit* eu) {
upgrade_lock<detail::shared_spinlock> guard{workers_mtx_}; upgrade_lock<detail::shared_spinlock> guard{workers_mtx_};
if (filter(guard, what->sender, what->mid, what->msg, eu)) { if (filter(guard, what->sender, what->mid, what->msg, eu))
return; return;
} policy_(home_system(), guard, workers_, what, eu);
policy_(system_, guard, workers_, what, eu);
} }
actor_pool::actor_pool(execution_unit* host) actor_pool::actor_pool(actor_config& cfg)
: abstract_actor(host, 0), : monitorable_actor(cfg),
planned_reason_(exit_reason::not_exited), planned_reason_(exit_reason::not_exited) {
system_(host->system()) {
is_registered(true); is_registered(true);
} }
...@@ -162,7 +160,7 @@ bool actor_pool::filter(upgrade_lock<detail::shared_spinlock>& guard, ...@@ -162,7 +160,7 @@ bool actor_pool::filter(upgrade_lock<detail::shared_spinlock>& guard,
for (auto& w : workers) { for (auto& w : workers) {
anon_send(w, msg); anon_send(w, msg);
} }
quit(); quit(eu);
return true; return true;
} }
if (msg.match_elements<down_msg>()) { if (msg.match_elements<down_msg>()) {
...@@ -177,7 +175,7 @@ bool actor_pool::filter(upgrade_lock<detail::shared_spinlock>& guard, ...@@ -177,7 +175,7 @@ bool actor_pool::filter(upgrade_lock<detail::shared_spinlock>& guard,
if (workers_.empty()) { if (workers_.empty()) {
planned_reason_ = exit_reason::out_of_workers; planned_reason_ = exit_reason::out_of_workers;
unique_guard.unlock(); unique_guard.unlock();
quit(); quit(eu);
} }
return true; return true;
} }
...@@ -223,10 +221,10 @@ bool actor_pool::filter(upgrade_lock<detail::shared_spinlock>& guard, ...@@ -223,10 +221,10 @@ bool actor_pool::filter(upgrade_lock<detail::shared_spinlock>& guard,
return false; return false;
} }
void actor_pool::quit() { void actor_pool::quit(execution_unit* host) {
// we can safely run our cleanup code here without holding // we can safely run our cleanup code here without holding
// workers_mtx_ because abstract_actor has its own lock // workers_mtx_ because abstract_actor has its own lock
cleanup(planned_reason_); cleanup(planned_reason_, host);
is_registered(false); is_registered(false);
} }
......
...@@ -67,7 +67,7 @@ actor_proxy::~actor_proxy() { ...@@ -67,7 +67,7 @@ actor_proxy::~actor_proxy() {
} }
actor_proxy::actor_proxy(actor_id aid, node_id nid) actor_proxy::actor_proxy(actor_id aid, node_id nid)
: abstract_actor(aid, nid), : monitorable_actor(aid, nid),
anchor_(make_counted<anchor>(this)) { anchor_(make_counted<anchor>(this)) {
// nop // nop
} }
......
...@@ -34,7 +34,7 @@ maybe<exit_reason> attachable::handle_exception(const std::exception_ptr&) { ...@@ -34,7 +34,7 @@ maybe<exit_reason> attachable::handle_exception(const std::exception_ptr&) {
return none; return none;
} }
void attachable::actor_exited(abstract_actor*, exit_reason) { void attachable::actor_exited(abstract_actor*, exit_reason, execution_unit*) {
// nop // nop
} }
......
...@@ -34,12 +34,13 @@ message make(abstract_actor* self, exit_reason reason) { ...@@ -34,12 +34,13 @@ message make(abstract_actor* self, exit_reason reason) {
} // namespace <anonymous> } // namespace <anonymous>
void default_attachable::actor_exited(abstract_actor* self, exit_reason rsn) { void default_attachable::actor_exited(abstract_actor* self, exit_reason rsn,
execution_unit* host) {
CAF_ASSERT(self->address() != observer_); CAF_ASSERT(self->address() != observer_);
auto factory = type_ == monitor ? &make<down_msg> : &make<exit_msg>; auto factory = type_ == monitor ? &make<down_msg> : &make<exit_msg>;
auto ptr = actor_cast<abstract_actor_ptr>(observer_); auto ptr = actor_cast<abstract_actor_ptr>(observer_);
ptr->enqueue(self->address(), message_id{}.with_high_priority(), ptr->enqueue(self->address(), message_id{}.with_high_priority(),
factory(self, rsn), self->context()); factory(self, rsn), host);
} }
bool default_attachable::matches(const token& what) { bool default_attachable::matches(const token& what) {
......
...@@ -120,9 +120,8 @@ void forwarding_actor_proxy::local_unlink_from(const actor_addr& other) { ...@@ -120,9 +120,8 @@ void forwarding_actor_proxy::local_unlink_from(const actor_addr& other) {
void forwarding_actor_proxy::kill_proxy(execution_unit* ctx, exit_reason rsn) { void forwarding_actor_proxy::kill_proxy(execution_unit* ctx, exit_reason rsn) {
CAF_ASSERT(ctx != nullptr); CAF_ASSERT(ctx != nullptr);
context(ctx);
manager(invalid_actor); manager(invalid_actor);
cleanup(rsn); cleanup(rsn, ctx);
} }
} // namespace caf } // namespace caf
...@@ -42,7 +42,8 @@ namespace caf { ...@@ -42,7 +42,8 @@ namespace caf {
// later on in spawn(); this prevents subtle bugs that lead to segfaults, // later on in spawn(); this prevents subtle bugs that lead to segfaults,
// e.g., when calling address() in the ctor of a derived class // e.g., when calling address() in the ctor of a derived class
local_actor::local_actor(actor_config& cfg) local_actor::local_actor(actor_config& cfg)
: abstract_actor(cfg), : monitorable_actor(cfg),
context_(cfg.host),
planned_exit_reason_(exit_reason::not_exited), planned_exit_reason_(exit_reason::not_exited),
timeout_id_(0), timeout_id_(0),
initial_behavior_fac_(std::move(cfg.init_fun)) { initial_behavior_fac_(std::move(cfg.init_fun)) {
...@@ -656,7 +657,7 @@ resumable::resume_result local_actor::resume(execution_unit* eu, ...@@ -656,7 +657,7 @@ resumable::resume_result local_actor::resume(execution_unit* eu,
// simply ignore exception // simply ignore exception
} }
// exit reason might have been changed by on_exit() // exit reason might have been changed by on_exit()
self->cleanup(self->planned_exit_reason()); self->cleanup(self->planned_exit_reason(), context());
return resumable::done; return resumable::done;
} }
if (is_initialized() if (is_initialized()
...@@ -921,7 +922,7 @@ response_promise local_actor::make_response_promise() { ...@@ -921,7 +922,7 @@ response_promise local_actor::make_response_promise() {
auto& mid = ptr->mid; auto& mid = ptr->mid;
if (mid.is_answered()) if (mid.is_answered())
return response_promise{}; return response_promise{};
response_promise result{address(), ptr->sender, mid.response_id()}; response_promise result{this, ptr->sender, mid.response_id()};
mid.mark_as_answered(); mid.mark_as_answered();
return result; return result;
} }
...@@ -955,11 +956,11 @@ bool local_actor::finished() { ...@@ -955,11 +956,11 @@ bool local_actor::finished() {
rsn = exit_reason::normal; rsn = exit_reason::normal;
planned_exit_reason(rsn); planned_exit_reason(rsn);
} }
cleanup(rsn); cleanup(rsn, context());
return true; return true;
} }
void local_actor::cleanup(exit_reason reason) { void local_actor::cleanup(exit_reason reason, execution_unit* host) {
CAF_LOG_TRACE(CAF_ARG(reason)); CAF_LOG_TRACE(CAF_ARG(reason));
current_mailbox_element().reset(); current_mailbox_element().reset();
detail::sync_request_bouncer f{reason}; detail::sync_request_bouncer f{reason};
...@@ -971,7 +972,7 @@ void local_actor::cleanup(exit_reason reason) { ...@@ -971,7 +972,7 @@ void local_actor::cleanup(exit_reason reason) {
subscription->unsubscribe(me); subscription->unsubscribe(me);
subscriptions_.clear(); subscriptions_.clear();
} }
abstract_actor::cleanup(reason); monitorable_actor::cleanup(reason, host);
// tell registry we're done // tell registry we're done
is_registered(false); is_registered(false);
} }
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/monitorable_actor.hpp"
#include "caf/logger.hpp"
#include "caf/system_messages.hpp"
#include "caf/default_attachable.hpp"
namespace caf {
void monitorable_actor::attach(attachable_ptr ptr) {
CAF_LOG_TRACE("");
if (ptr == nullptr)
return;
caf::exit_reason reason;
{ // lifetime scope of guard
std::unique_lock<std::mutex> guard{mtx_};
reason = exit_reason_;
if (reason == exit_reason::not_exited) {
attach_impl(ptr);
return;
}
}
CAF_LOG_DEBUG("cannot attach functor to terminated actor: call immediately");
ptr->actor_exited(this, reason, nullptr);
}
size_t monitorable_actor::detach(const attachable::token& what) {
CAF_LOG_TRACE("");
std::unique_lock<std::mutex> guard{mtx_};
return detach_impl(what, attachables_head_);
}
monitorable_actor::monitorable_actor(actor_config& cfg)
: abstract_actor(cfg),
exit_reason_(exit_reason::not_exited) {
// nop
}
/// Creates a new actor instance.
monitorable_actor::monitorable_actor(actor_id aid, node_id nid)
: abstract_actor(aid, nid),
exit_reason_(exit_reason::not_exited) {
// nop
}
void monitorable_actor::cleanup(exit_reason reason, execution_unit* host) {
CAF_LOG_TRACE(CAF_ARG(reason));
CAF_ASSERT(reason != exit_reason::not_exited);
// move everyhting out of the critical section before processing it
attachable_ptr head;
{ // lifetime scope of guard
std::unique_lock<std::mutex> guard{mtx_};
if (exit_reason_ != exit_reason::not_exited)
return;
exit_reason_ = reason;
attachables_head_.swap(head);
}
CAF_LOG_INFO_IF(node() == system().node(),
"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);
}
maybe<exit_reason> monitorable_actor::handle(const std::exception_ptr& eptr) {
std::unique_lock<std::mutex> guard{mtx_};
for (auto i = attachables_head_.get(); i != nullptr; i = i->next.get()) {
try {
auto result = i->handle_exception(eptr);
if (result)
return *result;
}
catch (...) {
// ignore exceptions
}
}
return none;
}
bool monitorable_actor::link_impl(linking_operation op, const actor_addr& other) {
CAF_LOG_TRACE(CAF_ARG(op) << CAF_ARG(other));
switch (op) {
case establish_link_op:
return establish_link_impl(other);
case establish_backlink_op:
return establish_backlink_impl(other);
case remove_link_op:
return remove_link_impl(other);
default:
CAF_ASSERT(op == remove_backlink_op);
return remove_backlink_impl(other);
}
}
bool monitorable_actor::establish_link_impl(const actor_addr& other) {
CAF_LOG_TRACE(CAF_ARG(other));
if (other && other != this) {
std::unique_lock<std::mutex> guard{mtx_};
auto ptr = actor_cast<abstract_actor_ptr>(other);
// send exit message if already exited
if (exited()) {
ptr->enqueue(address(), invalid_message_id,
make_message(exit_msg{address(), exit_reason_}), nullptr);
} 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);
attach_impl(tmp);
return true;
}
}
return false;
}
bool monitorable_actor::establish_backlink_impl(const actor_addr& other) {
CAF_LOG_TRACE(CAF_ARG(other));
auto reason = exit_reason::not_exited;
default_attachable::observe_token tk{other, default_attachable::link};
if (other && other != this) {
std::unique_lock<std::mutex> guard{mtx_};
reason = exit_reason_;
if (reason == exit_reason::not_exited) {
if (detach_impl(tk, attachables_head_, true, true) == 0) {
auto tmp = default_attachable::make_link(other);
attach_impl(tmp);
return true;
}
}
}
// send exit message without lock
if (reason != exit_reason::not_exited) {
auto ptr = actor_cast<abstract_actor_ptr>(other);
ptr->enqueue(address(), invalid_message_id,
make_message(exit_msg{address(), exit_reason_}), nullptr);
}
return false;
}
bool monitorable_actor::remove_link_impl(const actor_addr& other) {
CAF_LOG_TRACE(CAF_ARG(other));
if (other == invalid_actor_addr || other == this)
return false;
default_attachable::observe_token tk{other, default_attachable::link};
std::unique_lock<std::mutex> guard{mtx_};
// remove_backlink returns true if this actor is linked to other
auto ptr = actor_cast<abstract_actor_ptr>(other);
if (detach_impl(tk, attachables_head_, true) > 0) {
// tell remote side to remove link as well
ptr->remove_backlink(address());
return true;
}
return false;
}
bool monitorable_actor::remove_backlink_impl(const actor_addr& other) {
CAF_LOG_TRACE(CAF_ARG(other));
default_attachable::observe_token tk{other, default_attachable::link};
if (other && other != this) {
std::unique_lock<std::mutex> guard{mtx_};
return detach_impl(tk, attachables_head_, true) > 0;
}
return false;
}
size_t monitorable_actor::detach_impl(const attachable::token& what,
attachable_ptr& ptr, bool stop_on_hit,
bool dry_run) {
CAF_LOG_TRACE("");
if (! ptr) {
CAF_LOG_DEBUG("invalid ptr");
return 0;
}
if (ptr->matches(what)) {
if (! dry_run) {
CAF_LOG_DEBUG("removed element");
attachable_ptr next;
next.swap(ptr->next);
ptr.swap(next);
}
return stop_on_hit ? 1 : 1 + detach_impl(what, ptr, stop_on_hit, dry_run);
}
return detach_impl(what, ptr->next, stop_on_hit, dry_run);
}
} // namespace caf
...@@ -24,18 +24,16 @@ ...@@ -24,18 +24,16 @@
namespace caf { namespace caf {
response_promise::response_promise(const actor_addr& from, const actor_addr& to, response_promise::response_promise(local_actor* self, const actor_addr& to,
const message_id& id) const message_id& id)
: from_(from), to_(to), id_(id) { : self_(self), to_(to), id_(id) {
CAF_ASSERT(id.is_response() || ! id.valid()); CAF_ASSERT(id.is_response() || ! id.valid());
} }
void response_promise::deliver_impl(message msg) const { void response_promise::deliver_impl(message msg) const {
if (! to_) if (! to_)
return; return;
auto to = actor_cast<abstract_actor_ptr>(to_); to_->enqueue(self_->address(), id_, std::move(msg), self_->context());
auto from = actor_cast<abstract_actor_ptr>(from_);
to->enqueue(from_, id_, std::move(msg), from->context());
} }
void response_promise::deliver(error x) const { void response_promise::deliver(error x) const {
......
...@@ -55,7 +55,8 @@ scoped_actor::~scoped_actor() { ...@@ -55,7 +55,8 @@ scoped_actor::~scoped_actor() {
CAF_SET_AID(prev_); CAF_SET_AID(prev_);
} }
auto r = self_->planned_exit_reason(); auto r = self_->planned_exit_reason();
self_->cleanup(r == exit_reason::not_exited ? exit_reason::normal : r); self_->cleanup(r == exit_reason::not_exited ? exit_reason::normal : r,
&context_);
} }
std::string to_string(const scoped_actor& x) { std::string to_string(const scoped_actor& x) {
......
...@@ -89,7 +89,7 @@ public: ...@@ -89,7 +89,7 @@ public:
void enqueue(mailbox_element_ptr, execution_unit*) override; void enqueue(mailbox_element_ptr, execution_unit*) override;
/// Called after this broker has finished execution. /// Called after this broker has finished execution.
void cleanup(exit_reason reason) override; void cleanup(exit_reason reason, execution_unit* host) override;
/// Starts running this broker in the `middleman`. /// Starts running this broker in the `middleman`.
void launch(execution_unit* eu, bool lazy, bool hide); void launch(execution_unit* eu, bool lazy, bool hide);
......
...@@ -80,13 +80,13 @@ void abstract_broker::launch(execution_unit* eu, bool is_lazy, bool is_hidden) { ...@@ -80,13 +80,13 @@ void abstract_broker::launch(execution_unit* eu, bool is_lazy, bool is_hidden) {
eu->exec_later(this); eu->exec_later(this);
} }
void abstract_broker::cleanup(exit_reason reason) { void abstract_broker::cleanup(exit_reason reason, execution_unit* host) {
CAF_LOG_TRACE(CAF_ARG(reason)); CAF_LOG_TRACE(CAF_ARG(reason));
close_all(); close_all();
CAF_ASSERT(doormen_.empty()); CAF_ASSERT(doormen_.empty());
CAF_ASSERT(scribes_.empty()); CAF_ASSERT(scribes_.empty());
cache_.clear(); cache_.clear();
local_actor::cleanup(reason); local_actor::cleanup(reason, host);
} }
abstract_broker::~abstract_broker() { abstract_broker::~abstract_broker() {
......
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