Commit 8656db86 authored by Dominik Charousset's avatar Dominik Charousset

Get rid of `abstract_actor::m_link` member

Links now use attachables, just like monitors do; relates #180.
parent 216d1fae
......@@ -35,6 +35,7 @@ set (LIBCAF_CORE_SRCS
src/channel.cpp
src/continue_helper.cpp
src/decorated_tuple.cpp
src/default_attachable.cpp
src/demangle.cpp
src/deserializer.cpp
src/duration.cpp
......
......@@ -79,7 +79,7 @@ class abstract_actor : public abstract_channel {
* @returns `true` if `ptr` was successfully attached to the actor,
* otherwise (actor already exited) `false`.
*/
bool attach(attachable_ptr ptr);
void attach(attachable_ptr ptr);
/**
* Convenience function that attaches the functor `f` to this actor. The
......@@ -88,20 +88,20 @@ class abstract_actor : public abstract_channel {
* otherwise (actor already exited) `false`.
*/
template <class F>
bool attach_functor(F f) {
void attach_functor(F f) {
struct functor_attachable : attachable {
F m_functor;
functor_attachable(F arg) : m_functor(std::move(arg)) {
// nop
}
void actor_exited(uint32_t reason) {
void actor_exited(abstract_actor*, uint32_t reason) {
m_functor(reason);
}
bool matches(const attachable::token&) {
return false;
}
};
return attach(attachable_ptr{new functor_attachable(std::move(f))});
attach(attachable_ptr{new functor_attachable(std::move(f))});
}
/**
......@@ -194,6 +194,14 @@ class abstract_actor : public abstract_channel {
return static_cast<bool>(m_flags & static_cast<int>(mask));
}
/**
* Returns the execution unit currently used by this actor.
* @warning not thread safe
*/
inline execution_unit* host() const {
return m_host;
}
protected:
/**
* Creates a non-proxy instance.
......@@ -228,13 +236,6 @@ class abstract_actor : public abstract_channel {
return exit_reason() != exit_reason::not_exited;
}
/**
* Returns the execution unit currently used by this actor.
*/
inline execution_unit* host() const {
return m_host;
}
/**
* Sets the execution unit for this actor.
*/
......@@ -251,10 +252,7 @@ class abstract_actor : public abstract_channel {
// guards access to m_exit_reason, m_attachables, and m_links
std::mutex m_mtx;
// links to other actors
std::vector<abstract_actor_ptr> m_links;
// attached functors that are executed on cleanup
// attached functors that are executed on cleanup (for monitors, links, etc)
std::vector<attachable_ptr> m_attachables;
// identifies the execution unit this actor is currently executed by
......
......@@ -26,6 +26,8 @@
namespace caf {
class abstract_actor;
/**
* Callback utility class.
*/
......@@ -58,12 +60,20 @@ class attachable {
* Executed if the actor finished execution with given `reason`.
* The default implementation does nothing.
*/
virtual void actor_exited(uint32_t reason) = 0;
virtual void actor_exited(abstract_actor* self, uint32_t reason) = 0;
/**
* Returns `true` if `what` selects this instance, otherwise false.
* Returns `true` if `what` selects this instance, otherwise `false`.
*/
virtual bool matches(const token& what) = 0;
/**
* Returns `true` if `what` selects this instance, otherwise `false`.
*/
template <class T>
bool matches(const T& what) {
return matches(token{typeid(T), &what});
}
};
/**
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2014 *
* 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 LICENCE_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_DEFAULT_ATTACHABLE_HPP
#define CAF_DEFAULT_ATTACHABLE_HPP
#include "caf/actor_addr.hpp"
#include "caf/attachable.hpp"
namespace caf {
class default_attachable : public attachable {
public:
enum observe_type {
monitor,
link
};
struct observe_token {
actor_addr observer;
observe_type type;
};
void actor_exited(abstract_actor* self, uint32_t reason) 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_link(actor_addr observer) {
return attachable_ptr{new default_attachable(std::move(observer), link)};
}
class predicate {
public:
inline predicate(actor_addr observer, observe_type type)
: m_observer(std::move(observer)),
m_type(type) {
// nop
}
inline bool operator()(const attachable_ptr& ptr) const {
return ptr->matches(observe_token{m_observer, m_type});
}
private:
actor_addr m_observer;
observe_type m_type;
};
private:
default_attachable(actor_addr observer, observe_type type);
actor_addr m_observer;
observe_type m_type;
};
} // namespace caf
#endif // CAF_DEFAULT_ATTACHABLE_HPP
......@@ -345,7 +345,9 @@ class local_actor : public extend<abstract_actor>::with<mixin::memory_cached> {
/**
* @copydoc monitor(const actor_addr&)
*/
inline void monitor(const actor& whom) { monitor(whom.address()); }
inline void monitor(const actor& whom) {
monitor(whom.address());
}
/**
* @copydoc monitor(const actor_addr&)
......@@ -363,7 +365,9 @@ class local_actor : public extend<abstract_actor>::with<mixin::memory_cached> {
/**
* Removes a monitor from `whom`.
*/
inline void demonitor(const actor& whom) { demonitor(whom.address()); }
inline void demonitor(const actor& whom) {
demonitor(whom.address());
}
/**
* Can be overridden to perform cleanup code after an actor
......
......@@ -31,6 +31,7 @@
#include "caf/actor_cast.hpp"
#include "caf/abstract_actor.hpp"
#include "caf/system_messages.hpp"
#include "caf/default_attachable.hpp"
#include "caf/detail/logging.hpp"
#include "caf/detail/singletons.hpp"
......@@ -41,7 +42,7 @@ namespace caf {
namespace {
using guard_type = std::unique_lock<std::mutex>;
}
} // namespace <anonymous>
// m_exit_reason is guaranteed to be set to 0, i.e., exit_reason::not_exited,
// by std::atomic<> constructor
......@@ -76,17 +77,17 @@ bool abstract_actor::link_to_impl(const actor_addr& other) {
// add link if not already linked to other
// (checked by establish_backlink)
else if (ptr->establish_backlink(address())) {
m_links.push_back(ptr);
m_attachables.push_back(default_attachable::make_link(other));
return true;
}
}
return false;
}
bool abstract_actor::attach(attachable_ptr ptr) {
void abstract_actor::attach(attachable_ptr ptr) {
CAF_LOG_TRACE("");
if (ptr == nullptr) {
guard_type guard{m_mtx};
return m_exit_reason == exit_reason::not_exited;
return;
}
uint32_t reason;
{ // lifetime scope of guard
......@@ -94,11 +95,12 @@ bool abstract_actor::attach(attachable_ptr ptr) {
reason = m_exit_reason;
if (reason == exit_reason::not_exited) {
m_attachables.push_back(std::move(ptr));
return true;
CAF_LOG_DEBUG("actor has now " << m_attachables.size() << " attachables");
return;
}
}
ptr->actor_exited(reason);
return false;
CAF_LOG_DEBUG("cannot attach functor to terminated actor: call immediately");
ptr->actor_exited(this, reason);
}
void abstract_actor::detach(const attachable::token& what) {
......@@ -109,7 +111,7 @@ void abstract_actor::detach(const attachable::token& what) {
auto i = std::find_if(m_attachables.begin(), end,
[&](attachable_ptr& p) { return p->matches(what); });
if (i != end) {
ptr = std::move(*i);
ptr.swap(*i);
m_attachables.erase(i);
}
}
......@@ -125,11 +127,12 @@ void abstract_actor::unlink_from(const actor_addr& other) {
}
bool abstract_actor::remove_backlink(const actor_addr& other) {
default_attachable::predicate pred(other, default_attachable::link);
if (other && other != this) {
guard_type guard{m_mtx};
auto i = std::find(m_links.begin(), m_links.end(), other);
if (i != m_links.end()) {
m_links.erase(i);
auto i = std::find_if(m_attachables.begin(), m_attachables.end(), pred);
if (i != m_attachables.end()) {
m_attachables.erase(i);
return true;
}
}
......@@ -138,13 +141,14 @@ bool abstract_actor::remove_backlink(const actor_addr& other) {
bool abstract_actor::establish_backlink(const actor_addr& other) {
uint32_t reason = exit_reason::not_exited;
default_attachable::predicate pred(other, default_attachable::link);
if (other && other != this) {
guard_type guard{m_mtx};
reason = m_exit_reason;
if (reason == exit_reason::not_exited) {
auto i = std::find(m_links.begin(), m_links.end(), other);
if (i == m_links.end()) {
m_links.push_back(actor_cast<abstract_actor_ptr>(other));
auto i = std::find_if(m_attachables.begin(), m_attachables.end(), pred);
if (i == m_attachables.end()) {
m_attachables.push_back(default_attachable::make_link(other));
return true;
}
}
......@@ -162,13 +166,14 @@ bool abstract_actor::unlink_from_impl(const actor_addr& other) {
if (!other) {
return false;
}
default_attachable::predicate pred(other, default_attachable::link);
guard_type guard{m_mtx};
// remove_backlink returns true if this actor is linked to other
auto ptr = actor_cast<abstract_actor_ptr>(other);
if (!exited() && ptr->remove_backlink(address())) {
auto i = std::find(m_links.begin(), m_links.end(), ptr);
CAF_REQUIRE(i != m_links.end());
m_links.erase(i);
auto i = std::find_if(m_attachables.begin(), m_attachables.end(), pred);
CAF_REQUIRE(i != m_attachables.end());
m_attachables.erase(i);
return true;
}
return false;
......@@ -183,7 +188,6 @@ void abstract_actor::cleanup(uint32_t reason) {
CAF_LOGM_TRACE("caf::actor", CAF_ARG(m_id) << ", " << CAF_ARG(reason));
CAF_REQUIRE(reason != exit_reason::not_exited);
// move everyhting out of the critical section before processing it
decltype(m_links) mlinks;
decltype(m_attachables) mattachables;
{ // lifetime scope of guard
guard_type guard{m_mtx};
......@@ -192,27 +196,17 @@ void abstract_actor::cleanup(uint32_t reason) {
return;
}
m_exit_reason = reason;
mlinks = std::move(m_links);
mattachables = std::move(m_attachables);
// make sure lists are empty
m_links.clear();
m_attachables.clear();
mattachables.swap(m_attachables);
}
CAF_LOG_INFO_IF(!is_remote(), "actor with ID "
<< m_id << " had " << mlinks.size()
<< " links and " << mattachables.size()
<< " attached functors; exit reason = "
<< m_id << " had " << mattachables.size()
<< " attachables; exit reason = "
<< reason << ", class = "
<< detail::demangle(typeid(*this)));
// send exit messages
auto msg = make_message(exit_msg{address(), reason});
CAF_LOGM_DEBUG("caf::actor", "send EXIT to " << mlinks.size() << " links");
for (auto& aptr : mlinks) {
aptr->enqueue(address(), message_id {}.with_high_priority(), msg, m_host);
}
CAF_LOGM_DEBUG("caf::actor", "run " << mattachables.size() << " attachables");
for (attachable_ptr& ptr : mattachables) {
ptr->actor_exited(reason);
ptr->actor_exited(this, reason);
}
}
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2014 *
* 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 LICENCE_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/default_attachable.hpp"
#include "caf/message.hpp"
#include "caf/system_messages.hpp"
namespace caf {
namespace {
template <class MsgType>
message make(abstract_actor* self, uint32_t reason) {
return make_message(MsgType{self->address(), reason});
}
} // namespace <anonymous>
void default_attachable::actor_exited(abstract_actor* self, uint32_t reason) {
CAF_REQUIRE(self->address() != m_observer);
auto factory = m_type == monitor ? &make<down_msg> : &make<exit_msg>;
message msg = factory(self, reason);
auto ptr = actor_cast<abstract_actor_ptr>(m_observer);
ptr->enqueue(self->address(), message_id{}.with_high_priority(),
msg, self->host());
}
bool default_attachable::matches(const token& what) {
if (what.subtype != typeid(observe_token)) {
return false;
}
auto& ot = *reinterpret_cast<const observe_token*>(what.ptr);
return ot.observer == m_observer && ot.type == m_type;
}
default_attachable::default_attachable(actor_addr observer, observe_type type)
: m_observer(std::move(observer)),
m_type(type) {
// nop
}
} // namespace caf
......@@ -23,42 +23,12 @@
#include "caf/scheduler.hpp"
#include "caf/actor_cast.hpp"
#include "caf/local_actor.hpp"
#include "caf/default_attachable.hpp"
#include "caf/detail/logging.hpp"
namespace caf {
namespace {
class down_observer : public attachable {
public:
down_observer(actor_addr observer, actor_addr observed)
: m_observer(std::move(observer)), m_observed(std::move(observed)) {
CAF_REQUIRE(m_observer != invalid_actor_addr);
CAF_REQUIRE(m_observed != invalid_actor_addr);
}
void actor_exited(uint32_t reason) {
auto ptr = actor_cast<abstract_actor_ptr>(m_observer);
ptr->enqueue(m_observed, message_id {}.with_high_priority(),
make_message(down_msg{m_observed, reason}), nullptr);
}
bool matches(const attachable::token& match_token) {
if (match_token.subtype == typeid(down_observer)) {
auto ptr = reinterpret_cast<const local_actor*>(match_token.ptr);
return m_observer == ptr->address();
}
return false;
}
private:
actor_addr m_observer;
actor_addr m_observed;
};
} // namespace <anonymous>
local_actor::local_actor()
: m_dummy_node(),
m_current_node(&m_dummy_node),
......@@ -71,20 +41,20 @@ local_actor::~local_actor() {
}
void local_actor::monitor(const actor_addr& whom) {
if (!whom) {
if (whom == invalid_actor_addr) {
return;
}
auto ptr = actor_cast<abstract_actor_ptr>(whom);
ptr->attach(attachable_ptr{new down_observer(address(), whom)});
ptr->attach(default_attachable::make_monitor(address()));
}
void local_actor::demonitor(const actor_addr& whom) {
if (!whom) {
if (whom == invalid_actor_addr) {
return;
}
auto ptr = actor_cast<abstract_actor_ptr>(whom);
attachable::token mtoken{typeid(down_observer), this};
ptr->detach(mtoken);
default_attachable::observe_token tk{address(), default_attachable::monitor};
ptr->detach({typeid(default_attachable::observe_token), &tk});
}
void local_actor::on_exit() {
......
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