Commit 12cb692a authored by Dominik Charousset's avatar Dominik Charousset

Use plain reference counts for flow objects

Observer nor observable both assume to run in an event loop and are
not designed to be thread-safe. Hence, they do not need an atomic
reference count except in a few exceptions such as subscriptions to
asynchronous buffers.
parent c2758a4b
......@@ -105,6 +105,7 @@ caf_add_component(
src/deserializer.cpp
src/detail/abstract_worker.cpp
src/detail/abstract_worker_hub.cpp
src/detail/atomic_ref_counted.cpp
src/detail/base64.cpp
src/detail/behavior_impl.cpp
src/detail/behavior_stack.cpp
......@@ -125,11 +126,11 @@ caf_add_component(
src/detail/monotonic_buffer_resource.cpp
src/detail/parse.cpp
src/detail/parser/chars.cpp
src/detail/plain_ref_counted.cpp
src/detail/pretty_type_name.cpp
src/detail/print.cpp
src/detail/private_thread.cpp
src/detail/private_thread_pool.cpp
src/detail/ref_counted_base.cpp
src/detail/ripemd_160.cpp
src/detail/serialized_size.cpp
src/detail/set_thread_name.cpp
......
......@@ -33,14 +33,6 @@ public:
virtual void run() = 0;
virtual state current_state() const noexcept = 0;
friend void intrusive_ptr_add_ref(const impl* ptr) noexcept {
ptr->ref_disposable();
}
friend void intrusive_ptr_release(const impl* ptr) noexcept {
ptr->deref_disposable();
}
};
using impl_ptr = intrusive_ptr<impl>;
......
......@@ -11,14 +11,14 @@
namespace caf::detail {
// Base class for reference counted objects with an atomic reference count.
class CAF_CORE_EXPORT ref_counted_base {
/// Base class for reference counted objects with an atomic reference count.
class CAF_CORE_EXPORT atomic_ref_counted {
public:
virtual ~ref_counted_base();
virtual ~atomic_ref_counted();
ref_counted_base();
ref_counted_base(const ref_counted_base&);
ref_counted_base& operator=(const ref_counted_base&);
atomic_ref_counted();
atomic_ref_counted(const atomic_ref_counted&);
atomic_ref_counted& operator=(const atomic_ref_counted&);
/// Increases reference count by one.
void ref() const noexcept {
......@@ -34,6 +34,7 @@ public:
return rc_ == 1;
}
/// Queries the current reference count for this object.
size_t get_reference_count() const noexcept {
return rc_.load();
}
......
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include <cstddef>
#include "caf/detail/core_export.hpp"
namespace caf::detail {
/// Base class for reference counted objects with an plain (i.e., thread-unsafe)
/// reference count.
class CAF_CORE_EXPORT plain_ref_counted {
public:
virtual ~plain_ref_counted();
plain_ref_counted();
plain_ref_counted(const plain_ref_counted&);
plain_ref_counted& operator=(const plain_ref_counted&);
/// Increases reference count by one.
void ref() const noexcept {
++rc_;
}
/// Decreases reference count by one and calls `request_deletion`
/// when it drops to zero.
void deref() const noexcept {
if (rc_ > 1)
--rc_;
else
delete this;
}
/// Queries whether there is exactly one reference.
bool unique() const noexcept {
return rc_ == 1;
}
/// Queries the current reference count for this object.
size_t get_reference_count() const noexcept {
return rc_;
}
protected:
mutable size_t rc_;
};
} // namespace caf::detail
......@@ -8,7 +8,6 @@
#include "caf/detail/core_export.hpp"
#include "caf/intrusive_ptr.hpp"
#include "caf/ref_counted.hpp"
namespace caf {
......@@ -20,7 +19,13 @@ public:
/// Internal implementation class of a `disposable`.
class CAF_CORE_EXPORT impl {
public:
CAF_INTRUSIVE_PTR_FRIENDS_SFX(impl, _disposable)
friend void intrusive_ptr_add_ref(const impl* ptr) noexcept {
ptr->ref_disposable();
}
friend void intrusive_ptr_release(const impl* ptr) noexcept {
ptr->deref_disposable();
}
virtual ~impl();
......
......@@ -7,7 +7,6 @@
#include "caf/async/spsc_buffer.hpp"
#include "caf/defaults.hpp"
#include "caf/detail/core_export.hpp"
#include "caf/detail/ref_counted_base.hpp"
#include "caf/flow/coordinator.hpp"
#include "caf/flow/observable.hpp"
#include "caf/flow/op/defer.hpp"
......
......@@ -8,6 +8,7 @@
#include "caf/async/producer.hpp"
#include "caf/defaults.hpp"
#include "caf/detail/comparable.hpp"
#include "caf/detail/plain_ref_counted.hpp"
#include "caf/detail/type_traits.hpp"
#include "caf/disposable.hpp"
#include "caf/error.hpp"
......@@ -88,13 +89,6 @@ public:
pimpl_->on_next(item);
}
/// Creates a new observer from `Impl`.
template <class Impl, class... Ts>
[[nodiscard]] static observer make(Ts&&... xs) {
static_assert(std::is_base_of_v<impl, Impl>);
return observer{make_counted<Impl>(std::forward<Ts>(xs)...)};
}
bool valid() const noexcept {
return pimpl_ != nullptr;
}
......@@ -107,22 +101,6 @@ public:
return !valid();
}
impl* ptr() {
return pimpl_.get();
}
const impl* ptr() const {
return pimpl_.get();
}
const intrusive_ptr<impl>& as_intrusive_ptr() const& noexcept {
return pimpl_;
}
intrusive_ptr<impl>&& as_intrusive_ptr() && noexcept {
return std::move(pimpl_);
}
void swap(observer& other) noexcept {
pimpl_.swap(other.pimpl_);
}
......@@ -131,17 +109,67 @@ public:
return pimpl_.compare(other.pimpl_);
}
/// Returns an observer that ignores any of its inputs.
static observer ignore();
private:
intrusive_ptr<impl> pimpl_;
};
/// @relates observer
template <class T>
using observer_impl = typename observer<T>::impl;
/// Simple base type observer implementations that implements the reference
/// counting member functions with a plain (i.e., not thread-safe) reference
/// count.
/// @relates observer
template <class T>
class observer_impl_base : public detail::plain_ref_counted,
public observer_impl<T> {
public:
void ref_coordinated() const noexcept final {
this->ref();
}
void deref_coordinated() const noexcept final {
this->deref();
}
};
} // namespace caf::flow
namespace caf::detail {
template <class T>
class ignoring_observer : public flow::observer_impl_base<T> {
public:
void on_next(const T&) override {
if (sub_)
sub_.request(1);
}
void on_error(const error&) override {
sub_ = nullptr;
}
void on_complete() override {
sub_ = nullptr;
}
void on_subscribe(flow::subscription sub) override {
if (!sub_) {
sub_ = std::move(sub);
sub_.request(defaults::flow::buffer_size);
} else {
sub.dispose();
}
}
private:
flow::subscription sub_;
};
template <class OnNextSignature>
struct on_next_trait;
......@@ -164,8 +192,7 @@ using on_next_value_type = typename on_next_trait_t<F>::value_type;
template <class OnNext, class OnError = unit_t, class OnComplete = unit_t>
class default_observer_impl
: public detail::ref_counted_base,
public flow::observer_impl<on_next_value_type<OnNext>> {
: public flow::observer_impl_base<on_next_value_type<OnNext>> {
public:
static_assert(std::is_invocable_v<OnError, const error&>);
......@@ -197,14 +224,18 @@ public:
}
void on_error(const error& what) override {
if (sub_) {
on_error_(what);
sub_ = nullptr;
}
}
void on_complete() override {
if (sub_) {
on_complete_();
sub_ = nullptr;
}
}
void on_subscribe(flow::subscription sub) override {
if (!sub_) {
......@@ -215,14 +246,6 @@ public:
}
}
void ref_coordinated() const noexcept override {
this->ref();
}
void deref_coordinated() const noexcept override {
this->deref();
}
private:
OnNext on_next_;
OnError on_error_;
......@@ -234,6 +257,11 @@ private:
namespace caf::flow {
template <class T>
observer<T> observer<T>::ignore() {
return observer<T>{make_counted<detail::ignoring_observer<T>>()};
}
/// Creates an observer from given callbacks.
/// @param on_next Callback for handling incoming elements.
/// @param on_error Callback for handling an error.
......@@ -285,7 +313,7 @@ auto make_observer_from_ptr(SmartPointer ptr) {
/// Writes observed values to a bounded buffer.
template <class Buffer>
class buffer_writer_impl : public detail::ref_counted_base,
class buffer_writer_impl : public detail::atomic_ref_counted,
public observer_impl<typename Buffer::value_type>,
public async::producer {
public:
......@@ -295,10 +323,6 @@ public:
using value_type = typename Buffer::value_type;
// -- friends ----------------------------------------------------------------
CAF_INTRUSIVE_PTR_FRIENDS(buffer_writer_impl)
// -- constructors, destructors, and assignment operators --------------------
buffer_writer_impl(coordinator* ctx, buffer_ptr buf)
......@@ -312,7 +336,15 @@ public:
buf_->close();
}
// -- implementation of coordinated ------------------------------------------
// -- intrusive_ptr interface ------------------------------------------------
friend void intrusive_ptr_add_ref(const buffer_writer_impl* ptr) noexcept {
ptr->ref();
}
friend void intrusive_ptr_release(const buffer_writer_impl* ptr) noexcept {
ptr->deref();
}
void ref_coordinated() const noexcept final {
this->ref();
......@@ -322,6 +354,14 @@ public:
this->deref();
}
void ref_producer() const noexcept final {
this->ref();
}
void deref_producer() const noexcept final {
this->deref();
}
// -- implementation of observer<T>::impl ------------------------------------
void on_next(const value_type& item) override {
......@@ -381,14 +421,6 @@ public:
});
}
void ref_producer() const noexcept final {
this->ref();
}
void deref_producer() const noexcept final {
this->deref();
}
private:
void on_demand(size_t n) {
CAF_LOG_TRACE(CAF_ARG(n));
......@@ -416,9 +448,9 @@ private:
// -- utility observer ---------------------------------------------------------
/// Forwards all events to a parent.
/// Forwards all events to its parent.
template <class T, class Parent, class Token>
class forwarder : public detail::ref_counted_base, public observer_impl<T> {
class forwarder : public observer_impl_base<T> {
public:
// -- constructors, destructors, and assignment operators --------------------
......@@ -427,16 +459,6 @@ public:
// nop
}
// -- implementation of coordinated ------------------------------------------
void ref_coordinated() const noexcept final {
this->ref();
}
void deref_coordinated() const noexcept final {
this->deref();
}
// -- implementation of observer_impl<T> -------------------------------------
void on_complete() override {
......@@ -474,19 +496,8 @@ private:
/// An observer with minimal internal logic. Useful for writing unit tests.
template <class T>
class passive_observer : public detail::ref_counted_base,
public observer_impl<T> {
class passive_observer : public observer_impl_base<T> {
public:
// -- implementation of coordinated ------------------------------------------
void ref_coordinated() const noexcept final {
this->ref();
}
void deref_coordinated() const noexcept final {
this->deref();
}
// -- implementation of observer_impl<T> -------------------------------------
void on_complete() override {
......@@ -561,18 +572,8 @@ intrusive_ptr<passive_observer<T>> make_passive_observer() {
/// Similar to @ref passive_observer but automatically requests items until
/// completed. Useful for writing unit tests.
template <class T>
class auto_observer : public detail::ref_counted_base, public observer_impl<T> {
class auto_observer : public observer_impl_base<T> {
public:
// -- implementation of coordinated ------------------------------------------
void ref_coordinated() const noexcept final {
this->ref();
}
void deref_coordinated() const noexcept final {
this->deref();
}
// -- implementation of observer_impl<T> -------------------------------------
void on_complete() override {
......
......@@ -4,7 +4,6 @@
#pragma once
#include "caf/detail/ref_counted_base.hpp"
#include "caf/flow/observer.hpp"
#include "caf/flow/subscription.hpp"
......
......@@ -4,7 +4,7 @@
#pragma once
#include "caf/detail/ref_counted_base.hpp"
#include "caf/detail/plain_ref_counted.hpp"
#include "caf/flow/observer.hpp"
#include "caf/flow/op/base.hpp"
#include "caf/flow/subscription.hpp"
......@@ -13,7 +13,7 @@ namespace caf::flow::op {
/// Convenience base type for *cold* observable types.
template <class T>
class cold : public detail::ref_counted_base, public base<T> {
class cold : public detail::plain_ref_counted, public base<T> {
public:
// -- member types -----------------------------------------------------------
......
......@@ -4,7 +4,6 @@
#pragma once
#include "caf/detail/ref_counted_base.hpp"
#include "caf/flow/observer.hpp"
#include "caf/flow/op/cold.hpp"
#include "caf/flow/op/empty.hpp"
......
......@@ -4,7 +4,6 @@
#pragma once
#include "caf/detail/ref_counted_base.hpp"
#include "caf/detail/type_list.hpp"
#include "caf/flow/observer.hpp"
#include "caf/flow/op/hot.hpp"
......@@ -143,7 +142,6 @@ public:
// -- implementation of observable_impl<T> ---------------------------------
disposable subscribe(observer<output_type> out) override {
CAF_LOG_TRACE(CAF_ARG2("out", out.ptr()));
using impl_t = from_generator_sub<Generator, Steps...>;
auto ptr = make_counted<impl_t>(super::ctx_, out, gen_, steps_);
auto sub = subscription{std::move(ptr)};
......
......@@ -4,7 +4,7 @@
#pragma once
#include "caf/detail/ref_counted_base.hpp"
#include "caf/detail/atomic_ref_counted.hpp"
#include "caf/detail/scope_guard.hpp"
#include "caf/flow/observer.hpp"
#include "caf/flow/op/hot.hpp"
......@@ -17,19 +17,10 @@ namespace caf::flow::op {
/// Reads from an observable buffer and emits the consumed items.
template <class Buffer>
class from_resource_sub : public subscription::impl_base,
class from_resource_sub : public detail::atomic_ref_counted,
public subscription::impl,
public async::consumer {
public:
// -- intrusive_ptr interface ------------------------------------------------
friend void intrusive_ptr_add_ref(const from_resource_sub* ptr) noexcept {
ptr->ref();
}
friend void intrusive_ptr_release(const from_resource_sub* ptr) noexcept {
ptr->deref();
}
// -- member types -----------------------------------------------------------
using value_type = typename Buffer::value_type;
......@@ -89,11 +80,29 @@ public:
});
}
void ref_consumer() const noexcept override {
// -- intrusive_ptr interface ------------------------------------------------
friend void intrusive_ptr_add_ref(const from_resource_sub* ptr) noexcept {
ptr->ref();
}
friend void intrusive_ptr_release(const from_resource_sub* ptr) noexcept {
ptr->deref();
}
void ref_consumer() const noexcept final {
this->ref();
}
void deref_consumer() const noexcept final {
this->deref();
}
void ref_disposable() const noexcept final {
this->ref();
}
void deref_consumer() const noexcept override {
void deref_disposable() const noexcept final {
this->deref();
}
......
......@@ -5,7 +5,7 @@
#pragma once
#include "caf/defaults.hpp"
#include "caf/detail/ref_counted_base.hpp"
#include "caf/detail/plain_ref_counted.hpp"
#include "caf/detail/scope_guard.hpp"
#include "caf/detail/type_list.hpp"
......@@ -20,7 +20,7 @@ using from_steps_output_t =
typename detail::tl_back_t<detail::type_list<Steps...>>::output_type;
template <class Input, class... Steps>
class from_steps_sub : public detail::ref_counted_base,
class from_steps_sub : public detail::plain_ref_counted,
public observer_impl<Input>,
public subscription_impl {
public:
......@@ -267,7 +267,6 @@ public:
// -- implementation of observable_impl<T> -----------------------------------
disposable subscribe(observer<output_type> out) override {
CAF_LOG_TRACE(CAF_ARG2("out", out.ptr()));
using sub_t = from_steps_sub<Input, Steps...>;
auto ptr = make_counted<sub_t>(super::ctx_, out, steps_);
input_->subscribe(observer<input_type>{ptr});
......
......@@ -4,7 +4,7 @@
#pragma once
#include "caf/detail/ref_counted_base.hpp"
#include "caf/detail/plain_ref_counted.hpp"
#include "caf/flow/observer.hpp"
#include "caf/flow/op/base.hpp"
#include "caf/flow/subscription.hpp"
......@@ -13,7 +13,7 @@ namespace caf::flow::op {
/// Convenience base type for *hot* observable types.
template <class T>
class hot : public detail::ref_counted_base, public base<T> {
class hot : public detail::plain_ref_counted, public base<T> {
public:
// -- member types -----------------------------------------------------------
......
......@@ -4,7 +4,6 @@
#pragma once
#include "caf/detail/ref_counted_base.hpp"
#include "caf/flow/observer.hpp"
#include "caf/flow/op/cold.hpp"
#include "caf/flow/op/empty.hpp"
......
......@@ -65,8 +65,8 @@ public:
disposable subscribe(observer<output_type> out) override {
auto ptr = make_counted<never_sub<T>>(super::ctx_, out);
out.ptr()->on_subscribe(subscription{ptr});
return ptr->as_disposable();
out.on_subscribe(subscription{ptr});
return disposable{std::move(ptr)};
}
};
......
......@@ -16,7 +16,7 @@
namespace caf::flow::op {
template <class T>
class pipe_sub : public detail::ref_counted_base,
class pipe_sub : public detail::plain_ref_counted,
public observer_impl<T>,
public subscription_impl {
public:
......@@ -166,7 +166,7 @@ private:
};
template <class T>
class prefix_and_tail_fwd : public detail::ref_counted_base,
class prefix_and_tail_fwd : public detail::plain_ref_counted,
public observer_impl<T> {
public:
// -- constructors, destructors, and assignment operators --------------------
......
......@@ -4,7 +4,6 @@
#pragma once
#include "caf/detail/ref_counted_base.hpp"
#include "caf/flow/observable_decl.hpp"
#include "caf/flow/observer.hpp"
#include "caf/flow/op/cold.hpp"
......
......@@ -5,7 +5,7 @@
#pragma once
#include "caf/detail/core_export.hpp"
#include "caf/detail/ref_counted_base.hpp"
#include "caf/detail/plain_ref_counted.hpp"
#include "caf/disposable.hpp"
#include "caf/flow/coordinated.hpp"
#include "caf/flow/fwd.hpp"
......@@ -33,7 +33,7 @@ public:
/// Simple base type for all subscription implementations that implements the
/// reference counting member functions.
class CAF_CORE_EXPORT impl_base : public detail::ref_counted_base,
class CAF_CORE_EXPORT impl_base : public detail::plain_ref_counted,
public impl {
public:
void ref_disposable() const noexcept final;
......
......@@ -7,8 +7,8 @@
#include <atomic>
#include <cstddef>
#include "caf/detail/atomic_ref_counted.hpp"
#include "caf/detail/core_export.hpp"
#include "caf/detail/ref_counted_base.hpp"
namespace caf {
......@@ -16,9 +16,9 @@ namespace caf {
/// Serves the requirements of {@link intrusive_ptr}.
/// @note *All* instances of `ref_counted` start with a reference count of 1.
/// @relates intrusive_ptr
class CAF_CORE_EXPORT ref_counted : public detail::ref_counted_base {
class CAF_CORE_EXPORT ref_counted : public detail::atomic_ref_counted {
public:
using super = ref_counted_base;
using super = detail::atomic_ref_counted;
using super::super;
......
......@@ -2,28 +2,28 @@
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#include "caf/detail/ref_counted_base.hpp"
#include "caf/detail/atomic_ref_counted.hpp"
namespace caf::detail {
ref_counted_base::~ref_counted_base() {
atomic_ref_counted::~atomic_ref_counted() {
// nop
}
ref_counted_base::ref_counted_base() : rc_(1) {
atomic_ref_counted::atomic_ref_counted() : rc_(1) {
// nop
}
ref_counted_base::ref_counted_base(const ref_counted_base&) : rc_(1) {
// nop; intentionally don't copy the reference count
atomic_ref_counted::atomic_ref_counted(const atomic_ref_counted&) : rc_(1) {
// Intentionally don't copy the reference count.
}
ref_counted_base& ref_counted_base::operator=(const ref_counted_base&) {
// nop; intentionally don't copy the reference count
atomic_ref_counted& atomic_ref_counted::operator=(const atomic_ref_counted&) {
// Intentionally don't copy the reference count.
return *this;
}
void ref_counted_base::deref() const noexcept {
void atomic_ref_counted::deref() const noexcept {
if (unique() || rc_.fetch_sub(1, std::memory_order_acq_rel) == 1)
delete this;
}
......
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#include "caf/detail/plain_ref_counted.hpp"
namespace caf::detail {
plain_ref_counted::~plain_ref_counted() {
// nop
}
plain_ref_counted::plain_ref_counted() : rc_(1) {
// nop
}
plain_ref_counted::plain_ref_counted(const plain_ref_counted&) : rc_(1) {
// Intentionally don't copy the reference count.
}
plain_ref_counted& plain_ref_counted::operator=(const plain_ref_counted&) {
// Intentionally don't copy the reference count.
return *this;
}
} // namespace caf::detail
#include "caf/flow/op/interval.hpp"
#include "caf/detail/plain_ref_counted.hpp"
#include "caf/flow/subscription.hpp"
#include <limits>
......@@ -16,7 +17,7 @@ public:
using interval_sub_ptr = intrusive_ptr<interval_sub_base>;
class interval_action : public detail::ref_counted_base, public action::impl {
class interval_action : public detail::plain_ref_counted, public action::impl {
public:
interval_action(interval_sub_ptr sub)
: state_(action::state::scheduled), sub_(std::move(sub)) {
......
......@@ -29,17 +29,19 @@ SCENARIO("the never operator never invokes callbacks except when disposed") {
auto uut = ctx->make_observable().never<int32_t>();
auto snk1 = flow::make_auto_observer<int32_t>();
auto snk2 = flow::make_auto_observer<int32_t>();
auto sub = uut.subscribe(snk1->as_observer());
auto sub1 = uut.subscribe(snk1->as_observer());
ctx->run();
CHECK(snk1->buf.empty());
CHECK_EQ(snk1->state, flow::observer_state::subscribed);
sub.dispose();
sub1.dispose();
ctx->run();
CHECK_EQ(snk1->state, flow::observer_state::completed);
MESSAGE("dispose only affects the subscription, "
"the never operator remains unchanged");
uut.subscribe(snk2->as_observer());
auto sub2 = uut.subscribe(snk2->as_observer());
ctx->run();
CHECK_EQ(snk2->state, flow::observer_state::subscribed);
sub2.dispose();
}
}
}
......
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