Commit 943f51c8 authored by Dominik Charousset's avatar Dominik Charousset

Merge pull request #250 from actor-framework/ref-counting-overhaul

Always start ref counted objects with a count of 1
parents af15d8ff a027b522
...@@ -165,12 +165,12 @@ class abstract_actor : public abstract_channel { ...@@ -165,12 +165,12 @@ class abstract_actor : public abstract_channel {
/** /**
* Creates a non-proxy instance. * Creates a non-proxy instance.
*/ */
abstract_actor(size_t initial_ref_count = 0); abstract_actor();
/** /**
* Creates a proxy instance for a proxy running on `nid`. * Creates a proxy instance for a proxy running on `nid`.
*/ */
abstract_actor(actor_id aid, node_id nid, size_t initial_ref_count = 0); abstract_actor(actor_id aid, node_id nid);
/** /**
* Called by the runtime system to perform cleanup actions for this actor. * Called by the runtime system to perform cleanup actions for this actor.
......
...@@ -89,9 +89,9 @@ class abstract_channel : public ref_counted { ...@@ -89,9 +89,9 @@ class abstract_channel : public ref_counted {
private: private:
// can only be called from abstract_actor and abstract_group // can only be called from abstract_actor and abstract_group
abstract_channel(channel_type_flag subtype, size_t initial_ref_count = 0); abstract_channel(channel_type_flag subtype);
abstract_channel(channel_type_flag subtype, node_id nid, abstract_channel(channel_type_flag subtype, node_id nid);
size_t initial_ref_count = 0);
// identifies the node of this channel // identifies the node of this channel
node_id m_node; node_id m_node;
}; };
......
...@@ -72,23 +72,19 @@ class behavior { ...@@ -72,23 +72,19 @@ class behavior {
// nop // nop
} }
/**
* Creates a behavior using `d` and `f` as timeout definition
* without message handler.
*/
//template <class F>
//behavior(duration d, F f) : m_impl(detail::make_behavior(d, f)) {
// nop
//}
/** /**
* Assigns new handlers. * Assigns new handlers.
*/ */
template <class... Vs> template <class... Vs>
void assign(Vs... vs) { void assign(Vs... vs) {
static_assert(sizeof...(Vs) > 0, "assign() called without arguments");
m_impl = detail::make_behavior(vs...); m_impl = detail::make_behavior(vs...);
} }
void assign(intrusive_ptr<detail::behavior_impl> ptr) {
m_impl.swap(ptr);
}
/** /**
* Equal to `*this = other`. * Equal to `*this = other`.
*/ */
...@@ -142,8 +138,6 @@ class behavior { ...@@ -142,8 +138,6 @@ class behavior {
// nop // nop
} }
void assign(detail::behavior_impl*);
/** @endcond */ /** @endcond */
private: private:
......
...@@ -219,6 +219,8 @@ class blocking_actor ...@@ -219,6 +219,8 @@ class blocking_actor
// implemented by detail::proper_actor // implemented by detail::proper_actor
virtual void dequeue_response(behavior& bhvr, message_id mid) = 0; virtual void dequeue_response(behavior& bhvr, message_id mid) = 0;
void cleanup(uint32_t reason);
/** @endcond */ /** @endcond */
private: private:
...@@ -243,6 +245,8 @@ class blocking_actor::functor_based : public blocking_actor { ...@@ -243,6 +245,8 @@ class blocking_actor::functor_based : public blocking_actor {
create(dummy, tk, f, std::forward<Ts>(vs)...); create(dummy, tk, f, std::forward<Ts>(vs)...);
} }
void cleanup(uint32_t reason);
protected: protected:
void act() override; void act() override;
......
...@@ -27,6 +27,7 @@ ...@@ -27,6 +27,7 @@
#include "caf/variant.hpp" #include "caf/variant.hpp"
#include "caf/optional.hpp" #include "caf/optional.hpp"
#include "caf/match_case.hpp" #include "caf/match_case.hpp"
#include "caf/make_counted.hpp"
#include "caf/intrusive_ptr.hpp" #include "caf/intrusive_ptr.hpp"
#include "caf/atom.hpp" #include "caf/atom.hpp"
...@@ -132,7 +133,7 @@ class default_behavior_impl : public behavior_impl { ...@@ -132,7 +133,7 @@ class default_behavior_impl : public behavior_impl {
typename behavior_impl::pointer typename behavior_impl::pointer
copy(const generic_timeout_definition& tdef) const override { copy(const generic_timeout_definition& tdef) const override {
return new default_behavior_impl<Tuple>(m_cases, tdef); return make_counted<default_behavior_impl<Tuple>>(m_cases, tdef);
} }
void handle_timeout() override { void handle_timeout() override {
...@@ -155,13 +156,14 @@ class default_behavior_impl : public behavior_impl { ...@@ -155,13 +156,14 @@ class default_behavior_impl : public behavior_impl {
// ra = reorganize arguments // ra = reorganize arguments
template <class R, class... Ts> template <class R, class... Ts>
R* make_behavior_eor(std::tuple<Ts...> match_cases) { intrusive_ptr<R> make_behavior_eor(std::tuple<Ts...> match_cases) {
return new R(std::move(match_cases)); return make_counted<R>(std::move(match_cases));
} }
template <class R, class... Ts, class F> template <class R, class... Ts, class F>
R* make_behavior_eor(std::tuple<Ts...> match_cases, timeout_definition<F>& td) { intrusive_ptr<R> make_behavior_eor(std::tuple<Ts...> match_cases,
return new R(std::move(match_cases), std::move(td)); timeout_definition<F>& td) {
return make_counted<R>(std::move(match_cases), std::move(td));
} }
template <class F, bool IsMC = std::is_base_of<match_case, F>::value> template <class F, bool IsMC = std::is_base_of<match_case, F>::value>
...@@ -201,13 +203,13 @@ struct join_std_tuples<std::tuple<Ts...>, std::tuple<Us...>, Rs...> ...@@ -201,13 +203,13 @@ struct join_std_tuples<std::tuple<Ts...>, std::tuple<Us...>, Rs...>
// this function reorganizes its arguments to shift the timeout definition // this function reorganizes its arguments to shift the timeout definition
// to the front (receives it at the tail) // to the front (receives it at the tail)
template <class R, class F, class... Vs> template <class R, class F, class... Vs>
R* make_behavior_ra(timeout_definition<F>& tdef, intrusive_ptr<R> make_behavior_ra(timeout_definition<F>& tdef,
tail_argument_token&, Vs... vs) { tail_argument_token&, Vs... vs) {
return make_behavior_eor<R>(std::tuple_cat(to_match_case_tuple(vs)...), tdef); return make_behavior_eor<R>(std::tuple_cat(to_match_case_tuple(vs)...), tdef);
} }
template <class R, class... Vs> template <class R, class... Vs>
R* make_behavior_ra(tail_argument_token&, Vs... vs) { intrusive_ptr<R> make_behavior_ra(tail_argument_token&, Vs... vs) {
return make_behavior_eor<R>(std::tuple_cat(to_match_case_tuple(vs)...)); return make_behavior_eor<R>(std::tuple_cat(to_match_case_tuple(vs)...));
} }
...@@ -215,7 +217,7 @@ R* make_behavior_ra(tail_argument_token&, Vs... vs) { ...@@ -215,7 +217,7 @@ R* make_behavior_ra(tail_argument_token&, Vs... vs) {
template <class R, class V, class... Vs> template <class R, class V, class... Vs>
typename std::enable_if< typename std::enable_if<
!std::is_same<V, tail_argument_token>::value, !std::is_same<V, tail_argument_token>::value,
R* intrusive_ptr<R>
>::type >::type
make_behavior_ra(V& v, Vs&... vs) { make_behavior_ra(V& v, Vs&... vs) {
return make_behavior_ra<R>(vs..., v); return make_behavior_ra<R>(vs..., v);
...@@ -224,11 +226,12 @@ make_behavior_ra(V& v, Vs&... vs) { ...@@ -224,11 +226,12 @@ make_behavior_ra(V& v, Vs&... vs) {
// this function reorganizes its arguments to shift the timeout definition // this function reorganizes its arguments to shift the timeout definition
// to the front (receives it at the tail) // to the front (receives it at the tail)
template <class... Vs> template <class... Vs>
default_behavior_impl< intrusive_ptr<
default_behavior_impl<
typename join_std_tuples< typename join_std_tuples<
typename lift_to_mctuple<Vs>::type... typename lift_to_mctuple<Vs>::type...
>::type >::type
>* >>
make_behavior(Vs&... vs) { make_behavior(Vs&... vs) {
using result_type = using result_type =
default_behavior_impl< default_behavior_impl<
......
...@@ -104,6 +104,18 @@ class intrusive_partitioned_list { ...@@ -104,6 +104,18 @@ class intrusive_partitioned_list {
} }
} }
template <class F>
void clear(F f) {
while (!first_empty()) {
f(first_front());
erase(first_begin());
}
while (!second_empty()) {
f(second_front());
erase(second_begin());
}
}
iterator first_begin() { iterator first_begin() {
return m_head.next; return m_head.next;
} }
......
...@@ -65,8 +65,10 @@ template <class T> ...@@ -65,8 +65,10 @@ template <class T>
struct rc_storage : public ref_counted { struct rc_storage : public ref_counted {
T instance; T instance;
template <class... Vs> template <class... Vs>
rc_storage(Vs&&... vs) : instance(this, std::forward<Vs>(vs)...) { rc_storage(Vs&&... vs)
// nop : instance(intrusive_ptr<ref_counted>(this, false),
std::forward<Vs>(vs)...) {
CAF_REQUIRE(get_reference_count() >= 1);
} }
}; };
...@@ -159,7 +161,7 @@ class basic_memory_cache : public memory_cache { ...@@ -159,7 +161,7 @@ class basic_memory_cache : public memory_cache {
embedded_storage new_embedded_storage() override { embedded_storage new_embedded_storage() override {
// allocate cache on-the-fly // allocate cache on-the-fly
if (!m_cache) { if (!m_cache) {
m_cache.reset(new storage); m_cache.reset(new storage, false); // starts with ref count of 1
} }
auto res = m_cache->next(); auto res = m_cache->next();
if (m_cache->has_next()) { if (m_cache->has_next()) {
...@@ -168,7 +170,7 @@ class basic_memory_cache : public memory_cache { ...@@ -168,7 +170,7 @@ class basic_memory_cache : public memory_cache {
// we got the last element out of the cache; pass the reference to the // we got the last element out of the cache; pass the reference to the
// client to avoid pointless increase/decrease ops on the reference count // client to avoid pointless increase/decrease ops on the reference count
embedded_storage result; embedded_storage result;
result.first.reset(m_cache.release(), false); result.first = std::move(m_cache);
result.second = res; result.second = res;
return result; return result;
} }
......
...@@ -85,7 +85,7 @@ class message_data : public ref_counted { ...@@ -85,7 +85,7 @@ class message_data : public ref_counted {
ptr& operator=(ptr&&) = default; ptr& operator=(ptr&&) = default;
ptr& operator=(const ptr&) = default; ptr& operator=(const ptr&) = default;
inline explicit ptr(message_data* p) : m_ptr(p) { inline ptr(intrusive_ptr<message_data> p) : m_ptr(std::move(p)) {
// nop // nop
} }
......
...@@ -163,10 +163,8 @@ class single_reader_queue { ...@@ -163,10 +163,8 @@ class single_reader_queue {
* @warning Call only from the reader (owner). * @warning Call only from the reader (owner).
*/ */
void close() { void close() {
clear_cached_elements(); auto nop = [](const T&) { };
if (fetch_new_data(nullptr)) { close(nop);
clear_cached_elements();
}
} }
/** /**
...@@ -180,25 +178,19 @@ class single_reader_queue { ...@@ -180,25 +178,19 @@ class single_reader_queue {
if (fetch_new_data(nullptr)) { if (fetch_new_data(nullptr)) {
clear_cached_elements(f); clear_cached_elements(f);
} }
m_cache.clear(std::move(f));
} }
single_reader_queue() : m_head(nullptr) { single_reader_queue() : m_head(nullptr) {
m_stack = stack_empty_dummy(); m_stack = stack_empty_dummy();
} }
void clear() { ~single_reader_queue() {
if (!closed()) { if (!closed()) {
clear_cached_elements(); close();
if (fetch_new_data()) {
clear_cached_elements();
}
} }
} }
~single_reader_queue() {
clear();
}
size_t count(size_t max_count = std::numeric_limits<size_t>::max()) { size_t count(size_t max_count = std::numeric_limits<size_t>::max()) {
size_t res = m_cache.count(max_count); size_t res = m_cache.count(max_count);
if (res >= max_count) { if (res >= max_count) {
...@@ -328,14 +320,6 @@ class single_reader_queue { ...@@ -328,14 +320,6 @@ class single_reader_queue {
return nullptr; return nullptr;
} }
void clear_cached_elements() {
while (m_head != nullptr) {
auto next = m_head->next;
m_delete(m_head);
m_head = next;
}
}
template <class F> template <class F>
void clear_cached_elements(const F& f) { void clear_cached_elements(const F& f) {
while (m_head) { while (m_head) {
......
...@@ -62,11 +62,6 @@ class sync_handle_helper; ...@@ -62,11 +62,6 @@ class sync_handle_helper;
*/ */
class local_actor : public abstract_actor { class local_actor : public abstract_actor {
public: public:
/*
using del = detail::disposer;
using mailbox_type = detail::single_reader_queue<mailbox_element, del>;
*/
static constexpr auto memory_cache_flag = detail::needs_embedding; static constexpr auto memory_cache_flag = detail::needs_embedding;
~local_actor(); ~local_actor();
......
...@@ -76,7 +76,7 @@ class mailbox_element : public memory_managed { ...@@ -76,7 +76,7 @@ class mailbox_element : public memory_managed {
using storage = detail::pair_storage<mailbox_element, value_storage>; using storage = detail::pair_storage<mailbox_element, value_storage>;
auto ptr = detail::memory::create<storage>(tk, std::move(sender), id, auto ptr = detail::memory::create<storage>(tk, std::move(sender), id,
std::forward<Vs>(vs)...); std::forward<Vs>(vs)...);
ptr->first.msg.reset(&(ptr->second)); ptr->first.msg.reset(&(ptr->second), false);
return unique_ptr{&(ptr->first)}; return unique_ptr{&(ptr->first)};
} }
......
...@@ -17,38 +17,46 @@ ...@@ -17,38 +17,46 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_DETAIL_MAKE_COUNTED_HPP #ifndef CAF_MAKE_COUNTED_HPP
#define CAF_DETAIL_MAKE_COUNTED_HPP #define CAF_MAKE_COUNTED_HPP
#include "caf/intrusive_ptr.hpp"
#include "caf/ref_counted.hpp" #include "caf/ref_counted.hpp"
#include "caf/intrusive_ptr.hpp"
#include "caf/detail/memory.hpp" #include "caf/detail/memory.hpp"
#include "caf/detail/type_traits.hpp" #include "caf/detail/type_traits.hpp"
namespace caf { namespace caf {
namespace detail {
/**
* Constructs an object of type `T` in an `intrusive_ptr`.
* @relates ref_counted
* @relates intrusive_ptr
*/
template <class T, class... Ts> template <class T, class... Ts>
typename std::enable_if< typename std::enable_if<
detail::is_memory_cached<T>::value, detail::is_memory_cached<T>::value,
intrusive_ptr<T> intrusive_ptr<T>
>::type >::type
make_counted(Ts&&... args) { make_counted(Ts&&... args) {
return {detail::memory::create<T>(std::forward<Ts>(args)...)}; return intrusive_ptr<T>(detail::memory::create<T>(std::forward<Ts>(args)...),
false);
} }
/**
* Constructs an object of type `T` in an `intrusive_ptr`.
* @relates ref_counted
* @relates intrusive_ptr
*/
template <class T, class... Ts> template <class T, class... Ts>
typename std::enable_if< typename std::enable_if<
!detail::is_memory_cached<T>::value, !detail::is_memory_cached<T>::value,
intrusive_ptr<T> intrusive_ptr<T>
>::type >::type
make_counted(Ts&&... args) { make_counted(Ts&&... args) {
return {new T(std::forward<Ts>(args)...)}; return intrusive_ptr<T>(new T(std::forward<Ts>(args)...), false);
} }
} // namespace detail
} // namespace caf } // namespace caf
#endif // CAF_DETAIL_MAKE_COUNTED_HPP #endif // CAF_MAKE_COUNTED_HPP
...@@ -26,6 +26,7 @@ ...@@ -26,6 +26,7 @@
#include "caf/atom.hpp" #include "caf/atom.hpp"
#include "caf/config.hpp" #include "caf/config.hpp"
#include "caf/from_string.hpp" #include "caf/from_string.hpp"
#include "caf/make_counted.hpp"
#include "caf/skip_message.hpp" #include "caf/skip_message.hpp"
#include "caf/detail/int_list.hpp" #include "caf/detail/int_list.hpp"
...@@ -327,7 +328,7 @@ class message { ...@@ -327,7 +328,7 @@ class message {
m_vals.detach(); m_vals.detach();
} }
void reset(raw_ptr new_ptr = nullptr); void reset(raw_ptr new_ptr = nullptr, bool add_ref = true);
void swap(message& other); void swap(message& other);
...@@ -444,8 +445,8 @@ make_message(V&& v, Vs&&... vs) { ...@@ -444,8 +445,8 @@ make_message(V&& v, Vs&&... vs) {
typename unbox_message_element< typename unbox_message_element<
typename detail::strip_and_convert<Vs>::type typename detail::strip_and_convert<Vs>::type
>::type...>; >::type...>;
auto ptr = new storage(std::forward<V>(v), std::forward<Vs>(vs)...); auto ptr = make_counted<storage>(std::forward<V>(v), std::forward<Vs>(vs)...);
return message{detail::message_data::ptr{ptr}}; return message{detail::message_data::ptr{std::move(ptr)}};
} }
/** /**
......
...@@ -30,14 +30,16 @@ namespace caf { ...@@ -30,14 +30,16 @@ namespace caf {
/** /**
* Base class for reference counted objects with an atomic reference count. * Base class for reference counted objects with an atomic reference count.
* Serves the requirements of {@link intrusive_ptr}. * Serves the requirements of {@link intrusive_ptr}.
* @note *All* instances of `ref_counted` start with a reference count of 1.
* @relates intrusive_ptr * @relates intrusive_ptr
*/ */
class ref_counted : public memory_managed { class ref_counted : public memory_managed {
public: public:
~ref_counted(); ~ref_counted();
ref_counted();
ref_counted(const ref_counted&); ref_counted(const ref_counted&);
ref_counted& operator=(const ref_counted&); ref_counted& operator=(const ref_counted&);
explicit ref_counted(size_t initial_count = 0);
/** /**
* Increases reference count by one. * Increases reference count by one.
......
...@@ -24,6 +24,7 @@ ...@@ -24,6 +24,7 @@
#include "caf/spawn_fwd.hpp" #include "caf/spawn_fwd.hpp"
#include "caf/typed_actor.hpp" #include "caf/typed_actor.hpp"
#include "caf/make_counted.hpp"
#include "caf/spawn_options.hpp" #include "caf/spawn_options.hpp"
#include "caf/typed_event_based_actor.hpp" #include "caf/typed_event_based_actor.hpp"
...@@ -37,7 +38,6 @@ ...@@ -37,7 +38,6 @@
#include "caf/detail/logging.hpp" #include "caf/detail/logging.hpp"
#include "caf/detail/type_traits.hpp" #include "caf/detail/type_traits.hpp"
#include "caf/detail/make_counted.hpp"
#include "caf/detail/proper_actor.hpp" #include "caf/detail/proper_actor.hpp"
#include "caf/detail/typed_actor_util.hpp" #include "caf/detail/typed_actor_util.hpp"
#include "caf/detail/implicit_conversions.hpp" #include "caf/detail/implicit_conversions.hpp"
...@@ -98,10 +98,7 @@ intrusive_ptr<C> spawn_impl(execution_unit* host, ...@@ -98,10 +98,7 @@ intrusive_ptr<C> spawn_impl(execution_unit* host,
C, C,
detail::proper_actor<C, policy_token> detail::proper_actor<C, policy_token>
>::type; >::type;
auto ptr = detail::make_counted<actor_impl>(std::forward<Ts>(args)...); auto ptr = make_counted<actor_impl>(std::forward<Ts>(args)...);
// actors start with a reference count of 1, hence we need to deref ptr once
CAF_REQUIRE(!ptr->unique());
ptr->deref();
CAF_LOGF_DEBUG("spawned actor with ID " << ptr->id()); CAF_LOGF_DEBUG("spawned actor with ID " << ptr->id());
CAF_PUSH_AID(ptr->id()); CAF_PUSH_AID(ptr->id());
before_launch_fun(ptr.get()); before_launch_fun(ptr.get());
......
...@@ -225,7 +225,7 @@ class typed_behavior { ...@@ -225,7 +225,7 @@ class typed_behavior {
behavior& unbox() { return m_bhvr; } behavior& unbox() { return m_bhvr; }
template <class... Ts> template <class... Ts>
void set(detail::default_behavior_impl<std::tuple<Ts...>>* ptr) { void set(intrusive_ptr<detail::default_behavior_impl<std::tuple<Ts...>>> bp) {
using mpi = using mpi =
typename detail::tl_filter_not< typename detail::tl_filter_not<
detail::type_list<typename detail::deduce_mpi<Ts>::type...>, detail::type_list<typename detail::deduce_mpi<Ts>::type...>,
...@@ -233,7 +233,8 @@ class typed_behavior { ...@@ -233,7 +233,8 @@ class typed_behavior {
>::type; >::type;
detail::static_asserter<signatures, mpi, detail::ctm>::verify_match(); detail::static_asserter<signatures, mpi, detail::ctm>::verify_match();
// final (type-erasure) step // final (type-erasure) step
m_bhvr.assign(static_cast<detail::behavior_impl*>(ptr)); intrusive_ptr<detail::behavior_impl> ptr = std::move(bp);
m_bhvr.assign(std::move(ptr));
} }
behavior m_bhvr; behavior m_bhvr;
......
...@@ -47,18 +47,18 @@ using guard_type = std::unique_lock<std::mutex>; ...@@ -47,18 +47,18 @@ using guard_type = std::unique_lock<std::mutex>;
// m_exit_reason is guaranteed to be set to 0, i.e., exit_reason::not_exited, // m_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(actor_id aid, node_id nid, size_t initial_count) 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), initial_count), std::move(nid)),
m_id(aid), m_id(aid),
m_exit_reason(exit_reason::not_exited), m_exit_reason(exit_reason::not_exited),
m_host(nullptr) { m_host(nullptr) {
// nop // nop
} }
abstract_actor::abstract_actor(size_t initial_count) abstract_actor::abstract_actor()
: abstract_channel(abstract_channel::is_abstract_actor_flag, : abstract_channel(abstract_channel::is_abstract_actor_flag,
detail::singletons::get_node_id(), initial_count), detail::singletons::get_node_id()),
m_id(detail::singletons::get_actor_registry()->next_id()), m_id(detail::singletons::get_actor_registry()->next_id()),
m_exit_reason(exit_reason::not_exited), m_exit_reason(exit_reason::not_exited),
m_host(nullptr) { m_host(nullptr) {
......
...@@ -26,18 +26,14 @@ namespace caf { ...@@ -26,18 +26,14 @@ namespace caf {
using detail::singletons; using detail::singletons;
abstract_channel::abstract_channel(channel_type_flag subtype, abstract_channel::abstract_channel(channel_type_flag subtype)
size_t initial_ref_count) : m_flags(static_cast<int>(subtype)),
: ref_counted(initial_ref_count),
m_flags(static_cast<int>(subtype)),
m_node(singletons::get_node_id()) { m_node(singletons::get_node_id()) {
// nop // nop
} }
abstract_channel::abstract_channel(channel_type_flag subtype, node_id nid, abstract_channel::abstract_channel(channel_type_flag subtype, node_id nid)
size_t initial_ref_count) : m_flags(static_cast<int>(subtype)),
: ref_counted(initial_ref_count),
m_flags(static_cast<int>(subtype)),
m_node(std::move(nid)) { m_node(std::move(nid)) {
// nop // nop
} }
......
...@@ -214,11 +214,11 @@ void abstract_coordinator::initialize() { ...@@ -214,11 +214,11 @@ void abstract_coordinator::initialize() {
void abstract_coordinator::stop_actors() { void abstract_coordinator::stop_actors() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
scoped_actor self(true); scoped_actor self{true};
self->monitor(m_timer); self->monitor(m_timer);
self->monitor(m_printer); self->monitor(m_printer);
self->send_exit(m_timer, exit_reason::user_shutdown); anon_send_exit(m_timer, exit_reason::user_shutdown);
self->send_exit(m_printer, exit_reason::user_shutdown); anon_send_exit(m_printer, exit_reason::user_shutdown);
int i = 0; int i = 0;
self->receive_for(i, 2)( self->receive_for(i, 2)(
[](const down_msg&) { [](const down_msg&) {
......
...@@ -36,9 +36,4 @@ void behavior::assign(behavior other) { ...@@ -36,9 +36,4 @@ void behavior::assign(behavior other) {
m_impl.swap(other.m_impl); m_impl.swap(other.m_impl);
} }
void behavior::assign(detail::behavior_impl* ptr) {
CAF_REQUIRE(ptr != nullptr);
m_impl.reset(ptr);
}
} // namespace caf } // namespace caf
...@@ -38,6 +38,11 @@ void blocking_actor::await_all_other_actors_done() { ...@@ -38,6 +38,11 @@ void blocking_actor::await_all_other_actors_done() {
detail::singletons::get_actor_registry()->await_running_count_equal(1); detail::singletons::get_actor_registry()->await_running_count_equal(1);
} }
void blocking_actor::cleanup(uint32_t reason) {
m_sync_handler.clear();
mailbox_based_actor::cleanup(reason);
}
void blocking_actor::functor_based::create(blocking_actor*, act_fun fun) { void blocking_actor::functor_based::create(blocking_actor*, act_fun fun) {
m_act = fun; m_act = fun;
} }
...@@ -47,4 +52,9 @@ void blocking_actor::functor_based::act() { ...@@ -47,4 +52,9 @@ void blocking_actor::functor_based::act() {
m_act(this); m_act(this);
} }
void blocking_actor::functor_based::cleanup(uint32_t reason) {
m_act = nullptr;
blocking_actor::cleanup(reason);
}
} // namespace caf } // namespace caf
...@@ -31,9 +31,7 @@ namespace caf { ...@@ -31,9 +31,7 @@ namespace caf {
// local actors are created with a reference count of one that is adjusted // local actors are created with a reference count of one that is adjusted
// 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() local_actor::local_actor() : m_planned_exit_reason(exit_reason::not_exited) {
: abstract_actor(size_t{1}),
m_planned_exit_reason(exit_reason::not_exited) {
// nop // nop
} }
......
...@@ -58,17 +58,4 @@ mailbox_element_ptr mailbox_element::make(actor_addr sender, message_id id, ...@@ -58,17 +58,4 @@ mailbox_element_ptr mailbox_element::make(actor_addr sender, message_id id,
return mailbox_element_ptr{ptr}; return mailbox_element_ptr{ptr};
} }
/*
mailbox_element::joint::joint(ref_counted* v0, actor_addr&& v1, message_id v2)
: embedded<mailbox_element>(v0, std::move(v1), v2) {
// nop
}
void mailbox_element::joint::request_deletion() {
sender = invalid_actor_addr;
msg.reset();
m_storage->deref();
}
*/
} // namespace caf } // namespace caf
...@@ -46,8 +46,8 @@ message& message::operator=(message&& other) { ...@@ -46,8 +46,8 @@ message& message::operator=(message&& other) {
return *this; return *this;
} }
void message::reset(raw_ptr new_ptr) { void message::reset(raw_ptr new_ptr, bool add_ref) {
m_vals.reset(new_ptr); m_vals.reset(new_ptr, add_ref);
} }
void message::swap(message& other) { void message::swap(message& other) {
......
...@@ -69,8 +69,8 @@ message_data* message_data::ptr::get_detached() { ...@@ -69,8 +69,8 @@ message_data* message_data::ptr::get_detached() {
auto p = m_ptr.get(); auto p = m_ptr.get();
if (!p->unique()) { if (!p->unique()) {
auto np = p->copy(); auto np = p->copy();
m_ptr.reset(np);
CAF_REQUIRE(np->unique()); CAF_REQUIRE(np->unique());
m_ptr.reset(np, false);
return np; return np;
} }
return p; return p;
......
...@@ -24,11 +24,11 @@ ...@@ -24,11 +24,11 @@
#include <unistd.h> #include <unistd.h>
#include <sys/types.h> #include <sys/types.h>
#include "caf/string_algorithms.hpp"
#include "caf/config.hpp" #include "caf/config.hpp"
#include "caf/node_id.hpp" #include "caf/node_id.hpp"
#include "caf/serializer.hpp" #include "caf/serializer.hpp"
#include "caf/make_counted.hpp"
#include "caf/string_algorithms.hpp"
#include "caf/primitive_variant.hpp" #include "caf/primitive_variant.hpp"
#include "caf/detail/logging.hpp" #include "caf/detail/logging.hpp"
...@@ -179,9 +179,9 @@ node_id::data* node_id::data::create_singleton() { ...@@ -179,9 +179,9 @@ node_id::data* node_id::data::create_singleton() {
auto hd_serial_and_mac_addr = join(macs, "") + detail::get_root_uuid(); auto hd_serial_and_mac_addr = join(macs, "") + detail::get_root_uuid();
node_id::host_id_type nid; node_id::host_id_type nid;
detail::ripemd_160(nid, hd_serial_and_mac_addr); detail::ripemd_160(nid, hd_serial_and_mac_addr);
auto ptr = new node_id::data(static_cast<uint32_t>(getpid()), nid); auto ptr = make_counted<node_id::data>(static_cast<uint32_t>(getpid()), nid);
ptr->ref(); // implicit ref count held by detail::singletons // note: ptr has a ref count of 1 -> implicitly held by detail::singletons
return ptr; return ptr.release();
} }
uint32_t node_id::process_id() const { uint32_t node_id::process_id() const {
......
...@@ -25,7 +25,11 @@ ref_counted::~ref_counted() { ...@@ -25,7 +25,11 @@ ref_counted::~ref_counted() {
// nop // nop
} }
ref_counted::ref_counted(const ref_counted&) : m_rc(0) { ref_counted::ref_counted() : m_rc(1) {
// nop
}
ref_counted::ref_counted(const ref_counted&) : m_rc(1) {
// nop; don't copy reference count // nop; don't copy reference count
} }
...@@ -34,8 +38,4 @@ ref_counted& ref_counted::operator=(const ref_counted&) { ...@@ -34,8 +38,4 @@ ref_counted& ref_counted::operator=(const ref_counted&) {
return *this; return *this;
} }
ref_counted::ref_counted(size_t initial_count) : m_rc(initial_count) {
// nop
}
} // namespace caf } // namespace caf
...@@ -67,9 +67,8 @@ class middleman : public detail::abstract_singleton { ...@@ -67,9 +67,8 @@ class middleman : public detail::abstract_singleton {
if (i != m_named_brokers.end()) { if (i != m_named_brokers.end()) {
return static_cast<Impl*>(i->second.get()); return static_cast<Impl*>(i->second.get());
} }
auto result = detail::make_counted<Impl>(*this); auto result = make_counted<Impl>(*this);
CAF_REQUIRE(!result->unique()); CAF_REQUIRE(result->unique());
result->deref(); // local_actor starts with ref count of 1
result->launch(true, false, nullptr); result->launch(true, false, nullptr);
m_named_brokers.insert(std::make_pair(name, result)); m_named_brokers.insert(std::make_pair(name, result));
return result; return result;
......
...@@ -20,12 +20,12 @@ ...@@ -20,12 +20,12 @@
#include "caf/io/basp_broker.hpp" #include "caf/io/basp_broker.hpp"
#include "caf/exception.hpp" #include "caf/exception.hpp"
#include "caf/make_counted.hpp"
#include "caf/binary_serializer.hpp" #include "caf/binary_serializer.hpp"
#include "caf/binary_deserializer.hpp" #include "caf/binary_deserializer.hpp"
#include "caf/forwarding_actor_proxy.hpp" #include "caf/forwarding_actor_proxy.hpp"
#include "caf/detail/singletons.hpp" #include "caf/detail/singletons.hpp"
#include "caf/detail/make_counted.hpp"
#include "caf/detail/actor_registry.hpp" #include "caf/detail/actor_registry.hpp"
#include "caf/io/basp.hpp" #include "caf/io/basp.hpp"
...@@ -38,7 +38,6 @@ namespace caf { ...@@ -38,7 +38,6 @@ namespace caf {
namespace io { namespace io {
using detail::singletons; using detail::singletons;
using detail::make_counted;
basp_broker::payload_writer::~payload_writer() { basp_broker::payload_writer::~payload_writer() {
// nop // nop
......
...@@ -18,8 +18,8 @@ ...@@ -18,8 +18,8 @@
******************************************************************************/ ******************************************************************************/
#include "caf/none.hpp" #include "caf/none.hpp"
#include "caf/config.hpp" #include "caf/config.hpp"
#include "caf/make_counted.hpp"
#include "caf/detail/logging.hpp" #include "caf/detail/logging.hpp"
#include "caf/detail/singletons.hpp" #include "caf/detail/singletons.hpp"
...@@ -28,7 +28,6 @@ ...@@ -28,7 +28,6 @@
#include "caf/io/broker.hpp" #include "caf/io/broker.hpp"
#include "caf/io/middleman.hpp" #include "caf/io/middleman.hpp"
#include "caf/detail/make_counted.hpp"
#include "caf/detail/actor_registry.hpp" #include "caf/detail/actor_registry.hpp"
#include "caf/detail/sync_request_bouncer.hpp" #include "caf/detail/sync_request_bouncer.hpp"
......
...@@ -22,6 +22,7 @@ ...@@ -22,6 +22,7 @@
#include "caf/config.hpp" #include "caf/config.hpp"
#include "caf/optional.hpp" #include "caf/optional.hpp"
#include "caf/exception.hpp" #include "caf/exception.hpp"
#include "caf/make_counted.hpp"
#include "caf/io/broker.hpp" #include "caf/io/broker.hpp"
#include "caf/io/middleman.hpp" #include "caf/io/middleman.hpp"
...@@ -767,7 +768,7 @@ connection_handle default_multiplexer::add_tcp_scribe(broker* self, ...@@ -767,7 +768,7 @@ connection_handle default_multiplexer::add_tcp_scribe(broker* self,
bool m_launched; bool m_launched;
stream<default_socket> m_stream; stream<default_socket> m_stream;
}; };
broker::scribe_pointer ptr{new impl{self, std::move(sock)}}; broker::scribe_pointer ptr = make_counted<impl>(self, std::move(sock));
self->add_scribe(ptr); self->add_scribe(ptr);
return ptr->hdl(); return ptr->hdl();
} }
...@@ -804,7 +805,7 @@ default_multiplexer::add_tcp_doorman(broker* self, ...@@ -804,7 +805,7 @@ default_multiplexer::add_tcp_doorman(broker* self,
private: private:
network::acceptor<default_socket_acceptor> m_acceptor; network::acceptor<default_socket_acceptor> m_acceptor;
}; };
broker::doorman_pointer ptr{new impl{self, std::move(sock)}}; broker::doorman_pointer ptr = make_counted<impl>(self, std::move(sock));
self->add_doorman(ptr); self->add_doorman(ptr);
return ptr->hdl(); return ptr->hdl();
} }
......
...@@ -31,6 +31,7 @@ ...@@ -31,6 +31,7 @@
#include "caf/announce.hpp" #include "caf/announce.hpp"
#include "caf/to_string.hpp" #include "caf/to_string.hpp"
#include "caf/actor_proxy.hpp" #include "caf/actor_proxy.hpp"
#include "caf/make_counted.hpp"
#include "caf/scoped_actor.hpp" #include "caf/scoped_actor.hpp"
#include "caf/uniform_type_info.hpp" #include "caf/uniform_type_info.hpp"
...@@ -42,7 +43,6 @@ ...@@ -42,7 +43,6 @@
#include "caf/detail/ripemd_160.hpp" #include "caf/detail/ripemd_160.hpp"
#include "caf/detail/safe_equal.hpp" #include "caf/detail/safe_equal.hpp"
#include "caf/detail/singletons.hpp" #include "caf/detail/singletons.hpp"
#include "caf/detail/make_counted.hpp"
#include "caf/detail/get_root_uuid.hpp" #include "caf/detail/get_root_uuid.hpp"
#include "caf/detail/actor_registry.hpp" #include "caf/detail/actor_registry.hpp"
#include "caf/detail/get_mac_addresses.hpp" #include "caf/detail/get_mac_addresses.hpp"
...@@ -140,8 +140,6 @@ void do_announce(const char* tname) { ...@@ -140,8 +140,6 @@ void do_announce(const char* tname) {
} // namespace <anonymous> } // namespace <anonymous>
using detail::make_counted;
using middleman_actor_base = middleman_actor::extend< using middleman_actor_base = middleman_actor::extend<
reacts_to<ok_atom, int64_t>, reacts_to<ok_atom, int64_t>,
reacts_to<ok_atom, int64_t, actor_addr>, reacts_to<ok_atom, int64_t, actor_addr>,
......
...@@ -2,9 +2,9 @@ ...@@ -2,9 +2,9 @@
#include <cstddef> #include <cstddef>
#include "test.hpp" #include "test.hpp"
#include "caf/intrusive_ptr.hpp"
#include "caf/ref_counted.hpp" #include "caf/ref_counted.hpp"
#include "caf/detail/make_counted.hpp" #include "caf/make_counted.hpp"
#include "caf/intrusive_ptr.hpp"
using namespace caf; using namespace caf;
...@@ -13,7 +13,14 @@ namespace { ...@@ -13,7 +13,14 @@ namespace {
int class0_instances = 0; int class0_instances = 0;
int class1_instances = 0; int class1_instances = 0;
struct class0 : ref_counted { class class0;
class class1;
using class0ptr = intrusive_ptr<class0>;
using class1ptr = intrusive_ptr<class1>;
class class0 : public ref_counted {
public:
class0(bool subtype = false) : m_subtype(subtype) { class0(bool subtype = false) : m_subtype(subtype) {
if (!subtype) { if (!subtype) {
++class0_instances; ++class0_instances;
...@@ -30,14 +37,16 @@ struct class0 : ref_counted { ...@@ -30,14 +37,16 @@ struct class0 : ref_counted {
return m_subtype; return m_subtype;
} }
virtual class0* create() const { virtual class0ptr create() const {
return new class0; return make_counted<class0>();
} }
private:
bool m_subtype; bool m_subtype;
}; };
struct class1 : class0 { class class1 : public class0 {
public:
class1() : class0(true) { class1() : class0(true) {
++class1_instances; ++class1_instances;
} }
...@@ -46,19 +55,16 @@ struct class1 : class0 { ...@@ -46,19 +55,16 @@ struct class1 : class0 {
--class1_instances; --class1_instances;
} }
class1* create() const override { class0ptr create() const override {
return new class1; return make_counted<class1>();
} }
}; };
using class0_ptr = intrusive_ptr<class0>; class0ptr get_test_rc() {
using class1_ptr = intrusive_ptr<class1>; return make_counted<class0>();
class0* get_test_rc() {
return new class0;
} }
class0_ptr get_test_ptr() { class0ptr get_test_ptr() {
return get_test_rc(); return get_test_rc();
} }
...@@ -70,31 +76,31 @@ int main() { ...@@ -70,31 +76,31 @@ int main() {
CAF_TEST(test_intrusive_ptr); CAF_TEST(test_intrusive_ptr);
{ {
auto p = detail::make_counted<class0>(); auto p = make_counted<class0>();
CAF_CHECK_EQUAL(class0_instances, 1); CAF_CHECK_EQUAL(class0_instances, 1);
CAF_CHECK(p->unique()); CAF_CHECK(p->unique());
} }
CAF_CHECK_EQUAL(class0_instances, 0); CAF_CHECK_EQUAL(class0_instances, 0);
CAF_CHECK_EQUAL(class1_instances, 0); CAF_CHECK_EQUAL(class1_instances, 0);
{ {
class0_ptr p; class0ptr p;
p = new class0; p.reset(new class0, false);
CAF_CHECK_EQUAL(class0_instances, 1); CAF_CHECK_EQUAL(class0_instances, 1);
CAF_CHECK(p->unique()); CAF_CHECK(p->unique());
} }
CAF_CHECK_EQUAL(class0_instances, 0); CAF_CHECK_EQUAL(class0_instances, 0);
CAF_CHECK_EQUAL(class1_instances, 0); CAF_CHECK_EQUAL(class1_instances, 0);
{ {
class0_ptr p1; class0ptr p1;
p1 = get_test_rc(); p1 = get_test_rc();
class0_ptr p2 = p1; class0ptr p2 = p1;
CAF_CHECK_EQUAL(class0_instances, 1); CAF_CHECK_EQUAL(class0_instances, 1);
CAF_CHECK_EQUAL(p1->unique(), false); CAF_CHECK_EQUAL(p1->unique(), false);
} }
CAF_CHECK_EQUAL(class0_instances, 0); CAF_CHECK_EQUAL(class0_instances, 0);
CAF_CHECK_EQUAL(class1_instances, 0); CAF_CHECK_EQUAL(class1_instances, 0);
{ {
std::list<class0_ptr> pl; std::list<class0ptr> pl;
pl.push_back(get_test_ptr()); pl.push_back(get_test_ptr());
pl.push_back(get_test_rc()); pl.push_back(get_test_rc());
pl.push_back(pl.front()->create()); pl.push_back(pl.front()->create());
...@@ -104,17 +110,17 @@ int main() { ...@@ -104,17 +110,17 @@ int main() {
CAF_CHECK_EQUAL(class0_instances, 0); CAF_CHECK_EQUAL(class0_instances, 0);
CAF_CHECK_EQUAL(class1_instances, 0); CAF_CHECK_EQUAL(class1_instances, 0);
{ {
auto p1 = detail::make_counted<class0>(); auto p1 = make_counted<class0>();
CAF_CHECK_EQUAL(p1->is_subtype(), false); CAF_CHECK_EQUAL(p1->is_subtype(), false);
CAF_CHECK_EQUAL(p1->unique(), true); CAF_CHECK_EQUAL(p1->unique(), true);
CAF_CHECK_EQUAL(class0_instances, 1); CAF_CHECK_EQUAL(class0_instances, 1);
CAF_CHECK_EQUAL(class1_instances, 0); CAF_CHECK_EQUAL(class1_instances, 0);
p1.reset(new class1); p1.reset(new class1, false);
CAF_CHECK_EQUAL(p1->is_subtype(), true); CAF_CHECK_EQUAL(p1->is_subtype(), true);
CAF_CHECK_EQUAL(p1->unique(), true); CAF_CHECK_EQUAL(p1->unique(), true);
CAF_CHECK_EQUAL(class0_instances, 0); CAF_CHECK_EQUAL(class0_instances, 0);
CAF_CHECK_EQUAL(class1_instances, 1); CAF_CHECK_EQUAL(class1_instances, 1);
auto p2 = detail::make_counted<class1>(); auto p2 = make_counted<class1>();
p1 = p2; p1 = p2;
CAF_CHECK_EQUAL(p1->unique(), false); CAF_CHECK_EQUAL(p1->unique(), false);
CAF_CHECK_EQUAL(class0_instances, 0); CAF_CHECK_EQUAL(class0_instances, 0);
......
...@@ -772,9 +772,10 @@ void test_constructor_attach() { ...@@ -772,9 +772,10 @@ void test_constructor_attach() {
public: public:
testee(actor buddy) : m_buddy(buddy) { testee(actor buddy) : m_buddy(buddy) {
attach_functor([=](uint32_t reason) { attach_functor([=](uint32_t reason) {
send(m_buddy, atom("done"), reason); send(buddy, atom("done"), reason);
}); });
} }
behavior make_behavior() { behavior make_behavior() {
return { return {
on(atom("die")) >> [=] { on(atom("die")) >> [=] {
...@@ -782,12 +783,18 @@ void test_constructor_attach() { ...@@ -782,12 +783,18 @@ void test_constructor_attach() {
} }
}; };
} }
void on_exit() {
m_buddy = invalid_actor;
}
private: private:
actor m_buddy; actor m_buddy;
}; };
class spawner : public event_based_actor { class spawner : public event_based_actor {
public: public:
spawner() : m_downs(0) { spawner() : m_downs(0) {
// nop
} }
behavior make_behavior() { behavior make_behavior() {
m_testee = spawn<testee, monitored>(this); m_testee = spawn<testee, monitored>(this);
...@@ -809,6 +816,11 @@ void test_constructor_attach() { ...@@ -809,6 +816,11 @@ void test_constructor_attach() {
} }
}; };
} }
void on_exit() {
m_testee = invalid_actor;
}
private: private:
int m_downs; int m_downs;
actor m_testee; actor m_testee;
......
...@@ -199,11 +199,17 @@ class server : public event_based_actor { ...@@ -199,11 +199,17 @@ class server : public event_based_actor {
unbecome(); // await next idle message unbecome(); // await next idle message
}, },
on(idle_atom::value) >> skip_message, on(idle_atom::value) >> skip_message,
others >> die others >> [=] {
CAF_UNEXPECTED_MSG(this);
die();
}
); );
}, },
on(request_atom::value) >> skip_message, on(request_atom::value) >> skip_message,
others >> die others >> [=] {
CAF_UNEXPECTED_MSG(this);
die();
}
}; };
} }
}; };
...@@ -376,7 +382,7 @@ void test_sync_send() { ...@@ -376,7 +382,7 @@ void test_sync_send() {
); );
// first 'request', then 'idle' // first 'request', then 'idle'
auto handle = s->sync_send(serv, request_atom::value); auto handle = s->sync_send(serv, request_atom::value);
send_as(work, serv, idle_atom::value); send_as(work, serv, idle_atom::value, work);
handle.await( handle.await(
[=](response_atom) { [=](response_atom) {
CAF_CHECKPOINT(); CAF_CHECKPOINT();
...@@ -384,11 +390,7 @@ void test_sync_send() { ...@@ -384,11 +390,7 @@ void test_sync_send() {
}, },
others >> CAF_UNEXPECTED_MSG_CB(s) others >> CAF_UNEXPECTED_MSG_CB(s)
); );
s->send(s, "Ever danced with the devil in the pale moonlight?"); s->quit(exit_reason::user_shutdown);
// response: {'EXIT', exit_reason::user_shutdown}
s->receive_loop(
others >> CAF_UNEXPECTED_MSG_CB(s)
);
}); });
self->receive( self->receive(
[&](const down_msg& dm) { [&](const down_msg& dm) {
......
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