Commit 07523fc4 authored by Dominik Charousset's avatar Dominik Charousset

Backport of the CAF 0.19 flow API

parent fbeacbcf
......@@ -64,7 +64,6 @@ caf_add_component(
message_priority
pec
sec
stream_priority
HEADERS
${CAF_CORE_HEADERS}
SOURCES
......@@ -87,6 +86,7 @@ caf_add_component(
src/actor_system_config.cpp
src/async/batch.cpp
src/async/consumer.cpp
src/async/execution_context.cpp
src/async/producer.cpp
src/attachable.cpp
src/behavior.cpp
......@@ -99,12 +99,12 @@ caf_add_component(
src/config_value.cpp
src/config_value_reader.cpp
src/config_value_writer.cpp
src/credit_controller.cpp
src/decorator/sequencer.cpp
src/default_attachable.cpp
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,6 +125,7 @@ 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
......@@ -132,22 +133,21 @@ caf_add_component(
src/detail/ripemd_160.cpp
src/detail/serialized_size.cpp
src/detail/set_thread_name.cpp
src/detail/size_based_credit_controller.cpp
src/detail/stream_bridge.cpp
src/detail/stringification_inspector.cpp
src/detail/sync_request_bouncer.cpp
src/detail/test_actor_clock.cpp
src/detail/thread_safe_actor_clock.cpp
src/detail/tick_emitter.cpp
src/detail/token_based_credit_controller.cpp
src/detail/type_id_list_builder.cpp
src/disposable.cpp
src/downstream_manager.cpp
src/downstream_manager_base.cpp
src/error.cpp
src/event_based_actor.cpp
src/execution_unit.cpp
src/flow/coordinated.cpp
src/flow/coordinator.cpp
src/flow/observable_builder.cpp
src/flow/op/interval.cpp
src/flow/scoped_coordinator.cpp
src/flow/subscription.cpp
src/forwarding_actor_proxy.cpp
......@@ -155,7 +155,6 @@ caf_add_component(
src/group_manager.cpp
src/group_module.cpp
src/hash/sha1.cpp
src/inbound_path.cpp
src/init_global_meta_objects.cpp
src/ipv4_address.cpp
src/ipv4_endpoint.cpp
......@@ -175,8 +174,6 @@ caf_add_component(
src/message_handler.cpp
src/monitorable_actor.cpp
src/node_id.cpp
src/outbound_path.cpp
src/policy/downstream_messages.cpp
src/policy/unprofiled.cpp
src/policy/work_sharing.cpp
src/policy/work_stealing.cpp
......@@ -195,8 +192,6 @@ caf_add_component(
src/serializer.cpp
src/settings.cpp
src/skip.cpp
src/stream_aborter.cpp
src/stream_manager.cpp
src/string_algorithms.cpp
src/string_view.cpp
src/telemetry/collector/prometheus.cpp
......@@ -235,7 +230,6 @@ caf_add_component(
binary_deserializer
binary_serializer
blocking_actor
broadcast_downstream_manager
byte
composition
config_option
......@@ -245,7 +239,6 @@ caf_add_component(
config_value_writer
const_typed_message_view
constructor_attach
continuous_streaming
cow_string
cow_tuple
decorator.sequencer
......@@ -279,12 +272,10 @@ caf_add_component(
detail.tick_emitter
detail.type_id_list_builder
detail.unique_function
detail.unordered_flat_map
dictionary
dynamic_spawn
error
expected
flow.broadcaster
flow.concat
flow.concat_map
flow.flat_map
......@@ -296,7 +287,6 @@ caf_add_component(
flow.single
flow.zip_with
function_view
fused_downstream_manager
handles
hash.fnv
hash.sha1
......@@ -326,14 +316,10 @@ caf_add_component(
metaprogramming
mixin.requester
mixin.sender
mock_streaming_classes
mtl
native_streaming_classes
node_id
optional
or_else
pipeline_streaming
policy.categorized
policy.select_all
policy.select_any
request_timeout
......@@ -342,7 +328,6 @@ caf_add_component(
result
save_inspector
scheduled_actor
selective_streaming
serial_reply
serialization
settings
......
......@@ -6,12 +6,13 @@
#include "caf/allowed_unsafe_message_type.hpp"
#include "caf/config.hpp"
#include "caf/detail/atomic_ref_counted.hpp"
#include "caf/detail/core_export.hpp"
#include "caf/disposable.hpp"
#include "caf/make_counted.hpp"
#include "caf/ref_counted.hpp"
#include <atomic>
#include <cstddef>
namespace caf {
......@@ -33,14 +34,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>;
......@@ -59,6 +52,11 @@ public:
action& operator=(const action&) noexcept = default;
action& operator=(std::nullptr_t) noexcept {
pimpl_ = nullptr;
return *this;
}
// -- observers --------------------------------------------------------------
[[nodiscard]] bool disposed() const {
......@@ -108,6 +106,14 @@ public:
return pimpl_;
}
explicit operator bool() const noexcept {
return static_cast<bool>(pimpl_);
}
[[nodiscard]] bool operator!() const noexcept {
return !pimpl_;
}
private:
impl_ptr pimpl_;
};
......@@ -115,8 +121,8 @@ private:
} // namespace caf
namespace caf::detail {
template <class F>
struct default_action_impl : ref_counted, action::impl {
template <class F, bool IsSingleShot>
struct default_action_impl : detail::atomic_ref_counted, action::impl {
std::atomic<action::state> state_;
F f_;
......@@ -138,11 +144,12 @@ struct default_action_impl : ref_counted, action::impl {
}
void run() override {
// Note: we do *not* set the state to disposed after running the function
// object. This allows the action to re-register itself when needed, e.g.,
// to implement time-based loops.
if (state_.load() == action::state::scheduled) {
f_();
if constexpr (IsSingleShot)
state_ = action::state::disposed;
// else: allow the action to re-register itself when needed by *not*
// setting the state to disposed, e.g., to implement time loops.
}
}
......@@ -171,7 +178,15 @@ namespace caf {
/// @param f The body for the action.
template <class F>
action make_action(F f) {
using impl_t = detail::default_action_impl<F>;
using impl_t = detail::default_action_impl<F, false>;
return action{make_counted<impl_t>(std::move(f))};
}
/// Convenience function for creating an @ref action from a function object.
/// @param f The body for the action.
template <class F>
action make_single_shot_action(F f) {
using impl_t = detail::default_action_impl<F, true>;
return action{make_counted<impl_t>(std::move(f))};
}
......
......@@ -19,11 +19,6 @@
#include "caf/actor_system_config.hpp"
#include "caf/actor_traits.hpp"
#include "caf/after.hpp"
#include "caf/attach_continuous_stream_source.hpp"
#include "caf/attach_continuous_stream_stage.hpp"
#include "caf/attach_stream_sink.hpp"
#include "caf/attach_stream_source.hpp"
#include "caf/attach_stream_stage.hpp"
#include "caf/attachable.hpp"
#include "caf/behavior.hpp"
#include "caf/behavior_policy.hpp"
......@@ -42,7 +37,6 @@
#include "caf/deep_to_string.hpp"
#include "caf/defaults.hpp"
#include "caf/deserializer.hpp"
#include "caf/downstream_msg.hpp"
#include "caf/error.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/exec_main.hpp"
......@@ -51,7 +45,6 @@
#include "caf/expected.hpp"
#include "caf/extend.hpp"
#include "caf/function_view.hpp"
#include "caf/fused_downstream_manager.hpp"
#include "caf/group.hpp"
#include "caf/hash/fnv.hpp"
#include "caf/init_global_meta_objects.hpp"
......@@ -83,7 +76,6 @@
#include "caf/spawn_options.hpp"
#include "caf/stateful_actor.hpp"
#include "caf/stream.hpp"
#include "caf/stream_slot.hpp"
#include "caf/system_messages.hpp"
#include "caf/term.hpp"
#include "caf/thread_hook.hpp"
......@@ -99,7 +91,6 @@
#include "caf/typed_message_view.hpp"
#include "caf/typed_response_promise.hpp"
#include "caf/unit.hpp"
#include "caf/upstream_msg.hpp"
#include "caf/uuid.hpp"
#include "caf/decorator/sequencer.hpp"
......
......@@ -11,7 +11,6 @@
#include <memory>
#include "caf/async/fwd.hpp"
#include "caf/byte.hpp"
#include "caf/config.hpp"
#include "caf/detail/core_export.hpp"
#include "caf/fwd.hpp"
......@@ -40,7 +39,7 @@ public:
template <class T>
friend batch make_batch(span<const T> items);
using item_destructor = void (*)(type_id_t, uint16_t, size_t, byte*);
using item_destructor = void (*)(type_id_t, uint16_t, size_t, std::byte*);
batch() = default;
batch(batch&&) = default;
......@@ -143,11 +142,11 @@ private:
return size_;
}
byte* storage() noexcept {
std::byte* storage() noexcept {
return storage_;
}
const byte* storage() const noexcept {
const std::byte* storage() const noexcept {
return storage_;
}
......@@ -176,7 +175,7 @@ private:
type_id_t item_type_;
uint16_t item_size_;
size_t size_;
byte storage_[];
std::byte storage_[];
};
explicit batch(intrusive_ptr<data> ptr) : data_(std::move(ptr)) {
......@@ -205,7 +204,8 @@ batch make_batch(span<const T> items) {
auto vptr = malloc(total_size);
if (vptr == nullptr)
CAF_RAISE_ERROR(std::bad_alloc, "make_batch");
auto destroy_items = [](type_id_t, uint16_t, size_t size, byte* storage) {
auto destroy_items = [](type_id_t, uint16_t, size_t size,
std::byte* storage) {
auto ptr = reinterpret_cast<T*>(storage);
std::destroy(ptr, ptr + size);
};
......
......@@ -33,20 +33,24 @@ public:
template <class ErrorPolicy, class TimePoint>
read_result pull(ErrorPolicy policy, T& item, TimePoint timeout) {
if (!buf_) {
return abort_reason_ ? read_result::abort : read_result::stop;
}
val_ = &item;
std::unique_lock guard{buf_->mtx()};
if constexpr (std::is_same_v<TimePoint, none_t>) {
buf_->await_consumer_ready(guard, cv_);
} else {
if (!buf_->await_consumer_ready(guard, cv_, timeout))
return read_result::timeout;
return read_result::try_again_later;
}
auto [again, n] = buf_->pull_unsafe(guard, policy, 1u, *this);
CAF_IGNORE_UNUSED(again);
CAF_ASSERT(!again || n == 1);
if (!again) {
buf_ = nullptr;
}
if (n == 1) {
return read_result::ok;
} else if (buf_->abort_reason_unsafe()) {
} else if (abort_reason_) {
return read_result::abort;
} else {
return read_result::stop;
......@@ -58,16 +62,15 @@ public:
return pull(policy, item, none);
}
void on_next(span<const T> items) {
CAF_ASSERT(items.size() == 1);
*val_ = items[0];
void on_next(const T& item) {
*val_ = item;
}
void on_complete() {
}
void on_error(const caf::error&) {
// nop
void on_error(const caf::error& abort_reason) {
abort_reason_ = abort_reason;
}
void cancel() {
......@@ -94,8 +97,8 @@ public:
deref();
}
error abort_reason() const {
buf_->abort_reason()();
const error& abort_reason() const {
return abort_reason_;
}
CAF_INTRUSIVE_PTR_FRIENDS(impl)
......@@ -104,6 +107,7 @@ public:
spsc_buffer_ptr<T> buf_;
std::condition_variable cv_;
T* val_ = nullptr;
error abort_reason_;
};
using impl_ptr = intrusive_ptr<impl>;
......@@ -122,6 +126,10 @@ public:
// nop
}
explicit blocking_consumer(spsc_buffer_ptr<T> buf) {
impl_.emplace(std::move(buf));
}
~blocking_consumer() {
if (impl_)
impl_->cancel();
......
......@@ -4,14 +4,15 @@
#pragma once
#include <condition_variable>
#include <type_traits>
#include "caf/async/producer.hpp"
#include "caf/async/spsc_buffer.hpp"
#include "caf/detail/atomic_ref_counted.hpp"
#include "caf/intrusive_ptr.hpp"
#include "caf/make_counted.hpp"
#include "caf/ref_counted.hpp"
#include <condition_variable>
#include <optional>
#include <type_traits>
namespace caf::async {
......@@ -19,7 +20,7 @@ namespace caf::async {
template <class T>
class blocking_producer {
public:
class impl : public ref_counted, public producer {
class impl : public detail::atomic_ref_counted, public producer {
public:
impl() = delete;
impl(const impl&) = delete;
......@@ -66,7 +67,7 @@ public:
}
}
bool cancelled() const {
bool canceled() const {
std::unique_lock<std::mutex> guard{mtx_};
return demand_ == -1;
}
......@@ -99,8 +100,6 @@ public:
deref();
}
CAF_INTRUSIVE_PTR_FRIENDS(impl)
private:
spsc_buffer_ptr<T> buf_;
mutable std::mutex mtx_;
......@@ -124,6 +123,10 @@ public:
// nop
}
explicit blocking_producer(spsc_buffer_ptr<T> buf) {
impl_.emplace(std::move(buf));
}
~blocking_producer() {
if (impl_)
impl_->close();
......@@ -160,23 +163,35 @@ public:
}
}
/// Checks whether the consumer cancelled its subscription.
bool cancelled() const {
return impl_->cancelled();
/// Checks whether the consumer canceled its subscription.
bool canceled() const {
return impl_->canceled();
}
explicit operator bool() const noexcept {
return static_cast<bool>(impl_);
}
private:
intrusive_ptr<impl> impl_;
};
/// @pre `buf != nullptr`
/// @relates blocking_producer
template <class T>
blocking_producer<T> make_blocking_producer(spsc_buffer_ptr<T> buf) {
using impl_t = typename blocking_producer<T>::impl;
return blocking_producer<T>{make_counted<impl_t>(std::move(buf))};
}
/// @relates blocking_producer
template <class T>
expected<blocking_producer<T>>
std::optional<blocking_producer<T>>
make_blocking_producer(producer_resource<T> res) {
if (auto buf = res.try_open()) {
using impl_t = typename blocking_producer<T>::impl;
return {blocking_producer<T>{make_counted<impl_t>(std::move(buf))}};
return {make_blocking_producer(std::move(buf))};
} else {
return {make_error(sec::cannot_open_resource)};
return {};
}
}
......
// 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 "caf/async/consumer.hpp"
#include "caf/async/execution_context.hpp"
#include "caf/async/read_result.hpp"
#include "caf/async/spsc_buffer.hpp"
#include "caf/detail/atomic_ref_counted.hpp"
#include "caf/intrusive_ptr.hpp"
#include "caf/make_counted.hpp"
#include <optional>
namespace caf::async {
/// Integrates an SPSC buffer consumer into an asynchronous event loop.
template <class T>
class consumer_adapter {
public:
class impl : public detail::atomic_ref_counted, public consumer {
public:
impl() = delete;
impl(const impl&) = delete;
impl& operator=(const impl&) = delete;
impl(spsc_buffer_ptr<T> buf, execution_context_ptr ctx, action do_wakeup)
: buf_(std::move(buf)),
ctx_(std::move(ctx)),
do_wakeup_(std::move(do_wakeup)) {
buf_->set_consumer(this);
}
~impl() {
if (buf_) {
buf_->cancel();
do_wakeup_.dispose();
}
}
void cancel() {
if (buf_) {
buf_->cancel();
buf_ = nullptr;
do_wakeup_.dispose();
do_wakeup_ = nullptr;
}
}
template <class ErrorPolicy>
read_result pull(ErrorPolicy policy, T& item) {
if (buf_) {
val_ = &item;
auto [again, n] = buf_->pull(policy, 1u, *this);
if (!again) {
buf_ = nullptr;
}
if (n == 1) {
return read_result::ok;
} else if (again) {
CAF_ASSERT(n == 0);
return read_result::try_again_later;
} else {
CAF_ASSERT(n == 0);
return abort_reason_ ? read_result::abort : read_result::stop;
}
} else {
return abort_reason_ ? read_result::abort : read_result::stop;
}
}
const error& abort_reason() const noexcept {
return abort_reason_;
}
bool has_data() const noexcept {
return buf_ ? buf_->has_data() : false;
}
bool has_consumer_event() const noexcept {
return buf_ ? buf_->has_consumer_event() : false;
}
void on_next(const T& item) {
*val_ = item;
}
void on_complete() {
// nop
}
void on_error(const caf::error& what) {
abort_reason_ = what;
}
void on_producer_ready() override {
// nop
}
void on_producer_wakeup() override {
ctx_->schedule(do_wakeup_);
}
void ref_consumer() const noexcept override {
ref();
}
void deref_consumer() const noexcept override {
deref();
}
private:
spsc_buffer_ptr<T> buf_;
execution_context_ptr ctx_;
action do_wakeup_;
error abort_reason_;
T* val_ = nullptr;
};
using impl_ptr = intrusive_ptr<impl>;
consumer_adapter() = default;
consumer_adapter(const consumer_adapter&) = delete;
consumer_adapter& operator=(const consumer_adapter&) = delete;
consumer_adapter(consumer_adapter&&) = default;
consumer_adapter& operator=(consumer_adapter&&) = default;
explicit consumer_adapter(impl_ptr ptr) : impl_(std::move(ptr)) {
// nop
}
~consumer_adapter() {
if (impl_)
impl_->cancel();
}
consumer_adapter& operator=(std::nullptr_t) {
impl_ = nullptr;
return *this;
}
template <class Policy>
read_result pull(Policy policy, T& result) {
if (impl_)
return impl_->pull(policy, result);
else
return read_result::abort;
}
void cancel() {
if (impl_) {
impl_->cancel();
impl_ = nullptr;
}
}
error abort_reason() const noexcept {
if (impl_)
return impl_->abort_reason();
else
return make_error(sec::disposed);
}
explicit operator bool() const noexcept {
return static_cast<bool>(impl_);
}
bool has_data() const noexcept {
return impl_ ? impl_->has_data() : false;
}
bool has_consumer_event() const noexcept {
return impl_ ? impl_->has_consumer_event() : false;
}
static consumer_adapter make(spsc_buffer_ptr<T> buf,
execution_context_ptr ctx, action do_wakeup) {
if (buf) {
using impl_t = typename consumer_adapter<T>::impl;
auto impl = make_counted<impl_t>(std::move(buf), std::move(ctx),
std::move(do_wakeup));
return consumer_adapter<T>{std::move(impl)};
} else {
return {};
}
}
static std::optional<consumer_adapter>
make(consumer_resource<T> res, execution_context_ptr ctx, action do_wakeup) {
if (auto buf = res.try_open()) {
return {make(std::move(buf), std::move(ctx), std::move(do_wakeup))};
} else {
return {};
}
}
private:
intrusive_ptr<impl> impl_;
};
/// @pre `buf != nullptr`
/// @relates consumer_adapter
template <class T>
consumer_adapter<T> make_consumer_adapter(spsc_buffer_ptr<T> buf,
execution_context_ptr ctx,
action do_wakeup) {
return consumer_adapter<T>::make(std::move(buf), std::move(ctx),
std::move(do_wakeup));
}
/// @relates consumer_adapter
template <class T>
std::optional<consumer_adapter<T>>
make_consumer_adapter(consumer_resource<T> res, execution_context_ptr ctx,
action do_wakeup) {
if (auto buf = res.try_open()) {
return {consumer_adapter<T>::make(std::move(buf), std::move(ctx),
std::move(do_wakeup))};
} else {
return {};
}
}
} // namespace caf::async
// 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 "caf/action.hpp"
#include "caf/detail/core_export.hpp"
#include "caf/intrusive_ptr.hpp"
#include <type_traits>
namespace caf::async {
/// Represents a single execution context with an internal event-loop to
/// schedule @ref action objects.
class CAF_CORE_EXPORT execution_context {
public:
// -- constructors, destructors, and assignment operators --------------------
virtual ~execution_context();
// -- reference counting -----------------------------------------------------
/// Increases the reference count of the execution_context.
virtual void ref_execution_context() const noexcept = 0;
/// Decreases the reference count of the execution context and destroys the
/// object if necessary.
virtual void deref_execution_context() const noexcept = 0;
// -- scheduling of actions --------------------------------------------------
/// Schedules @p what to run on the event loop of the execution context. This
/// member function may get called from external sources or threads.
/// @thread-safe
virtual void schedule(action what) = 0;
///@copydoc schedule
template <class F>
void schedule_fn(F&& what) {
static_assert(std::is_invocable_v<F>);
return schedule(make_action(std::forward<F>(what)));
}
// -- lifetime management ----------------------------------------------------
/// Asks the coordinator to keep its event loop running until @p what becomes
/// disposed since it depends on external events or produces events that are
/// visible to outside observers. Must be called from within the event loop of
/// the execution context.
virtual void watch(disposable what) = 0;
};
/// @relates execution_context
inline void intrusive_ptr_add_ref(const execution_context* ptr) noexcept {
ptr->ref_execution_context();
}
/// @relates execution_context
inline void intrusive_ptr_release(const execution_context* ptr) noexcept {
ptr->deref_execution_context();
}
/// @relates execution_context
using execution_context_ptr = intrusive_ptr<execution_context>;
} // namespace caf::async
// 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 "caf/async/execution_context.hpp"
#include "caf/async/fwd.hpp"
#include "caf/detail/async_cell.hpp"
#include "caf/disposable.hpp"
#include "caf/error.hpp"
#include "caf/sec.hpp"
namespace caf::async {
/// Provides an interface for accessing the result of an asynchronous
/// computation on an asynchronous @ref execution_context.
template <class T>
class bound_future {
public:
friend class future<T>;
bound_future() noexcept = default;
bound_future(bound_future&&) noexcept = default;
bound_future(const bound_future&) noexcept = default;
bound_future& operator=(bound_future&&) noexcept = default;
bound_future& operator=(const bound_future&) noexcept = default;
/// Retrieves the result at some point in the future and then calls either
/// @p on_success if the asynchronous operation generated a result or
/// @p on_error if the asynchronous operation resulted in an error.
template <class OnSuccess, class OnError>
disposable then(OnSuccess on_success, OnError on_error) {
static_assert(std::is_invocable_v<OnSuccess, const T&>);
static_assert(std::is_invocable_v<OnError, const error&>);
auto cb = [cp = cell_, f = std::move(on_success),
g = std::move(on_error)]() mutable {
// Note: no need to lock the mutex. Once the cell has a value and actions
// are allowed to run, the value is immutable.
switch (cp->value.index()) {
default:
g(make_error(sec::broken_promise, "future found an invalid value"));
break;
case 1:
f(static_cast<const T&>(std::get<T>(cp->value)));
break;
case 2:
g(static_cast<const error&>(std::get<error>(cp->value)));
}
};
auto cb_action = make_single_shot_action(std::move(cb));
auto event = typename cell_type::event{ctx_, cb_action};
bool fire_immediately = false;
{ // Critical section.
std::unique_lock guard{cell_->mtx};
if (std::holds_alternative<none_t>(cell_->value)) {
cell_->events.push_back(std::move(event));
} else {
fire_immediately = true;
}
}
if (fire_immediately)
event.first->schedule(std::move(event.second));
auto res = std::move(cb_action).as_disposable();
ctx_->watch(res);
return res;
}
private:
using cell_type = detail::async_cell<T>;
using cell_ptr = std::shared_ptr<cell_type>;
bound_future(execution_context* ctx, cell_ptr cell)
: ctx_(ctx), cell_(std::move(cell)) {
// nop
}
execution_context* ctx_;
cell_ptr cell_;
};
/// Represents the result of an asynchronous computation.
template <class T>
class future {
public:
friend class promise<T>;
future() noexcept = default;
future(future&&) noexcept = default;
future(const future&) noexcept = default;
future& operator=(future&&) noexcept = default;
future& operator=(const future&) noexcept = default;
bool valid() const noexcept {
return cell_ != nullptr;
}
explicit operator bool() const noexcept {
return valid();
}
bool operator!() const noexcept {
return !valid();
}
/// Binds this future to an @ref execution_context to run callbacks.
/// @pre `valid()`
bound_future<T> bind_to(execution_context* ctx) && {
return {ctx, std::move(cell_)};
}
/// Binds this future to an @ref execution_context to run callbacks.
/// @pre `valid()`
bound_future<T> bind_to(execution_context& ctx) && {
return {&ctx, std::move(cell_)};
}
/// Binds this future to an @ref execution_context to run callbacks.
/// @pre `valid()`
bound_future<T> bind_to(execution_context* ctx) const& {
return {ctx, cell_};
}
/// Binds this future to an @ref execution_context to run callbacks.
/// @pre `valid()`
bound_future<T> bind_to(execution_context& ctx) const& {
return {&ctx, cell_};
}
/// Queries whether the result of the asynchronous computation is still
/// pending, i.e., neither `set_value` nor `set_error` has been called on the
/// @ref promise.
/// @pre `valid()`
bool pending() const {
CAF_ASSERT(valid());
std::unique_lock guard{cell_->mtx};
return std::holds_alternative<none_t>(cell_->value);
}
private:
using cell_ptr = std::shared_ptr<detail::async_cell<T>>;
explicit future(cell_ptr cell) : cell_(std::move(cell)) {
// nop
}
cell_ptr cell_;
};
} // namespace caf::async
......@@ -4,7 +4,7 @@
#pragma once
#include <caf/fwd.hpp>
#include "caf/fwd.hpp"
namespace caf::async {
......@@ -12,6 +12,7 @@ namespace caf::async {
class batch;
class consumer;
class execution_context;
class producer;
// -- template classes ---------------------------------------------------------
......@@ -25,6 +26,16 @@ class consumer_resource;
template <class T>
class producer_resource;
template <class T>
class promise;
template <class T>
class future;
// -- smart pointer aliases ----------------------------------------------------
using execution_context_ptr = intrusive_ptr<execution_context>;
// -- free function templates --------------------------------------------------
template <class T>
......
// 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 "caf/async/execution_context.hpp"
#include "caf/async/producer.hpp"
#include "caf/async/spsc_buffer.hpp"
#include "caf/detail/atomic_ref_counted.hpp"
#include "caf/intrusive_ptr.hpp"
#include "caf/make_counted.hpp"
#include <optional>
namespace caf::async {
/// Integrates an SPSC buffer producer into an asynchronous event loop.
template <class T>
class producer_adapter {
public:
class impl : public detail::atomic_ref_counted, public producer {
public:
impl() = delete;
impl(const impl&) = delete;
impl& operator=(const impl&) = delete;
impl(spsc_buffer_ptr<T> buf, execution_context_ptr ctx, action do_resume,
action do_cancel)
: buf_(std::move(buf)),
ctx_(std::move(ctx)),
do_resume_(std::move(do_resume)),
do_cancel_(std::move(do_cancel)) {
buf_->set_producer(this);
}
size_t push(span<const T> items) {
return buf_->push(items);
}
bool push(const T& item) {
return buf_->push(item);
}
void close() {
if (buf_) {
buf_->close();
buf_ = nullptr;
do_resume_.dispose();
do_resume_ = nullptr;
do_cancel_.dispose();
do_cancel_ = nullptr;
}
}
void abort(error reason) {
if (buf_) {
buf_->abort(std::move(reason));
buf_ = nullptr;
do_resume_.dispose();
do_resume_ = nullptr;
do_cancel_.dispose();
do_cancel_ = nullptr;
}
}
void on_consumer_ready() override {
// nop
}
void on_consumer_cancel() override {
ctx_->schedule(do_cancel_);
}
void on_consumer_demand(size_t) override {
ctx_->schedule(do_resume_);
}
void ref_producer() const noexcept override {
ref();
}
void deref_producer() const noexcept override {
deref();
}
private:
spsc_buffer_ptr<T> buf_;
execution_context_ptr ctx_;
action do_resume_;
action do_cancel_;
};
using impl_ptr = intrusive_ptr<impl>;
producer_adapter() = default;
producer_adapter(const producer_adapter&) = delete;
producer_adapter& operator=(const producer_adapter&) = delete;
producer_adapter(producer_adapter&&) = default;
producer_adapter& operator=(producer_adapter&&) = default;
explicit producer_adapter(impl_ptr ptr) : impl_(std::move(ptr)) {
// nop
}
producer_adapter& operator=(std::nullptr_t) {
impl_ = nullptr;
return *this;
}
~producer_adapter() {
if (impl_)
impl_->close();
}
/// Makes `item` available to the consumer.
/// @returns the remaining demand.
size_t push(const T& item) {
if (!impl_)
CAF_RAISE_ERROR("cannot push to a closed producer adapter");
return impl_->push(item);
}
/// Makes `items` available to the consumer.
/// @returns the remaining demand.
size_t push(span<const T> items) {
if (!impl_)
CAF_RAISE_ERROR("cannot push to a closed producer adapter");
return impl_->push(items);
}
void close() {
if (impl_) {
impl_->close();
impl_ = nullptr;
}
}
void abort(error reason) {
if (impl_) {
impl_->abort(std::move(reason));
impl_ = nullptr;
}
}
explicit operator bool() const noexcept {
return static_cast<bool>(impl_);
}
/// @pre `buf != nullptr`
static producer_adapter make(spsc_buffer_ptr<T> buf,
execution_context_ptr ctx, action do_resume,
action do_cancel) {
if (buf) {
using impl_t = typename producer_adapter<T>::impl;
auto impl = make_counted<impl_t>(std::move(buf), std::move(ctx),
std::move(do_resume),
std::move(do_cancel));
return producer_adapter<T>{std::move(impl)};
} else {
return {};
}
}
static std::optional<producer_adapter>
make(producer_resource<T> res, execution_context_ptr ctx, action do_resume,
action do_cancel) {
if (auto buf = res.try_open()) {
return {make(std::move(buf), std::move(ctx), std::move(do_resume),
std::move(do_cancel))};
} else {
return {};
}
}
private:
intrusive_ptr<impl> impl_;
};
/// @pre `buf != nullptr`
/// @relates producer_adapter
template <class T>
producer_adapter<T> make_producer_adapter(spsc_buffer_ptr<T> buf,
execution_context_ptr ctx,
action do_resume, action do_cancel) {
return producer_adapter<T>::make(std::move(buf), std::move(ctx),
std::move(do_resume), std::move(do_cancel));
}
/// @relates producer_adapter
template <class T>
std::optional<producer_adapter<T>>
make_producer_adapter(producer_resource<T> res, execution_context_ptr ctx,
action do_resume, action do_cancel) {
if (auto buf = res.try_open()) {
return {producer_adapter<T>::make(std::move(buf), std::move(ctx),
std::move(do_resume),
std::move(do_cancel))};
} else {
return {};
}
}
} // namespace caf::async
// 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 "caf/async/execution_context.hpp"
#include "caf/async/future.hpp"
#include "caf/detail/async_cell.hpp"
#include "caf/disposable.hpp"
#include "caf/raise_error.hpp"
namespace caf::async {
/// Provides a facility to store a value or an error that is later acquired
/// asynchronously via a @ref future object. A promise may deliver only one
/// value.
template <class T>
class promise {
public:
promise(promise&&) noexcept = default;
promise& operator=(promise&&) noexcept = default;
promise(const promise& other) noexcept : promise(other.cell_) {
// nop
}
promise& operator=(const promise& other) noexcept {
promise copy{other};
cell_.swap(copy.cell_);
return *this;
}
promise() : cell_(std::make_shared<cell_type>()) {
// nop
}
~promise() {
if (valid()) {
auto& cnt = cell_->promises;
if (cnt == 1 || cnt.fetch_sub(1, std::memory_order_acq_rel) == 1) {
typename cell_type::event_list events;
{ // Critical section.
std::unique_lock guard{cell_->mtx};
if (std::holds_alternative<none_t>(cell_->value)) {
cell_->value = make_error(sec::broken_promise);
cell_->events.swap(events);
}
}
for (auto& [listener, callback] : events)
listener->schedule(std::move(callback));
}
}
}
bool valid() const noexcept {
return cell_ != nullptr;
}
explicit operator bool() const noexcept {
return valid();
}
bool operator!() const noexcept {
return !valid();
}
/// @pre `valid()`
void set_value(T value) {
if (valid()) {
do_set(value);
cell_ = nullptr;
}
}
/// @pre `valid()`
void set_error(error reason) {
if (valid()) {
do_set(reason);
cell_ = nullptr;
}
}
/// @pre `valid()`
future<T> get_future() const {
return future<T>{cell_};
}
private:
using cell_type = detail::async_cell<T>;
using cell_ptr = std::shared_ptr<cell_type>;
explicit promise(cell_ptr cell) noexcept : cell_(std::move(cell)) {
CAF_ASSERT(cell_ != nullptr);
cell_->promises.fetch_add(1, std::memory_order_relaxed);
}
template <class What>
void do_set(What& what) {
typename cell_type::event_list events;
{ // Critical section.
std::unique_lock guard{cell_->mtx};
if (std::holds_alternative<none_t>(cell_->value)) {
cell_->value = std::move(what);
cell_->events.swap(events);
} else {
CAF_RAISE_ERROR("promise already satisfied");
}
}
for (auto& [listener, callback] : events)
listener->schedule(std::move(callback));
}
cell_ptr cell_;
};
} // namespace caf::async
......@@ -7,6 +7,7 @@
#include "caf/default_enum_inspect.hpp"
#include "caf/detail/core_export.hpp"
#include <string_view>
#include <type_traits>
namespace caf::async {
......@@ -19,15 +20,16 @@ enum class read_result {
stop,
/// Signals that the source failed with an error.
abort,
/// Signals that the read operation timed out.
timeout,
/// Signals that the read operation cannot produce a result at the moment,
/// e.g., because of a timeout.
try_again_later,
};
/// @relates read_result
CAF_CORE_EXPORT std::string to_string(read_result);
/// @relates read_result
CAF_CORE_EXPORT bool from_string(string_view, read_result&);
CAF_CORE_EXPORT bool from_string(std::string_view, read_result&);
/// @relates read_result
CAF_CORE_EXPORT bool from_integer(std::underlying_type_t<read_result>,
......
......@@ -62,7 +62,7 @@ public:
buf_.insert(buf_.end(), items.begin(), items.end());
if (buf_.size() == items.size() && consumer_)
consumer_->on_producer_wakeup();
if (capacity_ >= buf_.size())
if (capacity_ > buf_.size())
return capacity_ - buf_.size();
else
return 0;
......@@ -235,20 +235,18 @@ public:
consumer_buf_.assign(make_move_iterator(buf_.begin()),
make_move_iterator(buf_.begin() + n));
buf_.erase(buf_.begin(), buf_.begin() + n);
if (overflow == 0) {
signal_demand(n);
} else if (n <= overflow) {
overflow -= n;
} else {
if (n > overflow) {
signal_demand(n - overflow);
overflow = 0;
}
guard.unlock();
dst.on_next(span<const T>{consumer_buf_.data(), n});
auto items = span<const T>{consumer_buf_.data(), n};
for (auto& item : items)
dst.on_next(item);
demand -= n;
consumed += n;
consumer_buf_.clear();
guard.lock();
overflow = buf_.size() <= capacity_ ? 0u : buf_.size() - capacity_;
}
if (!buf_.empty() || !closed_) {
return {true, consumed};
......@@ -266,9 +264,13 @@ private:
void ready() {
producer_->on_consumer_ready();
consumer_->on_producer_ready();
if (!buf_.empty())
if (!buf_.empty()) {
consumer_->on_producer_wakeup();
signal_demand(capacity_);
if (capacity_ > buf_.size())
signal_demand(capacity_ - buf_.size());
} else {
signal_demand(capacity_);
}
}
void signal_demand(uint32_t new_demand) {
......@@ -327,7 +329,7 @@ struct resource_ctrl : ref_counted {
~resource_ctrl() {
if (buf) {
if constexpr (IsProducer) {
auto err = make_error(sec::invalid_upstream,
auto err = make_error(sec::disposed,
"producer_resource destroyed without opening it");
buf->abort(err);
} else {
......@@ -395,11 +397,19 @@ public:
}
}
/// Convenience function for calling
/// `ctx->make_observable().from_resource(*this)`.
template <class Coordinator>
auto observe_on(Coordinator* ctx) {
auto observe_on(Coordinator* ctx) const {
return ctx->make_observable().from_resource(*this);
}
/// Calls `try_open` and on success immediately calls `cancel` on the buffer.
void cancel() {
if (auto buf = try_open())
buf->cancel();
}
explicit operator bool() const noexcept {
return ctrl_ != nullptr;
}
......@@ -452,6 +462,12 @@ public:
}
}
/// Calls `try_open` and on success immediately calls `close` on the buffer.
void close() {
if (auto buf = try_open())
buf->close();
}
explicit operator bool() const noexcept {
return ctrl_ != nullptr;
}
......@@ -460,19 +476,23 @@ private:
intrusive_ptr<resource_ctrl<T, true>> ctrl_;
};
/// Creates spsc buffer and returns two resources connected by that buffer.
template <class T1, class T2 = T1>
using resource_pair = std::pair<consumer_resource<T1>, producer_resource<T2>>;
/// Creates an @ref spsc_buffer and returns two resources connected by that
/// buffer.
template <class T>
std::pair<consumer_resource<T>, producer_resource<T>>
resource_pair<T>
make_spsc_buffer_resource(size_t buffer_size, size_t min_request_size) {
using buffer_type = spsc_buffer<T>;
auto buf = make_counted<buffer_type>(buffer_size, min_request_size);
return {async::consumer_resource<T>{buf}, async::producer_resource<T>{buf}};
}
/// Creates spsc buffer and returns two resources connected by that buffer.
/// Creates an @ref spsc_buffer and returns two resources connected by that
/// buffer.
template <class T>
std::pair<consumer_resource<T>, producer_resource<T>>
make_spsc_buffer_resource() {
resource_pair<T> make_spsc_buffer_resource() {
return make_spsc_buffer_resource<T>(defaults::flow::buffer_size,
defaults::flow::min_demand);
}
......
......@@ -7,6 +7,7 @@
#include "caf/default_enum_inspect.hpp"
#include "caf/detail/core_export.hpp"
#include <string_view>
#include <type_traits>
namespace caf::async {
......@@ -28,7 +29,7 @@ enum class write_result {
CAF_CORE_EXPORT std::string to_string(write_result);
/// @relates write_result
CAF_CORE_EXPORT bool from_string(string_view, write_result&);
CAF_CORE_EXPORT bool from_string(std::string_view, write_result&);
/// @relates write_result
CAF_CORE_EXPORT bool from_integer(std::underlying_type_t<write_result>,
......
// 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 "caf/broadcast_downstream_manager.hpp"
#include "caf/detail/stream_source_driver_impl.hpp"
#include "caf/detail/stream_source_impl.hpp"
#include "caf/fwd.hpp"
#include "caf/policy/arg.hpp"
#include "caf/stream_source.hpp"
#include "caf/stream_source_driver.hpp"
#include "caf/stream_source_trait.hpp"
namespace caf {
/// Creates a new continuous stream source by instantiating the default source
/// implementation with `Driver`. The returned manager is not connected to any
/// slot and thus not stored by the actor automatically.
/// @param self Points to the hosting actor.
/// @param xs Parameter pack for constructing the driver.
/// @returns The new `stream_manager`.
template <class Driver, class... Ts>
typename Driver::source_ptr_type
attach_continuous_stream_source(scheduled_actor* self, Ts&&... xs) {
using detail::make_stream_source;
auto mgr = make_stream_source<Driver>(self, std::forward<Ts>(xs)...);
mgr->continuous(true);
return mgr;
}
/// Creates a new continuous stream source by instantiating the default source
/// implementation with `Driver`. The returned manager is not connected to any
/// slot and thus not stored by the actor automatically.
/// @param self Points to the hosting actor.
/// @param init Function object for initializing the state of the source.
/// @param pull Generator function object for producing downstream messages.
/// @param done Predicate returning `true` when generator is done.
/// @param fin Cleanup handler.
/// @returns The new `stream_manager`.
template <class Init, class Pull, class Done, class Finalize = unit_t,
class Trait = stream_source_trait_t<Pull>,
class DownstreamManager = broadcast_downstream_manager<
typename Trait::output>>
stream_source_ptr<DownstreamManager>
attach_continuous_stream_source(scheduled_actor* self, Init init, Pull pull,
Done done, Finalize fin = {},
policy::arg<DownstreamManager> = {}) {
using state_type = typename Trait::state;
static_assert(std::is_same<
void(state_type&),
typename detail::get_callable_trait<Init>::fun_sig>::value,
"Expected signature `void (State&)` for init function");
static_assert(std::is_same<
bool(const state_type&),
typename detail::get_callable_trait<Done>::fun_sig>::value,
"Expected signature `bool (const State&)` "
"for done predicate function");
using driver = detail::stream_source_driver_impl<DownstreamManager, Pull,
Done, Finalize>;
return attach_continuous_stream_source<driver>(self, std::move(init),
std::move(pull),
std::move(done),
std::move(fin));
}
} // namespace caf
// 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 "caf/default_downstream_manager.hpp"
#include "caf/detail/stream_stage_driver_impl.hpp"
#include "caf/detail/stream_stage_impl.hpp"
#include "caf/downstream_manager.hpp"
#include "caf/fwd.hpp"
#include "caf/make_stage_result.hpp"
#include "caf/policy/arg.hpp"
#include "caf/stream.hpp"
#include "caf/stream_stage.hpp"
#include "caf/unit.hpp"
namespace caf {
/// Returns a stream manager (implementing a continuous stage) without in- or
/// outbound path. The returned manager is not connected to any slot and thus
/// not stored by the actor automatically.
/// @param self Points to the hosting actor.
/// @param xs User-defined arguments for the downstream handshake.
/// @returns The new `stream_manager`.
template <class Driver, class... Ts>
typename Driver::stage_ptr_type
attach_continuous_stream_stage(scheduled_actor* self, Ts&&... xs) {
auto ptr = detail::make_stream_stage<Driver>(self, std::forward<Ts>(xs)...);
ptr->continuous(true);
return ptr;
}
/// @param self Points to the hosting actor.
/// @param init Function object for initializing the state of the stage.
/// @param fun Processing function.
/// @param fin Optional cleanup handler.
/// @param token Policy token for selecting a downstream manager
/// implementation.
template <class Init, class Fun, class Finalize = unit_t,
class DownstreamManager = default_downstream_manager_t<Fun>,
class Trait = stream_stage_trait_t<Fun>>
stream_stage_ptr<typename Trait::input, DownstreamManager>
attach_continuous_stream_stage(scheduled_actor* self, Init init, Fun fun,
Finalize fin = {},
policy::arg<DownstreamManager> token = {}) {
CAF_IGNORE_UNUSED(token);
using input_type = typename Trait::input;
using output_type = typename Trait::output;
using state_type = typename Trait::state;
static_assert(std::is_same<
void(state_type&),
typename detail::get_callable_trait<Init>::fun_sig>::value,
"Expected signature `void (State&)` for init function");
static_assert(std::is_same<
void(state_type&, downstream<output_type>&, input_type),
typename detail::get_callable_trait<Fun>::fun_sig>::value,
"Expected signature `void (State&, downstream<Out>&, In)` "
"for consume function");
using detail::stream_stage_driver_impl;
using driver = stream_stage_driver_impl<typename Trait::input,
DownstreamManager, Fun, Finalize>;
return attach_continuous_stream_stage<driver>(self, std::move(init),
std::move(fun), std::move(fin));
}
} // namespace caf
// 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 "caf/detail/stream_sink_driver_impl.hpp"
#include "caf/detail/stream_sink_impl.hpp"
#include "caf/fwd.hpp"
#include "caf/make_sink_result.hpp"
#include "caf/policy/arg.hpp"
#include "caf/stream.hpp"
#include "caf/stream_sink.hpp"
namespace caf {
/// Attaches a new stream sink to `self` by creating a default stream sink /
/// manager from given callbacks.
/// @param self Points to the hosting actor.
/// @param xs Additional constructor arguments for `Driver`.
/// @returns The new `stream_manager`, an inbound slot, and an outbound slot.
template <class Driver, class... Ts>
make_sink_result<typename Driver::input_type>
attach_stream_sink(scheduled_actor* self,
stream<typename Driver::input_type> in, Ts&&... xs) {
auto mgr = detail::make_stream_sink<Driver>(self, std::forward<Ts>(xs)...);
auto slot = mgr->add_inbound_path(in);
return {slot, std::move(mgr)};
}
/// Attaches a new stream sink to `self` by creating a default stream sink
/// manager from given callbacks.
/// @param self Points to the hosting actor.
/// @param in Stream handshake from upstream path.
/// @param init Function object for initializing the state of the sink.
/// @param fun Processing function.
/// @param fin Optional cleanup handler.
/// @returns The new `stream_manager` and the inbound slot.
template <class In, class Init, class Fun, class Finalize = unit_t,
class Trait = stream_sink_trait_t<Fun>>
make_sink_result<In> attach_stream_sink(scheduled_actor* self, stream<In> in,
Init init, Fun fun, Finalize fin = {}) {
using driver = detail::stream_sink_driver_impl<In, Fun, Finalize>;
return attach_stream_sink<driver>(self, in, std::move(init), std::move(fun),
std::move(fin));
}
} // namespace caf
// 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 <tuple>
#include "caf/broadcast_downstream_manager.hpp"
#include "caf/check_typed_input.hpp"
#include "caf/default_downstream_manager.hpp"
#include "caf/detail/implicit_conversions.hpp"
#include "caf/detail/stream_source_driver_impl.hpp"
#include "caf/detail/stream_source_impl.hpp"
#include "caf/detail/type_list.hpp"
#include "caf/detail/type_traits.hpp"
#include "caf/downstream_manager.hpp"
#include "caf/fwd.hpp"
#include "caf/is_actor_handle.hpp"
#include "caf/make_source_result.hpp"
#include "caf/policy/arg.hpp"
#include "caf/response_type.hpp"
#include "caf/stream_source.hpp"
namespace caf {
/// Attaches a new stream source to `self` by creating a default stream source
/// manager with `Driver`.
/// @param self Points to the hosting actor.
/// @param xs User-defined arguments for the stream handshake.
/// @param ctor_args Parameter pack for constructing the driver.
/// @returns The allocated `stream_manager` and the output slot.
template <class Driver, class... Ts, class... CtorArgs>
make_source_result_t<typename Driver::downstream_manager_type, Ts...>
attach_stream_source(scheduled_actor* self, std::tuple<Ts...> xs,
CtorArgs&&... ctor_args) {
using namespace detail;
auto mgr = make_stream_source<Driver>(self,
std::forward<CtorArgs>(ctor_args)...);
auto slot = mgr->add_outbound_path(std::move(xs));
return {slot, std::move(mgr)};
}
/// Attaches a new stream source to `self` by creating a default stream source
/// manager with the default driver.
/// @param self Points to the hosting actor.
/// @param xs User-defined arguments for the stream handshake.
/// @param init Function object for initializing the state of the source.
/// @param pull Generator function object for producing downstream messages.
/// @param done Predicate returning `true` when generator is done.
/// @param fin Cleanup handler.
/// @returns The allocated `stream_manager` and the output slot.
template <class... Ts, class Init, class Pull, class Done,
class Finalize = unit_t, class Trait = stream_source_trait_t<Pull>,
class DownstreamManager = broadcast_downstream_manager<
typename Trait::output>>
make_source_result_t<DownstreamManager, Ts...>
attach_stream_source(scheduled_actor* self, std::tuple<Ts...> xs, Init init,
Pull pull, Done done, Finalize fin = {},
policy::arg<DownstreamManager> = {}) {
using state_type = typename Trait::state;
static_assert(std::is_same<
void(state_type&),
typename detail::get_callable_trait<Init>::fun_sig>::value,
"Expected signature `void (State&)` for init function");
static_assert(std::is_same<
bool(const state_type&),
typename detail::get_callable_trait<Done>::fun_sig>::value,
"Expected signature `bool (const State&)` "
"for done predicate function");
using driver = detail::stream_source_driver_impl<DownstreamManager, Pull,
Done, Finalize>;
return attach_stream_source<driver>(self, std::move(xs), std::move(init),
std::move(pull), std::move(done),
std::move(fin));
}
/// Attaches a new stream source to `self` by creating a default stream source
/// manager with the default driver.
/// @param self Points to the hosting actor.
/// @param init Function object for initializing the state of the source.
/// @param pull Generator function object for producing downstream messages.
/// @param done Predicate returning `true` when generator is done.
/// @param fin Cleanup handler.
/// @returns The allocated `stream_manager` and the output slot.
template <class Init, class Pull, class Done, class Finalize = unit_t,
class DownstreamManager = default_downstream_manager_t<Pull>,
class Trait = stream_source_trait_t<Pull>>
detail::enable_if_t<!is_actor_handle<Init>::value && Trait::valid,
make_source_result_t<DownstreamManager>>
attach_stream_source(scheduled_actor* self, Init init, Pull pull, Done done,
Finalize fin = {},
policy::arg<DownstreamManager> token = {}) {
using output_type = typename Trait::output;
static_assert(detail::sendable<output_type>,
"the output type of the source has has no type ID, "
"did you forgot to announce it via CAF_ADD_TYPE_ID?");
static_assert(detail::sendable<stream<output_type>>,
"stream<T> for the output type has has no type ID, "
"did you forgot to announce it via CAF_ADD_TYPE_ID?");
return attach_stream_source(self, std::make_tuple(), init, pull, done, fin,
token);
}
/// Attaches a new stream source to `self` by creating a default stream source
/// manager with the default driver and starts sending to `dest` immediately.
/// @param self Points to the hosting actor.
/// @param dest Handle to the next stage in the pipeline.
/// @param xs User-defined arguments for the stream handshake.
/// @param init Function object for initializing the state of the source.
/// @param pull Generator function object for producing downstream messages.
/// @param done Predicate returning `true` when generator is done.
/// @param fin Cleanup handler.
/// @returns The allocated `stream_manager` and the output slot.
template <class ActorHandle, class... Ts, class Init, class Pull, class Done,
class Finalize = unit_t,
class DownstreamManager = default_downstream_manager_t<Pull>,
class Trait = stream_source_trait_t<Pull>>
detail::enable_if_t<is_actor_handle<ActorHandle>::value,
make_source_result_t<DownstreamManager>>
attach_stream_source(scheduled_actor* self, const ActorHandle& dest,
std::tuple<Ts...> xs, Init init, Pull pull, Done done,
Finalize fin = {}, policy::arg<DownstreamManager> = {}) {
using namespace detail;
using token = type_list<stream<typename DownstreamManager::output_type>,
strip_and_convert_t<Ts>...>;
static_assert(response_type_unbox<signatures_of_t<ActorHandle>, token>::valid,
"receiver does not accept the stream handshake");
using driver = detail::stream_source_driver_impl<DownstreamManager, Pull,
Done, Finalize>;
auto mgr = detail::make_stream_source<driver>(self, std::move(init),
std::move(pull),
std::move(done),
std::move(fin));
auto slot = mgr->add_outbound_path(dest, std::move(xs));
return {slot, std::move(mgr)};
}
/// Attaches a new stream source to `self` by creating a default stream source
/// manager with the default driver and starts sending to `dest` immediately.
/// @param self Points to the hosting actor.
/// @param dest Handle to the next stage in the pipeline.
/// @param init Function object for initializing the state of the source.
/// @param pull Generator function object for producing downstream messages.
/// @param done Predicate returning `true` when generator is done.
/// @param fin Cleanup handler.
/// @returns The allocated `stream_manager` and the output slot.
template <class ActorHandle, class Init, class Pull, class Done,
class Finalize = unit_t,
class DownstreamManager = default_downstream_manager_t<Pull>,
class Trait = stream_source_trait_t<Pull>>
detail::enable_if_t<is_actor_handle<ActorHandle>::value && Trait::valid,
make_source_result_t<DownstreamManager>>
attach_stream_source(scheduled_actor* self, const ActorHandle& dest, Init init,
Pull pull, Done done, Finalize fin = {},
policy::arg<DownstreamManager> token = {}) {
return attach_stream_source(self, dest, std::make_tuple(), std::move(init),
std::move(pull), std::move(done), std::move(fin),
token);
}
} // namespace caf
// 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 <vector>
#include "caf/default_downstream_manager.hpp"
#include "caf/detail/stream_stage_driver_impl.hpp"
#include "caf/detail/stream_stage_impl.hpp"
#include "caf/downstream_manager.hpp"
#include "caf/fwd.hpp"
#include "caf/make_stage_result.hpp"
#include "caf/policy/arg.hpp"
#include "caf/stream.hpp"
#include "caf/stream_stage.hpp"
#include "caf/unit.hpp"
namespace caf {
/// Attaches a new stream stage to `self` by creating a default stream stage
/// manager with `Driver`.
/// @param self Points to the hosting actor.
/// @param in Stream handshake from upstream path.
/// @param xs User-defined arguments for the downstream handshake.
/// @param ys Additional constructor arguments for `Driver`.
/// @returns The new `stream_manager`, an inbound slot, and an outbound slot.
template <class Driver, class In, class... Ts, class... Us>
make_stage_result_t<In, typename Driver::downstream_manager_type, Ts...>
attach_stream_stage(scheduled_actor* self, const stream<In>& in,
std::tuple<Ts...> xs, Us&&... ys) {
using detail::make_stream_stage;
auto mgr = make_stream_stage<Driver>(self, std::forward<Us>(ys)...);
auto islot = mgr->add_inbound_path(in);
auto oslot = mgr->add_outbound_path(std::move(xs));
return {islot, oslot, std::move(mgr)};
}
/// Attaches a new stream stage to `self` by creating a default stream stage
/// manager from given callbacks.
/// @param self Points to the hosting actor.
/// @param in Stream handshake from upstream path.
/// @param xs User-defined arguments for the downstream handshake.
/// @param init Function object for initializing the state of the stage.
/// @param fun Processing function.
/// @param fin Optional cleanup handler.
/// @param token Policy token for selecting a downstream manager
/// implementation.
/// @returns The new `stream_manager`, an inbound slot, and an outbound slot.
template <class In, class... Ts, class Init, class Fun, class Finalize = unit_t,
class DownstreamManager = default_downstream_manager_t<Fun>,
class Trait = stream_stage_trait_t<Fun>>
make_stage_result_t<In, DownstreamManager, Ts...>
attach_stream_stage(scheduled_actor* self, const stream<In>& in,
std::tuple<Ts...> xs, Init init, Fun fun, Finalize fin = {},
policy::arg<DownstreamManager> token = {}) {
CAF_IGNORE_UNUSED(token);
using output_type = typename stream_stage_trait_t<Fun>::output;
using state_type = typename stream_stage_trait_t<Fun>::state;
static_assert(
std::is_same<void(state_type&),
typename detail::get_callable_trait<Init>::fun_sig>::value,
"Expected signature `void (State&)` for init function");
using consume_one = void(state_type&, downstream<output_type>&, In);
using consume_all
= void(state_type&, downstream<output_type>&, std::vector<In>&);
using fun_sig = typename detail::get_callable_trait<Fun>::fun_sig;
static_assert(std::is_same<fun_sig, consume_one>::value
|| std::is_same<fun_sig, consume_all>::value,
"Expected signature `void (State&, downstream<Out>&, In)` "
"or `void (State&, downstream<Out>&, std::vector<In>&)` "
"for consume function");
using driver
= detail::stream_stage_driver_impl<typename Trait::input, DownstreamManager,
Fun, Finalize>;
return attach_stream_stage<driver>(self, in, std::move(xs), std::move(init),
std::move(fun), std::move(fin));
}
/// Attaches a new stream stage to `self` by creating a default stream stage
/// manager from given callbacks.
/// @param self Points to the hosting actor.
/// @param in Stream handshake from upstream path.
/// @param init Function object for initializing the state of the stage.
/// @param fun Processing function.
/// @param fin Optional cleanup handler.
/// @param token Policy token for selecting a downstream manager
/// implementation.
/// @returns The new `stream_manager`, an inbound slot, and an outbound slot.
template <class In, class Init, class Fun, class Finalize = unit_t,
class DownstreamManager = default_downstream_manager_t<Fun>,
class Trait = stream_stage_trait_t<Fun>>
make_stage_result_t<In, DownstreamManager>
attach_stream_stage(scheduled_actor* self, const stream<In>& in, Init init,
Fun fun, Finalize fin = {},
policy::arg<DownstreamManager> token = {}) {
return attach_stream_stage(self, in, std::make_tuple(), std::move(init),
std::move(fun), std::move(fin), token);
}
} // namespace caf
......@@ -4,10 +4,6 @@
#pragma once
#include <chrono>
#include <condition_variable>
#include <mutex>
#include "caf/actor_config.hpp"
#include "caf/actor_traits.hpp"
#include "caf/after.hpp"
......@@ -33,13 +29,15 @@
#include "caf/none.hpp"
#include "caf/policy/arg.hpp"
#include "caf/policy/categorized.hpp"
#include "caf/policy/downstream_messages.hpp"
#include "caf/policy/normal_messages.hpp"
#include "caf/policy/upstream_messages.hpp"
#include "caf/policy/urgent_messages.hpp"
#include "caf/send.hpp"
#include "caf/typed_actor.hpp"
#include <chrono>
#include <condition_variable>
#include <mutex>
namespace caf {
/// A thread-mapped or context-switching actor using a blocking
......@@ -388,8 +386,6 @@ public:
bool cleanup(error&& fail_state, execution_unit* host) override;
sec build_pipeline(stream_slot in, stream_slot out, stream_manager_ptr mgr);
// -- backwards compatibility ------------------------------------------------
mailbox_element_ptr next_message() {
......
This diff is collapsed.
// 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 <deque>
#include <iterator>
#include <vector>
#include "caf/downstream_manager_base.hpp"
#include "caf/logger.hpp"
namespace caf {
/// Mixin for streams with any number of downstreams. `Subtype` must provide a
/// member function `buf()` returning a queue with `std::deque`-like interface.
template <class T>
class buffered_downstream_manager : public downstream_manager_base {
public:
// -- member types -----------------------------------------------------------
using super = downstream_manager_base;
using output_type = T;
using buffer_type = std::deque<output_type>;
using chunk_type = std::vector<output_type>;
// -- sanity checks ----------------------------------------------------------
static_assert(detail::is_complete<type_id<std::vector<output_type>>>);
// -- constructors, destructors, and assignment operators --------------------
explicit buffered_downstream_manager(stream_manager* parent) : super(parent) {
// nop
}
buffered_downstream_manager(stream_manager* parent, type_id_t type)
: super(parent, type) {
// nop
}
template <class T0, class... Ts>
void push(T0&& x, Ts&&... xs) {
buf_.emplace_back(std::forward<T0>(x), std::forward<Ts>(xs)...);
this->generated_messages(1);
}
/// @pre `n <= buf_.size()`
static chunk_type get_chunk(buffer_type& buf, size_t n) {
CAF_LOG_TRACE(CAF_ARG(buf) << CAF_ARG(n));
chunk_type xs;
if (!buf.empty() && n > 0) {
xs.reserve(std::min(n, buf.size()));
if (n < buf.size()) {
auto first = buf.begin();
auto last = first + static_cast<ptrdiff_t>(n);
std::move(first, last, std::back_inserter(xs));
buf.erase(first, last);
} else {
std::move(buf.begin(), buf.end(), std::back_inserter(xs));
buf.clear();
}
}
return xs;
}
chunk_type get_chunk(size_t n) {
return get_chunk(buf_, n);
}
bool terminal() const noexcept override {
return false;
}
size_t capacity() const noexcept override {
// Our goal is to cache up to 2 full batches, whereby we pick the highest
// batch size available to us (optimistic estimate).
size_t desired = 1;
for (auto& kvp : this->paths_)
desired = std::max(static_cast<size_t>(kvp.second->desired_batch_size),
desired);
desired *= 2;
auto stored = buffered();
return desired > stored ? desired - stored : 0u;
}
size_t buffered() const noexcept override {
return buf_.size();
}
buffer_type& buf() {
return buf_;
}
const buffer_type& buf() const {
return buf_;
}
protected:
buffer_type buf_;
};
} // namespace caf
// 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 "caf/detail/comparable.hpp"
#include "caf/intrusive_cow_ptr.hpp"
#include "caf/make_counted.hpp"
#include "caf/ref_counted.hpp"
#include <initializer_list>
#include <vector>
namespace caf {
/// A copy-on-write vector implementation that wraps a `std::vector`.
template <class T>
class cow_vector {
public:
// -- member types -----------------------------------------------------------
using value_type = T;
using std_type = std::vector<value_type>;
using size_type = typename std_type::size_type;
using const_iterator = typename std_type::const_iterator;
using difference_type = typename std_type::difference_type;
using const_reverse_iterator = typename std_type::const_reverse_iterator;
// -- constructors, destructors, and assignment operators --------------------
cow_vector() {
impl_ = make_counted<impl>();
}
explicit cow_vector(std_type std) {
impl_ = make_counted<impl>(std::move(std));
}
explicit cow_vector(std::initializer_list<T> values) {
impl_ = make_counted<impl>(std_type{values});
}
cow_vector(cow_vector&&) noexcept = default;
cow_vector(const cow_vector&) noexcept = default;
cow_vector& operator=(cow_vector&&) noexcept = default;
cow_vector& operator=(const cow_vector&) noexcept = default;
// -- properties -------------------------------------------------------------
/// Returns a mutable reference to the managed vector. Copies the vector if
/// more than one reference to it exists to make sure the reference count is
/// exactly 1 when returning from this function.
std_type& unshared() {
return impl_.unshared().std;
}
/// Returns the managed STD container.
const std_type& std() const noexcept {
return impl_->std;
}
/// Returns whether the reference count of the managed object is 1.
[[nodiscard]] bool unique() const noexcept {
return impl_->unique();
}
[[nodiscard]] bool empty() const noexcept {
return impl_->std.empty();
}
size_type size() const noexcept {
return impl_->std.size();
}
size_type max_size() const noexcept {
return impl_->std.max_size();
}
// -- element access ---------------------------------------------------------
T at(size_type pos) const {
return impl_->std.at(pos);
}
T operator[](size_type pos) const {
return impl_->std[pos];
}
T front() const {
return impl_->std.front();
}
T back() const {
return impl_->std.back();
}
const T* data() const noexcept {
return impl_->std.data();
}
// -- iterator access --------------------------------------------------------
const_iterator begin() const noexcept {
return impl_->std.begin();
}
const_iterator cbegin() const noexcept {
return impl_->std.begin();
}
const_reverse_iterator rbegin() const noexcept {
return impl_->std.rbegin();
}
const_reverse_iterator crbegin() const noexcept {
return impl_->std.rbegin();
}
const_iterator end() const noexcept {
return impl_->std.end();
}
const_iterator cend() const noexcept {
return impl_->std.end();
}
const_reverse_iterator rend() const noexcept {
return impl_->std.rend();
}
const_reverse_iterator crend() const noexcept {
return impl_->std.rend();
}
// -- friends ----------------------------------------------------------------
template <class Inspector>
friend bool inspect(Inspector& f, cow_vector& x) {
if constexpr (Inspector::is_loading) {
return f.apply(x.unshared());
} else {
return f.apply(x.impl_->std);
}
}
private:
struct impl : ref_counted {
std_type std;
impl() = default;
explicit impl(std_type in) : std(std::move(in)) {
// nop
}
impl* copy() const {
return new impl{std};
}
};
intrusive_cow_ptr<impl> impl_;
};
// -- comparison ---------------------------------------------------------------
template <class T>
auto operator==(const cow_vector<T>& xs, const cow_vector<T>& ys)
-> decltype(xs.std() == ys.std()) {
return xs.std() == ys.std();
}
template <class T>
auto operator==(const cow_vector<T>& xs, const std::vector<T>& ys)
-> decltype(xs.std() == ys) {
return xs.std() == ys;
}
template <class T>
auto operator==(const std::vector<T>& xs, const cow_vector<T>& ys)
-> decltype(xs == ys.std()) {
return xs.std() == ys;
}
template <class T>
auto operator!=(const cow_vector<T>& xs, const cow_vector<T>& ys)
-> decltype(xs.std() != ys.std()) {
return xs.std() != ys.std();
}
template <class T>
auto operator!=(const cow_vector<T>& xs, const std::vector<T>& ys)
-> decltype(xs.std() != ys) {
return xs.std() != ys;
}
template <class T>
auto operator!=(const std::vector<T>& xs, const cow_vector<T>& ys)
-> decltype(xs != ys.std()) {
return xs.std() != ys;
}
} // namespace caf
// 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 "caf/broadcast_downstream_manager.hpp"
#include "caf/stream_source_trait.hpp"
#include "caf/stream_stage_trait.hpp"
#include "caf/detail/type_traits.hpp"
namespace caf {
/// Selects a downstream manager implementation based on the signature of
/// various handlers.
template <class F>
struct default_downstream_manager {
/// The function signature of `F`.
using fun_sig = typename detail::get_callable_trait<F>::fun_sig;
/// The source trait for `F`.
using source_trait = stream_source_trait<fun_sig>;
/// The stage trait for `F`.
using stage_trait = stream_stage_trait<fun_sig>;
/// The output type as returned by the source or stage trait.
using output_type =
typename std::conditional<
source_trait::valid,
typename source_trait::output,
typename stage_trait::output
>::type;
/// The default downstream manager deduced by this trait.
using type = broadcast_downstream_manager<output_type>;
};
template <class F>
using default_downstream_manager_t =
typename default_downstream_manager<F>::type;
} // namespace caf
// 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 <atomic>
#include <cstddef>
#include "caf/detail/core_export.hpp"
namespace caf::detail {
/// Base class for reference counted objects with an atomic reference count.
class CAF_CORE_EXPORT atomic_ref_counted {
public:
virtual ~atomic_ref_counted();
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 {
rc_.fetch_add(1, std::memory_order_relaxed);
}
/// Decreases reference count by one and calls `request_deletion`
/// when it drops to zero.
void deref() const noexcept;
/// 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_.load();
}
protected:
mutable std::atomic<size_t> rc_;
};
} // namespace caf::detail
......@@ -12,12 +12,7 @@
#include "caf/detail/overload.hpp"
#include "caf/detail/type_traits.hpp"
#include "caf/fwd.hpp"
#include "caf/make_sink_result.hpp"
#include "caf/make_source_result.hpp"
#include "caf/make_stage_result.hpp"
#include "caf/message.hpp"
#include "caf/none.hpp"
#include "caf/optional.hpp"
#include "caf/result.hpp"
#include "caf/skip.hpp"
#include "caf/unit.hpp"
......@@ -47,9 +42,6 @@ public:
/// Wraps arbitrary values into a `message` and calls the visitor recursively.
template <class... Ts>
void operator()(Ts&... xs) {
static_assert(detail::conjunction<!detail::is_stream<Ts>::value...>::value,
"returning a stream<T> from a message handler achieves not "
"what you would expect and is most likely a mistake");
auto tmp = make_message(std::move(xs)...);
(*this)(tmp);
}
......@@ -68,7 +60,7 @@ public:
/// Dispatches on the runtime-type of `x`.
template <class... Ts>
void operator()(result<Ts...>& res) {
caf::visit([this](auto& x) { (*this)(x); }, res);
visit([this](auto& x) { (*this)(x); }, res.get_data());
}
// -- special-purpose handlers that don't produce results --------------------
......@@ -86,31 +78,6 @@ public:
void operator()(delegated<Ts...>&) {
// nop
}
template <class Out, class... Ts>
void operator()(outbound_stream_slot<Out, Ts...>&) {
// nop
}
template <class In>
void operator()(inbound_stream_slot<In>&) {
// nop
}
template <class In>
void operator()(make_sink_result<In>&) {
// nop
}
template <class DownstreamManager, class... Ts>
void operator()(make_source_result<DownstreamManager, Ts...>&) {
// nop
}
template <class In, class DownstreamManager, class... Ts>
void operator()(make_stage_result<In, DownstreamManager, Ts...>&) {
// nop
}
};
} // namespace caf::detail
// 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
// 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 "caf/credit_controller.hpp"
#include "caf/detail/core_export.hpp"
#include "caf/detail/serialized_size.hpp"
#include "caf/downstream_msg.hpp"
#include "caf/stream.hpp"
namespace caf::detail {
/// A credit controller that estimates the bytes required to store incoming
/// batches and constrains credit based on upper bounds for memory usage.
class CAF_CORE_EXPORT size_based_credit_controller : public credit_controller {
public:
// -- constants --------------------------------------------------------------
/// Configures how many samples we require for recalculating buffer sizes.
static constexpr int32_t min_samples = 50;
/// Stores how many elements we buffer at most after the handshake.
int32_t initial_buffer_size = 10;
/// Stores how many elements we allow per batch after the handshake.
int32_t initial_batch_size = 2;
// -- constructors, destructors, and assignment operators --------------------
explicit size_based_credit_controller(local_actor* self);
~size_based_credit_controller() override;
// -- interface functions ----------------------------------------------------
calibration init() override;
calibration calibrate() override;
// -- factory functions ------------------------------------------------------
template <class T>
static auto make(local_actor* self, stream<T>) {
class impl : public size_based_credit_controller {
using size_based_credit_controller::size_based_credit_controller;
void before_processing(downstream_msg::batch& x) override {
if (++this->sample_counter_ == this->sampling_rate_) {
this->sample_counter_ = 0;
this->inspector_.result = 0;
this->sampled_elements_ += x.xs_size;
for (auto& element : x.xs.get_as<std::vector<T>>(0))
detail::save(this->inspector_, element);
this->sampled_total_size_
+= static_cast<int64_t>(this->inspector_.result);
}
}
};
return std::make_unique<impl>(self);
}
protected:
// -- member variables -------------------------------------------------------
local_actor* self_;
/// Keeps track of when to sample a batch.
int32_t sample_counter_ = 0;
/// Stores the last computed (moving) average for the serialized size per
/// element in the stream.
int32_t bytes_per_element_ = 0;
/// Stores how many elements were sampled since last calling `calibrate`.
int32_t sampled_elements_ = 0;
/// Stores how many bytes the sampled batches required when serialized.
int64_t sampled_total_size_ = 0;
/// Computes how many bytes elements require on the wire.
serialized_size_inspector inspector_;
/// Stores whether this is the first run.
bool initializing_ = true;
// -- see caf::defaults::stream::size_policy --------------------------------
int32_t bytes_per_batch_;
int32_t buffer_capacity_;
int32_t sampling_rate_ = 1;
int32_t calibration_interval_;
float smoothing_factor_;
};
} // namespace caf::detail
// 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 "caf/actor_control_block.hpp"
#include "caf/flow/observer.hpp"
#include "caf/flow/op/hot.hpp"
#include "caf/flow/subscription.hpp"
#include <cstddef>
#include <cstdint>
#include <deque>
namespace caf::detail {
class stream_bridge_sub : public flow::subscription::impl_base {
public:
stream_bridge_sub(scheduled_actor* self, strong_actor_ptr src,
flow::observer<async::batch> out, uint64_t snk_flow_id,
size_t max_in_flight, size_t request_threshold)
: self_(self),
src_(std::move(src)),
out_(std::move(out)),
snk_flow_id_(snk_flow_id),
max_in_flight_(max_in_flight),
request_threshold_(request_threshold) {
// nop
}
// -- callbacks for the actor ------------------------------------------------
void ack(uint64_t src_flow_id, uint32_t max_items_per_batch);
void drop();
void drop(const error& reason);
void push(const async::batch& input);
void push();
// -- implementation of subscription -----------------------------------------
bool disposed() const noexcept override;
void dispose() override;
void request(size_t n) override;
private:
bool initialized() const noexcept {
return src_flow_id_ != 0;
}
void do_abort(const error& reason);
void do_check_credit();
scheduled_actor* self_;
strong_actor_ptr src_;
flow::observer<async::batch> out_;
uint64_t src_flow_id_ = 0;
uint64_t snk_flow_id_;
size_t max_in_flight_batches_ = 0;
size_t in_flight_batches_ = 0;
size_t low_batches_threshold_ = 0;
size_t demand_ = 0;
std::deque<async::batch> buf_;
size_t max_in_flight_;
size_t request_threshold_;
};
using stream_bridge_sub_ptr = intrusive_ptr<stream_bridge_sub>;
class stream_bridge : public flow::op::hot<async::batch> {
public:
using super = flow::op::hot<async::batch>;
explicit stream_bridge(scheduled_actor* self, strong_actor_ptr src,
uint64_t stream_id, size_t buf_capacity,
size_t request_threshold);
disposable subscribe(flow::observer<async::batch> out) override;
private:
scheduled_actor* self_ptr();
strong_actor_ptr src_;
uint64_t stream_id_;
size_t buf_capacity_;
size_t request_threshold_;
};
} // namespace caf::detail
// 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 <memory>
#include <unordered_map>
#include "caf/logger.hpp"
#include "caf/ref_counted.hpp"
#include "caf/scheduled_actor.hpp"
#include "caf/stream_manager.hpp"
namespace caf::detail {
/// A stream distribution tree consist of peers forming an acyclic graph. The
/// user is responsible for making sure peers do not form a loop. Data is
/// flooded along the tree. Each peer serves any number of subscribers. The
/// policy of the tree enables subscriptions to different chunks of the whole
/// stream (substreams).
///
/// The tree uses two CAF streams between each pair of peers for transmitting
/// data. This automatically adds backpressure to the system, i.e., no peer can
/// overwhelm others.
///
/// Policies need to provide the following member types and functions:
///
/// ~~~{.cpp}
/// TODO
/// };
/// ~~~
template <class Policy>
class stream_distribution_tree : public stream_manager {
public:
// -- nested types -----------------------------------------------------------
using super = stream_manager;
using downstream_manager_type = typename Policy::downstream_manager_type;
// --- constructors and destructors ------------------------------------------
template <class... Ts>
stream_distribution_tree(scheduled_actor* selfptr, Ts&&... xs)
: super(selfptr), out_(this), policy_(this, std::forward<Ts>(xs)...) {
continuous(true);
}
~stream_distribution_tree() override {
// nop
}
// -- Accessors --------------------------------------------------------------
Policy& policy() {
return policy_;
}
const Policy& policy() const {
return policy_;
}
// -- overridden member functions of `stream_manager` ------------------------
void handle(inbound_path* path, downstream_msg::batch& x) override {
CAF_LOG_TRACE(CAF_ARG(path) << CAF_ARG(x));
auto slot = path->slots.receiver;
policy_.before_handle_batch(slot, path->hdl);
policy_.handle_batch(slot, path->hdl, x.xs);
policy_.after_handle_batch(slot, path->hdl);
}
void handle(inbound_path* path, downstream_msg::close& x) override {
CAF_LOG_TRACE(CAF_ARG(path) << CAF_ARG(x));
CAF_IGNORE_UNUSED(x);
policy_.path_closed(path->slots.receiver);
}
void handle(inbound_path* path, downstream_msg::forced_close& x) override {
CAF_LOG_TRACE(CAF_ARG(path) << CAF_ARG(x));
policy_.path_force_closed(path->slots.receiver, x.reason);
}
bool handle(stream_slots slots, upstream_msg::ack_open& x) override {
CAF_LOG_TRACE(CAF_ARG(slots) << CAF_ARG(x));
auto rebind_from = x.rebind_from;
auto rebind_to = x.rebind_to;
if (super::handle(slots, x)) {
policy_.ack_open_success(slots.receiver, rebind_from, rebind_to);
return true;
}
policy_.ack_open_failure(slots.receiver, rebind_from, rebind_to);
return false;
}
void handle(stream_slots slots, upstream_msg::drop& x) override {
CAF_LOG_TRACE(CAF_ARG(slots) << CAF_ARG(x));
CAF_IGNORE_UNUSED(x);
super::handle(slots, x);
}
void handle(stream_slots slots, upstream_msg::forced_drop& x) override {
CAF_LOG_TRACE(CAF_ARG(slots) << CAF_ARG(x));
CAF_IGNORE_UNUSED(x);
auto slot = slots.receiver;
if (out().remove_path(slots.receiver, x.reason, true))
policy_.path_force_dropped(slot, x.reason);
}
bool done() const override {
return !continuous() && pending_handshakes_ == 0 && inbound_paths_.empty()
&& out_.clean();
}
bool idle() const noexcept override {
// Same as `stream_stage<...>`::idle().
return out_.stalled() || (out_.clean() && this->inbound_paths_idle());
}
downstream_manager_type& out() override {
return out_;
}
private:
downstream_manager_type out_;
Policy policy_;
};
} // namespace caf::detail
// 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 "caf/none.hpp"
#include "caf/stream_finalize_trait.hpp"
#include "caf/stream_sink_driver.hpp"
#include "caf/stream_sink_trait.hpp"
namespace caf::detail {
/// Identifies an unbound sequence of messages.
template <class Input, class Process, class Finalize>
class stream_sink_driver_impl final : public stream_sink_driver<Input> {
public:
// -- member types -----------------------------------------------------------
using super = stream_sink_driver<Input>;
using typename super::input_type;
using trait = stream_sink_trait_t<Process>;
using state_type = typename trait::state;
template <class Init>
stream_sink_driver_impl(Init init, Process f, Finalize fin)
: process_(std::move(f)), fin_(std::move(fin)) {
init(state_);
}
void process(std::vector<input_type>& xs) override {
return trait::process::invoke(process_, state_, xs);
}
void finalize(const error& err) override {
stream_finalize_trait<Finalize, state_type>::invoke(fin_, state_, err);
}
private:
Process process_;
Finalize fin_;
state_type state_;
};
} // namespace caf::detail
// 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 "caf/config.hpp"
#include "caf/logger.hpp"
#include "caf/make_counted.hpp"
#include "caf/message_id.hpp"
#include "caf/policy/arg.hpp"
#include "caf/sec.hpp"
#include "caf/stream_manager.hpp"
#include "caf/stream_sink.hpp"
#include "caf/typed_message_view.hpp"
namespace caf::detail {
template <class Driver>
class stream_sink_impl : public Driver::sink_type {
public:
using super = typename Driver::sink_type;
using driver_type = Driver;
using input_type = typename driver_type::input_type;
template <class... Ts>
stream_sink_impl(scheduled_actor* self, Ts&&... xs)
: stream_manager(self), super(self), driver_(std::forward<Ts>(xs)...) {
// nop
}
using super::handle;
void handle(inbound_path*, downstream_msg::batch& x) override {
CAF_LOG_TRACE(CAF_ARG(x));
using vec_type = std::vector<input_type>;
if (auto view = make_typed_message_view<vec_type>(x.xs)) {
driver_.process(get<0>(view));
return;
}
CAF_LOG_ERROR("received unexpected batch type (dropped)");
}
int32_t acquire_credit(inbound_path* path, int32_t desired) override {
return driver_.acquire_credit(path, desired);
}
protected:
void finalize(const error& reason) override {
driver_.finalize(reason);
}
driver_type driver_;
};
template <class Driver, class... Ts>
typename Driver::sink_ptr_type
make_stream_sink(scheduled_actor* self, Ts&&... xs) {
using impl = stream_sink_impl<Driver>;
return make_counted<impl>(self, std::forward<Ts>(xs)...);
}
} // namespace caf::detail
// 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 "caf/none.hpp"
#include "caf/stream_finalize_trait.hpp"
#include "caf/stream_source_driver.hpp"
#include "caf/stream_source_trait.hpp"
namespace caf::detail {
/// Identifies an unbound sequence of messages.
template <class DownstreamManager, class Pull, class Done, class Finalize>
class stream_source_driver_impl final
: public stream_source_driver<DownstreamManager> {
public:
// -- member types -----------------------------------------------------------
using super = stream_source_driver<DownstreamManager>;
using output_type = typename super::output_type;
using trait = stream_source_trait_t<Pull>;
using state_type = typename trait::state;
template <class Init>
stream_source_driver_impl(Init init, Pull f, Done pred, Finalize fin)
: pull_(std::move(f)), done_(std::move(pred)), fin_(std::move(fin)) {
init(state_);
}
void pull(downstream<output_type>& out, size_t num) override {
return pull_(state_, out, num);
}
bool done() const noexcept override {
return done_(state_);
}
void finalize(const error& err) override {
stream_finalize_trait<Finalize, state_type>::invoke(fin_, state_, err);
}
private:
state_type state_;
Pull pull_;
Done done_;
Finalize fin_;
};
} // namespace caf::detail
// 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 "caf/downstream.hpp"
#include "caf/fwd.hpp"
#include "caf/logger.hpp"
#include "caf/make_counted.hpp"
#include "caf/make_source_result.hpp"
#include "caf/outbound_path.hpp"
#include "caf/stream_source.hpp"
#include "caf/stream_source_trait.hpp"
namespace caf::detail {
template <class Driver>
class stream_source_impl : public Driver::source_type {
public:
// -- member types -----------------------------------------------------------
using super = typename Driver::source_type;
using driver_type = Driver;
// -- constructors, destructors, and assignment operators --------------------
template <class... Ts>
stream_source_impl(scheduled_actor* self, Ts&&... xs)
: stream_manager(self),
super(self),
driver_(std::forward<Ts>(xs)...),
at_end_(false) {
// nop
}
// -- implementation of virtual functions ------------------------------------
void shutdown() override {
super::shutdown();
at_end_ = true;
}
bool done() const override {
return this->pending_handshakes_ == 0 && at_end_ && this->out_.clean();
}
bool generate_messages() override {
CAF_LOG_TRACE("");
if (at_end_)
return false;
auto hint = this->out_.capacity();
CAF_LOG_DEBUG(CAF_ARG(hint));
if (hint == 0)
return false;
auto old_size = this->out_.buf().size();
downstream<typename Driver::output_type> ds{this->out_.buf()};
driver_.pull(ds, hint);
if (driver_.done())
at_end_ = true;
auto new_size = this->out_.buf().size();
this->out_.generated_messages(new_size - old_size);
return new_size != old_size;
}
protected:
void finalize(const error& reason) override {
driver_.finalize(reason);
}
Driver driver_;
private:
bool at_end_;
};
template <class Driver, class... Ts>
typename Driver::source_ptr_type
make_stream_source(scheduled_actor* self, Ts&&... xs) {
using impl = stream_source_impl<Driver>;
return make_counted<impl>(self, std::forward<Ts>(xs)...);
}
} // namespace caf::detail
// 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 "caf/stream_finalize_trait.hpp"
#include "caf/stream_slot.hpp"
#include "caf/stream_stage_driver.hpp"
#include "caf/stream_stage_trait.hpp"
namespace caf::detail {
/// Default implementation for a `stream_stage_driver` that hardwires `message`
/// as result type and implements `process` and `finalize` using user-provided
/// function objects (usually lambdas).
template <class Input, class DownstreamManager, class Process, class Finalize>
class stream_stage_driver_impl final
: public stream_stage_driver<Input, DownstreamManager> {
public:
// -- member types -----------------------------------------------------------
using super = stream_stage_driver<Input, DownstreamManager>;
using typename super::input_type;
using typename super::output_type;
using typename super::stream_type;
using trait = stream_stage_trait_t<Process>;
using state_type = typename trait::state;
template <class Init>
stream_stage_driver_impl(DownstreamManager& out, Init init, Process f,
Finalize fin)
: super(out), process_(std::move(f)), fin_(std::move(fin)) {
init(state_);
}
void process(downstream<output_type>& out,
std::vector<input_type>& batch) override {
trait::process::invoke(process_, state_, out, batch);
}
void finalize(const error& err) override {
stream_finalize_trait<Finalize, state_type>::invoke(fin_, state_, err);
}
private:
state_type state_;
Process process_;
Finalize fin_;
};
} // namespace caf::detail
// 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 "caf/downstream.hpp"
#include "caf/logger.hpp"
#include "caf/make_counted.hpp"
#include "caf/outbound_path.hpp"
#include "caf/sec.hpp"
#include "caf/stream_manager.hpp"
#include "caf/stream_stage.hpp"
#include "caf/stream_stage_trait.hpp"
#include "caf/typed_message_view.hpp"
namespace caf::detail {
template <class Driver>
class stream_stage_impl : public Driver::stage_type {
public:
// -- member types -----------------------------------------------------------
using super = typename Driver::stage_type;
using driver_type = Driver;
using downstream_manager_type = typename Driver::downstream_manager_type;
using input_type = typename driver_type::input_type;
using output_type = typename driver_type::output_type;
// -- constructors, destructors, and assignment operators --------------------
template <class... Ts>
stream_stage_impl(scheduled_actor* self, Ts&&... xs)
: stream_manager(self),
super(self),
driver_(this->out_, std::forward<Ts>(xs)...) {
// nop
}
// -- implementation of virtual functions ------------------------------------
using super::handle;
void handle(inbound_path*, downstream_msg::batch& x) override {
CAF_LOG_TRACE(CAF_ARG(x));
using vec_type = std::vector<input_type>;
if (auto view = make_typed_message_view<vec_type>(x.xs)) {
auto old_size = this->out_.buf().size();
downstream<output_type> ds{this->out_.buf()};
driver_.process(ds, get<0>(view));
auto new_size = this->out_.buf().size();
this->out_.generated_messages(new_size - old_size);
return;
}
CAF_LOG_ERROR("received unexpected batch type (dropped)");
}
int32_t acquire_credit(inbound_path* path, int32_t desired) override {
return driver_.acquire_credit(path, desired);
}
protected:
void finalize(const error& reason) override {
driver_.finalize(reason);
}
driver_type driver_;
};
template <class Driver, class... Ts>
typename Driver::stage_ptr_type
make_stream_stage(scheduled_actor* self, Ts&&... xs) {
using impl = stream_stage_impl<Driver>;
return make_counted<impl>(self, std::forward<Ts>(xs)...);
}
} // namespace caf::detail
// 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 "caf/credit_controller.hpp"
#include "caf/detail/core_export.hpp"
namespace caf::detail {
/// A credit controller that estimates the bytes required to store incoming
/// batches and constrains credit based on upper bounds for memory usage.
class CAF_CORE_EXPORT token_based_credit_controller : public credit_controller {
public:
// -- constants --------------------------------------------------------------
/// Configures how many samples we require for recalculating buffer sizes.
static constexpr int32_t min_samples = 50;
/// Stores how many elements we buffer at most after the handshake.
int32_t initial_buffer_size = 10;
/// Stores how many elements we allow per batch after the handshake.
int32_t initial_batch_size = 2;
// -- constructors, destructors, and assignment operators --------------------
explicit token_based_credit_controller(local_actor* self);
~token_based_credit_controller() override;
// -- interface functions ----------------------------------------------------
void before_processing(downstream_msg::batch& batch) override;
calibration init() override;
calibration calibrate() override;
// -- factory functions ------------------------------------------------------
template <class T>
static auto make(local_actor* self, stream<T>) {
return std::make_unique<token_based_credit_controller>(self);
}
private:
// -- see caf::defaults::stream::token_policy -------------------------------
int32_t batch_size_;
int32_t buffer_size_;
};
} // namespace caf::detail
......@@ -7,6 +7,7 @@
#include <array>
#include <chrono>
#include <functional>
#include <optional>
#include <string>
#include <tuple>
#include <type_traits>
......@@ -221,6 +222,9 @@ public:
template <class T>
constexpr bool is_iterable<T>::value;
template <class T>
constexpr bool is_iterable_v = is_iterable<T>::value;
/// Checks whether `T` is a non-const reference.
template <class T>
struct is_mutable_ref : std::false_type { };
......@@ -528,12 +532,6 @@ struct transfer_const<const T, U> {
template <class T, class U>
using transfer_const_t = typename transfer_const<T, U>::type;
template <class T>
struct is_stream : std::false_type {};
template <class T>
struct is_stream<stream<T>> : std::true_type {};
template <class T>
struct is_result : std::false_type {};
......@@ -1029,6 +1027,29 @@ struct is_builtin_inspector_type<span<const byte>, false> {
static constexpr bool value = true;
};
/// Checks whether `T` is a `std::optional`.
template <class T>
struct is_optional : std::false_type {};
template <class T>
struct is_optional<std::optional<T>> : std::true_type {};
template <class T>
constexpr bool is_optional_v = is_optional<T>::value;
template <class T>
struct unboxed_oracle {
using type = T;
};
template <class T>
struct unboxed_oracle<std::optional<T>> {
using type = T;
};
template <class T>
using unboxed_t = typename unboxed_oracle<T>::type;
} // namespace caf::detail
#undef CAF_HAS_MEMBER_TRAIT
......
......@@ -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();
......@@ -35,13 +40,13 @@ public:
virtual void deref_disposable() const noexcept = 0;
};
// -- constructors, destructors, and assignment operators --------------------
explicit disposable(intrusive_ptr<impl> pimpl) noexcept
: pimpl_(std::move(pimpl)) {
// nop
}
// -- constructors, destructors, and assignment operators --------------------
disposable() noexcept = default;
disposable(disposable&&) noexcept = default;
......@@ -126,8 +131,17 @@ public:
return pimpl_.compare(other.pimpl_);
}
// -- utility ----------------------------------------------------------------
/// Erases each `x` from `xs` where `x.disposed()`.
/// @returns The number of erased elements.
static size_t erase_disposed(std::vector<disposable>& xs);
private:
intrusive_ptr<impl> pimpl_;
};
/// @relates disposable
using disposable_impl = disposable::impl;
} // namespace caf
// 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 <memory>
#include <vector>
#include "caf/actor_clock.hpp"
#include "caf/detail/core_export.hpp"
#include "caf/fwd.hpp"
#include "caf/stream_slot.hpp"
#include "caf/timespan.hpp"
namespace caf {
/// Manages downstream communication for a `stream_manager`. The downstream
/// manager owns the `outbound_path` objects, has a buffer for storing pending
/// output and is responsible for the dispatching policy (broadcasting, for
/// example). The default implementation terminates the stream and never
/// accepts any paths.
class CAF_CORE_EXPORT downstream_manager {
public:
// -- member types -----------------------------------------------------------
/// Outbound path.
using path_type = outbound_path;
/// Pointer to an outbound path.
using path_ptr = path_type*;
/// Pointer to an immutable outbound path.
using const_path_ptr = const path_type*;
/// Unique pointer to an outbound path.
using unique_path_ptr = std::unique_ptr<path_type>;
/// Discrete point in time, as reported by the actor clock.
using time_point = typename actor_clock::time_point;
/// Function object for iterating over all paths.
struct CAF_CORE_EXPORT path_visitor {
virtual ~path_visitor();
virtual void operator()(outbound_path& x) = 0;
};
/// Predicate object for paths.
struct CAF_CORE_EXPORT path_predicate {
virtual ~path_predicate();
virtual bool operator()(const outbound_path& x) const noexcept = 0;
};
/// Selects a check algorithms.
enum path_algorithm { all_of, any_of, none_of };
// -- constructors, destructors, and assignment operators --------------------
explicit downstream_manager(stream_manager* parent);
downstream_manager(const downstream_manager&) = delete;
downstream_manager& operator=(const downstream_manager&) = delete;
virtual ~downstream_manager();
// -- properties -------------------------------------------------------------
scheduled_actor* self() const noexcept;
stream_manager* parent() const noexcept;
/// Returns `true` if this manager belongs to a sink, i.e., terminates the
/// stream and never has outbound paths.
virtual bool terminal() const noexcept;
// -- time management --------------------------------------------------------
/// Forces underful batches after reaching the maximum delay.
void tick(time_point now, timespan max_batch_delay);
// -- path management --------------------------------------------------------
/// Applies `f` to each path.
template <class F>
void for_each_path(F f) {
struct impl : path_visitor {
F fun;
impl(F x) : fun(std::move(x)) {
// nop
}
void operator()(outbound_path& x) override {
fun(x);
}
};
impl g{std::move(f)};
for_each_path_impl(g);
}
/// Applies `f` to each path.
template <class F>
void for_each_path(F f) const {
struct impl : path_visitor {
F fun;
impl(F x) : fun(std::move(x)) {
// nop
}
void operator()(outbound_path& x) override {
fun(const_cast<const outbound_path&>(x));
}
};
impl g{std::move(f)};
// This const_cast is safe, because we restore the const in our overload for
// operator() above.
const_cast<downstream_manager*>(this)->for_each_path_impl(g);
}
/// Returns all used slots.
std::vector<stream_slot> path_slots();
/// Returns all open slots, i.e., slots assigned to outbound paths with
/// `closing == false`.
std::vector<stream_slot> open_path_slots();
/// Checks whether `predicate` holds true for all paths.
template <class Predicate>
bool all_paths(Predicate predicate) const noexcept {
return check_paths(path_algorithm::all_of, std::move(predicate));
}
/// Checks whether `predicate` holds true for any path.
template <class Predicate>
bool any_path(Predicate predicate) const noexcept {
return check_paths(path_algorithm::any_of, std::move(predicate));
}
/// Checks whether `predicate` holds true for no path.
template <class Predicate>
bool no_path(Predicate predicate) const noexcept {
return check_paths(path_algorithm::none_of, std::move(predicate));
}
/// Returns the current number of paths.
virtual size_t num_paths() const noexcept;
/// Adds a pending path to `target` to the manager.
/// @returns The added path on success, `nullptr` otherwise.
path_ptr add_path(stream_slot slot, strong_actor_ptr target);
/// Removes a path from the manager.
virtual bool
remove_path(stream_slot slot, error reason, bool silent) noexcept;
/// Returns the path associated to `slot` or `nullptr`.
virtual path_ptr path(stream_slot slot) noexcept;
/// Returns the path associated to `slot` or `nullptr`.
const_path_ptr path(stream_slot slot) const noexcept;
/// Returns `true` if there is no data pending and all batches are
/// acknowledged batch on all paths.
bool clean() const noexcept;
/// Returns `true` if `slot` is unknown or if there is no data pending and
/// all batches are acknowledged on `slot`. The default implementation
/// returns `false` for all paths, even if `clean()` return `true`.
bool clean(stream_slot slot) const noexcept;
/// Removes all paths gracefully.
virtual void close();
/// Removes path `slot` gracefully by sending pending batches before removing
/// it. Effectively calls `path(slot)->closing = true`.
virtual void close(stream_slot slot);
/// Removes all paths with an error message.
virtual void abort(error reason);
/// Returns `num_paths() == 0`.
bool empty() const noexcept {
return num_paths() == 0;
}
/// Returns the minimum amount of credit on all output paths.
size_t min_credit() const;
/// Returns the maximum amount of credit on all output paths.
size_t max_credit() const;
/// Returns the total amount of credit on all output paths, i.e., the sum of
/// all individual credits.
size_t total_credit() const;
/// Sends batches to sinks.
virtual void emit_batches();
/// Sends batches to sinks regardless of whether or not the batches reach the
/// desired batch size.
virtual void force_emit_batches();
/// Queries the currently available capacity for the output buffer.
virtual size_t capacity() const noexcept;
/// Queries the size of the output buffer.
virtual size_t buffered() const noexcept;
/// Queries an estimate of the size of the output buffer for `slot`.
virtual size_t buffered(stream_slot slot) const noexcept;
/// Queries whether the manager cannot make any progress, because its buffer
/// is full and no more credit is available.
bool stalled() const noexcept;
/// Silently removes all paths.
virtual void clear_paths();
protected:
// -- customization points ---------------------------------------------------
/// Inserts `ptr` to the implementation-specific container.
virtual bool insert_path(unique_path_ptr ptr);
/// Applies `f` to each path.
virtual void for_each_path_impl(path_visitor& f);
/// Dispatches the predicate to `std::all_of`, `std::any_of`, or
/// `std::none_of`.
virtual bool check_paths_impl(path_algorithm algo, path_predicate& pred) const
noexcept;
/// Emits a regular (`reason == nullptr`) or irregular (`reason != nullptr`)
/// shutdown if `silent == false`.
/// @warning moves `*reason` if `reason == nullptr`
virtual void about_to_erase(path_ptr ptr, bool silent, error* reason);
// -- helper functions -------------------------------------------------------
/// Delegates to `check_paths_impl`.
template <class Predicate>
bool check_paths(path_algorithm algorithm, Predicate predicate) const
noexcept {
struct impl : path_predicate {
Predicate fun;
impl(Predicate x) : fun(std::move(x)) {
// nop
}
bool operator()(const outbound_path& x) const noexcept override {
return fun(x);
}
};
impl g{std::move(predicate)};
return check_paths_impl(algorithm, g);
}
// -- member variables -------------------------------------------------------
stream_manager* parent_;
/// Stores the time stamp of our last batch.
time_point last_send_;
};
} // namespace caf
// 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 <memory>
#include "caf/detail/core_export.hpp"
#include "caf/detail/unordered_flat_map.hpp"
#include "caf/downstream_manager.hpp"
#include "caf/outbound_path.hpp"
#include "caf/telemetry/counter.hpp"
#include "caf/telemetry/gauge.hpp"
namespace caf {
/// The default downstream manager base stores outbound paths in an unordered
/// map. It always takes ownership of the paths by using unique pointers.
class CAF_CORE_EXPORT downstream_manager_base : public downstream_manager {
public:
// -- member types -----------------------------------------------------------
/// Base type.
using super = downstream_manager;
/// Maps slots to paths.
using map_type = detail::unordered_flat_map<stream_slot, unique_path_ptr>;
/// Optional metrics for outbound stream traffic.
struct metrics_t {
/// Counts the total number of elements that have been pushed downstream.
telemetry::int_counter* pushed_elements = nullptr;
/// Tracks how many stream elements are currently waiting in the output
/// buffer due to insufficient credit.
telemetry::int_gauge* output_buffer_size = nullptr;
};
// -- constructors, destructors, and assignment operators --------------------
explicit downstream_manager_base(stream_manager* parent);
downstream_manager_base(stream_manager* parent, type_id_t type);
~downstream_manager_base() override;
// -- properties -------------------------------------------------------------
const map_type& paths() const {
return paths_;
}
map_type& paths() {
return paths_;
}
// -- path management --------------------------------------------------------
size_t num_paths() const noexcept override;
bool remove_path(stream_slot slots, error reason,
bool silent) noexcept override;
path_ptr path(stream_slot slots) noexcept override;
void clear_paths() override;
// -- callbacks for actor metrics --------------------------------------------
void generated_messages(size_t num) {
if (num > 0 && metrics_.output_buffer_size)
metrics_.output_buffer_size->inc(static_cast<int64_t>(num));
}
void dropped_messages(size_t num) {
if (num > 0 && metrics_.output_buffer_size)
metrics_.output_buffer_size->dec(static_cast<int64_t>(num));
}
void shipped_messages(size_t num) {
if (num > 0 && metrics_.output_buffer_size) {
metrics_.output_buffer_size->dec(static_cast<int64_t>(num));
metrics_.pushed_elements->inc(static_cast<int64_t>(num));
}
}
protected:
bool insert_path(unique_path_ptr ptr) override;
void for_each_path_impl(path_visitor& f) override;
bool check_paths_impl(path_algorithm algo,
path_predicate& pred) const noexcept override;
// -- member variables -------------------------------------------------------
map_type paths_;
metrics_t metrics_;
};
} // namespace caf
// 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 <cstdint>
#include <utility>
#include <vector>
#include "caf/actor_addr.hpp"
#include "caf/actor_control_block.hpp"
#include "caf/detail/core_export.hpp"
#include "caf/detail/type_list.hpp"
#include "caf/message.hpp"
#include "caf/stream_priority.hpp"
#include "caf/stream_slot.hpp"
#include "caf/tag/boxing_type.hpp"
#include "caf/variant.hpp"
namespace caf {
/// Transmits stream data.
struct downstream_msg_batch {
/// Allows the testing DSL to unbox this type automagically.
using outer_type = downstream_msg;
/// Size of the type-erased vector<T> (used credit).
int32_t xs_size;
/// A type-erased vector<T> containing the elements of the batch.
message xs;
/// ID of this batch (ascending numbering).
int64_t id;
};
/// Orderly shuts down a stream after receiving an ACK for the last batch.
struct downstream_msg_close {
/// Allows the testing DSL to unbox this type automagically.
using outer_type = downstream_msg;
};
/// Propagates a fatal error from sources to sinks.
struct downstream_msg_forced_close {
/// Allows the testing DSL to unbox this type automagically.
using outer_type = downstream_msg;
/// Reason for shutting down the stream.
error reason;
};
/// Stream messages that travel downstream, i.e., batches and close messages.
struct CAF_CORE_EXPORT downstream_msg : tag::boxing_type {
// -- nested types -----------------------------------------------------------
using batch = downstream_msg_batch;
using close = downstream_msg_close;
using forced_close = downstream_msg_forced_close;
// -- member types -----------------------------------------------------------
/// Lists all possible options for the payload.
using alternatives = detail::type_list<batch, close, forced_close>;
/// Stores one of `alternatives`.
using content_type = variant<batch, close, forced_close>;
// -- constructors, destructors, and assignment operators --------------------
template <class T>
downstream_msg(stream_slots s, actor_addr addr, T&& x)
: slots(s), sender(std::move(addr)), content(std::forward<T>(x)) {
// nop
}
downstream_msg() = default;
downstream_msg(downstream_msg&&) = default;
downstream_msg(const downstream_msg&) = default;
downstream_msg& operator=(downstream_msg&&) = default;
downstream_msg& operator=(const downstream_msg&) = default;
// -- member variables -------------------------------------------------------
/// ID of the affected stream.
stream_slots slots;
/// Address of the sender. Identifies the up- or downstream actor sending
/// this message. Note that abort messages can get send after `sender`
/// already terminated. Hence, `current_sender()` can be `nullptr`, because
/// no strong pointers can be formed any more and receiver would receive an
/// anonymous message.
actor_addr sender;
/// Palyoad of the message.
content_type content;
};
/// Allows the testing DSL to unbox `downstream_msg` automagically.
template <class T>
const T& get(const downstream_msg& x) {
return get<T>(x.content);
}
/// Allows the testing DSL to check whether `downstream_msg` holds a `T`.
template <class T>
bool is(const downstream_msg& x) {
return holds_alternative<T>(x.content);
}
/// @relates downstream_msg
template <class T, class... Ts>
detail::enable_if_tt<detail::tl_contains<downstream_msg::alternatives, T>,
downstream_msg>
make(stream_slots slots, actor_addr addr, Ts&&... xs) {
return {slots, std::move(addr), T{std::forward<Ts>(xs)...}};
}
/// @relates downstream_msg::batch
template <class Inspector>
bool inspect(Inspector& f, downstream_msg::batch& x) {
return f.object(x).pretty_name("batch").fields(f.field("size", x.xs_size),
f.field("xs", x.xs),
f.field("id", x.id));
}
/// @relates downstream_msg::close
template <class Inspector>
bool inspect(Inspector& f, downstream_msg::close& x) {
return f.object(x).pretty_name("close").fields();
}
/// @relates downstream_msg::forced_close
template <class Inspector>
bool inspect(Inspector& f, downstream_msg::forced_close& x) {
return f.object(x)
.pretty_name("forced_close")
.fields(f.field("reason", x.reason));
}
/// @relates downstream_msg
template <class Inspector>
bool inspect(Inspector& f, downstream_msg& x) {
return f.object(x)
.pretty_name("downstream_msg")
.fields(f.field("slots", x.slots), f.field("sender", x.sender),
f.field("content", x.content));
}
} // namespace caf
......@@ -7,22 +7,4 @@
#include "caf/flow/observable.hpp"
#include "caf/flow/observer.hpp"
namespace caf::flow {
/// Combines the items emitted from `pub` and `pubs...` to appear as a single
/// stream of items.
template <class Observable, class... Observables>
observable<typename Observable::output_type>
concat(Observable x, Observables... xs) {
using output_type = output_type_t<Observable>;
static_assert(
(std::is_same_v<output_type, output_type_t<Observables>> && ...));
auto hdl = std::move(x).as_observable();
auto ptr = make_counted<merger_impl<output_type>>(hdl.ptr()->ctx());
ptr->concat_mode(true);
ptr->add(std::move(hdl));
(ptr->add(std::move(xs).as_observable()), ...);
return observable<output_type>{std::move(ptr)};
}
} // namespace caf::flow
namespace caf::flow {} // namespace caf::flow
// 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 "caf/detail/core_export.hpp"
namespace caf::flow {
/// An object that lives on a @ref coordinator.
class CAF_CORE_EXPORT coordinated {
public:
// -- constructors, destructors, and assignment operators --------------------
virtual ~coordinated();
// -- reference counting -----------------------------------------------------
/// Increases the reference count of the coordinated.
virtual void ref_coordinated() const noexcept = 0;
/// Decreases the reference count of the coordinated and destroys the object
/// if necessary.
virtual void deref_coordinated() const noexcept = 0;
friend void intrusive_ptr_add_ref(const coordinated* ptr) noexcept {
ptr->ref_coordinated();
}
friend void intrusive_ptr_release(const coordinated* ptr) noexcept {
ptr->deref_coordinated();
}
};
} // namespace caf::flow
......@@ -5,11 +5,11 @@
#pragma once
#include "caf/action.hpp"
#include "caf/async/execution_context.hpp"
#include "caf/detail/core_export.hpp"
#include "caf/flow/fwd.hpp"
#include "caf/flow/subscription.hpp"
#include "caf/intrusive_ptr.hpp"
#include "caf/ref_counted.hpp"
#include "caf/timespan.hpp"
#include <chrono>
......@@ -20,12 +20,8 @@ namespace caf::flow {
/// Coordinates any number of co-located observables and observers. The
/// co-located objects never need to synchronize calls to other co-located
/// objects since the coordinator guarantees synchronous execution.
class CAF_CORE_EXPORT coordinator {
class CAF_CORE_EXPORT coordinator : public async::execution_context {
public:
// -- friends ----------------------------------------------------------------
friend class subscription_impl;
// -- member types -----------------------------------------------------------
/// A time point of the monotonic clock.
......@@ -35,35 +31,11 @@ public:
virtual ~coordinator();
// -- reference counting -----------------------------------------------------
/// Increases the reference count of the coordinator.
virtual void ref_coordinator() const noexcept = 0;
/// Decreases the reference count of the coordinator and destroys the object
/// if necessary.
virtual void deref_coordinator() const noexcept = 0;
friend void intrusive_ptr_add_ref(const coordinator* ptr) noexcept {
ptr->ref_coordinator();
}
friend void intrusive_ptr_release(const coordinator* ptr) noexcept {
ptr->deref_coordinator();
}
// -- conversions ------------------------------------------------------------
/// Returns a factory object for new observable objects on this coordinator.
[[nodiscard]] observable_builder make_observable();
// -- lifetime management ----------------------------------------------------
/// Asks the coordinator to keep its event loop running until `obj` becomes
/// disposed since it depends on external events or produces events that are
/// visible to outside observers.
virtual void watch(disposable what) = 0;
// -- time -------------------------------------------------------------------
/// Returns the current time on the monotonic clock of this coordinator.
......@@ -71,17 +43,6 @@ public:
// -- scheduling of actions --------------------------------------------------
/// Schedules an action for execution on this coordinator. This member
/// function may get called from external sources or threads.
/// @thread-safe
virtual void schedule(action what) = 0;
///@copydoc schedule
template <class F>
void schedule_fn(F&& what) {
return schedule(make_action(std::forward<F>(what)));
}
/// Delays execution of an action until all pending actions were executed. May
/// call `schedule`.
/// @param what The action for delayed execution.
......
......@@ -38,14 +38,29 @@ class observer;
template <class T>
class observable;
template <class In, class Out>
class processor;
template <class Materializer, class... Steps>
class observable_def;
template <class Generator>
class generation_materializer;
/// A blueprint for an @ref observer that generates items and applies any number
/// of processing steps immediately before emitting them.
template <class Generator, class... Steps>
class generation;
using generation = observable_def<generation_materializer<Generator>, Steps...>;
template <class Input>
class transformation_materializer;
/// A blueprint for an @ref observer that applies a series of transformation
/// steps to its inputs and emits the results.
template <class Step, class... Steps>
class transformation;
using transformation
= observable_def<transformation_materializer<typename Step::input_type>, Step,
Steps...>;
template <class T>
class connectable;
template <class T>
struct is_observable {
......@@ -57,18 +72,8 @@ struct is_observable<observable<T>> {
static constexpr bool value = true;
};
template <class Step, class... Steps>
struct is_observable<transformation<Step, Steps...>> {
static constexpr bool value = true;
};
template <class Generator, class... Steps>
struct is_observable<generation<Generator, Steps...>> {
static constexpr bool value = true;
};
template <class In, class Out>
struct is_observable<processor<In, Out>> {
template <class Materializer, class... Steps>
struct is_observable<observable_def<Materializer, Steps...>> {
static constexpr bool value = true;
};
......@@ -80,29 +85,6 @@ struct is_observable<single<T>> {
template <class T>
constexpr bool is_observable_v = is_observable<T>::value;
template <class T>
struct is_observer {
static constexpr bool value = false;
};
template <class T>
struct is_observer<observer<T>> {
static constexpr bool value = true;
};
template <class Step, class... Steps>
struct is_observer<transformation<Step, Steps...>> {
static constexpr bool value = true;
};
template <class In, class Out>
struct is_observer<processor<In, Out>> {
static constexpr bool value = true;
};
template <class T>
constexpr bool is_observer_v = is_observer<T>::value;
class observable_builder;
template <class T>
......@@ -113,16 +95,19 @@ struct input_type_oracle {
template <class T>
using input_type_t = typename input_type_oracle<T>::type;
template <class...>
struct output_type_oracle;
template <class T>
struct output_type_oracle {
struct output_type_oracle<T> {
using type = typename T::output_type;
};
template <class T>
using output_type_t = typename output_type_oracle<T>::type;
template <class T0, class T1, class... Ts>
struct output_type_oracle<T0, T1, Ts...> : output_type_oracle<T1, Ts...> {};
template <class T>
class merger_impl;
template <class... Ts>
using output_type_t = typename output_type_oracle<Ts...>::type;
template <class>
struct has_impl_include {
......@@ -140,8 +125,8 @@ struct assert_scheduled_actor_hdr {
"include 'caf/scheduled_actor/flow.hpp' for this method");
};
template <class T>
template <class T, class V = T>
using assert_scheduled_actor_hdr_t
= std::enable_if_t<assert_scheduled_actor_hdr<T>::value, T>;
= std::enable_if_t<assert_scheduled_actor_hdr<T>::value, V>;
} // namespace caf::flow
// 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
namespace caf::flow::gen {
/// A generator that emits nothing and calls `on_complete` immediately.
template <class T>
class empty {
public:
using output_type = T;
template <class Step, class... Steps>
void pull(size_t, Step& step, Steps&... steps) {
step.on_complete(steps...);
}
};
} // namespace caf::flow::gen
// 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 "caf/detail/type_traits.hpp"
#include <type_traits>
namespace caf::flow::gen {
/// A generator that emits values from a function object.
template <class F>
class from_callable {
public:
using callable_res_t = std::invoke_result_t<F>;
static constexpr bool boxed_output = detail::is_optional_v<callable_res_t>;
using output_type = detail::unboxed_t<callable_res_t>;
explicit from_callable(F fn) : fn_(std::move(fn)) {
// nop
}
from_callable(from_callable&&) = default;
from_callable(const from_callable&) = default;
from_callable& operator=(from_callable&&) = default;
from_callable& operator=(const from_callable&) = default;
template <class Step, class... Steps>
void pull(size_t n, Step& step, Steps&... steps) {
for (size_t i = 0; i < n; ++i) {
if constexpr (boxed_output) {
auto val = fn_();
if (!val) {
step.on_complete(steps...);
return;
} else if (!step.on_next(*val, steps...))
return;
} else {
if (!step.on_next(fn_(), steps...))
return;
}
}
}
private:
F fn_;
};
} // namespace caf::flow::gen
// 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 <memory>
namespace caf::flow::gen {
/// A generator that emits values from a vector.
template <class Container>
class from_container {
public:
using output_type = typename Container::value_type;
explicit from_container(Container&& values) {
values_ = std::make_shared<Container>(std::move(values));
pos_ = values_->begin();
}
from_container() = default;
from_container(from_container&&) = default;
from_container(const from_container&) = default;
from_container& operator=(from_container&&) = default;
from_container& operator=(const from_container&) = default;
template <class Step, class... Steps>
void pull(size_t n, Step& step, Steps&... steps) {
auto end = values_->end();
while (pos_ != end && n > 0) {
if (!step.on_next(*pos_++, steps...))
return;
--n;
}
if (pos_ == end)
step.on_complete(steps...);
}
private:
std::shared_ptr<Container> values_;
typename Container::const_iterator pos_;
};
} // namespace caf::flow::gen
// 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
namespace caf::flow::gen {
/// A generator that emits ascending values.
template <class T>
class iota {
public:
using output_type = T;
explicit iota(T init) : value_(std::move(init)) {
// nop
}
iota(iota&&) = default;
iota(const iota&) = default;
iota& operator=(iota&&) = default;
iota& operator=(const iota&) = default;
template <class Step, class... Steps>
void pull(size_t n, Step& step, Steps&... steps) {
for (size_t i = 0; i < n; ++i)
if (!step.on_next(value_++, steps...))
return;
}
private:
T value_;
};
} // namespace caf::flow::gen
// 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
namespace caf::flow::gen {
/// A generator that emits a single value once.
template <class T>
class just {
public:
using output_type = T;
explicit just(T value) : value_(std::move(value)) {
// nop
}
just(just&&) = default;
just(const just&) = default;
just& operator=(just&&) = default;
just& operator=(const just&) = default;
template <class Step, class... Steps>
void pull([[maybe_unused]] size_t n, Step& step, Steps&... steps) {
if (step.on_next(value_, steps...))
step.on_complete(steps...);
}
private:
T value_;
};
} // namespace caf::flow::gen
// 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
namespace caf::flow::gen {
/// A generator that emits the same value repeatedly.
template <class T>
class repeat {
public:
using output_type = T;
explicit repeat(T value) : value_(std::move(value)) {
// nop
}
repeat(repeat&&) = default;
repeat(const repeat&) = default;
repeat& operator=(repeat&&) = default;
repeat& operator=(const repeat&) = default;
template <class Step, class... Steps>
void pull(size_t n, Step& step, Steps&... steps) {
for (size_t i = 0; i < n; ++i)
if (!step.on_next(value_, steps...))
return;
}
private:
T value_;
};
} // namespace caf::flow::gen
// 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 "caf/flow/fwd.hpp"
#include "caf/flow/observable_decl.hpp"
#include "caf/flow/op/mcast.hpp"
#include "caf/intrusive_ptr.hpp"
#include "caf/make_counted.hpp"
#include <cstdint>
namespace caf::flow {
template <class T>
class item_publisher {
public:
using impl_ptr = intrusive_ptr<op::mcast<T>>;
explicit item_publisher(coordinator* ctx) {
pimpl_ = make_counted<op::mcast<T>>(ctx);
}
explicit item_publisher(impl_ptr ptr) noexcept : pimpl_(std::move(ptr)) {
// nop
}
item_publisher(item_publisher&&) noexcept = default;
item_publisher& operator=(item_publisher&&) noexcept = default;
item_publisher(const item_publisher&) = delete;
item_publisher& operator=(const item_publisher&) = delete;
~item_publisher() {
if (pimpl_)
pimpl_->close();
}
/// Pushes an item to all subscribed observers. The publisher drops the item
/// if no subscriber exists.
void push(const T& item) {
pimpl_->push_all(item);
}
/// Pushes the items in range `[first, last)` to all subscribed observers. The
/// publisher drops the items if no subscriber exists.
template <class Iterator, class Sentinel>
void push(Iterator first, Sentinel last) {
while (first != last)
push(*first++);
}
/// Pushes the items from the initializer list to all subscribed observers.
/// The publisher drops the items if no subscriber exists.
void push(std::initializer_list<T> items) {
for (auto& item : items)
push(item);
}
/// Closes the publisher, eventually emitting on_complete on all observers.
void close() {
pimpl_->close();
}
/// Closes the publisher, eventually emitting on_error on all observers.
void abort(const error& reason) {
pimpl_->abort(reason);
}
/// Queries how many items the publisher may emit immediately to subscribed
/// observers.
size_t demand() const noexcept {
return pimpl_->min_demand();
}
/// Queries how many items are currently waiting in a buffer until the
/// observer requests additional items.
size_t buffered() const noexcept {
return pimpl_->max_buffered();
}
/// Queries whether there is at least one observer subscribed to the operator.
bool has_observers() const noexcept {
return pimpl_->has_observers();
}
/// Converts the publisher to an @ref observable.
observable<T> as_observable() const {
return observable<T>{pimpl_};
}
/// Subscribes a new @ref observer to the output of the publisher.
disposable subscribe(observer<T> out) {
return pimpl_->subscribe(out);
}
private:
impl_ptr pimpl_;
};
} // namespace caf::flow
......@@ -11,17 +11,5 @@ namespace caf::flow {
/// Combines the items emitted from `pub` and `pubs...` to appear as a single
/// stream of items.
template <class Observable, class... Observables>
observable<typename Observable::output_type>
merge(Observable x, Observables... xs) {
using output_type = output_type_t<Observable>;
static_assert(
(std::is_same_v<output_type, output_type_t<Observables>> && ...));
auto hdl = std::move(x).as_observable();
auto ptr = make_counted<merger_impl<output_type>>(hdl.ptr()->ctx());
ptr->add(std::move(hdl));
(ptr->add(std::move(xs).as_observable()), ...);
return observable<output_type>{std::move(ptr)};
}
} // namespace caf::flow
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
......@@ -41,7 +41,7 @@ constexpr bool is_active(observable_state x) noexcept {
CAF_CORE_EXPORT std::string to_string(observable_state);
/// @relates observable_state
CAF_CORE_EXPORT bool from_string(string_view, observable_state&);
CAF_CORE_EXPORT bool from_string(std::string_view, observable_state&);
/// @relates observable_state
CAF_CORE_EXPORT bool from_integer(std::underlying_type_t<observable_state>,
......
This diff is collapsed.
......@@ -18,16 +18,26 @@ enum class observer_state {
/// Indicates that on_complete was called.
completed,
/// Indicates that on_error was called.
aborted,
/// Indicates that dispose was called.
disposed,
aborted
};
/// Returns whether `x` represents a final state, i.e., `completed`, `aborted`
/// or `disposed`.
constexpr bool is_final(observer_state x) noexcept {
return static_cast<int>(x) >= static_cast<int>(observer_state::completed);
}
/// Returns whether `x` represents an active state, i.e., `idle` or
/// `subscribed`.
constexpr bool is_active(observer_state x) noexcept {
return static_cast<int>(x) <= static_cast<int>(observer_state::subscribed);
}
/// @relates sec
CAF_CORE_EXPORT std::string to_string(observer_state);
/// @relates observer_state
CAF_CORE_EXPORT bool from_string(string_view, observer_state&);
CAF_CORE_EXPORT bool from_string(std::string_view, observer_state&);
/// @relates observer_state
CAF_CORE_EXPORT bool from_integer(std::underlying_type_t<observer_state>,
......
The namespace `caf::flow::op` contains (private) implementation classes for
various operators.
// 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 "caf/flow/observer.hpp"
#include "caf/flow/subscription.hpp"
namespace caf::flow::op {
/// Abstract base type for all flow operators that implement the *observable*
/// concept.
template <class T>
class base : public coordinated {
public:
/// The type of observed values.
using output_type = T;
/// Returns the @ref coordinator that executes this flow operator.
virtual coordinator* ctx() const noexcept = 0;
/// Subscribes a new observer to the operator.
virtual disposable subscribe(observer<T> what) = 0;
};
} // namespace caf::flow::op
This diff is collapsed.
// 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 "caf/flow/observer.hpp"
#include "caf/flow/op/hot.hpp"
#include "caf/flow/subscription.hpp"
#include "caf/none.hpp"
#include <memory>
#include <utility>
#include <variant>
#include <vector>
namespace caf::flow::op {
/// State shared between one multicast operator and one subscribed observer.
template <class T>
struct cell_sub_state {
std::variant<none_t, unit_t, T, error> content;
std::vector<observer<T>> listeners;
void set_null() {
CAF_ASSERT(std::holds_alternative<none_t>(content));
content = unit;
std::vector<observer<T>> xs;
xs.swap(listeners);
for (auto& listener : xs) {
listener.on_complete();
}
}
void set_value(T item) {
CAF_ASSERT(std::holds_alternative<none_t>(content));
content = std::move(item);
auto& ref = std::get<T>(content);
std::vector<observer<T>> xs;
xs.swap(listeners);
for (auto& listener : xs) {
listener.on_next(ref);
listener.on_complete();
}
}
void set_error(error what) {
CAF_ASSERT(std::holds_alternative<none_t>(content));
content = std::move(what);
auto& ref = std::get<error>(content);
std::vector<observer<T>> xs;
xs.swap(listeners);
for (auto& listener : xs)
listener.on_error(ref);
}
void listen(observer<T> listener) {
switch (content.index()) {
case 1:
listener.on_complete();
break;
case 2:
listener.on_next(std::get<T>(content));
listener.on_complete();
break;
case 3:
listener.on_error(std::get<error>(content));
break;
default:
listeners.emplace_back(std::move(listener));
}
}
void drop(const observer<T>& listener) {
if (auto i = std::find(listeners.begin(), listeners.end(), listener);
i != listeners.end())
listeners.erase(i);
}
};
template <class T>
using cell_sub_state_ptr = std::shared_ptr<cell_sub_state<T>>;
template <class T>
class cell_sub : public subscription::impl_base {
public:
// -- constructors, destructors, and assignment operators --------------------
cell_sub(coordinator* ctx, cell_sub_state_ptr<T> state, observer<T> out)
: ctx_(ctx), state_(std::move(state)), out_(std::move(out)) {
// nop
}
// -- implementation of subscription -----------------------------------------
bool disposed() const noexcept override {
return !state_;
}
void dispose() override {
if (state_) {
state_->drop(out_);
state_ = nullptr;
out_ = nullptr;
}
}
void request(size_t) override {
if (!listening_) {
listening_ = true;
ctx_->delay_fn([state = state_, out = out_] { //
state->listen(std::move(out));
});
}
}
private:
coordinator* ctx_;
bool listening_ = false;
cell_sub_state_ptr<T> state_;
observer<T> out_;
};
// Base type for *hot* operators that multicast data to subscribed observers.
template <class T>
class cell : public hot<T> {
public:
// -- member types -----------------------------------------------------------
using super = hot<T>;
using state_type = cell_sub_state<T>;
using state_ptr_type = cell_sub_state_ptr<T>;
using observer_type = observer<T>;
// -- constructors, destructors, and assignment operators --------------------
explicit cell(coordinator* ctx)
: super(ctx), state_(std::make_shared<state_type>()) {
// nop
}
void set_null() {
state_->set_null();
}
void set_value(T item) {
state_->set_value(std::move(item));
}
void set_error(error what) {
state_->set_error(what);
}
disposable subscribe(observer<T> out) override {
auto ptr = make_counted<cell_sub<T>>(super::ctx_, state_, out);
out.on_subscribe(subscription{ptr});
return disposable{std::move(ptr)};
}
protected:
cell_sub_state_ptr<T> state_;
};
} // namespace caf::flow::op
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
......@@ -25,11 +25,13 @@ public:
void run();
size_t run_some();
// -- reference counting -----------------------------------------------------
void ref_coordinator() const noexcept override;
void ref_execution_context() const noexcept override;
void deref_coordinator() const noexcept override;
void deref_execution_context() const noexcept override;
friend void intrusive_ptr_add_ref(const scoped_coordinator* ptr) {
ptr->ref();
......
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
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