Unverified Commit 11a1b0f9 authored by Dominik Charousset's avatar Dominik Charousset Committed by GitHub

Merge pull request #1325

First chunk of the new flow API
parents bfa0f83d f0547cf7
......@@ -5,6 +5,13 @@ is based on [Keep a Changelog](https://keepachangelog.com).
## [Unreleased]
### Fixed
- Passing a response promise to a run-delayed continuation could result in a
heap-use-after-free if the actor terminates before the action runs. The
destructor of the promise now checks for this case.
- Accessing URI fields now always returns the normalized string.
## [0.18.6]
### Added
......
......@@ -316,18 +316,19 @@ function(caf_add_component name)
endif()
endforeach()
set(pub_lib_target "libcaf_${name}")
set(obj_lib_target "libcaf_${name}_obj")
set(targets ${pub_lib_target} ${obj_lib_target})
add_library(${obj_lib_target} OBJECT
${CAF_ADD_COMPONENT_HEADERS} ${CAF_ADD_COMPONENT_SOURCES})
set_property(TARGET ${obj_lib_target} PROPERTY POSITION_INDEPENDENT_CODE ON)
caf_target_link_libraries(${obj_lib_target}
${CAF_ADD_COMPONENT_DEPENDENCIES})
add_library(${pub_lib_target}
"${PROJECT_SOURCE_DIR}/cmake/dummy.cpp"
$<TARGET_OBJECTS:${obj_lib_target}>)
if(CAF_ENABLE_TESTING AND CAF_ADD_COMPONENT_TEST_SOURCES)
set(obj_lib_target "libcaf_${name}_obj")
set(tst_bin_target "caf-${name}-test")
set(targets ${pub_lib_target} ${obj_lib_target} ${tst_bin_target})
add_library(${obj_lib_target} OBJECT
${CAF_ADD_COMPONENT_HEADERS} ${CAF_ADD_COMPONENT_SOURCES})
set_property(TARGET ${obj_lib_target} PROPERTY POSITION_INDEPENDENT_CODE ON)
caf_target_link_libraries(${obj_lib_target}
${CAF_ADD_COMPONENT_DEPENDENCIES})
add_library(${pub_lib_target}
"${PROJECT_SOURCE_DIR}/cmake/dummy.cpp"
$<TARGET_OBJECTS:${obj_lib_target}>)
list(APPEND targets ${tst_bin_target})
add_executable(${tst_bin_target}
${CAF_ADD_COMPONENT_TEST_SOURCES}
$<TARGET_OBJECTS:${obj_lib_target}>)
......@@ -338,11 +339,6 @@ function(caf_add_component name)
if(CAF_ADD_COMPONENT_TEST_SUITES)
caf_add_test_suites(${tst_bin_target} ${CAF_ADD_COMPONENT_TEST_SUITES})
endif()
else()
set(targets ${pub_lib_target})
add_library(${pub_lib_target}
${CAF_ADD_COMPONENT_HEADERS} ${CAF_ADD_COMPONENT_SOURCES})
set_property(TARGET ${pub_lib_target} PROPERTY POSITION_INDEPENDENT_CODE ON)
endif()
target_link_libraries(${pub_lib_target} ${CAF_ADD_COMPONENT_DEPENDENCIES})
if(CAF_ADD_COMPONENT_ENUM_TYPES)
......
......@@ -2,6 +2,8 @@ add_custom_target(all_examples)
function(add_example folder name)
add_executable(${name} ${folder}/${name}.cpp ${ARGN})
set_target_properties(${name} PROPERTIES RUNTIME_OUTPUT_DIRECTORY
"${CMAKE_CURRENT_BINARY_DIR}/${folder}")
install(FILES ${folder}/${name}.cpp
DESTINATION ${CMAKE_INSTALL_DATADIR}/caf/examples/${folder})
add_dependencies(${name} all_examples)
......@@ -33,8 +35,10 @@ add_core_example(message_passing promises)
add_core_example(message_passing request)
add_core_example(message_passing typed_calculator)
# streaming API
add_core_example(streaming integer_stream)
# flow API
add_core_example(flow iota)
add_core_example(flow observe-on)
add_core_example(flow spsc-buffer-resource)
# dynamic behavior changes using 'become'
add_core_example(dynamic_behavior skip_messages)
......
// Non-interactive example to showcase `from_callable`.
#include "caf/actor_system.hpp"
#include "caf/caf_main.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/scheduled_actor/flow.hpp"
#include <iostream>
namespace {
struct config : caf::actor_system_config {
config() {
opt_group{custom_options_, "global"} //
.add(n, "num-values,n", "number of values produced by the source");
}
size_t n = 10;
};
// --(rst-main-begin)--
void caf_main(caf::actor_system& sys, const config& cfg) {
sys.spawn([n = cfg.n](caf::event_based_actor* self) {
self
// Get an observable factory.
->make_observable()
// Produce an integer sequence starting at 1, i.e., 1, 2, 3, ...
.iota(1)
// Only take the requested number of items from the infinite sequence.
.take(n)
// Print each integer.
.for_each([](int x) { std::cout << x << '\n'; });
});
}
// --(rst-main-end)--
} // namespace
CAF_MAIN()
// Non-interactive example to showcase `observe_on`.
#include "caf/actor_system.hpp"
#include "caf/caf_main.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/scheduled_actor/flow.hpp"
#include <iostream>
namespace {
struct config : caf::actor_system_config {
config() {
opt_group{custom_options_, "global"} //
.add(n, "num-values,n", "number of values produced by the source");
}
size_t n = 10;
};
// --(rst-main-begin)--
void caf_main(caf::actor_system& sys, const config& cfg) {
// Create two actors without actually running them yet.
using actor_t = caf::event_based_actor;
auto [src, launch_src] = sys.spawn_inactive<actor_t>();
auto [snk, launch_snk] = sys.spawn_inactive<actor_t>();
// Define our data flow: generate data on `src` and print it on `snk`.
src
// Get an observable factory.
->make_observable()
// Produce an integer sequence starting at 1, i.e., 1, 2, 3, ...
.iota(1)
// Only take the requested number of items from the infinite sequence.
.take(cfg.n)
// Switch to `snk` for further processing.
.observe_on(snk)
// Print each integer.
.for_each([](int x) { std::cout << x << '\n'; });
// Allow the actors to run. After this point, we may no longer dereference
// the `src` and `snk` pointers! Calling these manually is optional. When
// removing these two lines, CAF automatically launches the actors at scope
// exit.
launch_src();
launch_snk();
}
// --(rst-main-end)--
} // namespace
CAF_MAIN()
// Non-interactive example to illustrate how to connect flows over an
// asynchronous SPSC (Single Producer Single Consumer) buffer.
#include "caf/actor_system.hpp"
#include "caf/async/spsc_buffer.hpp"
#include "caf/caf_main.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/scheduled_actor/flow.hpp"
#include <iostream>
namespace {
// --(rst-source-begin)--
// Simple source for generating a stream of integers from 1 to n.
void source(caf::event_based_actor* self,
caf::async::producer_resource<int> out, size_t n) {
self
// Get an observable factory.
->make_observable()
// Produce an integer sequence starting at 1, i.e., 1, 2, 3, ...
.iota(1)
// Only take the requested number of items from the infinite sequence.
.take(n)
// Subscribe the resource to the sequence, thereby starting the stream.
.subscribe(out);
}
// --(rst-source-end)--
// --(rst-sink-begin)--
// Simple sink for consuming a stream of integers, printing it to stdout.
void sink(caf::event_based_actor* self, caf::async::consumer_resource<int> in) {
self
// Get an observable factory.
->make_observable()
// Lift the input to an observable flow.
.from_resource(std::move(in))
// Print each integer.
.for_each([](int x) { std::cout << x << '\n'; });
}
// --(rst-sink-end)--
struct config : caf::actor_system_config {
config() {
opt_group{custom_options_, "global"} //
.add(n, "num-values,n", "number of values produced by the source");
}
size_t n = 100;
};
// --(rst-main-begin)--
void caf_main(caf::actor_system& sys, const config& cfg) {
auto [snk_res, src_res] = caf::async::make_spsc_buffer_resource<int>();
sys.spawn(sink, std::move(snk_res));
sys.spawn(source, std::move(src_res), cfg.n);
}
// --(rst-main-end)--
} // namespace
CAF_MAIN()
/******************************************************************************
* Basic, non-interactive streaming example for processing integers. *
******************************************************************************/
#include <iostream>
#include <vector>
#include "caf/all.hpp"
CAF_BEGIN_TYPE_ID_BLOCK(integer_stream, first_custom_type_id)
CAF_ADD_TYPE_ID(integer_stream, (caf::stream<int32_t>) )
CAF_ADD_TYPE_ID(integer_stream, (std::vector<int32_t>) )
CAF_END_TYPE_ID_BLOCK(integer_stream)
using std::endl;
using namespace caf;
namespace {
// --(rst-source-begin)--
// Simple source for generating a stream of integers from [0, n).
behavior int_source(event_based_actor* self) {
return {
[=](open_atom, int32_t n) {
// Produce at least one value.
if (n <= 0)
n = 1;
// Create a stream manager for implementing a stream source. The
// streaming logic requires three functions: initializer, generator, and
// predicate.
return attach_stream_source(
self,
// Initializer. The type of the first argument (state) is freely
// chosen. If no state is required, `caf::unit_t` can be used here.
[](int32_t& x) { x = 0; },
// Generator. This function is called by CAF to produce new stream
// elements for downstream actors. The `x` argument is our state again
// (with our freely chosen type). The second argument `out` points to
// the output buffer. The template argument (here: int) determines what
// elements downstream actors receive in this stream. Finally, `num` is
// a hint from CAF how many elements we should ideally insert into
// `out`. We can always insert fewer or more items.
[n](int32_t& x, downstream<int32_t>& out, size_t num) {
auto max_x = std::min(x + static_cast<int>(num), n);
for (; x < max_x; ++x)
out.push(x);
},
// Predicate. This function tells CAF when we reached the end.
[n](const int32_t& x) { return x == n; });
},
};
}
// --(rst-source-end)--
// --(rst-stage-begin)--
// Simple stage that only selects even numbers.
behavior int_selector(event_based_actor* self) {
return {
[=](stream<int32_t> in) {
// Create a stream manager for implementing a stream stage. Similar to
// `make_source`, we need three functions: initialzer, processor, and
// finalizer.
return attach_stream_stage(
self,
// Our input source.
in,
// Initializer. Here, we don't need any state and simply use unit_t.
[](unit_t&) {
// nop
},
// Processor. This function takes individual input elements as `val`
// and forwards even integers to `out`.
[](unit_t&, downstream<int32_t>& out, int32_t val) {
if (val % 2 == 0)
out.push(val);
},
// Finalizer. Allows us to run cleanup code once the stream terminates.
[=](unit_t&, const error& err) {
if (err) {
aout(self) << "int_selector aborted with error: " << err
<< std::endl;
} else {
aout(self) << "int_selector finalized" << std::endl;
}
// else: regular stream shutdown
});
},
};
}
// --(rst-stage-end)--
// --(rst-sink-begin)--
behavior int_sink(event_based_actor* self) {
return {
[=](stream<int32_t> in) {
// Create a stream manager for implementing a stream sink. Once more, we
// have to provide three functions: Initializer, Consumer, Finalizer.
return attach_stream_sink(
self,
// Our input source.
in,
// Initializer. Here, we store all values we receive. Note that streams
// are potentially unbound, so this is usually a bad idea outside small
// examples like this one.
[](std::vector<int>&) {
// nop
},
// Consumer. Takes individual input elements as `val` and stores them
// in our history.
[](std::vector<int32_t>& xs, int32_t val) { xs.emplace_back(val); },
// Finalizer. Allows us to run cleanup code once the stream terminates.
[=](std::vector<int32_t>& xs, const error& err) {
if (err) {
aout(self) << "int_sink aborted with error: " << err << std::endl;
} else {
aout(self) << "int_sink finalized after receiving: " << xs
<< std::endl;
}
});
},
};
}
// --(rst-sink-end)--
struct config : actor_system_config {
config() {
opt_group{custom_options_, "global"}
.add(with_stage, "with-stage,s", "use a stage for filtering odd numbers")
.add(n, "num-values,n", "number of values produced by the source");
}
bool with_stage = false;
int32_t n = 100;
};
// --(rst-main-begin)--
void caf_main(actor_system& sys, const config& cfg) {
auto src = sys.spawn(int_source);
auto snk = sys.spawn(int_sink);
auto pipeline = cfg.with_stage ? snk * sys.spawn(int_selector) * src
: snk * src;
anon_send(pipeline, open_atom_v, cfg.n);
}
// --(rst-main-end)--
} // namespace
CAF_MAIN(id_block::integer_stream)
......@@ -53,14 +53,17 @@ caf_add_component(
PRIVATE
CAF::internal
ENUM_TYPES
async.read_result
async.write_result
exit_reason
flow.observable_state
flow.observer_state
intrusive.inbox_result
intrusive.task_result
invoke_message_result
message_priority
pec
sec
stream_priority
HEADERS
${CAF_CORE_HEADERS}
SOURCES
......@@ -81,6 +84,9 @@ caf_add_component(
src/actor_registry.cpp
src/actor_system.cpp
src/actor_system_config.cpp
src/async/batch.cpp
src/async/consumer.cpp
src/async/producer.cpp
src/attachable.cpp
src/behavior.cpp
src/binary_deserializer.cpp
......@@ -92,13 +98,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/append_percent_encoded.cpp
src/detail/atomic_ref_counted.cpp
src/detail/base64.cpp
src/detail/behavior_impl.cpp
src/detail/behavior_stack.cpp
......@@ -119,6 +124,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
......@@ -127,26 +133,26 @@ caf_add_component(
src/detail/serialized_size.cpp
src/detail/set_thread_name.cpp
src/detail/shared_spinlock.cpp
src/detail/size_based_credit_controller.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
src/group.cpp
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
......@@ -166,8 +172,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
......@@ -186,8 +190,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
......@@ -221,11 +223,11 @@ caf_add_component(
actor_system_config
actor_termination
aout
async.spsc_buffer
behavior
binary_deserializer
binary_serializer
blocking_actor
broadcast_downstream_manager
byte
composition
config_option
......@@ -235,7 +237,7 @@ caf_add_component(
config_value_writer
const_typed_message_view
constructor_attach
continuous_streaming
cow_string
cow_tuple
decorator.sequencer
deep_to_string
......@@ -265,7 +267,6 @@ caf_add_component(
detail.ringbuffer
detail.ripemd_160
detail.serialized_size
detail.tick_emitter
detail.type_id_list_builder
detail.unique_function
detail.unordered_flat_map
......@@ -274,8 +275,25 @@ caf_add_component(
dynamic_spawn
error
expected
flow.concat
flow.concat_map
flow.defer
flow.empty
flow.fail
flow.flat_map
flow.for_each
flow.generation
flow.interval
flow.item_publisher
flow.merge
flow.mixed
flow.never
flow.observe_on
flow.prefix_and_tail
flow.publish
flow.single
flow.zip_with
function_view
fused_downstream_manager
handles
hash.fnv
hash.sha1
......@@ -305,22 +323,19 @@ 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
response_handle
response_promise
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 {
......@@ -25,42 +26,21 @@ public:
enum class state {
disposed, /// The action may no longer run.
scheduled, /// The action is scheduled for execution.
invoked, /// The action fired and needs rescheduling before running again.
waiting, /// The action waits for reschedule but didn't run yet.
};
/// Describes the result of an attempted state transition.
enum class transition {
success, /// Transition completed as expected.
disposed, /// No transition since the action has been disposed.
failure, /// No transition since preconditions did not hold.
};
/// Internal interface of `action`.
class impl : public disposable::impl {
class CAF_CORE_EXPORT impl : public disposable::impl {
public:
virtual transition reschedule() = 0;
virtual transition run() = 0;
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>;
// -- constructors, destructors, and assignment operators --------------------
explicit action(impl_ptr ptr) noexcept : pimpl_(std::move(ptr)) {
// nop
}
explicit action(impl_ptr ptr) noexcept;
action() noexcept = default;
......@@ -72,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 {
......@@ -82,28 +67,18 @@ public:
return pimpl_->current_state() == state::scheduled;
}
[[nodiscard]] bool invoked() const {
return pimpl_->current_state() == state::invoked;
}
// -- mutators ---------------------------------------------------------------
/// Tries to transition from `scheduled` to `invoked`, running the body of the
/// internal function object as a side effect on success.
/// @return whether the transition took place.
transition run();
/// Triggers the action.
void run() {
pimpl_->run();
}
/// Cancel the action if it has not been invoked yet.
void dispose() {
pimpl_->dispose();
}
/// Tries setting the state from `invoked` back to `scheduled`.
/// @return whether the transition took place.
transition reschedule() {
return pimpl_->reschedule();
}
// -- conversion -------------------------------------------------------------
/// Returns a smart pointer to the implementation.
......@@ -131,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_;
};
......@@ -139,12 +122,12 @@ private:
namespace caf::detail {
template <class F>
struct default_action_impl : ref_counted, action::impl {
struct default_action_impl : detail::atomic_ref_counted, action::impl {
std::atomic<action::state> state_;
F f_;
default_action_impl(F fn, action::state init_state)
: state_(init_state), f_(std::move(fn)) {
default_action_impl(F fn)
: state_(action::state::scheduled), f_(std::move(fn)) {
// nop
}
......@@ -160,42 +143,12 @@ struct default_action_impl : ref_counted, action::impl {
return state_.load();
}
action::transition reschedule() override {
auto st = action::state::invoked;
for (;;) {
if (state_.compare_exchange_strong(st, action::state::scheduled))
return action::transition::success;
switch (st) {
case action::state::invoked:
case action::state::waiting:
break; // Try again.
case action::state::disposed:
return action::transition::disposed;
default:
return action::transition::failure;
}
}
}
action::transition run() override {
auto st = state_.load();
switch (st) {
case action::state::scheduled:
f_();
// No retry. If this action has been disposed while running, we stay
// in the state 'disposed'. We assume that only one thread may try to
// transition from scheduled to invoked, while other threads may only
// dispose the action.
if (state_.compare_exchange_strong(st, action::state::invoked)) {
return action::transition::success;
} else {
CAF_ASSERT(st == action::state::disposed);
return action::transition::disposed;
}
case action::state::disposed:
return action::transition::disposed;
default:
return action::transition::failure;
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_();
}
}
......@@ -220,17 +173,12 @@ struct default_action_impl : ref_counted, action::impl {
namespace caf {
/// Convenience function for creating @ref action objects from a function
/// object.
/// Convenience function for creating an @ref action from a function object.
/// @param f The body for the action.
/// @param init_state either `action::state::scheduled` or
/// `action::state::waiting`.
template <class F>
action make_action(F f, action::state init_state = action::state::scheduled) {
CAF_ASSERT(init_state == action::state::scheduled
|| init_state == action::state::waiting);
action make_action(F f) {
using impl_t = detail::default_action_impl<F>;
return action{make_counted<impl_t>(std::move(f), init_state)};
return action{make_counted<impl_t>(std::move(f))};
}
} // namespace caf
......
......@@ -53,18 +53,7 @@ public:
/// @param f The action to schedule.
/// @note The action runs on the thread of the clock worker and thus must
/// complete within a very short time in order to not delay other work.
disposable schedule(time_point t, action f);
/// Schedules an action for periodic execution.
/// @param first_run The local time at which the action should run initially.
/// @param f The action to schedule.
/// @param period The time to wait between two runs. A non-positive period
/// disables periodic execution.
/// @note The action runs on the thread of the clock worker and thus must
/// complete within a very short time in order to not delay other work.
virtual disposable
schedule_periodically(time_point first_run, action f, duration_type period)
= 0;
virtual disposable schedule(time_point t, action f) = 0;
/// Schedules an action for execution by an actor at a later time.
/// @param t The local time at which the action should get enqueued to the
......@@ -73,17 +62,6 @@ public:
/// @param target The actor that should run the action.
disposable schedule(time_point t, action f, strong_actor_ptr target);
/// Schedules an action for periodic execution by an actor.
/// @param first_run The local time at which the action should get enqueued to
/// the mailbox of the target for the first time.
/// @param f The action to schedule.
/// @param target The actor that should run the action.
/// @param period The time to wait between two messages to the actor.
/// @param policy The strategy for dealing with a stalling target.
disposable schedule_periodically(time_point first_run, action f,
strong_actor_ptr target,
duration_type period, stall_policy policy);
/// Schedules an action for execution by an actor at a later time.
/// @param target The actor that should run the action.
/// @param f The action to schedule.
......@@ -91,17 +69,6 @@ public:
/// mailbox of the target.
disposable schedule(time_point t, action f, weak_actor_ptr target);
/// Schedules an action for periodic execution by an actor.
/// @param first_run The local time at which the action should get enqueued to
/// the mailbox of the target for the first time.
/// @param f The action to schedule.
/// @param target The actor that should run the action.
/// @param period The time to wait between two messages to the actor.
/// @param policy The strategy for dealing with a stalling target.
disposable schedule_periodically(time_point first_run, action f,
weak_actor_ptr target, duration_type period,
stall_policy policy);
/// Schedules an arbitrary message to `receiver` for time point `t`.
disposable schedule_message(time_point t, strong_actor_ptr receiver,
mailbox_element_ptr content);
......
......@@ -587,6 +587,96 @@ public:
return res;
}
/// Utility function object that allows users to explicitly launch an action
/// by calling `operator()` but calls it implicitly at scope exit.
template <class F>
class launcher {
public:
launcher() : ready(false) {
// nop
}
explicit launcher(F&& f) : ready(true) {
new (&fn) F(std::move(f));
}
launcher(launcher&& other) : ready(other.ready) {
if (ready) {
new (&fn) F(std::move(other.fn));
other.reset();
}
}
launcher& operator=(launcher&& other) {
if (this != &other) {
if (ready)
reset();
if (other.ready) {
ready = true;
new (&fn) F(std::move(other.fn));
other.reset();
}
}
return *this;
}
launcher(const launcher&) = delete;
launcher& operator=(const launcher&) = delete;
~launcher() {
if (ready)
fn();
}
void operator()() {
if (ready) {
fn();
reset();
}
}
private:
// @pre `ready == true`
void reset() {
CAF_ASSERT(ready);
ready = false;
fn.~F();
}
bool ready;
union {
F fn;
};
};
/// 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, launcher<decltype(launch)>(std::move(launch)));
}
void profiler_add_actor(const local_actor& self, const local_actor* parent) {
if (profiler_)
profiler_->add_actor(self, parent);
......
......@@ -25,7 +25,6 @@
#include "caf/fwd.hpp"
#include "caf/is_typed_actor.hpp"
#include "caf/settings.hpp"
#include "caf/stream.hpp"
#include "caf/thread_hook.hpp"
namespace caf {
......@@ -196,14 +195,6 @@ public:
/// Credentials for connecting to the bootstrap node.
std::string bootstrap_node;
// -- stream parameters ------------------------------------------------------
/// @private
timespan stream_max_batch_delay;
/// @private
timespan stream_credit_round_interval;
// -- OpenSSL parameters -----------------------------------------------------
std::string openssl_certificate;
......@@ -260,8 +251,8 @@ public:
/// @param opts User-defined config options for type checking.
/// @returns A ::settings dictionary with the parsed content of `filename` on
/// success, an ::error otherwise.
static expected<settings>
parse_config_file(const char* filename, const config_option_set& opts);
static expected<settings> parse_config_file(const char* filename,
const config_option_set& opts);
/// Tries to open `filename`, parses its content as CAF config file and
/// stores all entries in `result` (overrides conflicting entries). Also
......@@ -272,9 +263,9 @@ public:
/// partial results if this function returns an error.
/// @returns A default-constructed ::error on success, the error code of the
/// parser otherwise.
static error
parse_config_file(const char* filename, const config_option_set& opts,
settings& result);
static error parse_config_file(const char* filename,
const config_option_set& opts,
settings& result);
/// Parses the content of `source` using CAF's config format.
/// @param source Character sequence in CAF's config format.
......@@ -288,8 +279,8 @@ public:
/// @param opts User-defined config options for type checking.
/// @returns A ::settings dictionary with the parsed content of `source` on
/// success, an ::error otherwise.
static expected<settings>
parse_config(std::istream& source, const config_option_set& opts);
static expected<settings> parse_config(std::istream& source,
const config_option_set& opts);
/// Parses the content of `source` using CAF's config format and stores all
/// entries in `result` (overrides conflicting entries). Also type-checks
......
......@@ -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"
......@@ -82,8 +75,6 @@
#include "caf/skip.hpp"
#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 +90,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"
......
// 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 "caf/async/consumer.hpp"
#include "caf/async/read_result.hpp"
#include "caf/async/spsc_buffer.hpp"
#include "caf/intrusive_ptr.hpp"
#include "caf/make_counted.hpp"
#include "caf/ref_counted.hpp"
#include <chrono>
#include <condition_variable>
#include <type_traits>
namespace caf::async {
/// Blocking interface for emitting items to an asynchronous consumer.
template <class T>
class blocking_consumer {
public:
class impl : public ref_counted, public consumer {
public:
impl() = delete;
impl(const impl&) = delete;
impl& operator=(const impl&) = delete;
explicit impl(spsc_buffer_ptr<T> buf) : buf_(std::move(buf)) {
buf_->set_consumer(this);
}
template <class ErrorPolicy, class TimePoint>
read_result pull(ErrorPolicy policy, T& item, TimePoint timeout) {
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;
}
auto [again, n] = buf_->pull_unsafe(guard, policy, 1u, *this);
CAF_IGNORE_UNUSED(again);
CAF_ASSERT(!again || n == 1);
if (n == 1) {
return read_result::ok;
} else if (buf_->abort_reason_unsafe()) {
return read_result::abort;
} else {
return read_result::stop;
}
}
template <class ErrorPolicy>
read_result pull(ErrorPolicy policy, T& item) {
return pull(policy, item, none);
}
void on_next(span<const T> items) {
CAF_ASSERT(items.size() == 1);
*val_ = items[0];
}
void on_complete() {
}
void on_error(const caf::error&) {
// nop
}
void dispose() {
if (buf_) {
buf_->cancel();
buf_ = nullptr;
}
}
void on_producer_ready() override {
// nop
}
void on_producer_wakeup() override {
// NOTE: buf_->mtx() is already locked at this point
cv_.notify_all();
}
void ref_consumer() const noexcept override {
ref();
}
void deref_consumer() const noexcept override {
deref();
}
error abort_reason() const {
buf_->abort_reason()();
}
CAF_INTRUSIVE_PTR_FRIENDS(impl)
private:
spsc_buffer_ptr<T> buf_;
std::condition_variable cv_;
T* val_ = nullptr;
};
using impl_ptr = intrusive_ptr<impl>;
blocking_consumer() = default;
blocking_consumer(const blocking_consumer&) = delete;
blocking_consumer& operator=(const blocking_consumer&) = delete;
blocking_consumer(blocking_consumer&&) = default;
blocking_consumer& operator=(blocking_consumer&&) = default;
explicit blocking_consumer(impl_ptr ptr) : impl_(std::move(ptr)) {
// nop
}
~blocking_consumer() {
if (impl_)
impl_->cancel();
}
/// Fetches the next item. If there is no item available, this functions
/// blocks unconditionally.
/// @param policy Either @ref instant_error, @ref delay_error or
/// @ref ignore_errors.
/// @param item Output parameter for storing the received item.
/// @returns the status of the read operation. The function writes to `item`
/// only when also returning `read_result::ok`.
template <class ErrorPolicy>
read_result pull(ErrorPolicy policy, T& item) {
return impl_->pull(policy, item);
}
/// Fetches the next item. If there is no item available, this functions
/// blocks until the absolute timeout was reached.
/// @param policy Either @ref instant_error, @ref delay_error or
/// @ref ignore_errors.
/// @param item Output parameter for storing the received item.
/// @param timeout Absolute timeout for the receive operation.
/// @returns the status of the read operation. The function writes to `item`
/// only when also returning `read_result::ok`.
template <class ErrorPolicy, class Clock,
class Duration = typename Clock::duration>
read_result pull(ErrorPolicy policy, T& item,
std::chrono::time_point<Clock, Duration> timeout) {
return impl_->pull(policy, item, timeout);
}
/// Fetches the next item. If there is no item available, this functions
/// blocks until the relative timeout was reached.
/// @param policy Either @ref instant_error, @ref delay_error or
/// @ref ignore_errors.
/// @param item Output parameter for storing the received item.
/// @param timeout Maximum duration before returning from the function.
/// @returns the status of the read operation. The function writes to `item`
/// only when also returning `read_result::ok`.
template <class ErrorPolicy, class Rep, class Period>
read_result pull(ErrorPolicy policy, T& item,
std::chrono::duration<Rep, Period> timeout) {
auto abs_timeout = std::chrono::system_clock::now() + timeout;
return impl_->pull(policy, item, abs_timeout);
}
error abort_reason() const {
impl_->abort_reason();
}
private:
intrusive_ptr<impl> impl_;
};
template <class T>
expected<blocking_consumer<T>>
make_blocking_consumer(consumer_resource<T> res) {
if (auto buf = res.try_open()) {
using impl_t = typename blocking_consumer<T>::impl;
return {blocking_consumer<T>{make_counted<impl_t>(std::move(buf))}};
} else {
return {make_error(sec::cannot_open_resource)};
}
}
} // 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 <condition_variable>
#include <type_traits>
#include "caf/async/producer.hpp"
#include "caf/async/spsc_buffer.hpp"
#include "caf/intrusive_ptr.hpp"
#include "caf/make_counted.hpp"
#include "caf/ref_counted.hpp"
namespace caf::async {
/// Blocking interface for emitting items to an asynchronous consumer.
template <class T>
class blocking_producer {
public:
class impl : public ref_counted, public producer {
public:
impl() = delete;
impl(const impl&) = delete;
impl& operator=(const impl&) = delete;
explicit impl(spsc_buffer_ptr<T> buf) : buf_(std::move(buf)) {
buf_->set_producer(this);
}
bool push(span<const T> items) {
std::unique_lock guard{mtx_};
while (items.size() > 0) {
while (demand_ == 0)
cv_.wait(guard);
if (demand_ < 0) {
return false;
} else {
auto n = std::min(static_cast<size_t>(demand_), items.size());
guard.unlock();
buf_->push(items.subspan(0, n));
guard.lock();
demand_ -= static_cast<ptrdiff_t>(n);
items = items.subspan(n);
}
}
return true;
}
bool push(const T& item) {
return push(make_span(&item, 1));
}
void close() {
if (buf_) {
buf_->close();
buf_ = nullptr;
}
}
void abort(error reason) {
if (buf_) {
buf_->abort(std::move(reason));
buf_ = nullptr;
}
}
bool canceled() const {
std::unique_lock<std::mutex> guard{mtx_};
return demand_ == -1;
}
void on_consumer_ready() override {
// nop
}
void on_consumer_cancel() override {
std::unique_lock<std::mutex> guard{mtx_};
demand_ = -1;
cv_.notify_all();
}
void on_consumer_demand(size_t demand) override {
std::unique_lock<std::mutex> guard{mtx_};
if (demand_ == 0) {
demand_ += static_cast<ptrdiff_t>(demand);
cv_.notify_all();
} else if (demand_ > 0) {
demand_ += static_cast<ptrdiff_t>(demand);
}
}
void ref_producer() const noexcept override {
ref();
}
void deref_producer() const noexcept override {
deref();
}
CAF_INTRUSIVE_PTR_FRIENDS(impl)
private:
spsc_buffer_ptr<T> buf_;
mutable std::mutex mtx_;
std::condition_variable cv_;
ptrdiff_t demand_ = 0;
};
using impl_ptr = intrusive_ptr<impl>;
blocking_producer() = default;
blocking_producer(const blocking_producer&) = delete;
blocking_producer& operator=(const blocking_producer&) = delete;
blocking_producer(blocking_producer&&) = default;
blocking_producer& operator=(blocking_producer&&) = default;
explicit blocking_producer(impl_ptr ptr) : impl_(std::move(ptr)) {
// nop
}
~blocking_producer() {
if (impl_)
impl_->close();
}
/// Pushes an item to the consumer. If there is no demand by the consumer to
/// deliver the item, this functions blocks unconditionally.
/// @returns `true` if the item was delivered to the consumer or `false` if
/// the consumer no longer receives any additional item.
bool push(const T& item) {
return impl_->push(item);
}
/// Pushes multiple items to the consumer. If there is no demand by the
/// consumer to deliver all items, this functions blocks unconditionally until
/// all items have been delivered.
/// @returns `true` if all items were delivered to the consumer or `false` if
/// the consumer no longer receives any additional item.
bool push(span<const T> items) {
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;
}
}
/// Checks whether the consumer canceled its subscription.
bool canceled() const {
return impl_->canceled();
}
private:
intrusive_ptr<impl> impl_;
};
template <class T>
expected<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))}};
} else {
return {make_error(sec::cannot_open_resource)};
}
}
} // 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 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 spsc_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
namespace caf::async {
/// Policy type for having `consume` call `on_error` immediately after the
/// producer has aborted even if the buffer still contains items.
struct prioritize_errors_t {
static constexpr bool calls_on_error = true;
};
/// @relates prioritize_errors_t
constexpr auto prioritize_errors = prioritize_errors_t{};
/// Policy type for having `consume` call `on_error` only after processing all
/// items from the buffer.
struct delay_errors_t {
static constexpr bool calls_on_error = true;
};
/// @relates delay_errors_t
constexpr auto delay_errors = delay_errors_t{};
} // 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,44 +4,39 @@
#pragma once
#include <cstdint>
#include <string>
#include <type_traits>
#include "caf/default_enum_inspect.hpp"
#include "caf/detail/core_export.hpp"
#include "caf/fwd.hpp"
namespace caf {
/// Categorizes individual streams.
enum class stream_priority : uint8_t {
/// Denotes soft-realtime traffic.
very_high,
/// Denotes time-sensitive traffic.
high,
/// Denotes traffic with moderate timing requirements.
normal,
/// Denotes uncritical traffic without timing requirements.
low,
/// Denotes best-effort traffic.
very_low
#include <type_traits>
namespace caf::async {
/// Encodes the result of an asynchronous read operation.
enum class read_result {
/// Signals that the read operation succeeded.
ok,
/// Signals that the reader reached the end of the input.
stop,
/// Signals that the source failed with an error.
abort,
/// Signals that the read operation timed out.
timeout,
};
/// @relates stream_priority
CAF_CORE_EXPORT std::string to_string(stream_priority x);
/// @relates read_result
CAF_CORE_EXPORT std::string to_string(read_result);
/// @relates stream_priority
CAF_CORE_EXPORT bool from_string(string_view, stream_priority&);
/// @relates read_result
CAF_CORE_EXPORT bool from_string(string_view, read_result&);
/// @relates stream_priority
CAF_CORE_EXPORT bool from_integer(std::underlying_type_t<stream_priority>,
stream_priority&);
/// @relates read_result
CAF_CORE_EXPORT bool from_integer(std::underlying_type_t<read_result>,
read_result&);
/// @relates stream_priority
/// @relates read_result
template <class Inspector>
bool inspect(Inspector& f, stream_priority& x) {
bool inspect(Inspector& f, read_result& x) {
return default_enum_inspect(f, x);
}
} // namespace caf
} // 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/default_enum_inspect.hpp"
#include "caf/detail/core_export.hpp"
#include <type_traits>
namespace caf::async {
/// Encodes the result of an asynchronous write operation.
enum class write_result {
/// Signals that the write operation succeeded.
ok,
/// Signals that the item must be dropped because the write operation failed
/// with an unrecoverable error. Retries will fail with the same result. When
/// writing to a @ref producer_resource, this usually means the consumer
/// closed its end of the buffer.
drop,
/// Signals that the write operation timed out.
timeout,
};
/// @relates write_result
CAF_CORE_EXPORT std::string to_string(write_result);
/// @relates write_result
CAF_CORE_EXPORT bool from_string(string_view, write_result&);
/// @relates write_result
CAF_CORE_EXPORT bool from_integer(std::underlying_type_t<write_result>,
write_result&);
/// @relates write_result
template <class Inspector>
bool inspect(Inspector& f, write_result& x) {
return default_enum_inspect(f, x);
}
} // 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/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
......@@ -4,6 +4,7 @@
#pragma once
#include "caf/allowed_unsafe_message_type.hpp"
#include "caf/config.hpp"
#include "caf/detail/type_traits.hpp"
#include "caf/error.hpp"
......@@ -91,4 +92,10 @@ auto make_shared_type_erased_callback(F fun) {
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
CAF_ALLOW_UNSAFE_MESSAGE_TYPE(caf::shared_action_ptr)
......@@ -242,6 +242,14 @@ struct IUnknown;
static_cast<void>(0)
#endif
// CAF_DEBUG_STMT(stmt): evaluates to stmt when compiling with runtime checks
// and to an empty expression otherwise.
#ifndef CAF_ENABLE_RUNTIME_CHECKS
# define CAF_DEBUG_STMT(stmt) static_cast<void>(0)
#else
# define CAF_DEBUG_STMT(stmt) stmt
#endif
// Convenience macros.
#define CAF_IGNORE_UNUSED(x) static_cast<void>(x)
......
// 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 "caf/string_algorithms.hpp"
#include <string>
#include <string_view>
namespace caf {
/// A copy-on-write string implementation that wraps a `std::basic_string`.
template <class CharT>
class basic_cow_string
: detail::comparable<basic_cow_string<CharT>>,
detail::comparable<basic_cow_string<CharT>, std::basic_string<CharT>>,
detail::comparable<basic_cow_string<CharT>, const CharT*> {
public:
// -- member types -----------------------------------------------------------
using std_type = std::basic_string<CharT>;
using view_type = std::basic_string_view<CharT>;
using size_type = typename std_type::size_type;
using const_iterator = typename std_type::const_iterator;
using const_reverse_iterator = typename std_type::const_reverse_iterator;
// -- constants --------------------------------------------------------------
static inline const size_type npos = std_type::npos;
// -- constructors, destructors, and assignment operators --------------------
basic_cow_string() {
impl_ = make_counted<impl>();
}
explicit basic_cow_string(std_type str) {
impl_ = make_counted<impl>(std::move(str));
}
explicit basic_cow_string(view_type str) {
impl_ = make_counted<impl>(std_type{str});
}
basic_cow_string(basic_cow_string&&) noexcept = default;
basic_cow_string(const basic_cow_string&) noexcept = default;
basic_cow_string& operator=(basic_cow_string&&) noexcept = default;
basic_cow_string& operator=(const basic_cow_string&) noexcept = default;
// -- properties -------------------------------------------------------------
/// Returns a mutable reference to the managed string. Copies the string 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().str;
}
/// Returns the managed string.
const std_type& str() const noexcept {
return impl_->str;
}
/// 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_->str.empty();
}
size_type size() const noexcept {
return impl_->str.size();
}
size_type length() const noexcept {
return impl_->str.length();
}
size_type max_size() const noexcept {
return impl_->str.max_size();
}
// -- element access ---------------------------------------------------------
CharT at(size_type pos) const {
return impl_->str.at(pos);
}
CharT operator[](size_type pos) const {
return impl_->str[pos];
}
CharT front() const {
return impl_->str.front();
}
CharT back() const {
return impl_->str.back();
}
const CharT* data() const noexcept {
return impl_->str.data();
}
const CharT* c_str() const noexcept {
return impl_->str.c_str();
}
// -- conversion and copying -------------------------------------------------
operator view_type() const noexcept {
return view_type{impl_->str};
}
basic_cow_string substr(size_type pos = 0, size_type count = npos) const {
return basic_cow_string{impl_->str.substr(pos, count)};
}
size_type copy(CharT* dest, size_type count, size_type pos = 0) const {
return impl_->str.copy(dest, count, pos);
}
// -- iterator access --------------------------------------------------------
const_iterator begin() const noexcept {
return impl_->str.begin();
}
const_iterator cbegin() const noexcept {
return impl_->str.begin();
}
const_reverse_iterator rbegin() const noexcept {
return impl_->str.rbegin();
}
const_reverse_iterator crbegin() const noexcept {
return impl_->str.rbegin();
}
const_iterator end() const noexcept {
return impl_->str.end();
}
const_iterator cend() const noexcept {
return impl_->str.end();
}
const_reverse_iterator rend() const noexcept {
return impl_->str.rend();
}
const_reverse_iterator crend() const noexcept {
return impl_->str.rend();
}
// -- predicates -------------------------------------------------------------
bool starts_with(view_type x) const noexcept {
return caf::starts_with(impl_->str, x);
}
bool starts_with(CharT x) const noexcept {
return empty() ? false : front() == x;
}
bool starts_with(const CharT* x) const {
return starts_with(view_type{x});
}
bool ends_with(view_type x) const noexcept {
return caf::ends_with(impl_->str, x);
}
bool ends_with(CharT x) const noexcept {
return empty() ? false : back() == x;
}
bool ends_with(const CharT* x) const {
return ends_with(view_type{x});
}
bool contains(std::string_view x) const noexcept {
return find(x) != npos;
}
bool contains(char x) const noexcept {
return find(x) != npos;
}
bool contains(const CharT* x) const {
return contains(view_type{x});
}
// -- search -----------------------------------------------------------------
size_type find(const std::string& str, size_type pos = 0) const noexcept {
return impl_->str.find(str, pos);
}
size_type find(const basic_cow_string& str,
size_type pos = 0) const noexcept {
return find(str.impl_->str, pos);
}
size_type find(const CharT* str, size_type pos, size_type count) const {
return impl_->str.find(str, pos, count);
}
size_type find(const CharT* str, size_type pos = 0) const {
return impl_->str.find(str, pos);
}
size_type find(char x, size_type pos = 0) const noexcept {
return impl_->str.find(x, pos);
}
template <class T>
std::enable_if_t<std::is_convertible_v<const T&, view_type> //
&& !std::is_convertible_v<const T&, const CharT*>,
size_type>
find(const T& x, size_type pos = 0) const noexcept {
return impl_->str.find(x, pos);
}
// -- comparison -------------------------------------------------------------
int compare(const CharT* x) const noexcept {
return str().compare(x);
}
int compare(const std_type& x) const noexcept {
return str().compare(x);
}
int compare(const cow_string& x) const noexcept {
return impl_ == x.impl_ ? 0 : compare(x.str());
}
// -- friends ----------------------------------------------------------------
template <class Inspector>
friend bool inspect(Inspector& f, basic_cow_string& x) {
if constexpr (Inspector::is_loading) {
return f.apply(x.unshared());
} else {
return f.apply(x.impl_->str);
}
}
private:
struct impl : ref_counted {
std_type str;
impl() = default;
explicit impl(std_type in) : str(std::move(in)) {
// nop
}
impl* copy() const {
return new impl{str};
}
};
intrusive_cow_ptr<impl> impl_;
};
/// A copy-on-write wrapper for a `std::string`;
/// @relates basic_cow_string
using cow_string = basic_cow_string<char>;
/// A copy-on-write wrapper for a `std::string`;
/// @relates basic_cow_string
using cow_u16string = basic_cow_string<char16_t>;
/// A copy-on-write wrapper for a `std::string`;
/// @relates basic_cow_string
using cow_u32string = basic_cow_string<char32_t>;
} // 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 <vector>
namespace caf {
/// A copy-on-write vector implementation that wraps a `std::vector`.
template <class T>
class cow_vector {
public:
// -- member types -----------------------------------------------------------
using std_type = std::vector<T>;
using size_type = typename std_type::size_type;
using const_iterator = typename std_type::const_iterator;
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));
}
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 <cstdint>
#include "caf/detail/core_export.hpp"
#include "caf/downstream_msg.hpp"
#include "caf/fwd.hpp"
namespace caf {
/// Computes credit for an attached source.
class CAF_CORE_EXPORT credit_controller {
public:
// -- member types -----------------------------------------------------------
/// Wraps an assignment of the controller to its source.
struct calibration {
/// Stores how much credit the path may emit at most.
int32_t max_credit;
/// Stores how many elements we demand per batch.
int32_t batch_size;
/// Stores how many batches the caller should wait before calling
/// `calibrate` again.
int32_t next_calibration;
};
// -- constructors, destructors, and assignment operators --------------------
virtual ~credit_controller();
// -- pure virtual functions -------------------------------------------------
/// Called before processing the batch `x` in order to allow the controller
/// to keep statistics on incoming batches.
virtual void before_processing(downstream_msg::batch& batch) = 0;
/// Returns an initial calibration for the path.
virtual calibration init() = 0;
/// Computes a credit assignment to the source after crossing the
/// low-threshold. May assign zero credit.
virtual calibration calibrate() = 0;
};
} // 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
......@@ -115,3 +115,11 @@ constexpr auto cached_udp_buffers = size_t{10};
constexpr auto max_pending_msgs = size_t{10};
} // 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
// 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,9 +12,6 @@
#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"
......@@ -47,9 +44,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);
}
......@@ -86,31 +80,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 <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
......@@ -15,15 +15,6 @@ namespace caf::detail {
class CAF_CORE_EXPORT test_actor_clock : public actor_clock {
public:
// -- member types -----------------------------------------------------------
struct schedule_entry {
action f;
duration_type period;
};
using schedule_map = std::multimap<time_point, schedule_entry>;
// -- constructors, destructors, and assignment operators --------------------
test_actor_clock();
......@@ -32,17 +23,14 @@ public:
time_point now() const noexcept override;
disposable schedule_periodically(time_point first_run, action f,
duration_type period) override;
disposable schedule(time_point abs_time, action f) override;
// -- testing DSL API --------------------------------------------------------
/// Returns whether the actor clock has at least one pending timeout.
bool has_pending_timeout() const {
auto not_disposed = [](const auto& kvp) {
return !kvp.second.f.disposed();
};
return std::any_of(schedule.begin(), schedule.end(), not_disposed);
auto not_disposed = [](const auto& kvp) { return !kvp.second.disposed(); };
return std::any_of(actions.begin(), actions.end(), not_disposed);
}
/// Triggers the next pending timeout regardless of its timestamp. Sets
......@@ -61,11 +49,18 @@ public:
/// @returns The number of triggered timeouts.
size_t advance_time(duration_type x);
// -- properties -------------------------------------------------------------
/// @pre has_pending_timeout()
time_point next_timeout() const {
return actions.begin()->first;
}
// -- member variables -------------------------------------------------------
time_point current_time;
schedule_map schedule;
std::multimap<time_point, action> actions;
private:
bool try_trigger_once();
......
......@@ -25,11 +25,12 @@ public:
// -- member types -----------------------------------------------------------
using super = actor_clock;
/// Stores actions along with their scheduling period.
struct schedule_entry {
time_point t;
action f;
duration_type period;
};
/// @relates schedule_entry
......@@ -41,8 +42,9 @@ public:
// -- overrides --------------------------------------------------------------
disposable schedule_periodically(time_point first_run, action f,
duration_type period) override;
using super::schedule;
disposable schedule(time_point abs_time, action f) override;
// -- thread management ------------------------------------------------------
......
// 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 <chrono>
#include <initializer_list>
#include "caf/config.hpp"
#include "caf/detail/core_export.hpp"
namespace caf::detail {
/// Converts realtime into a series of ticks, whereas each tick represents a
/// preconfigured timespan. For example, a tick emitter configured with a
/// timespan of 25ms generates a tick every 25ms after starting it.
class CAF_CORE_EXPORT tick_emitter {
public:
// -- member types -----------------------------------------------------------
/// Discrete point in time.
using clock_type = std::chrono::steady_clock;
/// Discrete point in time.
using time_point = typename clock_type::time_point;
/// Difference between two points in time.
using duration_type = typename time_point::duration;
// -- constructors, destructors, and assignment operators --------------------
tick_emitter();
tick_emitter(time_point now);
/// Queries whether the start time is non-default constructed.
bool started() const;
/// Configures the start time.
void start(time_point now);
/// Resets the start time to 0.
void stop();
/// Configures the time interval per tick.
void interval(duration_type x);
/// Returns the time interval per tick.
duration_type interval() const {
return interval_;
}
/// Advances time and calls `consumer` for each emitted tick.
template <class F>
void update(time_point now, F& consumer) {
CAF_ASSERT(interval_.count() != 0);
auto d = now - start_;
auto current_tick_id = static_cast<size_t>(d.count() / interval_.count());
while (last_tick_id_ < current_tick_id)
consumer(++last_tick_id_);
}
/// Advances time by `t` and returns all triggered periods as bitmask.
size_t timeouts(time_point t, std::initializer_list<size_t> periods);
/// Returns the next time point after `t` that would trigger any of the tick
/// periods, i.e., returns the next time where any of the tick periods
/// triggers `tick id % period == 0`.
time_point next_timeout(time_point t, std::initializer_list<size_t> periods);
private:
time_point start_;
duration_type interval_;
size_t last_tick_id_;
};
} // 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
This diff is collapsed.
......@@ -8,16 +8,25 @@
#include "caf/detail/core_export.hpp"
#include "caf/intrusive_ptr.hpp"
#include "caf/ref_counted.hpp"
namespace caf {
/// Represents a disposable resource.
class CAF_CORE_EXPORT disposable {
class CAF_CORE_EXPORT disposable : detail::comparable<disposable> {
public:
// -- member types -----------------------------------------------------------
/// Internal implementation class of a `disposable`.
class impl {
class CAF_CORE_EXPORT impl {
public:
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();
virtual void dispose() = 0;
......@@ -29,32 +38,39 @@ public:
virtual void ref_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();
}
};
// -- constructors, destructors, and assignment operators --------------------
explicit disposable(intrusive_ptr<impl> pimpl) noexcept
: pimpl_(std::move(pimpl)) {
// nop
}
disposable() noexcept = default;
disposable(disposable&&) noexcept = default;
disposable(const disposable&) noexcept = default;
disposable& operator=(disposable&&) noexcept = default;
disposable& operator=(const disposable&) noexcept = default;
disposable& operator=(std::nullptr_t) noexcept {
pimpl_ = nullptr;
return *this;
}
// -- factories --------------------------------------------------------------
/// Combines multiple disposables into a single disposable. The new disposable
/// is disposed if all of its elements are disposed. Disposing the composite
/// disposes all elements individually.
static disposable make_composite(std::vector<disposable> entries);
// -- mutators ---------------------------------------------------------------
/// Disposes the resource. Calling `dispose()` on a disposed resource is a
/// no-op.
void dispose() {
......@@ -64,6 +80,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.
[[nodiscard]] bool disposed() const noexcept {
return pimpl_ ? pimpl_->disposed() : true;
......@@ -89,6 +112,8 @@ public:
return pimpl_.get();
}
// -- conversions ------------------------------------------------------------
/// Returns a smart pointer to the implementation.
[[nodiscard]] intrusive_ptr<impl>&& as_intrusive_ptr() && noexcept {
return std::move(pimpl_);
......@@ -99,13 +124,18 @@ public:
return pimpl_;
}
/// Exchanges the content of this handle with `other`.
void swap(disposable& other) {
pimpl_.swap(other.pimpl_);
// -- comparisons ------------------------------------------------------------
/// Compares the internal pointers.
[[nodiscard]] intptr_t compare(const disposable& other) const noexcept {
return pimpl_.compare(other.pimpl_);
}
private:
intrusive_ptr<impl> pimpl_;
};
/// @relates disposable
using disposable_impl = disposable::impl;
} // namespace caf
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
......@@ -48,6 +48,8 @@ public:
/// Required by `spawn` for type deduction.
using behavior_type = behavior;
using handle_type = actor;
// -- constructors, destructors ----------------------------------------------
explicit event_based_actor(actor_config& cfg);
......
......@@ -4,12 +4,7 @@
#pragma once
#include "caf/detail/type_traits.hpp"
#include "caf/flow/observable.hpp"
#include "caf/flow/observer.hpp"
namespace caf {
struct invalid_stream_t {};
constexpr invalid_stream_t invalid_stream = invalid_stream_t{};
} // namespace caf
namespace caf::flow {} // namespace caf::flow
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 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.
} // namespace caf::flow
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
The namespace `caf::flow::op` contains (private) implementation classes for
various operators.
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.
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