Commit 4c68deb8 authored by Dominik Charousset's avatar Dominik Charousset

Add infrastructure for a new flow API

parent b10ee098
...@@ -42,7 +42,7 @@ configure_file("${PROJECT_SOURCE_DIR}/cmake/build_config.hpp.in" ...@@ -42,7 +42,7 @@ configure_file("${PROJECT_SOURCE_DIR}/cmake/build_config.hpp.in"
install(FILES "${CMAKE_CURRENT_BINARY_DIR}/caf/detail/build_config.hpp" install(FILES "${CMAKE_CURRENT_BINARY_DIR}/caf/detail/build_config.hpp"
DESTINATION "${CMAKE_INSTALL_INCLUDEDIR}/caf/detail/") DESTINATION "${CMAKE_INSTALL_INCLUDEDIR}/caf/detail/")
# -- add targets --------------------------------------------------------------- #-- add targets ----------------------------------------------------------------
caf_add_component( caf_add_component(
core core
...@@ -81,6 +81,9 @@ caf_add_component( ...@@ -81,6 +81,9 @@ caf_add_component(
src/actor_registry.cpp src/actor_registry.cpp
src/actor_system.cpp src/actor_system.cpp
src/actor_system_config.cpp src/actor_system_config.cpp
src/async/batch.cpp
src/async/consumer.cpp
src/async/producer.cpp
src/attachable.cpp src/attachable.cpp
src/behavior.cpp src/behavior.cpp
src/binary_deserializer.cpp src/binary_deserializer.cpp
...@@ -141,6 +144,9 @@ caf_add_component( ...@@ -141,6 +144,9 @@ caf_add_component(
src/error.cpp src/error.cpp
src/event_based_actor.cpp src/event_based_actor.cpp
src/execution_unit.cpp src/execution_unit.cpp
src/flow/coordinator.cpp
src/flow/scoped_coordinator.cpp
src/flow/subscription.cpp
src/forwarding_actor_proxy.cpp src/forwarding_actor_proxy.cpp
src/group.cpp src/group.cpp
src/group_manager.cpp src/group_manager.cpp
...@@ -221,6 +227,7 @@ caf_add_component( ...@@ -221,6 +227,7 @@ caf_add_component(
actor_system_config actor_system_config
actor_termination actor_termination
aout aout
async.bounded_buffer
behavior behavior
binary_deserializer binary_deserializer
binary_serializer binary_serializer
...@@ -273,6 +280,13 @@ caf_add_component( ...@@ -273,6 +280,13 @@ caf_add_component(
dynamic_spawn dynamic_spawn
error error
expected expected
flow.concat
flow.concat_map
flow.flat_map
flow.for_each
flow.merge
flow.observe_on
flow.single
function_view function_view
fused_downstream_manager fused_downstream_manager
handles handles
...@@ -315,6 +329,7 @@ caf_add_component( ...@@ -315,6 +329,7 @@ caf_add_component(
policy.select_all policy.select_all
policy.select_any policy.select_any
request_timeout request_timeout
response_handle
response_promise response_promise
result result
save_inspector save_inspector
......
...@@ -587,6 +587,61 @@ public: ...@@ -587,6 +587,61 @@ public:
return res; return res;
} }
/// Creates a new, cooperatively scheduled `flow::coordinator`. The returned
/// coordinator is constructed but has not been added to the scheduler yet to
/// allow the caller to set up flows.
/// @returns A pointer to the new coordinator and a function object that the
/// caller must invoke to launch the coordinator. After the
/// coordinator started running, the caller *must not* access the
/// pointer again.
template <class Impl, spawn_options = no_spawn_options, class... Ts>
auto make_flow_coordinator(Ts&&... xs) {
static_assert(std::is_base_of_v<scheduled_actor, Impl>,
"make_flow_coordinator only supports scheduled actors ATM");
CAF_SET_LOGGER_SYS(this);
actor_config cfg{dummy_execution_unit(), nullptr};
auto res = make_actor<Impl>(next_actor_id(), node(), this, cfg,
std::forward<Ts>(xs)...);
auto ptr = static_cast<Impl*>(actor_cast<abstract_actor*>(res));
#ifdef CAF_ENABLE_ACTOR_PROFILER
profiler_add_actor(*ptr, cfg.parent);
#endif
auto launch = [res, host{cfg.host}] {
// Note: we pass `res` to this lambda instead of `ptr` to keep a strong
// reference to the actor.
static_cast<Impl*>(actor_cast<abstract_actor*>(res))
->launch(host, false, false);
};
return std::make_tuple(ptr, launch);
}
/// Creates a new, cooperatively scheduled actor. The returned actor is
/// constructed but has not been added to the scheduler yet to allow the
/// caller to set up any additional logic on the actor before it starts.
/// @returns A pointer to the new actor and a function object that the caller
/// must invoke to launch the actor. After the actor started running,
/// the caller *must not* access the pointer again.
template <class Impl, spawn_options = no_spawn_options, class... Ts>
auto spawn_inactive(Ts&&... xs) {
static_assert(std::is_base_of_v<scheduled_actor, Impl>,
"only scheduled actors may get spawned inactively");
CAF_SET_LOGGER_SYS(this);
actor_config cfg{dummy_execution_unit(), nullptr};
auto res = make_actor<Impl>(next_actor_id(), node(), this, cfg,
std::forward<Ts>(xs)...);
auto ptr = static_cast<Impl*>(actor_cast<abstract_actor*>(res));
#ifdef CAF_ENABLE_ACTOR_PROFILER
profiler_add_actor(*ptr, cfg.parent);
#endif
auto launch = [res, host{cfg.host}] {
// Note: we pass `res` to this lambda instead of `ptr` to keep a strong
// reference to the actor.
static_cast<Impl*>(actor_cast<abstract_actor*>(res))
->launch(host, false, false);
};
return std::make_tuple(ptr, launch);
}
void profiler_add_actor(const local_actor& self, const local_actor* parent) { void profiler_add_actor(const local_actor& self, const local_actor* parent) {
if (profiler_) if (profiler_)
profiler_->add_actor(self, parent); profiler_->add_actor(self, parent);
......
// 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 <cstdint>
#include <cstdlib>
#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"
#include "caf/intrusive_ptr.hpp"
#include "caf/raise_error.hpp"
#include "caf/span.hpp"
#include "caf/type_id.hpp"
#ifdef CAF_CLANG
# pragma clang diagnostic push
# pragma clang diagnostic ignored "-Wc99-extensions"
#elif defined(CAF_GCC)
# pragma GCC diagnostic push
# pragma GCC diagnostic ignored "-Wpedantic"
#elif defined(CAF_MSVC)
# pragma warning(push)
# pragma warning(disable : 4200)
#endif
namespace caf::async {
/// A reference-counted container for transferring items from publishers to
/// subscribers.
class CAF_CORE_EXPORT batch {
public:
template <class T>
friend batch make_batch(span<const T> items);
using item_destructor = void (*)(type_id_t, uint16_t, size_t, byte*);
batch() = default;
batch(batch&&) = default;
batch(const batch&) = default;
batch& operator=(batch&&) = default;
batch& operator=(const batch&) = default;
size_t size() const noexcept {
return data_ ? data_->size() : 0;
}
bool empty() const noexcept {
return data_ ? data_->size() == 0 : true;
}
template <class T>
span<T> items() noexcept {
return data_ ? data_->items<T>() : span<T>{};
}
template <class T>
span<const T> items() const noexcept {
return data_ ? data_->items<T>() : span<const T>{};
}
bool save(serializer& f) const;
bool save(binary_serializer& f) const;
bool load(deserializer& f);
bool load(binary_deserializer& f);
void swap(batch& other) {
data_.swap(other.data_);
}
private:
template <class Inspector>
bool save_impl(Inspector& f) const;
template <class Inspector>
bool load_impl(Inspector& f);
class data {
public:
data() = delete;
data(const data&) = delete;
data& operator=(const data&) = delete;
data(item_destructor destroy_items, type_id_t item_type, uint16_t item_size,
size_t size)
: rc_(1),
destroy_items_(destroy_items),
item_type_(item_type),
item_size_(item_size),
size_(size) {
// nop
}
~data() {
if (size_ > 0)
destroy_items_(item_type_, item_size_, size_, storage_);
}
// -- reference counting ---------------------------------------------------
bool unique() const noexcept {
return rc_ == 1;
}
void ref() const noexcept {
rc_.fetch_add(1, std::memory_order_relaxed);
}
void deref() noexcept {
if (unique() || rc_.fetch_sub(1, std::memory_order_acq_rel) == 1) {
this->~data();
free(this);
}
}
friend void intrusive_ptr_add_ref(const data* ptr) {
ptr->ref();
}
friend void intrusive_ptr_release(data* ptr) {
ptr->deref();
}
// -- properties -----------------------------------------------------------
type_id_t item_type() {
return item_type_;
}
size_t size() noexcept {
return size_;
}
byte* storage() noexcept {
return storage_;
}
const byte* storage() const noexcept {
return storage_;
}
template <class T>
span<T> items() noexcept {
return {reinterpret_cast<T*>(storage_), size_};
}
template <class T>
span<const T> items() const noexcept {
return {reinterpret_cast<const T*>(storage_), size_};
}
// -- serialization --------------------------------------------------------
/// @pre `size() > 0`
template <class Inspector>
bool save(Inspector& sink) const;
template <class Inspector>
bool load(Inspector& source);
private:
mutable std::atomic<size_t> rc_;
item_destructor destroy_items_;
type_id_t item_type_;
uint16_t item_size_;
size_t size_;
byte storage_[];
};
explicit batch(intrusive_ptr<data> ptr) : data_(std::move(ptr)) {
// nop
}
intrusive_ptr<data> data_;
};
template <class Inspector>
auto inspect(Inspector& f, batch& x)
-> std::enable_if_t<!Inspector::is_loading, decltype(x.save(f))> {
return x.save(f);
}
template <class Inspector>
auto inspect(Inspector& f, batch& x)
-> std::enable_if_t<Inspector::is_loading, decltype(x.load(f))> {
return x.load(f);
}
template <class T>
batch make_batch(span<const T> items) {
static_assert(sizeof(T) < 0xFFFF);
auto total_size = sizeof(batch::data) + (items.size() * sizeof(T));
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 ptr = reinterpret_cast<T*>(storage);
std::destroy(ptr, ptr + size);
};
intrusive_ptr<batch::data> ptr{
new (vptr) batch::data(destroy_items, type_id_or_invalid<T>(),
static_cast<uint16_t>(sizeof(T)), items.size()),
false};
std::uninitialized_copy(items.begin(), items.end(),
reinterpret_cast<T*>(ptr->storage()));
return batch{std::move(ptr)};
}
} // namespace caf::async
#ifdef CAF_CLANG
# pragma clang diagnostic pop
#elif defined(CAF_GCC)
# pragma GCC diagnostic pop
#elif defined(CAF_MSVC)
# pragma warning(pop)
#endif
// 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 <condition_variable>
#include <type_traits>
#include "caf/async/observer_buffer.hpp"
namespace caf::async {
/// Consumes all elements from a publisher and blocks the current thread until
/// completion.
template <class T>
class blocking_observer : public observer_buffer<T> {
public:
using super = observer_buffer<T>;
using super::super;
template <class OnNext, class OnError, class OnComplete>
auto run(OnNext fun, OnError err, OnComplete fin) {
static_assert(std::is_invocable_v<OnNext, T>,
"OnNext handlers must have signature 'void(T)' or 'bool(T)'");
static_assert(std::is_invocable_v<OnError, error>,
"OnError handlers must have signature 'void(const error&)'");
static_assert(std::is_invocable_v<OnComplete>,
"OnComplete handlers must have signature 'void()'");
auto wait_fn = [this](std::unique_lock<std::mutex>& guard) {
cv_.wait(guard);
};
for (;;) {
auto [val, done, what] = super::wait_with(wait_fn);
if (val) {
using fun_res = decltype(fun(*val));
if constexpr (std::is_same_v<bool, fun_res>) {
if (!fun(*val)) {
std::unique_lock guard{super::mtx_};
if (super::sub_)
super::sub_->cancel();
return;
}
} else {
static_assert(
std::is_same_v<void, fun_res>,
"OnNext handlers must have signature 'void(T)' or 'bool(T)'");
fun(*val);
}
} else if (done) {
if (!what)
fin();
else
err(*what);
return;
}
}
}
private:
std::condition_variable cv_;
void wakeup(std::unique_lock<std::mutex>&) override {
cv_.notify_all();
}
};
} // namespace caf::async
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/detail/core_export.hpp"
#include "caf/intrusive_ptr.hpp"
namespace caf::async {
/// Base type for asynchronous consumers of an event source.
class CAF_CORE_EXPORT consumer {
public:
virtual ~consumer();
/// Called to signal that the producer started emitting items.
virtual void on_producer_ready() = 0;
/// Called to signal to the consumer that the producer added an item to a
/// previously empty source or completed the flow of events.
virtual void on_producer_wakeup() = 0;
/// Increases the reference count of the consumer.
virtual void ref_consumer() const noexcept = 0;
/// Decreases the reference count of the consumer and destroys the object if
/// necessary.
virtual void deref_consumer() const noexcept = 0;
};
/// @relates consumer
inline void intrusive_ptr_add_ref(const consumer* p) noexcept {
p->ref_consumer();
}
/// @relates consumer
inline void intrusive_ptr_release(const consumer* p) noexcept {
p->deref_consumer();
}
/// @relates consumer
using consumer_ptr = intrusive_ptr<consumer>;
} // 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/fwd.hpp>
namespace caf::async {
// -- classes ------------------------------------------------------------------
class batch;
class consumer;
class producer;
// -- template classes ---------------------------------------------------------
template <class T>
class bounded_buffer;
template <class T>
class consumer_resource;
template <class T>
class producer_resource;
// -- free function templates --------------------------------------------------
template <class T>
batch make_batch(span<const T> items);
} // 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 <list>
#include <mutex>
#include <tuple>
#include "caf/async/batch.hpp"
#include "caf/defaults.hpp"
#include "caf/flow/observer.hpp"
#include "caf/flow/subscription.hpp"
namespace caf::async {
/// Enables buffered consumption of published items.
template <class T>
class observer_buffer : public flow::observer_impl<T> {
public:
observer_buffer() {
// nop
}
void on_complete() override {
std::unique_lock guard{mtx_};
if (!done_) {
sub_ = nullptr;
done_ = true;
deinit(guard);
}
}
void on_error(const error& what) override {
std::unique_lock guard{mtx_};
if (!done_) {
sub_ = nullptr;
done_ = true;
err_ = what;
deinit(guard);
}
}
void on_next(span<const T> items) override {
on_batch(make_batch(items));
}
void on_batch(const batch& buf) override {
std::list<batch> ls;
ls.emplace_back(buf);
std::unique_lock guard{mtx_};
batches_.splice(batches_.end(), std::move(ls));
if (batches_.size() == 1)
wakeup(guard);
}
void on_attach(flow::subscription sub) override {
CAF_ASSERT(sub.valid());
std::unique_lock guard{mtx_};
sub_ = std::move(sub);
init(guard);
}
bool has_data() {
if (local.pos_ != local.end_) {
return true;
} else {
std::unique_lock guard{mtx_};
return !batches_.empty();
}
}
/// Tries to fetch the next value. If no value exists, the first element in
/// the tuple is `nullptr`. The second value indicates whether the stream was
/// closed. If the stream was closed, the third value is `nullptr` if
/// `on_complete` was called and a pointer to the error if `on_error` was
/// called.
std::tuple<const T*, bool, const error*> poll() {
if (local.pos_ != local.end_) {
auto res = local.pos_;
if (++local.pos_ == local.end_) {
std::unique_lock guard{mtx_};
if (sub_)
sub_.request(local.cache_.size());
}
return {res, false, nullptr};
} else if (std::unique_lock guard{mtx_}; !batches_.empty()) {
batches_.front().swap(local.cache_);
batches_.pop_front();
guard.unlock();
auto items = local.cache_.template items<T>();
CAF_ASSERT(!items.empty());
local.pos_ = items.begin();
local.end_ = items.end();
return {local.pos_++, false, nullptr};
} else if (!err_) {
return {nullptr, done_, nullptr};
} else {
return {nullptr, true, &err_};
}
}
void dispose() override {
on_complete();
}
bool disposed() const noexcept override {
std::unique_lock guard{mtx_};
return done_;
}
protected:
template <class WaitFn>
std::tuple<const T*, bool, const error*> wait_with(WaitFn wait_fn) {
if (local.pos_ != local.end_) {
auto res = local.pos_;
if (++local.pos_ == local.end_) {
std::unique_lock guard{mtx_};
if (sub_)
sub_.request(local.cache_.size());
}
return {res, false, nullptr};
} else {
std::unique_lock guard{mtx_};
while (batches_.empty() && !done_)
wait_fn(guard);
if (!batches_.empty()) {
batches_.front().swap(local.cache_);
batches_.pop_front();
guard.unlock();
auto items = local.cache_.template items<T>();
local.pos_ = items.begin();
local.end_ = items.end();
return {local.pos_++, false, nullptr};
} else if (!err_) {
return {nullptr, done_, nullptr};
} else {
return {nullptr, true, &err_};
}
}
}
/// Wraps fields that we only access from the consumer's thread.
struct local_t {
const T* pos_ = nullptr;
const T* end_ = nullptr;
batch cache_;
} local;
static_assert(sizeof(local_t) < CAF_CACHE_LINE_SIZE);
/// Avoids false sharing.
char pad_[CAF_CACHE_LINE_SIZE - sizeof(local_t)];
// -- state for consumer and publisher ---------------------------------------
/// Protects fields that we access with both the consumer and the producer.
mutable std::mutex mtx_;
flow::subscription sub_;
bool done_ = false;
error err_;
std::list<batch> batches_;
private:
virtual void init(std::unique_lock<std::mutex>&) {
sub_.request(defaults::flow::buffer_size);
}
virtual void deinit(std::unique_lock<std::mutex>& guard) {
wakeup(guard);
}
virtual void wakeup(std::unique_lock<std::mutex>&) {
// Customization point.
}
};
} // 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/detail/core_export.hpp"
#include "caf/intrusive_ptr.hpp"
namespace caf::async {
/// Base type for asynchronous producers of events.
class CAF_CORE_EXPORT producer {
public:
virtual ~producer();
/// Called to signal that the consumer started handling events.
virtual void on_consumer_ready() = 0;
/// Called to signal that the consumer stopped handling events.
virtual void on_consumer_cancel() = 0;
/// Called to signal that the consumer requests more events.
virtual void on_consumer_demand(size_t demand) = 0;
/// Increases the reference count of the producer.
virtual void ref_producer() const noexcept = 0;
/// Decreases the reference count of the producer and destroys the object if
/// necessary.
virtual void deref_producer() const noexcept = 0;
};
/// @relates producer
inline void intrusive_ptr_add_ref(const producer* p) noexcept {
p->ref_producer();
}
/// @relates producer
inline void intrusive_ptr_release(const producer* p) noexcept {
p->deref_producer();
}
/// @relates producer
using producer_ptr = intrusive_ptr<producer>;
} // namespace caf::async
...@@ -4,6 +4,7 @@ ...@@ -4,6 +4,7 @@
#pragma once #pragma once
#include "caf/allowed_unsafe_message_type.hpp"
#include "caf/config.hpp" #include "caf/config.hpp"
#include "caf/detail/type_traits.hpp" #include "caf/detail/type_traits.hpp"
#include "caf/error.hpp" #include "caf/error.hpp"
...@@ -91,4 +92,10 @@ auto make_shared_type_erased_callback(F fun) { ...@@ -91,4 +92,10 @@ auto make_shared_type_erased_callback(F fun) {
return shared_callback_ptr<signature>{std::move(res)}; return shared_callback_ptr<signature>{std::move(res)};
} }
/// Convenience type alias for an action with shared ownership.
/// @relates callback
using shared_action_ptr = shared_callback_ptr<void()>;
} // namespace caf } // namespace caf
CAF_ALLOW_UNSAFE_MESSAGE_TYPE(caf::shared_action_ptr)
...@@ -115,3 +115,11 @@ constexpr auto cached_udp_buffers = size_t{10}; ...@@ -115,3 +115,11 @@ constexpr auto cached_udp_buffers = size_t{10};
constexpr auto max_pending_msgs = size_t{10}; constexpr auto max_pending_msgs = size_t{10};
} // namespace caf::defaults::middleman } // namespace caf::defaults::middleman
namespace caf::defaults::flow {
constexpr auto min_demand = size_t{8};
constexpr auto batch_size = size_t{32};
constexpr auto buffer_size = size_t{128};
} // namespace caf::defaults::flow
...@@ -15,9 +15,13 @@ namespace caf { ...@@ -15,9 +15,13 @@ namespace caf {
/// Represents a disposable resource. /// Represents a disposable resource.
class CAF_CORE_EXPORT disposable { class CAF_CORE_EXPORT disposable {
public: public:
// -- member types -----------------------------------------------------------
/// Internal implementation class of a `disposable`. /// Internal implementation class of a `disposable`.
class impl { class impl {
public: public:
CAF_INTRUSIVE_PTR_FRIENDS_SFX(impl, _disposable)
virtual ~impl(); virtual ~impl();
virtual void dispose() = 0; virtual void dispose() = 0;
...@@ -29,14 +33,6 @@ public: ...@@ -29,14 +33,6 @@ public:
virtual void ref_disposable() const noexcept = 0; virtual void ref_disposable() const noexcept = 0;
virtual void deref_disposable() const noexcept = 0; virtual void deref_disposable() 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();
}
}; };
explicit disposable(intrusive_ptr<impl> pimpl) noexcept explicit disposable(intrusive_ptr<impl> pimpl) noexcept
...@@ -44,17 +40,27 @@ public: ...@@ -44,17 +40,27 @@ public:
// nop // nop
} }
// -- constructors, destructors, and assignment operators --------------------
disposable() noexcept = default; disposable() noexcept = default;
disposable(disposable&&) noexcept = default; disposable(disposable&&) noexcept = default;
disposable(const disposable&) noexcept = default; disposable(const disposable&) noexcept = default;
disposable& operator=(disposable&&) noexcept = default; disposable& operator=(disposable&&) noexcept = default;
disposable& operator=(const disposable&) noexcept = default; disposable& operator=(const disposable&) noexcept = default;
// -- factories --------------------------------------------------------------
/// Combines multiple disposables into a single disposable. The new disposable /// Combines multiple disposables into a single disposable. The new disposable
/// is disposed if all of its elements are disposed. Disposing the composite /// is disposed if all of its elements are disposed. Disposing the composite
/// disposes all elements individually. /// disposes all elements individually.
static disposable make_composite(std::vector<disposable> entries); static disposable make_composite(std::vector<disposable> entries);
// -- mutators ---------------------------------------------------------------
/// Disposes the resource. Calling `dispose()` on a disposed resource is a /// Disposes the resource. Calling `dispose()` on a disposed resource is a
/// no-op. /// no-op.
void dispose() { void dispose() {
...@@ -64,6 +70,13 @@ public: ...@@ -64,6 +70,13 @@ public:
} }
} }
/// Exchanges the content of this handle with `other`.
void swap(disposable& other) {
pimpl_.swap(other.pimpl_);
}
// -- properties -------------------------------------------------------------
/// Returns whether the resource has been disposed. /// Returns whether the resource has been disposed.
[[nodiscard]] bool disposed() const noexcept { [[nodiscard]] bool disposed() const noexcept {
return pimpl_ ? pimpl_->disposed() : true; return pimpl_ ? pimpl_->disposed() : true;
...@@ -89,6 +102,8 @@ public: ...@@ -89,6 +102,8 @@ public:
return pimpl_.get(); return pimpl_.get();
} }
// -- conversions ------------------------------------------------------------
/// Returns a smart pointer to the implementation. /// Returns a smart pointer to the implementation.
[[nodiscard]] intrusive_ptr<impl>&& as_intrusive_ptr() && noexcept { [[nodiscard]] intrusive_ptr<impl>&& as_intrusive_ptr() && noexcept {
return std::move(pimpl_); return std::move(pimpl_);
...@@ -99,11 +114,6 @@ public: ...@@ -99,11 +114,6 @@ public:
return pimpl_; return pimpl_;
} }
/// Exchanges the content of this handle with `other`.
void swap(disposable& other) {
pimpl_.swap(other.pimpl_);
}
private: private:
intrusive_ptr<impl> pimpl_; intrusive_ptr<impl> pimpl_;
}; };
......
...@@ -48,6 +48,8 @@ public: ...@@ -48,6 +48,8 @@ public:
/// Required by `spawn` for type deduction. /// Required by `spawn` for type deduction.
using behavior_type = behavior; using behavior_type = behavior;
using handle_type = actor;
// -- constructors, destructors ---------------------------------------------- // -- constructors, destructors ----------------------------------------------
explicit event_based_actor(actor_config& cfg); explicit event_based_actor(actor_config& cfg);
......
// 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/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
// 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/action.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"
namespace caf::flow {
/// Coordinates any number of co-located publishers 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 {
public:
// class CAF_CORE_EXPORT subscription_impl : public subscription::impl {
// public:
// friend class coordinator;
// subscription_impl(coordinator* ctx, observable_base_ptr src,
// observer_base_ptr snk)
// : ctx_(ctx), src_(std::move(src)), snk_(std::move(snk)) {
// // nop
// }
// void request(size_t n) final;
// void cancel() final;
// bool disposed() const noexcept final;
// auto* ctx() const noexcept {
// return ctx_;
// }
// private:
// coordinator* ctx_;
// observable_base_ptr src_;
// observer_base_ptr snk_;
// };
// using subscription_impl_ptr = intrusive_ptr<subscription_impl>;
friend class subscription_impl;
virtual ~coordinator();
[[nodiscard]] observable_builder make_observable();
/// 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;
/// 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)));
}
/// Schedules an action for execution from within the coordinator. May call
/// `schedule` for coordinators that use a single work queue.
virtual void post_internally(action what) = 0;
///@copydoc schedule
template <class F>
void post_internally_fn(F&& what) {
return post_internally(make_action(std::forward<F>(what)));
}
/// 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;
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();
}
};
/// @relates coordinator
using coordinator_ptr = intrusive_ptr<coordinator>;
} // 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/fwd.hpp"
#include <type_traits>
namespace caf::detail {
template <class Left, class Right>
struct left_oracle {
using type = Left;
};
/// Useful to force dependent evaluation of Left. For example in template
/// functions where only Right is actually a template parameter.
template <class Left, class Right>
using left_t = typename left_oracle<Left, Right>::type;
} // namespace caf::detail
namespace caf::flow {
class coordinator;
class subscription;
template <class T>
class single;
template <class T>
class observer;
template <class T>
class observable;
template <class In, class Out>
class processor;
template <class Generator, class... Steps>
class generation;
template <class Step, class... Steps>
class transformation;
template <class T>
struct is_observable {
static constexpr bool value = false;
};
template <class T>
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>> {
static constexpr bool value = true;
};
template <class T>
struct is_observable<single<T>> {
static constexpr bool value = true;
};
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>
struct input_type_oracle {
using type = typename T::input_type;
};
template <class T>
using input_type_t = typename input_type_oracle<T>::type;
template <class T>
struct output_type_oracle {
using type = typename T::output_type;
};
template <class T>
using output_type_t = typename output_type_oracle<T>::type;
template <class T>
class merger_impl;
template <class>
struct has_impl_include {
static constexpr bool value = false;
};
template <class T>
constexpr bool has_impl_include_v = has_impl_include<T>::value;
template <class T>
struct assert_scheduled_actor_hdr {
static constexpr bool value
= has_impl_include_v<detail::left_t<scheduled_actor, T>>;
static_assert(value,
"include 'caf/scheduled_actor/flow.hpp' for this method");
};
template <class T>
using assert_scheduled_actor_hdr_t
= std::enable_if_t<assert_scheduled_actor_hdr<T>::value, T>;
} // 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/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>
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 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/bounded_buffer.hpp"
#include "caf/defaults.hpp"
#include "caf/flow/coordinator.hpp"
#include "caf/flow/observable.hpp"
namespace caf::flow {
template <class Container>
class container_source {
public:
using output_type = typename Container::value_type;
explicit container_source(Container&& values) : values_(std::move(values)) {
pos_ = values_.begin();
}
container_source(container_source&&) = default;
container_source& operator=(container_source&&) = default;
container_source() = delete;
container_source(const container_source&) = delete;
container_source& operator=(const container_source&) = delete;
template <class Step, class... Steps>
void pull(size_t n, Step& step, Steps&... steps) {
while (pos_ != values_.end() && n > 0) {
if (!step.on_next(*pos_++, steps...))
return;
--n;
}
if (pos_ == values_.end())
step.on_complete(steps...);
}
private:
Container values_;
typename Container::const_iterator pos_;
};
template <class T>
class repeater_source {
public:
using output_type = T;
explicit repeater_source(T value) : value_(std::move(value)) {
// nop
}
repeater_source(repeater_source&&) = default;
repeater_source(const repeater_source&) = default;
repeater_source& operator=(repeater_source&&) = default;
repeater_source& operator=(const repeater_source&) = 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_;
};
template <class F>
class callable_source {
public:
using output_type = std::decay_t<decltype(std::declval<F&>()())>;
explicit callable_source(F fn) : fn_(std::move(fn)) {
// nop
}
callable_source(callable_source&&) = default;
callable_source& operator=(callable_source&&) = default;
callable_source(const callable_source&) = delete;
callable_source& operator=(const callable_source&) = delete;
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(fn_(), steps...))
return;
}
private:
F fn_;
};
class observable_builder {
public:
friend class coordinator;
observable_builder(const observable_builder&) noexcept = default;
observable_builder& operator=(const observable_builder&) noexcept = default;
template <class T>
[[nodiscard]] generation<repeater_source<T>> repeat(T value) const;
template <class Container>
[[nodiscard]] generation<container_source<Container>>
from_container(Container values) const;
template <class T>
[[nodiscard]] auto just(T value) const;
template <class F>
[[nodiscard]] generation<callable_source<F>> from_callable(F fn) const;
/// Opens an asynchronous, buffered resource and emits all inputs from the
/// buffer.
template <class T>
[[nodiscard]] observable<T>
from_resource(async::consumer_resource<T> hdl) const;
template <class Pullable>
[[nodiscard]] generation<Pullable> lift(Pullable pullable) const;
private:
explicit observable_builder(coordinator* ctx) : ctx_(ctx) {
// nop
}
coordinator* ctx_;
};
// -- generation ---------------------------------------------------------------
template <class Generator, class... Steps>
class generation final
: public observable_def<
transform_processor_output_type_t<Generator, Steps...>> {
public:
using output_type = transform_processor_output_type_t<Generator, Steps...>;
class impl : public buffered_observable_impl<output_type> {
public:
using super = buffered_observable_impl<output_type>;
template <class... Ts>
impl(coordinator* ctx, Generator gen, Ts&&... steps)
: super(ctx), gen_(std::move(gen)), steps_(std::forward<Ts>(steps)...) {
// nop
}
private:
virtual void pull(size_t n) {
auto fn = [this, n](auto&... steps) {
term_step<output_type> term{this};
gen_.pull(n, steps..., term);
};
std::apply(fn, steps_);
}
Generator gen_;
std::tuple<Steps...> steps_;
};
template <class... Ts>
generation(coordinator* ctx, Generator gen, Ts&&... steps)
: ctx_(ctx), gen_(std::move(gen)), steps_(std::forward<Ts>(steps)...) {
// nop
}
generation() = delete;
generation(const generation&) = delete;
generation& operator=(const generation&) = delete;
generation(generation&&) = default;
generation& operator=(generation&&) = default;
/// @copydoc observable::transform
template <class NewStep>
generation<Generator, Steps..., NewStep> transform(NewStep step) && {
static_assert(std::is_same_v<typename NewStep::input_type, output_type>,
"step object does not match the output type");
return {ctx_, std::move(gen_),
std::tuple_cat(std::move(steps_),
std::make_tuple(std::move(step)))};
}
auto take(size_t n) && {
return std::move(*this).transform(limit_step<output_type>{n});
}
template <class Predicate>
auto filter(Predicate predicate) && {
return std::move(*this).transform(
filter_step<Predicate>{std::move(predicate)});
}
template <class Fn>
auto map(Fn fn) && {
return std::move(*this).transform(map_step<Fn>{std::move(fn)});
}
observable<output_type> as_observable() && override {
auto pimpl = make_counted<impl>(ctx_, std::move(gen_), std::move(steps_));
return observable<output_type>{std::move(pimpl)};
}
private:
coordinator* ctx_;
Generator gen_;
std::tuple<Steps...> steps_;
};
// -- observable_builder::repeat -----------------------------------------------
template <class T>
generation<repeater_source<T>> observable_builder::repeat(T value) const {
return {ctx_, repeater_source<T>{std::move(value)}};
}
// -- observable_builder::from_container ---------------------------------------
template <class Container>
generation<container_source<Container>>
observable_builder::from_container(Container values) const {
return {ctx_, container_source<Container>{std::move(values)}};
}
// -- observable_builder::just -------------------------------------------------
template <class T>
auto observable_builder::just(T value) const {
return repeat(std::move(value)).take(1);
}
// -- observable_builder::from_callable ----------------------------------------
template <class F>
generation<callable_source<F>> observable_builder::from_callable(F fn) const {
return {ctx_, callable_source<F>{std::move(fn)}};
}
// -- observable_builder::from_resource ----------------------------------------
template <class T>
observable<T>
observable_builder::from_resource(async::consumer_resource<T> hdl) const {
using buffer_type = typename async::consumer_resource<T>::buffer_type;
using res_t = observable<T>;
if (auto buf = hdl.try_open()) {
auto adapter = make_counted<observable_buffer_impl<buffer_type>>(ctx_, buf);
buf->set_consumer(adapter);
ctx_->watch(adapter->as_disposable());
return res_t{std::move(adapter)};
} else {
auto err = make_error(sec::invalid_observable,
"from_resource: failed to open the resource");
return res_t{make_counted<observable_error_impl<T>>(ctx_, std::move(err))};
}
}
// -- observable_builder::lift -------------------------------------------------
template <class Pullable>
generation<Pullable> observable_builder::lift(Pullable pullable) const {
return {ctx_, std::move(pullable)};
}
} // namespace caf::flow
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 <condition_variable>
#include <mutex>
#include "caf/flow/coordinator.hpp"
#include "caf/make_counted.hpp"
#include "caf/ref_counted.hpp"
namespace caf::flow {
class CAF_CORE_EXPORT scoped_coordinator final : public ref_counted,
public coordinator {
public:
void run();
void ref_coordinator() const noexcept override;
void deref_coordinator() const noexcept override;
void schedule(action what) override;
void post_internally(action what) override;
void watch(disposable what) override;
friend void intrusive_ptr_add_ref(const scoped_coordinator* ptr) {
ptr->ref();
}
friend void intrusive_ptr_release(scoped_coordinator* ptr) {
ptr->deref();
}
static intrusive_ptr<scoped_coordinator> make();
private:
scoped_coordinator() = default;
void drop_disposed_flows();
std::vector<disposable> watched_disposables_;
mutable std::mutex mtx_;
std::condition_variable cv_;
std::vector<action> actions_;
};
using scoped_coordinator_ptr = intrusive_ptr<scoped_coordinator>;
inline scoped_coordinator_ptr make_scoped_coordinator() {
return scoped_coordinator::make();
}
} // 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 <algorithm>
#include <utility>
#include <variant>
#include "caf/detail/overload.hpp"
#include "caf/disposable.hpp"
#include "caf/error.hpp"
#include "caf/flow/observable.hpp"
#include "caf/none.hpp"
#include "caf/ref_counted.hpp"
namespace caf::flow {
/// Similar to an `observable`, but always emits either a single value or an
/// error.
template <class T>
class single {
public:
using output_type = T;
/// Internal interface of a `single`.
class impl : public ref_counted, public observable_impl<T> {
public:
using super = observable_impl<T>;
CAF_INTRUSIVE_PTR_FRIENDS(impl)
explicit impl(coordinator* ctx) : ctx_(ctx) {
// nop
}
coordinator* ctx() const noexcept override {
return ctx_;
}
disposable subscribe(observer<T> what) override {
if (!std::holds_alternative<error>(value_)) {
auto res = super::do_subscribe(what.ptr());
observers_.emplace_back(std::move(what), 0u);
return res;
} else {
what.on_error(std::get<error>(value_));
return disposable{};
}
}
void on_request(observer_impl<T>* sink, size_t n) override {
auto pred = [sink](auto& entry) { return entry.first.ptr() == sink; };
if (auto i = std::find_if(observers_.begin(), observers_.end(), pred);
i != observers_.end()) {
auto f = detail::make_overload( //
[i, n](none_t) { i->second += n; },
[this, i](const T& val) {
i->first.on_next(make_span(&val, 1));
i->first.on_complete();
observers_.erase(i);
},
[this, i](const error& err) {
i->first.on_error(err);
observers_.erase(i);
});
std::visit(f, value_);
}
}
void on_cancel(observer_impl<T>* sink) override {
auto pred = [sink](auto& entry) { return entry.first.ptr() == sink; };
if (auto i = std::find_if(observers_.begin(), observers_.end(), pred);
i != observers_.end())
observers_.erase(i);
}
void dispose() override {
if (!std::holds_alternative<error>(value_))
set_error(make_error(sec::discarded));
}
bool disposed() const noexcept override {
return observers_.empty() && !std::holds_alternative<none_t>(value_);
}
void ref_disposable() const noexcept final {
this->ref();
}
void deref_disposable() const noexcept final {
this->deref();
}
void set_value(T val) {
if (std::holds_alternative<none_t>(value_)) {
value_ = std::move(val);
auto& ref = std::get<T>(value_);
auto pred = [](auto& entry) { return entry.second == 0; };
if (auto first = std::partition(observers_.begin(), observers_.end(),
pred);
first != observers_.end()) {
for (auto i = first; i != observers_.end(); ++i) {
i->first.on_next(make_span(&ref, 1));
i->first.on_complete();
}
observers_.erase(first, observers_.end());
}
}
}
void set_error(error err) {
value_ = std::move(err);
auto& ref = std::get<error>(value_);
for (auto& entry : observers_)
entry.first.on_error(ref);
observers_.clear();
}
private:
coordinator* ctx_;
std::variant<none_t, T, error> value_;
std::vector<std::pair<observer<T>, size_t>> observers_;
};
explicit single(intrusive_ptr<impl> pimpl) noexcept
: pimpl_(std::move(pimpl)) {
// nop
}
single& operator=(std::nullptr_t) noexcept {
pimpl_.reset();
return *this;
}
single() noexcept = default;
single(single&&) noexcept = default;
single(const single&) noexcept = default;
single& operator=(single&&) noexcept = default;
single& operator=(const single&) noexcept = default;
disposable as_disposable() && {
return disposable{std::move(pimpl_)};
}
disposable as_disposable() const& {
return disposable{pimpl_};
}
observable<T> as_observable() && {
return observable<T>{std::move(pimpl_)};
}
observable<T> as_observable() const& {
return observable<T>{pimpl_};
}
void subscribe(observer<T> what) {
if (pimpl_)
pimpl_->subscribe(std::move(what));
else
what.on_error(make_error(sec::invalid_observable));
}
template <class OnSuccess, class OnError>
void subscribe(OnSuccess on_success, OnError on_error) {
static_assert(std::is_invocable_v<OnSuccess, const T&>);
as_observable().for_each(
[f{std::move(on_success)}](span<const T> items) mutable {
CAF_ASSERT(items.size() == 1);
f(items[0]);
},
std::move(on_error));
}
void set_value(T val) {
if (pimpl_)
pimpl_->set_value(std::move(val));
}
void set_error(error err) {
if (pimpl_)
pimpl_->set_error(std::move(err));
}
bool valid() const noexcept {
return pimpl_ != nullptr;
}
explicit operator bool() const noexcept {
return valid();
}
bool operator!() const noexcept {
return !valid();
}
impl* ptr() {
return pimpl_.get();
}
const impl* ptr() const {
return pimpl_.get();
}
const intrusive_ptr<impl>& as_intrusive_ptr() const& noexcept {
return pimpl_;
}
intrusive_ptr<impl>&& as_intrusive_ptr() && noexcept {
return std::move(pimpl_);
}
void swap(single& other) {
pimpl_.swap(other.pimpl_);
}
private:
intrusive_ptr<impl> pimpl_;
};
template <class T>
using single_impl = typename single<T>::impl;
template <class T>
single<T> make_single(coordinator* ctx) {
return single<T>{make_counted<single_impl<T>>(ctx)};
}
} // 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 <type_traits>
#include "caf/detail/type_list.hpp"
#include "caf/detail/type_traits.hpp"
namespace caf::flow {
template <class T>
struct limit_step {
size_t remaining;
using input_type = T;
using output_type = T;
template <class Next, class... Steps>
bool on_next(const input_type& item, Next& next, Steps&... steps) {
if (remaining > 0) {
if (next.on_next(item, steps...)) {
if (--remaining > 0) {
return true;
} else {
next.on_complete(steps...);
return false;
}
}
}
return false;
}
template <class Next, class... Steps>
void on_complete(Next& next, Steps&... steps) {
next.on_complete(steps...);
}
template <class Next, class... Steps>
void on_error(const error& what, Next& next, Steps&... steps) {
next.on_error(what, steps...);
}
};
template <class Predicate>
struct filter_step {
using trait = detail::get_callable_trait_t<Predicate>;
static_assert(std::is_convertible_v<typename trait::result_type, bool>,
"predicates must return a boolean value");
static_assert(trait::num_args == 1,
"predicates must take exactly one argument");
using input_type = std::decay_t<detail::tl_head_t<typename trait::arg_types>>;
using output_type = input_type;
Predicate predicate;
template <class Next, class... Steps>
bool on_next(const input_type& item, Next& next, Steps&... steps) {
if (predicate(item))
return next.on_next(item, steps...);
else
return true;
}
template <class Next, class... Steps>
void on_complete(Next& next, Steps&... steps) {
next.on_complete(steps...);
}
template <class Next, class... Steps>
void on_error(const error& what, Next& next, Steps&... steps) {
next.on_error(what, steps...);
}
};
template <class Fn>
struct map_step {
using trait = detail::get_callable_trait_t<Fn>;
static_assert(!std::is_same_v<typename trait::result_type, void>,
"map functions may not return void");
static_assert(trait::num_args == 1,
"map functions must take exactly one argument");
using input_type = std::decay_t<detail::tl_head_t<typename trait::arg_types>>;
using output_type = std::decay_t<typename trait::result_type>;
Fn fn;
template <class Next, class... Steps>
bool on_next(const input_type& item, Next& next, Steps&... steps) {
return next.on_next(fn(item), steps...);
}
template <class Next, class... Steps>
void on_complete(Next& next, Steps&... steps) {
next.on_complete(steps...);
}
template <class Next, class... Steps>
void on_error(const error& what, Next& next, Steps&... steps) {
next.on_error(what, steps...);
}
};
template <class T, class Fn>
struct on_complete_step {
using input_type = T;
using output_type = T;
Fn fn;
template <class Next, class... Steps>
bool on_next(const input_type& item, Next& next, Steps&... steps) {
return next.on_next(item, steps...);
}
template <class Next, class... Steps>
void on_complete(Next& next, Steps&... steps) {
fn();
next.on_complete(steps...);
}
template <class Next, class... Steps>
void on_error(const error& what, Next& next, Steps&... steps) {
next.on_error(what, steps...);
}
};
template <class T, class Fn>
struct on_error_step {
using input_type = T;
using output_type = T;
Fn fn;
template <class Next, class... Steps>
bool on_next(const input_type& item, Next& next, Steps&... steps) {
return next.on_next(item, steps...);
}
template <class Next, class... Steps>
void on_complete(Next& next, Steps&... steps) {
next.on_complete(steps...);
}
template <class Next, class... Steps>
void on_error(const error& what, Next& next, Steps&... steps) {
fn();
next.on_error(what, steps...);
}
};
template <class T, class Fn>
struct finally_step {
using input_type = T;
using output_type = T;
Fn fn;
template <class Next, class... Steps>
bool on_next(const input_type& item, Next& next, Steps&... steps) {
return next.on_next(item, steps...);
}
template <class Next, class... Steps>
void on_complete(Next& next, Steps&... steps) {
fn();
next.on_complete(steps...);
}
template <class Next, class... Steps>
void on_error(const error& what, Next& next, Steps&... steps) {
fn();
next.on_error(what, steps...);
}
};
} // 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 <cstddef>
#include "caf/detail/core_export.hpp"
#include "caf/disposable.hpp"
#include "caf/intrusive_ptr.hpp"
#include "caf/ref_counted.hpp"
namespace caf::flow {
/// Controls the flow of items from publishers to subscribers.
class CAF_CORE_EXPORT subscription {
public:
// -- nested types -----------------------------------------------------------
/// Internal impl of a `disposable`.
class impl : public disposable::impl {
public:
~impl() override;
/// Causes the publisher to stop producing items for the subscriber. Any
/// in-flight items may still get dispatched.
virtual void cancel() = 0;
/// Signals demand for `n` more items.
virtual void request(size_t n) = 0;
void dispose() final;
};
// -- constructors, destructors, and assignment operators --------------------
explicit subscription(intrusive_ptr<impl> pimpl) noexcept
: pimpl_(std::move(pimpl)) {
// nop
}
subscription& operator=(std::nullptr_t) noexcept {
pimpl_.reset();
return *this;
}
subscription() noexcept = default;
subscription(subscription&&) noexcept = default;
subscription(const subscription&) noexcept = default;
subscription& operator=(subscription&&) noexcept = default;
subscription& operator=(const subscription&) noexcept = default;
// -- demand signaling -------------------------------------------------------
/// @copydoc impl::cancel
void cancel() {
if (pimpl_) {
pimpl_->cancel();
pimpl_ = nullptr;
}
}
/// @copydoc impl::request
/// @pre `valid()`
void request(size_t n) {
pimpl_->request(n);
}
// -- properties -------------------------------------------------------------
bool valid() const noexcept {
return pimpl_ != nullptr;
}
explicit operator bool() const noexcept {
return valid();
}
bool operator!() const noexcept {
return !valid();
}
impl* ptr() noexcept {
return pimpl_.get();
}
const impl* ptr() const noexcept {
return pimpl_.get();
}
intrusive_ptr<impl> as_intrusive_ptr() const& noexcept {
return pimpl_;
}
intrusive_ptr<impl>&& as_intrusive_ptr() && noexcept {
return std::move(pimpl_);
}
disposable as_disposable() const& noexcept {
return disposable{pimpl_};
}
disposable as_disposable() && noexcept {
return disposable{std::move(pimpl_)};
}
// -- swapping ---------------------------------------------------------------
void swap(subscription& other) noexcept {
pimpl_.swap(other.pimpl_);
}
private:
intrusive_ptr<impl> pimpl_;
};
} // namespace caf::flow
...@@ -381,4 +381,8 @@ using strong_actor_ptr = intrusive_ptr<actor_control_block>; ...@@ -381,4 +381,8 @@ using strong_actor_ptr = intrusive_ptr<actor_control_block>;
using mailbox_element_ptr = std::unique_ptr<mailbox_element>; using mailbox_element_ptr = std::unique_ptr<mailbox_element>;
using tracing_data_ptr = std::unique_ptr<tracing_data>; using tracing_data_ptr = std::unique_ptr<tracing_data>;
// -- shared pointer aliases ---------------------------------------------------
using shared_action_ptr = std::shared_ptr<callback<void()>>;
} // namespace caf } // namespace caf
...@@ -271,3 +271,22 @@ std::string to_string(const intrusive_ptr<T>& x) { ...@@ -271,3 +271,22 @@ std::string to_string(const intrusive_ptr<T>& x) {
} // namespace caf } // namespace caf
/// Convenience macro for adding `intrusive_ptr_add_ref` and
/// `intrusive_ptr_release` as free friend functions.
#define CAF_INTRUSIVE_PTR_FRIENDS(class_name) \
friend void intrusive_ptr_add_ref(const class_name* ptr) noexcept { \
ptr->ref(); \
} \
friend void intrusive_ptr_release(const class_name* ptr) noexcept { \
ptr->deref(); \
}
/// Convenience macro for adding `intrusive_ptr_add_ref` and
/// `intrusive_ptr_release` as free friend functions with a custom suffix.
#define CAF_INTRUSIVE_PTR_FRIENDS_SFX(class_name, suffix) \
friend void intrusive_ptr_add_ref(const class_name* ptr) noexcept { \
ptr->ref##suffix(); \
} \
friend void intrusive_ptr_release(const class_name* ptr) noexcept { \
ptr->deref##suffix(); \
}
...@@ -41,18 +41,16 @@ public: ...@@ -41,18 +41,16 @@ public:
return rc_; return rc_;
} }
friend void intrusive_ptr_add_ref(const ref_counted* p) noexcept {
p->ref();
}
friend void intrusive_ptr_release(const ref_counted* p) noexcept {
p->deref();
}
protected: protected:
mutable std::atomic<size_t> rc_; mutable std::atomic<size_t> rc_;
}; };
/// @relates ref_counted
inline void intrusive_ptr_add_ref(const ref_counted* p) {
p->ref();
}
/// @relates ref_counted
inline void intrusive_ptr_release(const ref_counted* p) {
p->deref();
}
} // namespace caf } // namespace caf
...@@ -8,6 +8,7 @@ ...@@ -8,6 +8,7 @@
#include "caf/actor_traits.hpp" #include "caf/actor_traits.hpp"
#include "caf/catch_all.hpp" #include "caf/catch_all.hpp"
#include "caf/flow/fwd.hpp"
#include "caf/message_id.hpp" #include "caf/message_id.hpp"
#include "caf/none.hpp" #include "caf/none.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
...@@ -104,6 +105,18 @@ public: ...@@ -104,6 +105,18 @@ public:
then(std::move(f), [self](error& err) { self->call_error_handler(err); }); then(std::move(f), [self](error& err) { self->call_error_handler(err); });
} }
template <class T>
flow::assert_scheduled_actor_hdr_t<flow::single<T>> as_single() && {
static_assert(std::is_same_v<response_type, message>);
return self_->template single_from_response<T>(policy_);
}
template <class T>
flow::assert_scheduled_actor_hdr_t<flow::observable<T>> as_observable() && {
static_assert(std::is_same_v<response_type, message>);
return self_->template single_from_response<T>(policy_).as_observable();
}
// -- blocking API ----------------------------------------------------------- // -- blocking API -----------------------------------------------------------
template <class T = traits, class F = none_t, class OnError = none_t, template <class T = traits, class F = none_t, class OnError = none_t,
......
...@@ -20,8 +20,11 @@ ...@@ -20,8 +20,11 @@
#include "caf/detail/behavior_stack.hpp" #include "caf/detail/behavior_stack.hpp"
#include "caf/detail/core_export.hpp" #include "caf/detail/core_export.hpp"
#include "caf/detail/unordered_flat_map.hpp" #include "caf/detail/unordered_flat_map.hpp"
#include "caf/disposable.hpp"
#include "caf/error.hpp" #include "caf/error.hpp"
#include "caf/extend.hpp" #include "caf/extend.hpp"
#include "caf/flow/coordinator.hpp"
#include "caf/flow/fwd.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/inbound_path.hpp" #include "caf/inbound_path.hpp"
#include "caf/intrusive/drr_cached_queue.hpp" #include "caf/intrusive/drr_cached_queue.hpp"
...@@ -72,8 +75,14 @@ CAF_CORE_EXPORT skippable_result drop(scheduled_actor*, message&); ...@@ -72,8 +75,14 @@ CAF_CORE_EXPORT skippable_result drop(scheduled_actor*, message&);
/// A cooperatively scheduled, event-based actor implementation. /// A cooperatively scheduled, event-based actor implementation.
class CAF_CORE_EXPORT scheduled_actor : public local_actor, class CAF_CORE_EXPORT scheduled_actor : public local_actor,
public resumable, public resumable,
public non_blocking_actor_base { public non_blocking_actor_base,
public flow::coordinator {
public: public:
// -- friends ----------------------------------------------------------------
template <class, class>
friend class response_handle;
// -- nested enums ----------------------------------------------------------- // -- nested enums -----------------------------------------------------------
/// Categorizes incoming messages. /// Categorizes incoming messages.
...@@ -472,6 +481,18 @@ public: ...@@ -472,6 +481,18 @@ public:
/// Returns the queue of the mailbox that stores `downstream_msg` messages. /// Returns the queue of the mailbox that stores `downstream_msg` messages.
downstream_queue& get_downstream_queue(); downstream_queue& get_downstream_queue();
// -- caf::flow API ----------------------------------------------------------
void ref_coordinator() const noexcept override;
void deref_coordinator() const noexcept override;
void schedule(action what) override;
void post_internally(action what) override;
void watch(disposable what) override;
// -- inbound_path management ------------------------------------------------ // -- inbound_path management ------------------------------------------------
/// Creates a new path for incoming stream traffic from `sender`. /// Creates a new path for incoming stream traffic from `sender`.
...@@ -660,7 +681,8 @@ public: ...@@ -660,7 +681,8 @@ public:
bool alive() const noexcept { bool alive() const noexcept {
return !bhvr_stack_.empty() || !awaited_responses_.empty() return !bhvr_stack_.empty() || !awaited_responses_.empty()
|| !multiplexed_responses_.empty() || !stream_managers_.empty() || !multiplexed_responses_.empty() || !stream_managers_.empty()
|| !pending_stream_managers_.empty(); || !pending_stream_managers_.empty()
|| !watched_disposables_.empty();
} }
auto max_batch_delay() const noexcept { auto max_batch_delay() const noexcept {
...@@ -671,6 +693,13 @@ public: ...@@ -671,6 +693,13 @@ public:
std::vector<stream_manager*> active_stream_managers(); std::vector<stream_manager*> active_stream_managers();
/// Runs all pending actions.
void run_actions();
std::vector<disposable> watched_disposables() const {
return watched_disposables_;
}
/// @endcond /// @endcond
protected: protected:
...@@ -735,6 +764,8 @@ protected: ...@@ -735,6 +764,8 @@ protected:
#endif // CAF_ENABLE_EXCEPTIONS #endif // CAF_ENABLE_EXCEPTIONS
private: private:
// -- utilities for instrumenting actors -------------------------------------
template <class F> template <class F>
intrusive::task_result run_with_metrics(mailbox_element& x, F body) { intrusive::task_result run_with_metrics(mailbox_element& x, F body) {
if (metrics_.mailbox_time) { if (metrics_.mailbox_time) {
...@@ -752,9 +783,31 @@ private: ...@@ -752,9 +783,31 @@ private:
} }
} }
// -- timeout management -----------------------------------------------------
disposable run_scheduled(timestamp when, action what); disposable run_scheduled(timestamp when, action what);
disposable run_scheduled(actor_clock::time_point when, action what); disposable run_scheduled(actor_clock::time_point when, action what);
disposable run_delayed(timespan delay, action what); disposable run_delayed(timespan delay, action what);
// -- caf::flow bindings -----------------------------------------------------
template <class T, class Policy>
flow::single<T> single_from_response(Policy& policy) {
return single_from_response_impl<T>(policy);
}
template <class T, class Policy>
flow::single<T> single_from_response_impl(Policy& policy);
/// Removes any watched object that became disposed since the last update.
void update_watched_disposables();
/// Stores actions that the actor executes after processing the current
/// message.
std::vector<action> actions_;
/// Stores resources that block the actor from terminating.
std::vector<disposable> watched_disposables_;
}; };
} // namespace caf } // 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/actor.hpp"
#include "caf/disposable.hpp"
#include "caf/flow/coordinator.hpp"
#include "caf/flow/observable.hpp"
#include "caf/flow/observable_builder.hpp"
#include "caf/flow/observer.hpp"
#include "caf/flow/single.hpp"
#include "caf/scheduled_actor.hpp"
namespace caf::flow {
template <>
struct has_impl_include<scheduled_actor> {
static constexpr bool value = true;
};
} // namespace caf::flow
namespace caf {
template <class T, class Policy>
flow::single<T> scheduled_actor::single_from_response_impl(Policy& policy) {
using output_type = T;
using impl_type = typename flow::single<output_type>::impl;
auto ptr = make_counted<impl_type>(this);
policy.then(
this,
[this, ptr](T& val) {
ptr->set_value(std::move(val));
run_actions();
},
[this, ptr](error& err) {
ptr->set_error(std::move(err));
run_actions();
});
return flow::single<output_type>{std::move(ptr)};
}
} // namespace caf
...@@ -166,6 +166,11 @@ enum class sec : uint8_t { ...@@ -166,6 +166,11 @@ enum class sec : uint8_t {
/// Signals that an actor fell behind a periodic action trigger. After raising /// Signals that an actor fell behind a periodic action trigger. After raising
/// this error, an @ref actor_clock stops scheduling the action. /// this error, an @ref actor_clock stops scheduling the action.
action_reschedule_failed, action_reschedule_failed,
/// Attaching to an observable failed because the target is invalid.
invalid_observable,
/// Signals to a component that is has been discarded by its parent or the
/// consumer/producer that was attached to the component.
discarded = 70,
}; };
// --(rst-sec-end)-- // --(rst-sec-end)--
......
...@@ -406,6 +406,7 @@ CAF_BEGIN_TYPE_ID_BLOCK(core_module, 0) ...@@ -406,6 +406,7 @@ CAF_BEGIN_TYPE_ID_BLOCK(core_module, 0)
CAF_ADD_TYPE_ID(core_module, (caf::open_stream_msg)) CAF_ADD_TYPE_ID(core_module, (caf::open_stream_msg))
CAF_ADD_TYPE_ID(core_module, (caf::pec)) CAF_ADD_TYPE_ID(core_module, (caf::pec))
CAF_ADD_TYPE_ID(core_module, (caf::sec)) CAF_ADD_TYPE_ID(core_module, (caf::sec))
CAF_ADD_TYPE_ID(core_module, (caf::shared_action_ptr))
CAF_ADD_TYPE_ID(core_module, (caf::stream_slots)) CAF_ADD_TYPE_ID(core_module, (caf::stream_slots))
CAF_ADD_TYPE_ID(core_module, (caf::strong_actor_ptr)) CAF_ADD_TYPE_ID(core_module, (caf::strong_actor_ptr))
CAF_ADD_TYPE_ID(core_module, (caf::timespan)) CAF_ADD_TYPE_ID(core_module, (caf::timespan))
......
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#include "caf/async/batch.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/deserializer.hpp"
#include "caf/detail/meta_object.hpp"
#include "caf/serializer.hpp"
namespace caf::async {
// -- batch::data --------------------------------------------------------------
namespace {
// void dynamic_item_destructor(type_id_t item_type, uint16_t item_size,
// size_t array_size, byte* data_ptr) {
// auto meta = detail::global_meta_object(item_type);
// do {
// meta->destroy(data_ptr);
// data_ptr += item_size;
// --array_size;
// } while (array_size > 0);
// }
} // namespace
template <class Inspector>
bool batch::data::save(Inspector& sink) const {
CAF_ASSERT(size_ > 0);
if (item_type_ == invalid_type_id) {
sink.emplace_error(sec::unsafe_type);
return false;
}
auto meta = detail::global_meta_object(item_type_);
auto ptr = storage_;
if (!sink.begin_sequence(size_))
return false;
auto len = size_;
do {
if constexpr (std::is_same_v<Inspector, binary_serializer>) {
if (!meta->save_binary(sink, ptr))
return false;
} else {
if (!meta->save(sink, ptr))
return false;
}
ptr += item_size_;
--len;
} while (len > 0);
return sink.end_sequence();
}
// -- batch --------------------------------------------------------------------
template <class Inspector>
bool batch::save_impl(Inspector& f) const {
if (data_)
return data_->save(f);
else
return f.begin_sequence(0) && f.end_sequence();
}
bool batch::save(serializer& f) const {
return save_impl(f);
}
bool batch::save(binary_serializer& f) const {
return save_impl(f);
}
template <class Inspector>
bool batch::load_impl(Inspector&) {
// TODO: implement me
return false;
/*
auto len = size_t{0};
if (!f.begin_sequence(len))
return false;
if (len == 0) {
data_.reset();
return f.end_sequence();
}
*/
}
bool batch::load(deserializer& f) {
return load_impl(f);
}
bool batch::load(binary_deserializer& f) {
return load_impl(f);
}
} // 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.
#include "caf/async/consumer.hpp"
namespace caf::async {
consumer::~consumer() {
// nop
}
} // 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.
#include "caf/async/producer.hpp"
namespace caf::async {
producer::~producer() {
// nop
}
} // 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.
#include "caf/flow/coordinator.hpp"
#include "caf/config.hpp"
#include "caf/flow/observable_builder.hpp"
namespace caf::flow {
coordinator::~coordinator() {
// nop
}
observable_builder coordinator::make_observable() {
return observable_builder{this};
}
// void coordinator::subscription_impl::request(size_t n) {
// CAF_ASSERT(n != 0);
// if (src_) {
// ctx_->dispatch_request(src_.get(), snk_.get(), n);
// }
// }
//
// void coordinator::subscription_impl::cancel() {
// if (src_) {
// ctx_->dispatch_cancel(src_.get(), snk_.get());
// src_.reset();
// snk_.reset();
// }
// }
//
// bool coordinator::subscription_impl::disposed() const noexcept {
// return src_ == nullptr;
// }
} // 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.
#include "caf/flow/scoped_coordinator.hpp"
namespace caf::flow {
void scoped_coordinator::run() {
auto next = [this](bool blocking) {
std::unique_lock guard{mtx_};
if (blocking) {
while (actions_.empty())
cv_.wait(guard);
} else if (actions_.empty()) {
return action{};
}
auto res = std::move(actions_[0]);
actions_.erase(actions_.begin());
return res;
};
for (;;) {
auto hdl = next(!watched_disposables_.empty());
if (hdl.ptr() != nullptr)
hdl.run();
else
return;
}
}
void scoped_coordinator::ref_coordinator() const noexcept {
ref();
}
void scoped_coordinator::deref_coordinator() const noexcept {
deref();
}
void scoped_coordinator::schedule(action what) {
std::unique_lock guard{mtx_};
actions_.emplace_back(std::move(what));
if (actions_.size() == 1)
cv_.notify_all();
}
void scoped_coordinator::post_internally(action what) {
schedule(std::move(what));
}
void scoped_coordinator::watch(disposable what) {
watched_disposables_.emplace_back(std::move(what));
}
intrusive_ptr<scoped_coordinator> scoped_coordinator::make() {
return {new scoped_coordinator, false};
}
void scoped_coordinator::drop_disposed_flows() {
auto disposed = [](auto& hdl) { return hdl.disposed(); };
auto& xs = watched_disposables_;
if (auto e = std::remove_if(xs.begin(), xs.end(), disposed); e != xs.end())
xs.erase(e, xs.end());
}
} // 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.
#include "caf/flow/subscription.hpp"
namespace caf::flow {
subscription::impl::~impl() {
// nop
}
void subscription::impl::dispose() {
cancel();
}
} // namespace caf::flow
...@@ -10,6 +10,7 @@ ...@@ -10,6 +10,7 @@
#include "caf/actor_control_block.hpp" #include "caf/actor_control_block.hpp"
#include "caf/actor_system.hpp" #include "caf/actor_system.hpp"
#include "caf/byte_buffer.hpp" #include "caf/byte_buffer.hpp"
#include "caf/callback.hpp"
#include "caf/config_value.hpp" #include "caf/config_value.hpp"
#include "caf/downstream_msg.hpp" #include "caf/downstream_msg.hpp"
#include "caf/error.hpp" #include "caf/error.hpp"
......
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