Commit 36d5a8bd authored by Dominik Charousset's avatar Dominik Charousset

Delete obsolete stream support

After implementing the new flow API, we drop the legacy streaming API in
preparation of re-implementing a fresh API based on the flows.
parent 12cb692a
......@@ -42,7 +42,7 @@ configure_file("${PROJECT_SOURCE_DIR}/cmake/build_config.hpp.in"
install(FILES "${CMAKE_CURRENT_BINARY_DIR}/caf/detail/build_config.hpp"
DESTINATION "${CMAKE_INSTALL_INCLUDEDIR}/caf/detail/")
#-- add targets ----------------------------------------------------------------
# -- add targets ---------------------------------------------------------------
caf_add_component(
core
......@@ -64,7 +64,6 @@ caf_add_component(
message_priority
pec
sec
stream_priority
HEADERS
${CAF_CORE_HEADERS}
SOURCES
......@@ -99,7 +98,6 @@ 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
......@@ -135,17 +133,12 @@ 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
......@@ -160,7 +153,6 @@ caf_add_component(
src/group_manager.cpp
src/group_module.cpp
src/hash/sha1.cpp
src/inbound_path.cpp
src/init_global_meta_objects.cpp
src/ipv4_address.cpp
src/ipv4_endpoint.cpp
......@@ -180,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
......@@ -200,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
......@@ -240,7 +228,6 @@ caf_add_component(
binary_deserializer
binary_serializer
blocking_actor
broadcast_downstream_manager
byte
composition
config_option
......@@ -250,7 +237,6 @@ caf_add_component(
config_value_writer
const_typed_message_view
constructor_attach
continuous_streaming
cow_string
cow_tuple
decorator.sequencer
......@@ -281,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
......@@ -307,7 +292,6 @@ caf_add_component(
flow.single
flow.zip_with
function_view
fused_downstream_manager
handles
hash.fnv
hash.sha1
......@@ -337,13 +321,10 @@ caf_add_component(
metaprogramming
mixin.requester
mixin.sender
mock_streaming_classes
mtl
native_streaming_classes
node_id
optional
or_else
pipeline_streaming
policy.categorized
policy.select_all
policy.select_any
......@@ -353,7 +334,6 @@ caf_add_component(
result
save_inspector
scheduled_actor
selective_streaming
serial_reply
serialization
settings
......
......@@ -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 "caf/broadcast_downstream_manager.hpp"
#include "caf/detail/stream_source_driver_impl.hpp"
#include "caf/detail/stream_source_impl.hpp"
#include "caf/fwd.hpp"
#include "caf/policy/arg.hpp"
#include "caf/stream_source.hpp"
#include "caf/stream_source_driver.hpp"
#include "caf/stream_source_trait.hpp"
namespace caf {
/// Creates a new continuous stream source by instantiating the default source
/// implementation with `Driver`. The returned manager is not connected to any
/// slot and thus not stored by the actor automatically.
/// @param self Points to the hosting actor.
/// @param xs Parameter pack for constructing the driver.
/// @returns The new `stream_manager`.
template <class Driver, class... Ts>
typename Driver::source_ptr_type
attach_continuous_stream_source(scheduled_actor* self, Ts&&... xs) {
using detail::make_stream_source;
auto mgr = make_stream_source<Driver>(self, std::forward<Ts>(xs)...);
mgr->continuous(true);
return mgr;
}
/// Creates a new continuous stream source by instantiating the default source
/// implementation with `Driver`. The returned manager is not connected to any
/// slot and thus not stored by the actor automatically.
/// @param self Points to the hosting actor.
/// @param init Function object for initializing the state of the source.
/// @param pull Generator function object for producing downstream messages.
/// @param done Predicate returning `true` when generator is done.
/// @param fin Cleanup handler.
/// @returns The new `stream_manager`.
template <class Init, class Pull, class Done, class Finalize = unit_t,
class Trait = stream_source_trait_t<Pull>,
class DownstreamManager = broadcast_downstream_manager<
typename Trait::output>>
stream_source_ptr<DownstreamManager>
attach_continuous_stream_source(scheduled_actor* self, Init init, Pull pull,
Done done, Finalize fin = {},
policy::arg<DownstreamManager> = {}) {
using state_type = typename Trait::state;
static_assert(std::is_same<
void(state_type&),
typename detail::get_callable_trait<Init>::fun_sig>::value,
"Expected signature `void (State&)` for init function");
static_assert(std::is_same<
bool(const state_type&),
typename detail::get_callable_trait<Done>::fun_sig>::value,
"Expected signature `bool (const State&)` "
"for done predicate function");
using driver = detail::stream_source_driver_impl<DownstreamManager, Pull,
Done, Finalize>;
return attach_continuous_stream_source<driver>(self, std::move(init),
std::move(pull),
std::move(done),
std::move(fin));
}
} // namespace caf
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/default_downstream_manager.hpp"
#include "caf/detail/stream_stage_driver_impl.hpp"
#include "caf/detail/stream_stage_impl.hpp"
#include "caf/downstream_manager.hpp"
#include "caf/fwd.hpp"
#include "caf/make_stage_result.hpp"
#include "caf/policy/arg.hpp"
#include "caf/stream.hpp"
#include "caf/stream_stage.hpp"
#include "caf/unit.hpp"
namespace caf {
/// Returns a stream manager (implementing a continuous stage) without in- or
/// outbound path. The returned manager is not connected to any slot and thus
/// not stored by the actor automatically.
/// @param self Points to the hosting actor.
/// @param xs User-defined arguments for the downstream handshake.
/// @returns The new `stream_manager`.
template <class Driver, class... Ts>
typename Driver::stage_ptr_type
attach_continuous_stream_stage(scheduled_actor* self, Ts&&... xs) {
auto ptr = detail::make_stream_stage<Driver>(self, std::forward<Ts>(xs)...);
ptr->continuous(true);
return ptr;
}
/// @param self Points to the hosting actor.
/// @param init Function object for initializing the state of the stage.
/// @param fun Processing function.
/// @param fin Optional cleanup handler.
/// @param token Policy token for selecting a downstream manager
/// implementation.
template <class Init, class Fun, class Finalize = unit_t,
class DownstreamManager = default_downstream_manager_t<Fun>,
class Trait = stream_stage_trait_t<Fun>>
stream_stage_ptr<typename Trait::input, DownstreamManager>
attach_continuous_stream_stage(scheduled_actor* self, Init init, Fun fun,
Finalize fin = {},
policy::arg<DownstreamManager> token = {}) {
CAF_IGNORE_UNUSED(token);
using input_type = typename Trait::input;
using output_type = typename Trait::output;
using state_type = typename Trait::state;
static_assert(std::is_same<
void(state_type&),
typename detail::get_callable_trait<Init>::fun_sig>::value,
"Expected signature `void (State&)` for init function");
static_assert(std::is_same<
void(state_type&, downstream<output_type>&, input_type),
typename detail::get_callable_trait<Fun>::fun_sig>::value,
"Expected signature `void (State&, downstream<Out>&, In)` "
"for consume function");
using detail::stream_stage_driver_impl;
using driver = stream_stage_driver_impl<typename Trait::input,
DownstreamManager, Fun, Finalize>;
return attach_continuous_stream_stage<driver>(self, std::move(init),
std::move(fun), std::move(fin));
}
} // namespace caf
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/detail/stream_sink_driver_impl.hpp"
#include "caf/detail/stream_sink_impl.hpp"
#include "caf/fwd.hpp"
#include "caf/make_sink_result.hpp"
#include "caf/policy/arg.hpp"
#include "caf/stream.hpp"
#include "caf/stream_sink.hpp"
namespace caf {
/// Attaches a new stream sink to `self` by creating a default stream sink /
/// manager from given callbacks.
/// @param self Points to the hosting actor.
/// @param xs Additional constructor arguments for `Driver`.
/// @returns The new `stream_manager`, an inbound slot, and an outbound slot.
template <class Driver, class... Ts>
make_sink_result<typename Driver::input_type>
attach_stream_sink(scheduled_actor* self,
stream<typename Driver::input_type> in, Ts&&... xs) {
auto mgr = detail::make_stream_sink<Driver>(self, std::forward<Ts>(xs)...);
auto slot = mgr->add_inbound_path(in);
return {slot, std::move(mgr)};
}
/// Attaches a new stream sink to `self` by creating a default stream sink
/// manager from given callbacks.
/// @param self Points to the hosting actor.
/// @param in Stream handshake from upstream path.
/// @param init Function object for initializing the state of the sink.
/// @param fun Processing function.
/// @param fin Optional cleanup handler.
/// @returns The new `stream_manager` and the inbound slot.
template <class In, class Init, class Fun, class Finalize = unit_t,
class Trait = stream_sink_trait_t<Fun>>
make_sink_result<In> attach_stream_sink(scheduled_actor* self, stream<In> in,
Init init, Fun fun, Finalize fin = {}) {
using driver = detail::stream_sink_driver_impl<In, Fun, Finalize>;
return attach_stream_sink<driver>(self, in, std::move(init), std::move(fun),
std::move(fin));
}
} // namespace caf
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include <tuple>
#include "caf/broadcast_downstream_manager.hpp"
#include "caf/check_typed_input.hpp"
#include "caf/default_downstream_manager.hpp"
#include "caf/detail/implicit_conversions.hpp"
#include "caf/detail/stream_source_driver_impl.hpp"
#include "caf/detail/stream_source_impl.hpp"
#include "caf/detail/type_list.hpp"
#include "caf/detail/type_traits.hpp"
#include "caf/downstream_manager.hpp"
#include "caf/fwd.hpp"
#include "caf/is_actor_handle.hpp"
#include "caf/make_source_result.hpp"
#include "caf/policy/arg.hpp"
#include "caf/response_type.hpp"
#include "caf/stream_source.hpp"
namespace caf {
/// Attaches a new stream source to `self` by creating a default stream source
/// manager with `Driver`.
/// @param self Points to the hosting actor.
/// @param xs User-defined arguments for the stream handshake.
/// @param ctor_args Parameter pack for constructing the driver.
/// @returns The allocated `stream_manager` and the output slot.
template <class Driver, class... Ts, class... CtorArgs>
make_source_result_t<typename Driver::downstream_manager_type, Ts...>
attach_stream_source(scheduled_actor* self, std::tuple<Ts...> xs,
CtorArgs&&... ctor_args) {
using namespace detail;
auto mgr = make_stream_source<Driver>(self,
std::forward<CtorArgs>(ctor_args)...);
auto slot = mgr->add_outbound_path(std::move(xs));
return {slot, std::move(mgr)};
}
/// Attaches a new stream source to `self` by creating a default stream source
/// manager with the default driver.
/// @param self Points to the hosting actor.
/// @param xs User-defined arguments for the stream handshake.
/// @param init Function object for initializing the state of the source.
/// @param pull Generator function object for producing downstream messages.
/// @param done Predicate returning `true` when generator is done.
/// @param fin Cleanup handler.
/// @returns The allocated `stream_manager` and the output slot.
template <class... Ts, class Init, class Pull, class Done,
class Finalize = unit_t, class Trait = stream_source_trait_t<Pull>,
class DownstreamManager = broadcast_downstream_manager<
typename Trait::output>>
make_source_result_t<DownstreamManager, Ts...>
attach_stream_source(scheduled_actor* self, std::tuple<Ts...> xs, Init init,
Pull pull, Done done, Finalize fin = {},
policy::arg<DownstreamManager> = {}) {
using state_type = typename Trait::state;
static_assert(std::is_same<
void(state_type&),
typename detail::get_callable_trait<Init>::fun_sig>::value,
"Expected signature `void (State&)` for init function");
static_assert(std::is_same<
bool(const state_type&),
typename detail::get_callable_trait<Done>::fun_sig>::value,
"Expected signature `bool (const State&)` "
"for done predicate function");
using driver = detail::stream_source_driver_impl<DownstreamManager, Pull,
Done, Finalize>;
return attach_stream_source<driver>(self, std::move(xs), std::move(init),
std::move(pull), std::move(done),
std::move(fin));
}
/// Attaches a new stream source to `self` by creating a default stream source
/// manager with the default driver.
/// @param self Points to the hosting actor.
/// @param init Function object for initializing the state of the source.
/// @param pull Generator function object for producing downstream messages.
/// @param done Predicate returning `true` when generator is done.
/// @param fin Cleanup handler.
/// @returns The allocated `stream_manager` and the output slot.
template <class Init, class Pull, class Done, class Finalize = unit_t,
class DownstreamManager = default_downstream_manager_t<Pull>,
class Trait = stream_source_trait_t<Pull>>
detail::enable_if_t<!is_actor_handle<Init>::value && Trait::valid,
make_source_result_t<DownstreamManager>>
attach_stream_source(scheduled_actor* self, Init init, Pull pull, Done done,
Finalize fin = {},
policy::arg<DownstreamManager> token = {}) {
using output_type = typename Trait::output;
static_assert(detail::sendable<output_type>,
"the output type of the source has has no type ID, "
"did you forgot to announce it via CAF_ADD_TYPE_ID?");
static_assert(detail::sendable<stream<output_type>>,
"stream<T> for the output type has has no type ID, "
"did you forgot to announce it via CAF_ADD_TYPE_ID?");
return attach_stream_source(self, std::make_tuple(), init, pull, done, fin,
token);
}
/// Attaches a new stream source to `self` by creating a default stream source
/// manager with the default driver and starts sending to `dest` immediately.
/// @param self Points to the hosting actor.
/// @param dest Handle to the next stage in the pipeline.
/// @param xs User-defined arguments for the stream handshake.
/// @param init Function object for initializing the state of the source.
/// @param pull Generator function object for producing downstream messages.
/// @param done Predicate returning `true` when generator is done.
/// @param fin Cleanup handler.
/// @returns The allocated `stream_manager` and the output slot.
template <class ActorHandle, class... Ts, class Init, class Pull, class Done,
class Finalize = unit_t,
class DownstreamManager = default_downstream_manager_t<Pull>,
class Trait = stream_source_trait_t<Pull>>
detail::enable_if_t<is_actor_handle<ActorHandle>::value,
make_source_result_t<DownstreamManager>>
attach_stream_source(scheduled_actor* self, const ActorHandle& dest,
std::tuple<Ts...> xs, Init init, Pull pull, Done done,
Finalize fin = {}, policy::arg<DownstreamManager> = {}) {
using namespace detail;
using token = type_list<stream<typename DownstreamManager::output_type>,
strip_and_convert_t<Ts>...>;
static_assert(response_type_unbox<signatures_of_t<ActorHandle>, token>::valid,
"receiver does not accept the stream handshake");
using driver = detail::stream_source_driver_impl<DownstreamManager, Pull,
Done, Finalize>;
auto mgr = detail::make_stream_source<driver>(self, std::move(init),
std::move(pull),
std::move(done),
std::move(fin));
auto slot = mgr->add_outbound_path(dest, std::move(xs));
return {slot, std::move(mgr)};
}
/// Attaches a new stream source to `self` by creating a default stream source
/// manager with the default driver and starts sending to `dest` immediately.
/// @param self Points to the hosting actor.
/// @param dest Handle to the next stage in the pipeline.
/// @param init Function object for initializing the state of the source.
/// @param pull Generator function object for producing downstream messages.
/// @param done Predicate returning `true` when generator is done.
/// @param fin Cleanup handler.
/// @returns The allocated `stream_manager` and the output slot.
template <class ActorHandle, class Init, class Pull, class Done,
class Finalize = unit_t,
class DownstreamManager = default_downstream_manager_t<Pull>,
class Trait = stream_source_trait_t<Pull>>
detail::enable_if_t<is_actor_handle<ActorHandle>::value && Trait::valid,
make_source_result_t<DownstreamManager>>
attach_stream_source(scheduled_actor* self, const ActorHandle& dest, Init init,
Pull pull, Done done, Finalize fin = {},
policy::arg<DownstreamManager> token = {}) {
return attach_stream_source(self, dest, std::make_tuple(), std::move(init),
std::move(pull), std::move(done), std::move(fin),
token);
}
} // namespace caf
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include <vector>
#include "caf/default_downstream_manager.hpp"
#include "caf/detail/stream_stage_driver_impl.hpp"
#include "caf/detail/stream_stage_impl.hpp"
#include "caf/downstream_manager.hpp"
#include "caf/fwd.hpp"
#include "caf/make_stage_result.hpp"
#include "caf/policy/arg.hpp"
#include "caf/stream.hpp"
#include "caf/stream_stage.hpp"
#include "caf/unit.hpp"
namespace caf {
/// Attaches a new stream stage to `self` by creating a default stream stage
/// manager with `Driver`.
/// @param self Points to the hosting actor.
/// @param in Stream handshake from upstream path.
/// @param xs User-defined arguments for the downstream handshake.
/// @param ys Additional constructor arguments for `Driver`.
/// @returns The new `stream_manager`, an inbound slot, and an outbound slot.
template <class Driver, class In, class... Ts, class... Us>
make_stage_result_t<In, typename Driver::downstream_manager_type, Ts...>
attach_stream_stage(scheduled_actor* self, const stream<In>& in,
std::tuple<Ts...> xs, Us&&... ys) {
using detail::make_stream_stage;
auto mgr = make_stream_stage<Driver>(self, std::forward<Us>(ys)...);
auto islot = mgr->add_inbound_path(in);
auto oslot = mgr->add_outbound_path(std::move(xs));
return {islot, oslot, std::move(mgr)};
}
/// Attaches a new stream stage to `self` by creating a default stream stage
/// manager from given callbacks.
/// @param self Points to the hosting actor.
/// @param in Stream handshake from upstream path.
/// @param xs User-defined arguments for the downstream handshake.
/// @param init Function object for initializing the state of the stage.
/// @param fun Processing function.
/// @param fin Optional cleanup handler.
/// @param token Policy token for selecting a downstream manager
/// implementation.
/// @returns The new `stream_manager`, an inbound slot, and an outbound slot.
template <class In, class... Ts, class Init, class Fun, class Finalize = unit_t,
class DownstreamManager = default_downstream_manager_t<Fun>,
class Trait = stream_stage_trait_t<Fun>>
make_stage_result_t<In, DownstreamManager, Ts...>
attach_stream_stage(scheduled_actor* self, const stream<In>& in,
std::tuple<Ts...> xs, Init init, Fun fun, Finalize fin = {},
policy::arg<DownstreamManager> token = {}) {
CAF_IGNORE_UNUSED(token);
using output_type = typename stream_stage_trait_t<Fun>::output;
using state_type = typename stream_stage_trait_t<Fun>::state;
static_assert(
std::is_same<void(state_type&),
typename detail::get_callable_trait<Init>::fun_sig>::value,
"Expected signature `void (State&)` for init function");
using consume_one = void(state_type&, downstream<output_type>&, In);
using consume_all
= void(state_type&, downstream<output_type>&, std::vector<In>&);
using fun_sig = typename detail::get_callable_trait<Fun>::fun_sig;
static_assert(std::is_same<fun_sig, consume_one>::value
|| std::is_same<fun_sig, consume_all>::value,
"Expected signature `void (State&, downstream<Out>&, In)` "
"or `void (State&, downstream<Out>&, std::vector<In>&)` "
"for consume function");
using driver
= detail::stream_stage_driver_impl<typename Trait::input, DownstreamManager,
Fun, Finalize>;
return attach_stream_stage<driver>(self, in, std::move(xs), std::move(init),
std::move(fun), std::move(fin));
}
/// Attaches a new stream stage to `self` by creating a default stream stage
/// manager from given callbacks.
/// @param self Points to the hosting actor.
/// @param in Stream handshake from upstream path.
/// @param init Function object for initializing the state of the stage.
/// @param fun Processing function.
/// @param fin Optional cleanup handler.
/// @param token Policy token for selecting a downstream manager
/// implementation.
/// @returns The new `stream_manager`, an inbound slot, and an outbound slot.
template <class In, class Init, class Fun, class Finalize = unit_t,
class DownstreamManager = default_downstream_manager_t<Fun>,
class Trait = stream_stage_trait_t<Fun>>
make_stage_result_t<In, DownstreamManager>
attach_stream_stage(scheduled_actor* self, const stream<In>& in, Init init,
Fun fun, Finalize fin = {},
policy::arg<DownstreamManager> token = {}) {
return attach_stream_stage(self, in, std::make_tuple(), std::move(init),
std::move(fun), std::move(fin), token);
}
} // namespace caf
......@@ -4,10 +4,6 @@
#pragma once
#include <chrono>
#include <condition_variable>
#include <mutex>
#include "caf/actor_config.hpp"
#include "caf/actor_traits.hpp"
#include "caf/after.hpp"
......@@ -33,13 +29,15 @@
#include "caf/none.hpp"
#include "caf/policy/arg.hpp"
#include "caf/policy/categorized.hpp"
#include "caf/policy/downstream_messages.hpp"
#include "caf/policy/normal_messages.hpp"
#include "caf/policy/upstream_messages.hpp"
#include "caf/policy/urgent_messages.hpp"
#include "caf/send.hpp"
#include "caf/typed_actor.hpp"
#include <chrono>
#include <condition_variable>
#include <mutex>
namespace caf {
/// A thread-mapped or context-switching actor using a blocking
......@@ -388,8 +386,6 @@ public:
bool cleanup(error&& fail_state, execution_unit* host) override;
sec build_pipeline(stream_slot in, stream_slot out, stream_manager_ptr mgr);
// -- backwards compatibility ------------------------------------------------
mailbox_element_ptr next_message() {
......
This diff is collapsed.
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include <cstddef>
#include <deque>
#include <iterator>
#include <vector>
#include "caf/downstream_manager_base.hpp"
#include "caf/logger.hpp"
namespace caf {
/// Mixin for streams with any number of downstreams. `Subtype` must provide a
/// member function `buf()` returning a queue with `std::deque`-like interface.
template <class T>
class buffered_downstream_manager : public downstream_manager_base {
public:
// -- member types -----------------------------------------------------------
using super = downstream_manager_base;
using output_type = T;
using buffer_type = std::deque<output_type>;
using chunk_type = std::vector<output_type>;
// -- sanity checks ----------------------------------------------------------
static_assert(detail::is_complete<type_id<std::vector<output_type>>>);
// -- constructors, destructors, and assignment operators --------------------
explicit buffered_downstream_manager(stream_manager* parent) : super(parent) {
// nop
}
buffered_downstream_manager(stream_manager* parent, type_id_t type)
: super(parent, type) {
// nop
}
template <class T0, class... Ts>
void push(T0&& x, Ts&&... xs) {
buf_.emplace_back(std::forward<T0>(x), std::forward<Ts>(xs)...);
this->generated_messages(1);
}
/// @pre `n <= buf_.size()`
static chunk_type get_chunk(buffer_type& buf, size_t n) {
CAF_LOG_TRACE(CAF_ARG(buf) << CAF_ARG(n));
chunk_type xs;
if (!buf.empty() && n > 0) {
xs.reserve(std::min(n, buf.size()));
if (n < buf.size()) {
auto first = buf.begin();
auto last = first + static_cast<ptrdiff_t>(n);
std::move(first, last, std::back_inserter(xs));
buf.erase(first, last);
} else {
std::move(buf.begin(), buf.end(), std::back_inserter(xs));
buf.clear();
}
}
return xs;
}
chunk_type get_chunk(size_t n) {
return get_chunk(buf_, n);
}
bool terminal() const noexcept override {
return false;
}
size_t capacity() const noexcept override {
// Our goal is to cache up to 2 full batches, whereby we pick the highest
// batch size available to us (optimistic estimate).
size_t desired = 1;
for (auto& kvp : this->paths_)
desired = std::max(static_cast<size_t>(kvp.second->desired_batch_size),
desired);
desired *= 2;
auto stored = buffered();
return desired > stored ? desired - stored : 0u;
}
size_t buffered() const noexcept override {
return buf_.size();
}
buffer_type& buf() {
return buf_;
}
const buffer_type& buf() const {
return buf_;
}
protected:
buffer_type buf_;
};
} // namespace caf
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include <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
......@@ -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 "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
// 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
......@@ -159,8 +159,8 @@ class is_comparable {
// silences float-equal warnings caused by decltype(*arg0 == *arg1)
template <class A, class B>
static bool cmp_help_fun(const A*, const B*, bool*,
std::integral_constant<bool, true>);
static bool
cmp_help_fun(const A*, const B*, bool*, std::integral_constant<bool, true>);
template <class A, class B, class C>
static void cmp_help_fun(const A*, const B*, void*, C);
......@@ -203,11 +203,12 @@ template <class T>
class is_iterable {
// this horrible code would just disappear if we had concepts
template <class C>
static bool sfinae(C* cc,
// check if 'C::begin()' returns a forward iterator
enable_if_tt<is_forward_iterator<decltype(cc->begin())>>* = nullptr,
// check if begin() and end() both exist and are comparable
decltype(cc->begin() != cc->end())* = nullptr);
static bool
sfinae(C* cc,
// check if 'C::begin()' returns a forward iterator
enable_if_tt<is_forward_iterator<decltype(cc->begin())>>* = nullptr,
// check if begin() and end() both exist and are comparable
decltype(cc->begin() != cc->end())* = nullptr);
// SFNINAE default
static void sfinae(void*);
......@@ -227,13 +228,13 @@ constexpr bool is_iterable_v = is_iterable<T>::value;
/// Checks whether `T` is a non-const reference.
template <class T>
struct is_mutable_ref : std::false_type { };
struct is_mutable_ref : std::false_type {};
template <class T>
struct is_mutable_ref<const T&> : std::false_type { };
struct is_mutable_ref<const T&> : std::false_type {};
template <class T>
struct is_mutable_ref<T&> : std::true_type { };
struct is_mutable_ref<T&> : std::true_type {};
/// Defines `result_type,` `arg_types,` and `fun_type`. Functor is
/// (a) a member function pointer, (b) a function,
......@@ -247,7 +248,7 @@ struct callable_trait;
// good ol' function
template <class R, class... Ts>
struct callable_trait<R (Ts...)> {
struct callable_trait<R(Ts...)> {
/// The result type as returned by the function.
using result_type = R;
......@@ -258,10 +259,10 @@ struct callable_trait<R (Ts...)> {
using decayed_arg_types = type_list<std::decay_t<Ts>...>;
/// The signature of the function.
using fun_sig = R (Ts...);
using fun_sig = R(Ts...);
/// The signature of the function, wrapped into a `std::function`.
using fun_type = std::function<R (Ts...)>;
using fun_type = std::function<R(Ts...)>;
/// Tells whether the function takes mutable references as argument.
static constexpr bool mutates_args = (is_mutable_ref<Ts>::value || ...);
......@@ -286,11 +287,11 @@ struct callable_trait<R (C::*)(Ts...) noexcept> : callable_trait<R(Ts...)> {};
// member const function pointer
template <class C, typename R, class... Ts>
struct callable_trait<R (C::*)(Ts...) const> : callable_trait<R (Ts...)> {};
struct callable_trait<R (C::*)(Ts...) const> : callable_trait<R(Ts...)> {};
// member function pointer
template <class C, typename R, class... Ts>
struct callable_trait<R (C::*)(Ts...)> : callable_trait<R (Ts...)> {};
struct callable_trait<R (C::*)(Ts...)> : callable_trait<R(Ts...)> {};
// good ol' noexcept function pointer
template <class R, class... Ts>
......@@ -298,7 +299,7 @@ struct callable_trait<R (*)(Ts...) noexcept> : callable_trait<R(Ts...)> {};
// good ol' function pointer
template <class R, class... Ts>
struct callable_trait<R (*)(Ts...)> : callable_trait<R (Ts...)> {};
struct callable_trait<R (*)(Ts...)> : callable_trait<R(Ts...)> {};
template <class T>
struct has_apply_operator {
......@@ -314,11 +315,10 @@ struct has_apply_operator {
// matches (IsFun || IsMemberFun)
template <class T,
bool IsFun = std::is_function<T>::value
|| std::is_function<
typename std::remove_pointer<T>::type
>::value
|| std::is_member_function_pointer<T>::value,
bool IsFun
= std::is_function<T>::value
|| std::is_function<typename std::remove_pointer<T>::type>::value
|| std::is_member_function_pointer<T>::value,
bool HasApplyOp = has_apply_operator<T>::value>
struct get_callable_trait_helper {
using type = callable_trait<T>;
......@@ -371,7 +371,7 @@ template <class F, class... Ts>
struct is_callable_with {
template <class U>
static auto sfinae(U*)
-> decltype((std::declval<U&>())(std::declval<Ts>()...), std::true_type());
-> decltype((std::declval<U&>())(std::declval<Ts>()...), std::true_type());
template <class U>
static auto sfinae(...) -> std::false_type;
......@@ -386,8 +386,9 @@ struct is_callable_with {
/// mutable references.
template <class F>
struct is_manipulator {
static constexpr bool value =
tl_exists<typename get_callable_trait<F>::arg_types, is_mutable_ref>::value;
static constexpr bool value
= tl_exists<typename get_callable_trait<F>::arg_types,
is_mutable_ref>::value;
};
/// Gets the Nth element of the template parameter pack `Ts`.
......@@ -439,8 +440,8 @@ template <class T>
class has_peek_all {
private:
template <class U>
static int fun(const U*,
decltype(std::declval<U&>().peek_all(unit))* = nullptr);
static int
fun(const U*, decltype(std::declval<U&>().peek_all(unit))* = nullptr);
static char fun(const void*);
......@@ -457,9 +458,9 @@ CAF_HAS_MEMBER_TRAIT(size);
/// or `std::function<void (const T&)>`.
template <class F, class T>
struct is_handler_for {
static constexpr bool value =
std::is_convertible<F, std::function<void (T&)>>::value
|| std::is_convertible<F, std::function<void (const T&)>>::value;
static constexpr bool value
= std::is_convertible<F, std::function<void(T&)>>::value
|| std::is_convertible<F, std::function<void(const T&)>>::value;
};
template <class T>
......@@ -494,10 +495,9 @@ struct strip_reference_wrapper<std::reference_wrapper<T>> {
template <class T>
constexpr bool can_insert_elements_impl(
T*,
decltype(std::declval<T&>()
.insert(std::declval<T&>().end(),
std::declval<typename T::value_type>()))* = nullptr) {
T*, decltype(std::declval<T&>().insert(
std::declval<T&>().end(),
std::declval<typename T::value_type>()))* = nullptr) {
return true;
}
......@@ -513,10 +513,10 @@ constexpr bool can_insert_elements() {
/// Checks whether `Tpl` is a specialization of `T` or not.
template <template <class...> class Tpl, class T>
struct is_specialization : std::false_type { };
struct is_specialization : std::false_type {};
template <template <class...> class T, class... Ts>
struct is_specialization<T, T<Ts...>> : std::true_type { };
struct is_specialization<T, T<Ts...>> : std::true_type {};
/// Transfers const from `T` to `U`. `U` remains unchanged if `T` is not const.
template <class T, class U>
......@@ -532,12 +532,6 @@ struct transfer_const<const T, U> {
template <class T, class U>
using transfer_const_t = typename transfer_const<T, U>::type;
template <class T>
struct is_stream : std::false_type {};
template <class T>
struct is_stream<stream<T>> : std::true_type {};
template <class T>
struct is_result : std::false_type {};
......@@ -561,23 +555,19 @@ template <class T, class U,
struct is_equal_int_type {
static constexpr bool value = sizeof(T) == sizeof(U)
&& std::is_signed<T>::value
== std::is_signed<U>::value;
== std::is_signed<U>::value;
};
template <class T, typename U>
struct is_equal_int_type<T, U, false> : std::false_type { };
struct is_equal_int_type<T, U, false> : std::false_type {};
/// Compares `T` to `U` und evaluates to `true_type` if either
/// `T == U or if T and U are both integral types of the
/// same size and signedness. This works around the issue that
/// `uint8_t != unsigned char on some compilers.
template <class T, typename U>
struct is_same_ish
: std::conditional<
std::is_same<T, U>::value,
std::true_type,
is_equal_int_type<T, U>
>::type { };
struct is_same_ish : std::conditional<std::is_same<T, U>::value, std::true_type,
is_equal_int_type<T, U>>::type {};
/// Utility for fallbacks calling `static_assert`.
template <class>
......@@ -628,9 +618,9 @@ struct all_constructible<type_list<>, type_list<>> : std::true_type {};
template <class T, class... Ts, class U, class... Us>
struct all_constructible<type_list<T, Ts...>, type_list<U, Us...>> {
static constexpr bool value = std::is_constructible<T, U>::value
&& all_constructible<type_list<Ts...>,
type_list<Us...>>::value;
static constexpr bool value
= std::is_constructible<T, U>::value
&& all_constructible<type_list<Ts...>, type_list<Us...>>::value;
};
/// Checks whether T behaves like `std::map`.
......
// 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 <deque>
#include <vector>
#include "caf/message.hpp"
namespace caf {
/// Grants access to an output stream buffer.
template <class T>
class downstream {
public:
// -- member types -----------------------------------------------------------
/// A queue of items for temporary storage before moving them into chunks.
using queue_type = std::deque<T>;
// -- constructors, destructors, and assignment operators --------------------
downstream(queue_type& q) : buf_(q) {
// nop
}
// -- queue access -----------------------------------------------------------
template <class... Ts>
void push(Ts&&... xs) {
buf_.emplace_back(std::forward<Ts>(xs)...);
}
template <class Iterator, class Sentinel>
void append(Iterator first, Sentinel last) {
buf_.insert(buf_.end(), first, last);
}
// @private
queue_type& buf() {
return buf_;
}
protected:
queue_type& buf_;
};
} // namespace caf
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include <memory>
#include <vector>
#include "caf/actor_clock.hpp"
#include "caf/detail/core_export.hpp"
#include "caf/fwd.hpp"
#include "caf/stream_slot.hpp"
#include "caf/timespan.hpp"
namespace caf {
/// Manages downstream communication for a `stream_manager`. The downstream
/// manager owns the `outbound_path` objects, has a buffer for storing pending
/// output and is responsible for the dispatching policy (broadcasting, for
/// example). The default implementation terminates the stream and never
/// accepts any paths.
class CAF_CORE_EXPORT downstream_manager {
public:
// -- member types -----------------------------------------------------------
/// Outbound path.
using path_type = outbound_path;
/// Pointer to an outbound path.
using path_ptr = path_type*;
/// Pointer to an immutable outbound path.
using const_path_ptr = const path_type*;
/// Unique pointer to an outbound path.
using unique_path_ptr = std::unique_ptr<path_type>;
/// Discrete point in time, as reported by the actor clock.
using time_point = typename actor_clock::time_point;
/// Function object for iterating over all paths.
struct CAF_CORE_EXPORT path_visitor {
virtual ~path_visitor();
virtual void operator()(outbound_path& x) = 0;
};
/// Predicate object for paths.
struct CAF_CORE_EXPORT path_predicate {
virtual ~path_predicate();
virtual bool operator()(const outbound_path& x) const noexcept = 0;
};
/// Selects a check algorithms.
enum path_algorithm { all_of, any_of, none_of };
// -- constructors, destructors, and assignment operators --------------------
explicit downstream_manager(stream_manager* parent);
downstream_manager(const downstream_manager&) = delete;
downstream_manager& operator=(const downstream_manager&) = delete;
virtual ~downstream_manager();
// -- properties -------------------------------------------------------------
scheduled_actor* self() const noexcept;
stream_manager* parent() const noexcept;
/// Returns `true` if this manager belongs to a sink, i.e., terminates the
/// stream and never has outbound paths.
virtual bool terminal() const noexcept;
// -- time management --------------------------------------------------------
/// Forces underful batches after reaching the maximum delay.
void tick(time_point now, timespan max_batch_delay);
// -- path management --------------------------------------------------------
/// Applies `f` to each path.
template <class F>
void for_each_path(F f) {
struct impl : path_visitor {
F fun;
impl(F x) : fun(std::move(x)) {
// nop
}
void operator()(outbound_path& x) override {
fun(x);
}
};
impl g{std::move(f)};
for_each_path_impl(g);
}
/// Applies `f` to each path.
template <class F>
void for_each_path(F f) const {
struct impl : path_visitor {
F fun;
impl(F x) : fun(std::move(x)) {
// nop
}
void operator()(outbound_path& x) override {
fun(const_cast<const outbound_path&>(x));
}
};
impl g{std::move(f)};
// This const_cast is safe, because we restore the const in our overload for
// operator() above.
const_cast<downstream_manager*>(this)->for_each_path_impl(g);
}
/// Returns all used slots.
std::vector<stream_slot> path_slots();
/// Returns all open slots, i.e., slots assigned to outbound paths with
/// `closing == false`.
std::vector<stream_slot> open_path_slots();
/// Checks whether `predicate` holds true for all paths.
template <class Predicate>
bool all_paths(Predicate predicate) const noexcept {
return check_paths(path_algorithm::all_of, std::move(predicate));
}
/// Checks whether `predicate` holds true for any path.
template <class Predicate>
bool any_path(Predicate predicate) const noexcept {
return check_paths(path_algorithm::any_of, std::move(predicate));
}
/// Checks whether `predicate` holds true for no path.
template <class Predicate>
bool no_path(Predicate predicate) const noexcept {
return check_paths(path_algorithm::none_of, std::move(predicate));
}
/// Returns the current number of paths.
virtual size_t num_paths() const noexcept;
/// Adds a pending path to `target` to the manager.
/// @returns The added path on success, `nullptr` otherwise.
path_ptr add_path(stream_slot slot, strong_actor_ptr target);
/// Removes a path from the manager.
virtual bool
remove_path(stream_slot slot, error reason, bool silent) noexcept;
/// Returns the path associated to `slot` or `nullptr`.
virtual path_ptr path(stream_slot slot) noexcept;
/// Returns the path associated to `slot` or `nullptr`.
const_path_ptr path(stream_slot slot) const noexcept;
/// Returns `true` if there is no data pending and all batches are
/// acknowledged batch on all paths.
bool clean() const noexcept;
/// Returns `true` if `slot` is unknown or if there is no data pending and
/// all batches are acknowledged on `slot`. The default implementation
/// returns `false` for all paths, even if `clean()` return `true`.
bool clean(stream_slot slot) const noexcept;
/// Removes all paths gracefully.
virtual void close();
/// Removes path `slot` gracefully by sending pending batches before removing
/// it. Effectively calls `path(slot)->closing = true`.
virtual void close(stream_slot slot);
/// Removes all paths with an error message.
virtual void abort(error reason);
/// Returns `num_paths() == 0`.
bool empty() const noexcept {
return num_paths() == 0;
}
/// Returns the minimum amount of credit on all output paths.
size_t min_credit() const;
/// Returns the maximum amount of credit on all output paths.
size_t max_credit() const;
/// Returns the total amount of credit on all output paths, i.e., the sum of
/// all individual credits.
size_t total_credit() const;
/// Sends batches to sinks.
virtual void emit_batches();
/// Sends batches to sinks regardless of whether or not the batches reach the
/// desired batch size.
virtual void force_emit_batches();
/// Queries the currently available capacity for the output buffer.
virtual size_t capacity() const noexcept;
/// Queries the size of the output buffer.
virtual size_t buffered() const noexcept;
/// Queries an estimate of the size of the output buffer for `slot`.
virtual size_t buffered(stream_slot slot) const noexcept;
/// Queries whether the manager cannot make any progress, because its buffer
/// is full and no more credit is available.
bool stalled() const noexcept;
/// Silently removes all paths.
virtual void clear_paths();
protected:
// -- customization points ---------------------------------------------------
/// Inserts `ptr` to the implementation-specific container.
virtual bool insert_path(unique_path_ptr ptr);
/// Applies `f` to each path.
virtual void for_each_path_impl(path_visitor& f);
/// Dispatches the predicate to `std::all_of`, `std::any_of`, or
/// `std::none_of`.
virtual bool check_paths_impl(path_algorithm algo, path_predicate& pred) const
noexcept;
/// Emits a regular (`reason == nullptr`) or irregular (`reason != nullptr`)
/// shutdown if `silent == false`.
/// @warning moves `*reason` if `reason == nullptr`
virtual void about_to_erase(path_ptr ptr, bool silent, error* reason);
// -- helper functions -------------------------------------------------------
/// Delegates to `check_paths_impl`.
template <class Predicate>
bool check_paths(path_algorithm algorithm, Predicate predicate) const
noexcept {
struct impl : path_predicate {
Predicate fun;
impl(Predicate x) : fun(std::move(x)) {
// nop
}
bool operator()(const outbound_path& x) const noexcept override {
return fun(x);
}
};
impl g{std::move(predicate)};
return check_paths_impl(algorithm, g);
}
// -- member variables -------------------------------------------------------
stream_manager* parent_;
/// Stores the time stamp of our last batch.
time_point last_send_;
};
} // namespace caf
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include <cstddef>
#include <memory>
#include "caf/detail/core_export.hpp"
#include "caf/detail/unordered_flat_map.hpp"
#include "caf/downstream_manager.hpp"
#include "caf/outbound_path.hpp"
#include "caf/telemetry/counter.hpp"
#include "caf/telemetry/gauge.hpp"
namespace caf {
/// The default downstream manager base stores outbound paths in an unordered
/// map. It always takes ownership of the paths by using unique pointers.
class CAF_CORE_EXPORT downstream_manager_base : public downstream_manager {
public:
// -- member types -----------------------------------------------------------
/// Base type.
using super = downstream_manager;
/// Maps slots to paths.
using map_type = detail::unordered_flat_map<stream_slot, unique_path_ptr>;
/// Optional metrics for outbound stream traffic.
struct metrics_t {
/// Counts the total number of elements that have been pushed downstream.
telemetry::int_counter* pushed_elements = nullptr;
/// Tracks how many stream elements are currently waiting in the output
/// buffer due to insufficient credit.
telemetry::int_gauge* output_buffer_size = nullptr;
};
// -- constructors, destructors, and assignment operators --------------------
explicit downstream_manager_base(stream_manager* parent);
downstream_manager_base(stream_manager* parent, type_id_t type);
~downstream_manager_base() override;
// -- properties -------------------------------------------------------------
const map_type& paths() const {
return paths_;
}
map_type& paths() {
return paths_;
}
// -- path management --------------------------------------------------------
size_t num_paths() const noexcept override;
bool remove_path(stream_slot slots, error reason,
bool silent) noexcept override;
path_ptr path(stream_slot slots) noexcept override;
void clear_paths() override;
// -- callbacks for actor metrics --------------------------------------------
void generated_messages(size_t num) {
if (num > 0 && metrics_.output_buffer_size)
metrics_.output_buffer_size->inc(static_cast<int64_t>(num));
}
void dropped_messages(size_t num) {
if (num > 0 && metrics_.output_buffer_size)
metrics_.output_buffer_size->dec(static_cast<int64_t>(num));
}
void shipped_messages(size_t num) {
if (num > 0 && metrics_.output_buffer_size) {
metrics_.output_buffer_size->dec(static_cast<int64_t>(num));
metrics_.pushed_elements->inc(static_cast<int64_t>(num));
}
}
protected:
bool insert_path(unique_path_ptr ptr) override;
void for_each_path_impl(path_visitor& f) override;
bool check_paths_impl(path_algorithm algo,
path_predicate& pred) const noexcept override;
// -- member variables -------------------------------------------------------
map_type paths_;
metrics_t metrics_;
};
} // namespace caf
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include <cstdint>
#include <utility>
#include <vector>
#include "caf/actor_addr.hpp"
#include "caf/actor_control_block.hpp"
#include "caf/detail/core_export.hpp"
#include "caf/detail/type_list.hpp"
#include "caf/message.hpp"
#include "caf/stream_priority.hpp"
#include "caf/stream_slot.hpp"
#include "caf/tag/boxing_type.hpp"
#include "caf/variant.hpp"
namespace caf {
/// Transmits stream data.
struct downstream_msg_batch {
/// Allows the testing DSL to unbox this type automagically.
using outer_type = downstream_msg;
/// Size of the type-erased vector<T> (used credit).
int32_t xs_size;
/// A type-erased vector<T> containing the elements of the batch.
message xs;
/// ID of this batch (ascending numbering).
int64_t id;
};
/// Orderly shuts down a stream after receiving an ACK for the last batch.
struct downstream_msg_close {
/// Allows the testing DSL to unbox this type automagically.
using outer_type = downstream_msg;
};
/// Propagates a fatal error from sources to sinks.
struct downstream_msg_forced_close {
/// Allows the testing DSL to unbox this type automagically.
using outer_type = downstream_msg;
/// Reason for shutting down the stream.
error reason;
};
/// Stream messages that travel downstream, i.e., batches and close messages.
struct CAF_CORE_EXPORT downstream_msg : tag::boxing_type {
// -- nested types -----------------------------------------------------------
using batch = downstream_msg_batch;
using close = downstream_msg_close;
using forced_close = downstream_msg_forced_close;
// -- member types -----------------------------------------------------------
/// Lists all possible options for the payload.
using alternatives = detail::type_list<batch, close, forced_close>;
/// Stores one of `alternatives`.
using content_type = variant<batch, close, forced_close>;
// -- constructors, destructors, and assignment operators --------------------
template <class T>
downstream_msg(stream_slots s, actor_addr addr, T&& x)
: slots(s), sender(std::move(addr)), content(std::forward<T>(x)) {
// nop
}
downstream_msg() = default;
downstream_msg(downstream_msg&&) = default;
downstream_msg(const downstream_msg&) = default;
downstream_msg& operator=(downstream_msg&&) = default;
downstream_msg& operator=(const downstream_msg&) = default;
// -- member variables -------------------------------------------------------
/// ID of the affected stream.
stream_slots slots;
/// Address of the sender. Identifies the up- or downstream actor sending
/// this message. Note that abort messages can get send after `sender`
/// already terminated. Hence, `current_sender()` can be `nullptr`, because
/// no strong pointers can be formed any more and receiver would receive an
/// anonymous message.
actor_addr sender;
/// Palyoad of the message.
content_type content;
};
/// Allows the testing DSL to unbox `downstream_msg` automagically.
template <class T>
const T& get(const downstream_msg& x) {
return get<T>(x.content);
}
/// Allows the testing DSL to check whether `downstream_msg` holds a `T`.
template <class T>
bool is(const downstream_msg& x) {
return holds_alternative<T>(x.content);
}
/// @relates downstream_msg
template <class T, class... Ts>
detail::enable_if_tt<detail::tl_contains<downstream_msg::alternatives, T>,
downstream_msg>
make(stream_slots slots, actor_addr addr, Ts&&... xs) {
return {slots, std::move(addr), T{std::forward<Ts>(xs)...}};
}
/// @relates downstream_msg::batch
template <class Inspector>
bool inspect(Inspector& f, downstream_msg::batch& x) {
return f.object(x).pretty_name("batch").fields(f.field("size", x.xs_size),
f.field("xs", x.xs),
f.field("id", x.id));
}
/// @relates downstream_msg::close
template <class Inspector>
bool inspect(Inspector& f, downstream_msg::close& x) {
return f.object(x).pretty_name("close").fields();
}
/// @relates downstream_msg::forced_close
template <class Inspector>
bool inspect(Inspector& f, downstream_msg::forced_close& x) {
return f.object(x)
.pretty_name("forced_close")
.fields(f.field("reason", x.reason));
}
/// @relates downstream_msg
template <class Inspector>
bool inspect(Inspector& f, downstream_msg& x) {
return f.object(x)
.pretty_name("downstream_msg")
.fields(f.field("slots", x.slots), f.field("sender", x.sender),
f.field("content", x.content));
}
} // namespace caf
// 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 <cstddef>
#include "caf/downstream_manager.hpp"
#include "caf/logger.hpp"
#include "caf/outbound_path.hpp"
#include "caf/sec.hpp"
#include "caf/stream.hpp"
#include "caf/stream_slot.hpp"
#include "caf/detail/type_list.hpp"
#include "caf/detail/unordered_flat_map.hpp"
namespace caf {
namespace detail {
/// Utility function for repeating `x` for a given template parameter pack.
template <class T, class U>
U pack_repeat(U x) {
return x;
}
template <class Iter>
class ptr_array_initializer {
public:
ptr_array_initializer(Iter first) : i_(first) {
// nop
}
void operator()() {
// end of recursion
}
template <class T, class... Ts>
void operator()(T& x, Ts&... xs) {
*i_ = &x;
++i_;
(*this)(xs...);
}
private:
Iter i_;
};
struct downstream_manager_selector {
downstream_manager* operator()(const message&) {
return nullptr;
}
template <class T, class... Ts>
downstream_manager* operator()(const message& msg, T& x, Ts&... xs) {
if (msg.size() > 1
&& msg.type_at(0) == type_id_v<stream<typename T::value_type>>)
return &x;
return (*this)(msg, xs...);
}
};
template <size_t I, size_t E>
struct init_ptr_array {
template <class... Ts>
static void apply(downstream_manager* (&xs)[E], std::tuple<Ts...>& ys) {
xs[I] = &std::get<I>(ys);
init_ptr_array<I + 1, E>::apply(xs, ys);
}
};
template <size_t I>
struct init_ptr_array<I, I> {
template <class... Ts>
static void apply(downstream_manager* (&)[I], std::tuple<Ts...>&) {
// nop
}
};
} // namespace detail
/// A downstream manager that delegates to any number of sub-managers.
template <class T, class... Ts>
class fused_downstream_manager : public downstream_manager {
public:
// -- member and nested types ------------------------------------------------
/// Base type.
using super = downstream_manager;
/// A tuple holding all nested managers.
using nested_managers = std::tuple<T, Ts...>;
/// Pointer to an outbound path.
using typename super::path_ptr;
/// Unique pointer to an outbound path.
using typename super::unique_path_ptr;
// Lists all template parameters `[T, Ts...]`;
using param_list = detail::type_list<T, Ts...>;
/// State held for each slot.
struct non_owning_ptr {
path_ptr ptr;
downstream_manager* owner;
};
/// Maps slots to path and nested managers.
using map_type = detail::unordered_flat_map<stream_slot, non_owning_ptr>;
/// Maps slots to paths that haven't a managers assigned yet.
using unassigned_map_type = detail::unordered_flat_map<stream_slot,
unique_path_ptr>;
// -- constructors, destructors, and assignment operators --------------------
fused_downstream_manager(stream_manager* parent)
: super(parent),
nested_(parent, detail::pack_repeat<Ts>(parent)...) {
detail::init_ptr_array<0, sizeof...(Ts) + 1>::apply(ptrs_, nested_);
}
// -- properties -------------------------------------------------------------
template <class U>
U& get() {
return std::get<U>(nested_);
}
template <class U>
const U& get() const {
return std::get<U>(nested_);
}
/// Requires a previous call to `add_path` for given slot.
template <class U>
void assign(stream_slot slot) {
// Fetch pointer from the unassigned paths.
auto i = unassigned_paths_.find(slot);
if (i == unassigned_paths_.end()) {
CAF_LOG_ERROR("cannot assign nested manager to unknown slot");
return;
}
// Error or not, remove entry from unassigned_paths_ before leaving.
auto cleanup = detail::make_scope_guard([&] {
unassigned_paths_.erase(i);
});
// Transfer ownership to nested manager.
auto ptr = i->second.get();
CAF_ASSERT(ptr != nullptr);
auto owner = &get<U>();
if (!owner->insert_path(std::move(i->second))) {
CAF_LOG_ERROR("slot exists as unassigned and assigned");
return;
}
// Store owner and path in our map.
auto er = paths_.emplace(slot, non_owning_ptr{ptr, owner});
if (!er.second) {
CAF_LOG_ERROR("slot already mapped");
owner->remove_path(slot, sec::invalid_stream_state, false);
return;
}
}
// -- overridden functions ---------------------------------------------------
bool terminal() const noexcept override {
return false;
}
size_t num_paths() const noexcept override {
return paths_.size();
}
bool remove_path(stream_slot slot, error reason,
bool silent) noexcept override {
CAF_LOG_TRACE(CAF_ARG(slot) << CAF_ARG(reason) << CAF_ARG(silent));
auto i = paths_.find(slot);
if (i == paths_.end())
return false;
auto owner = i->second.owner;
paths_.erase(i);
return owner->remove_path(slot, std::move(reason), silent);
}
path_ptr path(stream_slot slot) noexcept override {
auto i = paths_.find(slot);
if (i == paths_.end())
return nullptr;
return i->second.ptr;
}
using downstream_manager::close;
void close() override {
CAF_LOG_TRACE(CAF_ARG(paths_));
for (auto ptr : ptrs_)
ptr->close();
paths_.clear();
}
void abort(error reason) override {
CAF_LOG_TRACE(CAF_ARG(paths_));
for (auto ptr : ptrs_)
ptr->abort(reason);
paths_.clear();
}
void emit_batches() override {
CAF_LOG_TRACE("");
for (auto ptr : ptrs_)
ptr->emit_batches();
}
void force_emit_batches() override {
CAF_LOG_TRACE("");
for (auto ptr : ptrs_)
ptr->force_emit_batches();
}
size_t capacity() const noexcept override {
// Get the minimum of all available capacities.
size_t result = std::numeric_limits<size_t>::max();
for (auto ptr : ptrs_)
result = std::min(result, ptr->capacity());
return result;
}
size_t buffered() const noexcept override {
// Get the maximum of all available buffer sizes.
size_t result = 0;
for (auto ptr : ptrs_)
result = std::max(result, ptr->buffered());
return result;
}
size_t buffered(stream_slot slot) const noexcept override {
// We don't know which nested manager stores this path. Only one will give a
// valid answer, though. Everyone else always responds with 0. Hence, we can
// simply call all managers and sum up the results.
size_t result = 0;
for (auto ptr : ptrs_)
result += ptr->buffered(slot);
return result;
}
void clear_paths() override {
CAF_LOG_TRACE("");
for (auto ptr : ptrs_)
ptr->clear_paths();
paths_.clear();
}
protected:
bool insert_path(unique_path_ptr ptr) override {
CAF_LOG_TRACE(CAF_ARG(ptr));
CAF_ASSERT(ptr != nullptr);
auto slot = ptr->slots.sender;
return unassigned_paths_.emplace(slot, std::move(ptr)).second;
}
void for_each_path_impl(path_visitor& f) override {
for (auto& kvp : paths_)
f(*kvp.second.ptr);
}
bool check_paths_impl(path_algorithm algo,
path_predicate& pred) const noexcept override {
auto f = [&](const typename map_type::value_type& x) {
return pred(*x.second.ptr);
};
switch (algo) {
default: // all_of
return std::all_of(paths_.begin(), paths_.end(), f);
case path_algorithm::any_of:
return std::any_of(paths_.begin(), paths_.end(), f);
case path_algorithm::none_of:
return std::none_of(paths_.begin(), paths_.end(), f);
}
}
private:
nested_managers nested_;
downstream_manager* ptrs_[sizeof...(Ts) + 1];
map_type paths_;
unassigned_map_type unassigned_paths_;
};
} // namespace caf
......@@ -21,17 +21,12 @@ template <class> class basic_cow_string;
template <class> class behavior_type_of;
template <class> class callback;
template <class> class dictionary;
template <class> class downstream;
template <class> class expected;
template <class> class inbound_stream_slot;
template <class> class intrusive_cow_ptr;
template <class> class intrusive_ptr;
template <class> class optional;
template <class> class param;
template <class> class span;
template <class> class stream;
template <class> class stream_sink;
template <class> class stream_source;
template <class> class weak_intrusive_ptr;
template <class> struct inspector_access;
......@@ -43,16 +38,12 @@ template <uint16_t> struct type_name_by_id;
// -- 2 param templates --------------------------------------------------------
template <class, class> class stream_stage;
template <class Iterator, class Sentinel = Iterator> struct parser_state;
// -- 3 param templates --------------------------------------------------------
template <class, class, int> class actor_cast_access;
template <class, class, class> class broadcast_downstream_manager;
// -- variadic templates -------------------------------------------------------
template <class...> class const_typed_message_view;
......@@ -67,8 +58,6 @@ template <class...> class typed_message_view;
template <class...> class typed_response_promise;
template <class...> class variant;
template <class, class...> class outbound_stream_slot;
// clang-format on
// -- classes ------------------------------------------------------------------
......@@ -99,15 +88,12 @@ class config_option_set;
class config_value;
class deserializer;
class disposable;
class downstream_manager;
class downstream_manager_base;
class event_based_actor;
class execution_unit;
class forwarding_actor_proxy;
class group;
class group_module;
class hashed_node_id;
class inbound_path;
class ipv4_address;
class ipv4_endpoint;
class ipv4_subnet;
......@@ -122,7 +108,6 @@ class message_handler;
class message_id;
class node_id;
class node_id_data;
class outbound_path;
class proxy_registry;
class ref_counted;
class response_promise;
......@@ -131,7 +116,6 @@ class scheduled_actor;
class scoped_actor;
class serializer;
class skip_t;
class stream_manager;
class string_view;
class tracing_data;
class tracing_data_factory;
......@@ -148,10 +132,6 @@ class stateful_actor;
// -- structs ------------------------------------------------------------------
struct down_msg;
struct downstream_msg;
struct downstream_msg_batch;
struct downstream_msg_close;
struct downstream_msg_forced_close;
struct exit_msg;
struct group_down_msg;
struct illegal_message_element;
......@@ -159,15 +139,8 @@ struct invalid_actor_addr_t;
struct invalid_actor_t;
struct node_down_msg;
struct none_t;
struct open_stream_msg;
struct prohibit_top_level_spawn_marker;
struct stream_slots;
struct unit_t;
struct upstream_msg;
struct upstream_msg_ack_batch;
struct upstream_msg_ack_open;
struct upstream_msg_drop;
struct upstream_msg_forced_drop;
// -- free template functions --------------------------------------------------
......@@ -186,7 +159,6 @@ enum class exit_reason : uint8_t;
enum class invoke_message_result;
enum class pec : uint8_t;
enum class sec : uint8_t;
enum class stream_priority : uint8_t;
// -- aliases ------------------------------------------------------------------
......@@ -202,7 +174,6 @@ using ip_endpoint = ipv6_endpoint;
using ip_subnet = ipv6_subnet;
using settings = dictionary<config_value>;
using skippable_result = variant<delegated<message>, message, error, skip_t>;
using stream_slot = uint16_t;
using type_id_t = uint16_t;
// -- functions ----------------------------------------------------------------
......@@ -347,9 +318,6 @@ class manager;
namespace detail {
template <class>
class stream_distribution_tree;
class abstract_worker;
class abstract_worker_hub;
class disposer;
......@@ -377,7 +345,6 @@ using weak_actor_ptr = weak_intrusive_ptr<actor_control_block>;
// -- intrusive pointer aliases ------------------------------------------------
using group_module_ptr = intrusive_ptr<group_module>;
using stream_manager_ptr = intrusive_ptr<stream_manager>;
using strong_actor_ptr = intrusive_ptr<actor_control_block>;
// -- unique pointer aliases ---------------------------------------------------
......
// 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 <cstdint>
#include "caf/actor_clock.hpp"
#include "caf/actor_control_block.hpp"
#include "caf/credit_controller.hpp"
#include "caf/detail/core_export.hpp"
#include "caf/detail/size_based_credit_controller.hpp"
#include "caf/detail/token_based_credit_controller.hpp"
#include "caf/downstream_msg.hpp"
#include "caf/logger.hpp"
#include "caf/settings.hpp"
#include "caf/stream_aborter.hpp"
#include "caf/stream_priority.hpp"
#include "caf/stream_slot.hpp"
#include "caf/telemetry/counter.hpp"
#include "caf/telemetry/gauge.hpp"
#include "caf/timestamp.hpp"
#include "caf/upstream_msg.hpp"
namespace caf {
/// State for a path to an upstream actor (source).
class CAF_CORE_EXPORT inbound_path {
public:
// -- member types -----------------------------------------------------------
/// Message type for propagating graceful shutdowns.
using regular_shutdown = upstream_msg::drop;
/// Message type for propagating errors.
using irregular_shutdown = upstream_msg::forced_drop;
/// Wraps optional actor metrics collected by this path.
struct metrics_t {
telemetry::int_counter* processed_elements;
telemetry::int_gauge* input_buffer_size;
};
/// Discrete point in time, as reported by the actor clock.
using time_point = typename actor_clock::time_point;
/// Time interval, as reported by the actor clock.
using duration_type = typename actor_clock::duration_type;
// -- constructors, destructors, and assignment operators --------------------
template <class T>
inbound_path(stream_manager* mgr, stream<T> in)
: inbound_path(mgr, type_id_v<T>) {
auto& cfg = config();
auto set_default = [this, in] {
controller_ = detail::size_based_credit_controller::make(self(), in);
};
if (auto str = get_if<std::string>(&cfg, "caf.stream.credit-policy")) {
if (*str == "token-based")
controller_ = detail::token_based_credit_controller::make(self(), in);
else if (*str == "size-based")
set_default();
else {
set_default();
CAF_LOG_WARNING("unrecognized credit policy:"
<< *str << "(falling back to 'size-based')");
}
} else {
set_default();
}
}
void init(strong_actor_ptr source_hdl, stream_slots id);
~inbound_path();
// -- member variables -------------------------------------------------------
/// Points to the manager responsible for incoming traffic.
stream_manager* mgr;
/// Handle to the source.
strong_actor_ptr hdl;
/// Stores slot IDs for sender (hdl) and receiver (self).
stream_slots slots;
/// Stores pointers to optional telemetry objects.
metrics_t metrics;
/// Stores the last computed desired batch size. Adjusted at run-time by the
/// controller.
int32_t desired_batch_size = 0;
/// Amount of credit we have signaled upstream.
int32_t assigned_credit = 0;
/// Maximum amount of credit that the path may signal upstream. Adjusted at
/// run-time by the controller.
int32_t max_credit = 0;
/// Decremented whenever receiving a batch. Triggers a re-calibration by the
/// controller when reaching zero.
int32_t calibration_countdown = 10;
/// Priority of incoming batches from this source.
stream_priority prio = stream_priority::normal;
/// ID of the last acknowledged batch ID.
int64_t last_acked_batch_id = 0;
/// ID of the last received batch.
int64_t last_batch_id = 0;
/// Controller for assigning credit to the source.
std::unique_ptr<credit_controller> controller_;
/// Stores when the last ACK was emitted.
time_point last_ack_time;
// -- properties -------------------------------------------------------------
/// Returns whether the path received no input since last emitting
/// `ack_batch`, i.e., `last_acked_batch_id == last_batch_id`.
bool up_to_date() const noexcept;
/// Returns a pointer to the parent actor.
scheduled_actor* self() const noexcept;
/// Returns currently unassigned credit that we could assign to the source.
int32_t available_credit() const noexcept;
/// Returns the system-wide configuration.
const settings& config() const noexcept;
// -- callbacks --------------------------------------------------------------
/// Updates `last_batch_id` and `assigned_credit` before dispatching to the
/// manager.
void handle(downstream_msg::batch& x);
/// Forward the `close` message to the manager.
void handle(downstream_msg::close& x);
/// Forward the `forced_close` message to the manager.
void handle(downstream_msg::forced_close& x);
/// Forces an ACK message after receiving no input for a considerable amount
/// of time.
void tick(time_point now, duration_type max_batch_delay);
// -- messaging --------------------------------------------------------------
/// Emits an `upstream_msg::ack_batch`.
void emit_ack_open(local_actor* self, actor_addr rebind_from);
/// Sends an `upstream_msg::ack_batch` for granting new credit.
/// @param self Points to the parent actor, i.e., sender of the message.
/// @param new_credit Amount of new credit to assign to the source.
void emit_ack_batch(local_actor* self, int32_t new_credit);
/// Sends an `upstream_msg::drop` on this path.
void emit_regular_shutdown(local_actor* self);
/// Sends an `upstream_msg::forced_drop` on this path.
void emit_irregular_shutdown(local_actor* self, error reason);
/// Sends an `upstream_msg::forced_drop`.
static void
emit_irregular_shutdown(local_actor* self, stream_slots slots,
const strong_actor_ptr& hdl, error reason);
private:
inbound_path(stream_manager* mgr_ptr, type_id_t input_type);
};
/// @relates inbound_path
template <class Inspector>
bool inspect(Inspector& f, inbound_path& x) {
return f.object(x).fields(
f.field("hdl", x.hdl), f.field("slots", x.slots), f.field("prio", x.prio),
f.field("last_acked_batch_id", x.last_acked_batch_id),
f.field("last_batch_id", x.last_batch_id),
f.field("assigned_credit", x.assigned_credit));
}
} // 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/type_traits.hpp"
namespace caf {
struct invalid_stream_t {};
constexpr invalid_stream_t invalid_stream = invalid_stream_t{};
} // namespace caf
......@@ -533,14 +533,16 @@ CAF_CORE_EXPORT bool operator==(const logger::field& x, const logger::field& y);
<< "; GROUPS =" << ::caf::logger::joined_groups_of(ref))
# define CAF_LOG_SEND_EVENT(ptr) \
CAF_LOG_IMPL( \
CAF_LOG_FLOW_COMPONENT, CAF_LOG_LEVEL_DEBUG, \
"SEND ; TO =" \
<< ::caf::deep_to_string(::caf::strong_actor_ptr{this->ctrl()}) \
.c_str() \
<< "; FROM =" << ::caf::deep_to_string(ptr->sender).c_str() \
<< "; STAGES =" << ::caf::deep_to_string(ptr->stages).c_str() \
<< "; CONTENT =" << ::caf::deep_to_string(ptr->content()).c_str())
CAF_LOG_IMPL(CAF_LOG_FLOW_COMPONENT, CAF_LOG_LEVEL_DEBUG, \
"SEND ; TO =" \
<< ::caf::deep_to_string( \
::caf::strong_actor_ptr{this->ctrl()}) \
.c_str() \
<< "; FROM =" << ::caf::deep_to_string(ptr->sender).c_str() \
<< "; STAGES =" \
<< ::caf::deep_to_string(ptr->stages).c_str() \
<< "; CONTENT =" \
<< ::caf::deep_to_string(ptr->content()).c_str())
# define CAF_LOG_RECEIVE_EVENT(ptr) \
CAF_LOG_IMPL(CAF_LOG_FLOW_COMPONENT, CAF_LOG_LEVEL_DEBUG, \
......@@ -604,20 +606,3 @@ CAF_CORE_EXPORT bool operator==(const logger::field& x, const logger::field& y);
# define CAF_LOG_TERMINATE_EVENT(thisptr, rsn) CAF_VOID_STMT
#endif // CAF_LOG_LEVEL >= CAF_LOG_LEVEL_DEBUG
// -- macros for logging streaming-related events ------------------------------
/// The log component for logging streaming-related events that are crucial for
/// understanding handshaking, credit decisions, etc.
#define CAF_LOG_STREAM_COMPONENT "caf_stream"
#if CAF_LOG_LEVEL >= CAF_LOG_LEVEL_DEBUG
# define CAF_STREAM_LOG_DEBUG(output) \
CAF_LOG_IMPL(CAF_LOG_STREAM_COMPONENT, CAF_LOG_LEVEL_DEBUG, output)
# define CAF_STREAM_LOG_DEBUG_IF(condition, output) \
if (condition) \
CAF_LOG_IMPL(CAF_LOG_STREAM_COMPONENT, CAF_LOG_LEVEL_DEBUG, output)
#else
# define CAF_STREAM_LOG_DEBUG(unused) CAF_VOID_STMT
# define CAF_STREAM_LOG_DEBUG_IF(unused1, unused2) CAF_VOID_STMT
#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/delegated.hpp"
#include "caf/fwd.hpp"
#include "caf/stream_sink.hpp"
#include "caf/stream_slot.hpp"
namespace caf {
/// Returns a stream sink with the slot ID of its first inbound path.
template <class In>
class make_sink_result : public delegated<void> {
public:
// -- member types -----------------------------------------------------------
/// Type of a single element.
using input_type = In;
/// Fully typed stream manager as returned by `make_source`.
using sink_type = stream_sink<In>;
/// Pointer to a fully typed stream manager.
using sink_ptr_type = intrusive_ptr<sink_type>;
// -- constructors, destructors, and assignment operators --------------------
make_sink_result() noexcept : slot_(0) {
// nop
}
make_sink_result(stream_slot slot, sink_ptr_type ptr) noexcept
: slot_(slot), ptr_(std::move(ptr)) {
// nop
}
make_sink_result(make_sink_result&&) = default;
make_sink_result(const make_sink_result&) = default;
make_sink_result& operator=(make_sink_result&&) = default;
make_sink_result& operator=(const make_sink_result&) = default;
// -- properties -------------------------------------------------------------
stream_slot inbound_slot() const noexcept {
return slot_;
}
sink_ptr_type& ptr() noexcept {
return ptr_;
}
const sink_ptr_type& ptr() const noexcept {
return ptr_;
}
private:
// -- member variables -------------------------------------------------------
stream_slot slot_;
sink_ptr_type ptr_;
};
} // 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/delegated.hpp"
#include "caf/detail/implicit_conversions.hpp"
#include "caf/fwd.hpp"
#include "caf/stream.hpp"
#include "caf/stream_slot.hpp"
#include "caf/stream_source.hpp"
namespace caf {
/// Returns a stream source with the slot ID of its first outbound path.
template <class DownstreamManager, class... Ts>
class make_source_result
: public delegated<stream<typename DownstreamManager::output_type>, Ts...> {
public:
// -- member types -----------------------------------------------------------
/// Type of a single element.
using output_type = typename DownstreamManager::output_type;
/// Fully typed stream manager as returned by `make_source`.
using source_type = stream_source<DownstreamManager>;
/// Pointer to a fully typed stream manager.
using source_ptr_type = intrusive_ptr<source_type>;
/// The return type for `scheduled_actor::make_stage`.
using stream_type = stream<output_type>;
/// Type of user-defined handshake arguments.
using handshake_arguments = std::tuple<Ts...>;
// -- constructors, destructors, and assignment operators --------------------
make_source_result() noexcept : slot_(0) {
// nop
}
make_source_result(stream_slot slot, source_ptr_type ptr) noexcept
: slot_(slot), ptr_(std::move(ptr)) {
// nop
}
make_source_result(make_source_result&&) = default;
make_source_result(const make_source_result&) = default;
make_source_result& operator=(make_source_result&&) = default;
make_source_result& operator=(const make_source_result&) = default;
// -- properties -------------------------------------------------------------
stream_slot outbound_slot() const noexcept {
return slot_;
}
source_ptr_type& ptr() noexcept {
return ptr_;
}
const source_ptr_type& ptr() const noexcept {
return ptr_;
}
private:
stream_slot slot_;
source_ptr_type ptr_;
};
/// @relates make_source_result
template <class DownstreamManager, class... Ts>
using make_source_result_t
= make_source_result<DownstreamManager, detail::strip_and_convert_t<Ts>...>;
} // 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/delegated.hpp"
#include "caf/detail/implicit_conversions.hpp"
#include "caf/fwd.hpp"
#include "caf/stream.hpp"
#include "caf/stream_slot.hpp"
#include "caf/stream_stage.hpp"
namespace caf {
/// Returns a stream stage with the slot IDs of its first in- and outbound
/// paths.
template <class In, class DownstreamManager, class... Ts>
class make_stage_result
: public delegated<stream<typename DownstreamManager::output_type>, Ts...> {
public:
// -- member types -----------------------------------------------------------
/// Type of a single element.
using input_type = In;
/// Type of a single element.
using output_type = typename DownstreamManager::output_type;
/// Fully typed stream manager as returned by `make_source`.
using stage_type = stream_stage<In, DownstreamManager>;
/// Pointer to a fully typed stream manager.
using stage_ptr_type = intrusive_ptr<stage_type>;
/// The return type for `scheduled_actor::make_stage`.
using stream_type = stream<output_type>;
/// Type of user-defined handshake arguments.
using handshake_arguments = std::tuple<Ts...>;
// -- constructors, destructors, and assignment operators --------------------
make_stage_result() noexcept : inbound_slot_(0), outbound_slot_(0) {
// nop
}
make_stage_result(stream_slot in, stream_slot out, stage_ptr_type ptr)
: inbound_slot_(in), outbound_slot_(out), ptr_(std::move(ptr)) {
// nop
}
make_stage_result(make_stage_result&&) = default;
make_stage_result(const make_stage_result&) = default;
make_stage_result& operator=(make_stage_result&&) = default;
make_stage_result& operator=(const make_stage_result&) = default;
// -- properties -------------------------------------------------------------
stream_slot inbound_slot() const noexcept {
return inbound_slot_;
}
stream_slot outbound_slot() const noexcept {
return outbound_slot_;
}
stage_ptr_type& ptr() noexcept {
return ptr_;
}
const stage_ptr_type& ptr() const noexcept {
return ptr_;
}
private:
stream_slot inbound_slot_;
stream_slot outbound_slot_;
stage_ptr_type ptr_;
};
/// Helper type for defining a `make_stage_result` from a downstream manager
/// plus additional handshake types. Hardwires `message` as result type.
template <class In, class DownstreamManager, class... Ts>
using make_stage_result_t
= make_stage_result<In, DownstreamManager,
detail::strip_and_convert_t<Ts>...>;
} // namespace caf
......@@ -40,14 +40,6 @@ public:
/// Identifies one-to-one messages with normal priority.
static constexpr uint64_t normal_message_category = 1;
/// Identifies stream messages that flow upstream, e.g.,
/// `upstream_msg::ack_batch`.
static constexpr uint64_t upstream_message_category = 2;
/// Identifies stream messages that flow downstream, e.g.,
/// `downstream_msg::batch`.
static constexpr uint64_t downstream_message_category = 3;
/// Number of bits trailing the category.
static constexpr uint64_t category_offset = 60;
......@@ -71,8 +63,7 @@ public:
// -- properties ------------------------------------------------------------
/// Returns the message category, i.e., one of `normal_message_category`,
/// `upstream_message_category`, `downstream_message_category`, or
/// Returns the message category, i.e., `normal_message_category` or
/// `urgent_message_category`.
constexpr uint64_t category() const noexcept {
return (value_ & category_flag_mask) >> category_offset;
......@@ -83,8 +74,7 @@ public:
return message_id{(value_ & ~category_flag_mask) | (x << category_offset)};
}
/// Returns whether a message is asynchronous, i.e., neither a request, nor a
/// response, nor a stream message.
/// Returns whether a message is asynchronous, i.e., not a request.
constexpr bool is_async() const noexcept {
return value_ == 0 || value_ == default_async_value;
}
......@@ -114,21 +104,6 @@ public:
return category() == normal_message_category;
}
/// Returns whether `category()` is an up- or downstream message.
constexpr bool is_stream_message() const noexcept {
return category() > 1;
}
/// Returns whether `category() == upstream_message_category`.
constexpr bool is_upstream_message() const noexcept {
return category() == upstream_message_category;
}
/// Returns whether `category() == downstream_message_category`.
constexpr bool is_downstream_message() const noexcept {
return category() == downstream_message_category;
}
/// Returns a response ID for the current request or an asynchronous ID with
/// the same priority as this ID.
constexpr message_id response_id() const noexcept {
......@@ -143,13 +118,11 @@ public:
}
/// Returns the same ID but high message priority.
/// @pre `!is_stream_message()`
constexpr message_id with_high_priority() const noexcept {
return message_id{value_ & ~category_flag_mask};
}
/// Returns the same ID with normal message priority.
/// @pre `!is_stream_message()`
constexpr message_id with_normal_priority() const noexcept {
return message_id{value_ | default_async_value};
}
......@@ -187,15 +160,15 @@ private:
/// Generates a `message_id` with given integer value.
/// @relates message_id
constexpr message_id
make_message_id(normal_message_priority_constant, uint64_t value) {
constexpr message_id make_message_id(normal_message_priority_constant,
uint64_t value) {
return message_id{value | message_id::default_async_value};
}
/// Generates a `message_id` with given integer value.
/// @relates message_id
constexpr message_id
make_message_id(high_message_priority_constant, uint64_t value) {
constexpr message_id make_message_id(high_message_priority_constant,
uint64_t value) {
return message_id{value};
}
......
// 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 <cstdint>
#include <deque>
#include <vector>
#include "caf/actor_control_block.hpp"
#include "caf/detail/core_export.hpp"
#include "caf/detail/type_traits.hpp"
#include "caf/downstream_msg.hpp"
#include "caf/fwd.hpp"
#include "caf/logger.hpp"
#include "caf/stream_aborter.hpp"
#include "caf/stream_slot.hpp"
#include "caf/system_messages.hpp"
namespace caf {
/// State for a single path to a sink of a `downstream_manager`.
class CAF_CORE_EXPORT outbound_path {
public:
// -- member types -----------------------------------------------------------
/// Propagates graceful shutdowns.
using regular_shutdown = downstream_msg::close;
/// Propagates errors.
using irregular_shutdown = downstream_msg::forced_close;
/// Stores batches until receiving corresponding ACKs.
using cache_type = std::deque<std::pair<int64_t, downstream_msg::batch>>;
// -- constants --------------------------------------------------------------
/// Stream aborter flag to monitor a path.
static constexpr const auto aborter_type = stream_aborter::sink_aborter;
// -- constructors, destructors, and assignment operators --------------------
/// Constructs a pending path for given slot and handle.
outbound_path(stream_slot sender_slot, strong_actor_ptr receiver_hdl);
~outbound_path();
// -- downstream communication -----------------------------------------------
/// Sends a `downstream_msg::batch` on this path. Decrements `open_credit` by
/// `xs_size` and increments `next_batch_id` by 1.
void emit_batch(local_actor* self, int32_t xs_size, message xs);
template <class Iterator>
Iterator emit_batches_impl(local_actor* self, Iterator i, Iterator e,
bool force_underfull) {
CAF_LOG_TRACE(CAF_ARG(force_underfull));
CAF_ASSERT(desired_batch_size > 0);
using type = detail::decay_t<decltype(*i)>;
// Ship full batches.
while (std::distance(i, e) >= desired_batch_size) {
std::vector<type> tmp(std::make_move_iterator(i),
std::make_move_iterator(i + desired_batch_size));
emit_batch(self, desired_batch_size, make_message(std::move(tmp)));
i += desired_batch_size;
}
// Ship underful batch only if `force_underful` is set.
if (i != e && force_underfull) {
std::vector<type> tmp(std::make_move_iterator(i),
std::make_move_iterator(e));
auto tmp_size = static_cast<int32_t>(tmp.size());
emit_batch(self, tmp_size, make_message(std::move(tmp)));
return e;
}
return i;
}
/// Calls `emit_batch` for each chunk in the cache, whereas each chunk is of
/// size `desired_batch_size`. Does nothing for pending paths.
template <class T>
void
emit_batches(local_actor* self, std::vector<T>& cache, bool force_underfull) {
CAF_LOG_TRACE(CAF_ARG(slots) << CAF_ARG(open_credit) << CAF_ARG(cache)
<< CAF_ARG(force_underfull));
if (pending())
return;
CAF_ASSERT(open_credit >= 0);
CAF_ASSERT(desired_batch_size > 0);
CAF_ASSERT(cache.size()
<= static_cast<size_t>(std::numeric_limits<int32_t>::max()));
auto first = cache.begin();
auto last
= first + std::min(open_credit, static_cast<int32_t>(cache.size()));
if (first == last)
return;
auto i = emit_batches_impl(self, first, last, force_underfull);
if (i == cache.end()) {
cache.clear();
} else if (i != first) {
cache.erase(first, i);
}
}
/// Sends a `downstream_msg::close` on this path.
void emit_regular_shutdown(local_actor* self);
/// Sends a `downstream_msg::forced_close` on this path.
void emit_irregular_shutdown(local_actor* self, error reason);
/// Sends a `downstream_msg::forced_close`.
static void
emit_irregular_shutdown(local_actor* self, stream_slots slots,
const strong_actor_ptr& hdl, error reason);
// -- properties -------------------------------------------------------------
/// Returns whether this path is pending, i.e., didn't receive an `ack_open`
/// yet.
bool pending() const noexcept {
return slots.receiver == invalid_stream_slot;
}
/// Returns whether no pending ACKs exist.
bool clean() const noexcept {
return next_batch_id == next_ack_id;
}
void set_desired_batch_size(int32_t value) noexcept;
// -- member variables -------------------------------------------------------
/// Slot IDs for sender (self) and receiver (hdl).
stream_slots slots;
/// Handle to the sink.
strong_actor_ptr hdl;
/// Next expected batch ID.
int64_t next_batch_id;
/// Currently available credit on this path.
int32_t open_credit;
/// Ideal batch size. Configured by the sink.
int32_t desired_batch_size;
/// ID of the first unacknowledged batch. Note that CAF uses accumulative
/// ACKs, i.e., receiving an ACK with a higher ID is not an error.
int64_t next_ack_id;
/// Stores whether an outbound path is marked for removal. The
/// `downstream_manger` no longer sends new batches to a closing path, but
/// buffered batches are still shipped. The owning `stream_manager` removes
/// the path when receiving an `upstream_msg::ack_batch` and no pending
/// batches for this path exist.
bool closing;
};
/// @relates outbound_path
template <class Inspector>
bool inspect(Inspector& f, outbound_path& x) {
return f.object(x).fields(f.field("slots", x.slots), f.field("hdl", x.hdl),
f.field("next_batch_id", x.next_batch_id),
f.field("open_credit", x.open_credit),
f.field("desired_batch_size", x.desired_batch_size),
f.field("next_ack_id", x.next_ack_id));
}
} // namespace caf
......@@ -8,9 +8,7 @@
#include "caf/fwd.hpp"
#include "caf/mailbox_element.hpp"
#include "caf/message_priority.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/unit.hpp"
......@@ -52,14 +50,6 @@ public:
return x + x;
}
template <template <class> class Queue>
static deficit_type
quantum(const Queue<upstream_messages>& q, deficit_type) noexcept {
// Allow actors to consume *all* upstream messages. They are lightweight by
// design and require little processing.
return q.total_task_size();
}
template <class Queue>
static deficit_type quantum(const Queue&, deficit_type x) noexcept {
return 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 <map>
#include "caf/detail/core_export.hpp"
#include "caf/fwd.hpp"
#include "caf/intrusive/drr_queue.hpp"
#include "caf/mailbox_element.hpp"
#include "caf/stream_slot.hpp"
#include "caf/unit.hpp"
namespace caf::policy {
/// Configures a dynamic WDRR queue for holding downstream messages.
class CAF_CORE_EXPORT downstream_messages {
public:
// -- nested types -----------------------------------------------------------
/// Configures a nested DRR queue.
class CAF_CORE_EXPORT nested {
public:
// -- member types ---------------------------------------------------------
using mapped_type = mailbox_element;
using task_size_type = size_t;
using deficit_type = size_t;
using unique_pointer = mailbox_element_ptr;
using handler_type = std::unique_ptr<inbound_path>;
static task_size_type task_size(const downstream_msg_batch& x) noexcept;
static constexpr task_size_type
task_size(const downstream_msg_close&) noexcept {
return 1;
}
static constexpr task_size_type
task_size(const downstream_msg_forced_close&) noexcept {
return 1;
}
static task_size_type task_size(const mailbox_element& x) noexcept;
// -- constructors, destructors, and assignment operators ------------------
template <class T>
nested(T&& x) : handler(std::forward<T>(x)) {
// nop
}
nested() = default;
nested(nested&&) = default;
nested& operator=(nested&&) = default;
nested(const nested&) = delete;
nested& operator=(const nested&) = delete;
// -- member variables -----------------------------------------------------
handler_type handler;
size_t bulk_inserted_size = 0;
};
// -- member types -----------------------------------------------------------
using mapped_type = mailbox_element;
using task_size_type = size_t;
using deficit_type = size_t;
using pointer = mapped_type*;
using unique_pointer = mailbox_element_ptr;
using key_type = stream_slot;
using nested_queue_type = intrusive::drr_queue<nested>;
using queue_map_type = std::map<key_type, nested_queue_type>;
// -- required functions for wdrr_dynamic_multiplexed_queue ------------------
static key_type id_of(mailbox_element& x) noexcept;
static bool enabled(const nested_queue_type& q) noexcept;
static deficit_type quantum(const nested_queue_type& q,
deficit_type x) noexcept;
// -- constructors, destructors, and assignment operators --------------------
downstream_messages() = default;
downstream_messages(const downstream_messages&) = default;
downstream_messages& operator=(const downstream_messages&) = default;
constexpr downstream_messages(unit_t) {
// nop
}
// -- required functions for drr_queue ---------------------------------------
static task_size_type task_size(const mailbox_element&) noexcept {
return 1;
}
// -- required functions for wdrr_dynamic_multiplexed_queue ------------------
static void cleanup(nested_queue_type&) noexcept;
static bool push_back(nested_queue_type& sub_queue, pointer ptr) noexcept;
static void lifo_append(nested_queue_type& sub_queue, pointer ptr) noexcept;
static void stop_lifo_append(nested_queue_type& sub_queue) noexcept;
};
} // namespace caf::policy
// 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/fwd.hpp"
#include "caf/mailbox_element.hpp"
#include "caf/unit.hpp"
namespace caf::policy {
/// Configures a DRR queue for holding upstream messages.
class CAF_CORE_EXPORT upstream_messages {
public:
// -- member types -----------------------------------------------------------
using mapped_type = mailbox_element;
using task_size_type = size_t;
using deficit_type = size_t;
using unique_pointer = mailbox_element_ptr;
// -- constructors, destructors, and assignment operators --------------------
upstream_messages() = default;
upstream_messages(const upstream_messages&) = default;
upstream_messages& operator=(const upstream_messages&) = default;
constexpr upstream_messages(unit_t) {
// nop
}
// -- interface required by drr_queue ----------------------------------------
static task_size_type task_size(const mailbox_element&) noexcept {
return 1;
}
};
} // namespace caf::policy
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/fwd.hpp"
#include "caf/type_id.hpp"
namespace caf {
/// Empty marker type for streaming handshakes.
template <class T>
class stream {
public:
using value_type = T;
};
/// @relates stream
template <class Inspector, class T>
auto inspect(Inspector& f, stream<T>& x) {
return f.object(x).fields();
}
} // namespace caf
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/actor_addr.hpp"
#include "caf/attachable.hpp"
#include "caf/detail/core_export.hpp"
#include "caf/fwd.hpp"
#include "caf/stream_slot.hpp"
namespace caf {
class CAF_CORE_EXPORT stream_aborter : public attachable {
public:
enum mode { source_aborter, sink_aborter };
struct token {
const actor_addr& observer;
stream_slot slot;
mode m;
static constexpr size_t token_type = attachable::token::stream_aborter;
};
stream_aborter(actor_addr&& observed, actor_addr&& observer, stream_slot slot,
mode m);
~stream_aborter() override;
void actor_exited(const error& rsn, execution_unit* host) override;
bool matches(const attachable::token& what) override;
/// Adds a stream aborter to `observed`.
static void
add(strong_actor_ptr observed, actor_addr observer, stream_slot slot, mode m);
/// Removes a stream aborter from `observed`.
static void del(strong_actor_ptr observed, const actor_addr& observer,
stream_slot slot, mode m);
private:
actor_addr observed_;
actor_addr observer_;
stream_slot slot_;
mode mode_;
};
CAF_CORE_EXPORT attachable_ptr make_stream_aborter(actor_addr observed,
actor_addr observer,
stream_slot slot,
stream_aborter::mode m);
} // 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/type_traits.hpp"
#include "caf/message.hpp"
#include "caf/stream_sink.hpp"
namespace caf {
/// Dispatches a finalize call to a function taking either one or two arguments.
template <class Fun, class State,
bool AcceptsTwoArgs
= detail::is_callable_with<Fun, State&, const error&>::value>
struct stream_finalize_trait;
/// Specializes the trait for callbacks that only take the state.
template <class Fun, class State>
struct stream_finalize_trait<Fun, State, false> {
static void invoke(Fun& f, State& st, const error&) {
static_assert(detail::is_callable_with<Fun, State&>::value,
"Finalize function neither accepts (State&, const error&) "
"nor (State&)");
f(st);
}
};
/// Specializes the trait for callbacks that take state and error.
template <class Fun, class State>
struct stream_finalize_trait<Fun, State, true> {
static void invoke(Fun& f, State& st, const error& err) {
f(st, err);
}
};
} // namespace caf
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 <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
};
/// @relates stream_priority
CAF_CORE_EXPORT std::string to_string(stream_priority x);
/// @relates stream_priority
CAF_CORE_EXPORT bool from_string(string_view, stream_priority&);
/// @relates stream_priority
CAF_CORE_EXPORT bool from_integer(std::underlying_type_t<stream_priority>,
stream_priority&);
/// @relates stream_priority
template <class Inspector>
bool inspect(Inspector& f, stream_priority& x) {
return default_enum_inspect(f, x);
}
} // 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 <algorithm>
#include <tuple>
#include <typeinfo>
#include "caf/detail/type_traits.hpp"
#include "caf/downstream_manager.hpp"
#include "caf/inbound_path.hpp"
#include "caf/intrusive_ptr.hpp"
#include "caf/stream_manager.hpp"
namespace caf {
template <class In>
class stream_sink : public virtual stream_manager {
public:
// -- member types -----------------------------------------------------------
using input_type = In;
// -- constructors, destructors, and assignment operators --------------------
stream_sink(scheduled_actor* self) : stream_manager(self), dummy_out_(this) {
// nop
}
// -- overridden member functions --------------------------------------------
bool done() const override {
return !this->continuous() && this->inbound_paths_.empty();
}
bool idle() const noexcept override {
// A sink is idle if there's no pending batch and a new credit round would
// emit no `ack_batch` messages.
return this->inbound_paths_idle();
}
downstream_manager& out() override {
return dummy_out_;
}
// -- properties -------------------------------------------------------------
/// Creates a new input path to the current sender.
inbound_stream_slot<input_type> add_inbound_path(stream<input_type> in) {
return {this->add_unchecked_inbound_path(in)};
}
private:
// -- member variables -------------------------------------------------------
downstream_manager dummy_out_;
};
template <class In>
using stream_sink_ptr = intrusive_ptr<stream_sink<In>>;
} // namespace caf
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
......@@ -650,9 +650,6 @@ public:
cfg.set("caf.middleman.workers", size_t{0});
cfg.set("caf.middleman.heartbeat-interval", caf::timespan{0});
}
cfg.set("caf.stream.credit-policy", "token-based");
cfg.set("caf.stream.token-based-policy.batch-size", 50);
cfg.set("caf.stream.token-based-policy.buffer-size", 200);
return cfg;
}
......
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