Commit 7c60431b authored by Dominik Charousset's avatar Dominik Charousset

Merge branch 'master' into topic/dll

parents 5730b77a 27a0009d
...@@ -363,10 +363,10 @@ else() ...@@ -363,10 +363,10 @@ else()
endif() endif()
# enable address sanitizer if requested by the user # enable address sanitizer if requested by the user
if(CAF_ENABLE_ADDRESS_SANITIZER AND NOT WIN32) if(CAF_SANITIZERS)
add_compile_options("-fsanitize=address" add_compile_options("-fsanitize=${CAF_SANITIZERS}"
"-fno-omit-frame-pointer") "-fno-omit-frame-pointer")
list(APPEND CAF_EXTRA_LDFLAGS "-fsanitize=address") list(APPEND CAF_EXTRA_LDFLAGS "-fsanitize=${CAF_SANITIZERS}")
endif() endif()
# -pthread is ignored on MacOSX but required on other platforms # -pthread is ignored on MacOSX but required on other platforms
if(NOT APPLE AND NOT WIN32) if(NOT APPLE AND NOT WIN32)
......
...@@ -6,12 +6,11 @@ ...@@ -6,12 +6,11 @@
defaultReleaseBuildFlags = [ defaultReleaseBuildFlags = [
'CAF_ENABLE_RUNTIME_CHECKS:BOOL=yes', 'CAF_ENABLE_RUNTIME_CHECKS:BOOL=yes',
'CAF_NO_OPENCL:BOOL=yes', 'CAF_NO_OPENCL:BOOL=yes',
'CAF_INSTALL_UNIT_TESTS:BOOL=yes',
] ]
// Default CMake flags for debug builds. // Default CMake flags for debug builds.
defaultDebugBuildFlags = defaultReleaseBuildFlags + [ defaultDebugBuildFlags = defaultReleaseBuildFlags + [
'CAF_ENABLE_ADDRESS_SANITIZER:BOOL=yes', 'CAF_SANITIZERS:STRING=address,undefined',
'CAF_LOG_LEVEL:STRING=TRACE', 'CAF_LOG_LEVEL:STRING=TRACE',
'CAF_ENABLE_ACTOR_PROFILER:BOOL=yes', 'CAF_ENABLE_ACTOR_PROFILER:BOOL=yes',
] ]
......
...@@ -75,16 +75,14 @@ Usage: $0 [OPTION]... [VAR=VALUE]... ...@@ -75,16 +75,14 @@ Usage: $0 [OPTION]... [VAR=VALUE]...
- INFO - INFO
- DEBUG - DEBUG
- TRACE - TRACE
--with-address-sanitizer build with address sanitizer if available --with-sanitizers=LIST build with this list of sanitizers enabled
--with-asan alias for --with-address-sanitier
--enable-asan alias for --with-address-sanitier
--with-gcov build with gcov coverage enabled
--with-actor-profiler enables the (experimental) actor_profiler API --with-actor-profiler enables the (experimental) actor_profiler API
Convenience options: Convenience options:
--dev-mode sets --build-type=debug, --no-examples, --dev-mode sets --build-type=debug, --no-examples,
--no-tools, --with-runtime-checks, --no-tools, --with-runtime-checks,
--log-level=trace, and --enable-asan --log-level=trace, and
--sanitizers=address,undefined
Influential Environment Variables (only on first invocation): Influential Environment Variables (only on first invocation):
CXX C++ compiler command CXX C++ compiler command
...@@ -237,17 +235,8 @@ while [ $# -ne 0 ]; do ...@@ -237,17 +235,8 @@ while [ $# -ne 0 ]; do
--with-runtime-checks) --with-runtime-checks)
append_cache_entry CAF_ENABLE_RUNTIME_CHECKS BOOL yes append_cache_entry CAF_ENABLE_RUNTIME_CHECKS BOOL yes
;; ;;
--with-address-sanitizer) --with-sanitizers=*)
append_cache_entry CAF_ENABLE_ADDRESS_SANITIZER BOOL yes append_cache_entry CAF_SANITIZERS STRING "$optarg"
;;
--with-asan)
append_cache_entry CAF_ENABLE_ADDRESS_SANITIZER BOOL yes
;;
--enable-asan)
append_cache_entry CAF_ENABLE_ADDRESS_SANITIZER BOOL yes
;;
--with-gcov)
append_cache_entry CAF_ENABLE_GCOV BOOL yes
;; ;;
--with-actor-profiler) --with-actor-profiler)
append_cache_entry CAF_ENABLE_ACTOR_PROFILER BOOL yes append_cache_entry CAF_ENABLE_ACTOR_PROFILER BOOL yes
...@@ -370,12 +359,12 @@ while [ $# -ne 0 ]; do ...@@ -370,12 +359,12 @@ while [ $# -ne 0 ]; do
done done
;; ;;
--dev-mode) --dev-mode)
append_cache_entry CMAKE_BUILD_TYPE STRING Debug for var in CAF_NO_TOOLS CAF_NO_EXAMPLES CAF_ENABLE_RUNTIME_CHECKS ; do
append_cache_entry CAF_NO_EXAMPLES BOOL yes append_cache_entry $var BOOL yes
append_cache_entry CAF_NO_TOOLS BOOL yes done
append_cache_entry CMAKE_BUILD_TYPE STRING debug
append_cache_entry CAF_LOG_LEVEL STRING TRACE append_cache_entry CAF_LOG_LEVEL STRING TRACE
append_cache_entry CAF_ENABLE_RUNTIME_CHECKS BOOL yes append_cache_entry CAF_SANITIZERS STRING address,undefined
append_cache_entry CAF_ENABLE_ADDRESS_SANITIZER BOOL yes
;; ;;
*) *)
echo "Invalid option '$1'. Try $0 --help to see available options." echo "Invalid option '$1'. Try $0 --help to see available options."
......
...@@ -47,10 +47,12 @@ set(CAF_CORE_SOURCES ...@@ -47,10 +47,12 @@ set(CAF_CORE_SOURCES
src/binary_deserializer.cpp src/binary_deserializer.cpp
src/binary_serializer.cpp src/binary_serializer.cpp
src/blocking_actor.cpp src/blocking_actor.cpp
src/complexity_based_credit_controller.cpp
src/config_option.cpp src/config_option.cpp
src/config_option_adder.cpp src/config_option_adder.cpp
src/config_option_set.cpp src/config_option_set.cpp
src/config_value.cpp src/config_value.cpp
src/credit_controller.cpp
src/decorator/sequencer.cpp src/decorator/sequencer.cpp
src/default_attachable.cpp src/default_attachable.cpp
src/defaults.cpp src/defaults.cpp
...@@ -75,6 +77,7 @@ set(CAF_CORE_SOURCES ...@@ -75,6 +77,7 @@ set(CAF_CORE_SOURCES
src/detail/pretty_type_name.cpp src/detail/pretty_type_name.cpp
src/detail/private_thread.cpp src/detail/private_thread.cpp
src/detail/ripemd_160.cpp src/detail/ripemd_160.cpp
src/detail/serialized_size.cpp
src/detail/set_thread_name.cpp src/detail/set_thread_name.cpp
src/detail/shared_spinlock.cpp src/detail/shared_spinlock.cpp
src/detail/simple_actor_clock.cpp src/detail/simple_actor_clock.cpp
...@@ -138,14 +141,18 @@ set(CAF_CORE_SOURCES ...@@ -138,14 +141,18 @@ set(CAF_CORE_SOURCES
src/sec.cpp src/sec.cpp
src/serializer.cpp src/serializer.cpp
src/settings.cpp src/settings.cpp
src/size_based_credit_controller.cpp
src/skip.cpp src/skip.cpp
src/stream_aborter.cpp src/stream_aborter.cpp
src/stream_manager.cpp src/stream_manager.cpp
src/string_algorithms.cpp src/string_algorithms.cpp
src/string_view.cpp src/string_view.cpp
src/term.cpp src/term.cpp
src/test_credit_controller.cpp
src/thread_hook.cpp src/thread_hook.cpp
src/timestamp.cpp src/timestamp.cpp
src/tracing_data.cpp
src/tracing_data_factory.cpp
src/type_erased_tuple.cpp src/type_erased_tuple.cpp
src/type_erased_value.cpp src/type_erased_value.cpp
src/uniform_type_info_map.cpp src/uniform_type_info_map.cpp
...@@ -156,115 +163,120 @@ set(CAF_CORE_SOURCES ...@@ -156,115 +163,120 @@ set(CAF_CORE_SOURCES
# -- list cpp files for caf-core-test ------------------------------------------ # -- list cpp files for caf-core-test ------------------------------------------
set(CAF_CORE_TEST_SOURCES set(CAF_CORE_TEST_SOURCES
test/actor_clock.cpp test/constructor_attach.cpp
test/actor_factory.cpp test/make_config_value_field.cpp
test/actor_lifetime.cpp test/sum_type.cpp
test/actor_pool.cpp test/binary_deserializer.cpp
test/config_option_set.cpp
test/actor_profiler.cpp test/actor_profiler.cpp
test/actor_registry.cpp test/sum_type_token.cpp
test/actor_system_config.cpp
test/actor_termination.cpp
test/aout.cpp
test/atom.cpp
test/behavior.cpp test/behavior.cpp
test/blocking_actor.cpp test/custom_exception_handler.cpp
test/broadcast_downstream_manager.cpp test/config_value.cpp
test/byte.cpp test/ipv4_address.cpp
test/mock_streaming_classes.cpp
test/ipv4_endpoint.cpp
test/composable_behavior.cpp test/composable_behavior.cpp
test/composition.cpp test/decorator/sequencer.cpp
test/local_migration.cpp
test/message.cpp
test/typed_spawn.cpp
test/native_streaming_classes.cpp
test/actor_system_config.cpp
test/error.cpp
test/config_option.cpp test/config_option.cpp
test/config_option_set.cpp test/string_algorithms.cpp
test/config_value.cpp
test/config_value_adaptor.cpp
test/constructor_attach.cpp
test/continuous_streaming.cpp test/continuous_streaming.cpp
test/serial_reply.cpp
test/atom.cpp
test/aout.cpp
test/byte.cpp
test/function_view.cpp
test/fused_downstream_manager.cpp
test/type_erased_tuple.cpp
test/optional.cpp
test/mixin/sender.cpp
test/mixin/requester.cpp
test/binary_serializer.cpp
test/composition.cpp
test/string_view.cpp
test/local_group.cpp
test/match.cpp
test/runtime_settings_map.cpp
test/uri.cpp
test/mailbox_element.cpp
test/actor_lifetime.cpp
test/cow_tuple.cpp test/cow_tuple.cpp
test/custom_exception_handler.cpp test/selective_streaming.cpp
test/decorator/sequencer.cpp test/blocking_actor.cpp
test/deep_to_string.cpp test/typed_response_promise.cpp
test/detached_actors.cpp test/actor_termination.cpp
test/actor_clock.cpp
test/dynamic_spawn.cpp
test/ipv6_subnet.cpp
test/or_else.cpp
test/message_lifetime.cpp
test/detail/bounds_checker.cpp test/detail/bounds_checker.cpp
test/detail/parse.cpp
test/detail/ini_consumer.cpp test/detail/ini_consumer.cpp
test/detail/tick_emitter.cpp
test/detail/limited_vector.cpp test/detail/limited_vector.cpp
test/detail/parse.cpp test/detail/unique_function.cpp
test/detail/serialized_size.cpp
test/detail/parser/read_atom.cpp test/detail/parser/read_atom.cpp
test/detail/parser/read_bool.cpp
test/detail/parser/read_floating_point.cpp test/detail/parser/read_floating_point.cpp
test/detail/parser/read_ini.cpp
test/detail/parser/read_number.cpp test/detail/parser/read_number.cpp
test/detail/parser/read_number_or_timespan.cpp
test/detail/parser/read_signed_integer.cpp test/detail/parser/read_signed_integer.cpp
test/detail/parser/read_string.cpp test/detail/parser/read_number_or_timespan.cpp
test/detail/parser/read_timespan.cpp
test/detail/parser/read_unsigned_integer.cpp test/detail/parser/read_unsigned_integer.cpp
test/detail/parser/read_bool.cpp
test/detail/parser/read_timespan.cpp
test/detail/parser/read_ini.cpp
test/detail/parser/read_string.cpp
test/detail/ringbuffer.cpp test/detail/ringbuffer.cpp
test/detail/ripemd_160.cpp test/detail/ripemd_160.cpp
test/detail/tick_emitter.cpp
test/detail/unique_function.cpp
test/detail/unordered_flat_map.cpp test/detail/unordered_flat_map.cpp
test/dictionary.cpp
test/dynamic_spawn.cpp
test/expected.cpp
test/function_view.cpp
test/fused_downstream_manager.cpp
test/handles.cpp
test/inbound_path.cpp
test/inspector.cpp
test/intrusive/drr_cached_queue.cpp test/intrusive/drr_cached_queue.cpp
test/intrusive/drr_queue.cpp
test/intrusive/fifo_inbox.cpp
test/intrusive/lifo_inbox.cpp test/intrusive/lifo_inbox.cpp
test/intrusive/fifo_inbox.cpp
test/intrusive/drr_queue.cpp
test/intrusive/wdrr_fixed_multiplexed_queue.cpp
test/intrusive/task_queue.cpp test/intrusive/task_queue.cpp
test/intrusive/wdrr_dynamic_multiplexed_queue.cpp test/intrusive/wdrr_dynamic_multiplexed_queue.cpp
test/intrusive/wdrr_fixed_multiplexed_queue.cpp test/config_value_adaptor.cpp
test/serialization.cpp
test/deep_to_string.cpp
test/pipeline_streaming.cpp
test/actor_factory.cpp
test/intrusive_ptr.cpp test/intrusive_ptr.cpp
test/ipv4_address.cpp test/inspector.cpp
test/ipv4_endpoint.cpp test/variant.cpp
test/ipv4_subnet.cpp test/tracing_data.cpp
test/ipv6_address.cpp test/unit.cpp
test/actor_registry.cpp
test/rtti_pair.cpp
test/result.cpp
test/ipv6_endpoint.cpp test/ipv6_endpoint.cpp
test/ipv6_subnet.cpp test/request_timeout.cpp
test/local_group.cpp test/ipv4_subnet.cpp
test/local_migration.cpp test/to_string.cpp
test/logger.cpp test/logger.cpp
test/mailbox_element.cpp
test/make_config_value_field.cpp
test/match.cpp
test/message.cpp
test/message_id.cpp test/message_id.cpp
test/message_lifetime.cpp test/stateful_actor.cpp
test/metaprogramming.cpp test/ipv6_address.cpp
test/mixin/requester.cpp test/actor_pool.cpp
test/mixin/sender.cpp
test/mock_streaming_classes.cpp
test/native_streaming_classes.cpp
test/optional.cpp
test/or_else.cpp
test/pipeline_streaming.cpp
test/policy/categorized.cpp
test/request_timeout.cpp
test/result.cpp
test/rtti_pair.cpp
test/runtime_settings_map.cpp
test/selective_streaming.cpp
test/serial_reply.cpp
test/serialization.cpp
test/settings.cpp
test/simple_timeout.cpp test/simple_timeout.cpp
test/detached_actors.cpp
test/span.cpp test/span.cpp
test/stateful_actor.cpp test/handles.cpp
test/string_algorithms.cpp
test/string_view.cpp
test/sum_type.cpp
test/sum_type_token.cpp
test/thread_hook.cpp test/thread_hook.cpp
test/to_string.cpp test/settings.cpp
test/type_erased_tuple.cpp test/broadcast_downstream_manager.cpp
test/typed_response_promise.cpp test/dictionary.cpp
test/typed_spawn.cpp test/policy/categorized.cpp
test/unit.cpp test/policy/fan_in_responses.cpp
test/uri.cpp test/expected.cpp
test/variant.cpp test/metaprogramming.cpp
) )
# -- add library target -------------------------------------------------------- # -- add library target --------------------------------------------------------
......
...@@ -582,6 +582,10 @@ public: ...@@ -582,6 +582,10 @@ public:
profiler_->before_sending_scheduled(self, timeout, element); profiler_->before_sending_scheduled(self, timeout, element);
} }
tracing_data_factory* tracing_context() const noexcept {
return tracing_context_;
}
/// @endcond /// @endcond
private: private:
...@@ -665,6 +669,9 @@ private: ...@@ -665,6 +669,9 @@ private:
/// Stores custom, system-wide key-value pairs. /// Stores custom, system-wide key-value pairs.
runtime_settings_map settings_; runtime_settings_map settings_;
/// Stores the system-wide factory for deserializing tracing data.
tracing_data_factory* tracing_context_;
}; };
} // namespace caf } // namespace caf
...@@ -305,6 +305,11 @@ public: ...@@ -305,6 +305,11 @@ public:
/// @note Has no effect unless building CAF with CAF_ENABLE_ACTOR_PROFILER. /// @note Has no effect unless building CAF with CAF_ENABLE_ACTOR_PROFILER.
actor_profiler* profiler = nullptr; actor_profiler* profiler = nullptr;
/// Enables CAF to deserialize application-specific tracing information.
/// @experimental
/// @note Has no effect unless building CAF with CAF_ENABLE_ACTOR_PROFILER.
tracing_data_factory* tracing_context = nullptr;
// -- run-time type information ---------------------------------------------- // -- run-time type information ----------------------------------------------
portable_name_map type_names_by_rtti; portable_name_map type_names_by_rtti;
......
...@@ -18,6 +18,8 @@ ...@@ -18,6 +18,8 @@
#pragma once #pragma once
#include <vector>
#include "caf/default_downstream_manager.hpp" #include "caf/default_downstream_manager.hpp"
#include "caf/detail/stream_stage_driver_impl.hpp" #include "caf/detail/stream_stage_driver_impl.hpp"
#include "caf/detail/stream_stage_impl.hpp" #include "caf/detail/stream_stage_impl.hpp"
...@@ -70,17 +72,22 @@ attach_stream_stage(scheduled_actor* self, const stream<In>& in, ...@@ -70,17 +72,22 @@ attach_stream_stage(scheduled_actor* self, const stream<In>& in,
CAF_IGNORE_UNUSED(token); CAF_IGNORE_UNUSED(token);
using output_type = typename stream_stage_trait_t<Fun>::output; using output_type = typename stream_stage_trait_t<Fun>::output;
using state_type = typename stream_stage_trait_t<Fun>::state; using state_type = typename stream_stage_trait_t<Fun>::state;
static_assert(std::is_same< static_assert(
void(state_type&), std::is_same<void(state_type&),
typename detail::get_callable_trait<Init>::fun_sig>::value, typename detail::get_callable_trait<Init>::fun_sig>::value,
"Expected signature `void (State&)` for init function"); "Expected signature `void (State&)` for init function");
static_assert(std::is_same< using consume_one = void(state_type&, downstream<output_type>&, In);
void(state_type&, downstream<output_type>&, In), using consume_all
typename detail::get_callable_trait<Fun>::fun_sig>::value, = 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)` " "Expected signature `void (State&, downstream<Out>&, In)` "
"or `void (State&, downstream<Out>&, std::vector<In>&)` "
"for consume function"); "for consume function");
using driver = detail::stream_stage_driver_impl< using driver
typename Trait::input, DownstreamManager, Fun, Finalize>; = detail::stream_stage_driver_impl<typename Trait::input, DownstreamManager,
Fun, Finalize>;
return attach_stream_stage<driver>(self, in, std::move(xs), std::move(init), return attach_stream_stage<driver>(self, in, std::move(xs), std::move(init),
std::move(fun), std::move(fin)); std::move(fun), std::move(fin));
} }
......
...@@ -28,6 +28,7 @@ ...@@ -28,6 +28,7 @@
#include "caf/byte.hpp" #include "caf/byte.hpp"
#include "caf/byte_buffer.hpp" #include "caf/byte_buffer.hpp"
#include "caf/detail/core_export.hpp" #include "caf/detail/core_export.hpp"
#include "caf/detail/squashed_int.hpp"
#include "caf/error_code.hpp" #include "caf/error_code.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/read_inspector.hpp" #include "caf/read_inspector.hpp"
...@@ -129,7 +130,9 @@ public: ...@@ -129,7 +130,9 @@ public:
template <class T> template <class T>
std::enable_if_t<std::is_integral<T>::value && std::is_signed<T>::value> std::enable_if_t<std::is_integral<T>::value && std::is_signed<T>::value>
apply(T x) { apply(T x) {
return apply(static_cast<std::make_unsigned_t<T>>(x)); using unsigned_type = std::make_unsigned_t<T>;
using squashed_unsigned_type = detail::squashed_int_t<unsigned_type>;
return apply(static_cast<squashed_unsigned_type>(x));
} }
template <class Enum> template <class Enum>
......
...@@ -149,6 +149,9 @@ ...@@ -149,6 +149,9 @@
# ifdef __APPLE__ # ifdef __APPLE__
# define CAF_NO_THREAD_LOCAL # define CAF_NO_THREAD_LOCAL
# endif # endif
# ifdef __MINGW32__
# define CAF_NO_THREAD_LOCAL
# endif
#elif defined(_MSC_VER) #elif defined(_MSC_VER)
# define CAF_MSVC # define CAF_MSVC
# define CAF_LIKELY(x) x # define CAF_LIKELY(x) x
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#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 CAF_CORE_EXPORT assignment {
/// Stores how much credit we assign to the source.
int32_t credit;
/// Stores how many elements we demand per batch.
int32_t batch_size;
};
// -- constructors, destructors, and assignment operators --------------------
explicit credit_controller(scheduled_actor* self);
virtual ~credit_controller();
// -- properties -------------------------------------------------------------
scheduled_actor* self() noexcept {
return self_;
}
// -- 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& x) = 0;
/// Called after processing the batch `x` in order to allow the controller to
/// keep statistics on incoming batches.
/// @note The consumer may alter the batch while processing it, for example
/// by moving each element of the batch downstream.
virtual void after_processing(downstream_msg::batch& x) = 0;
/// Assigns initial credit during the stream handshake.
/// @returns The initial credit for the new sources.
virtual assignment compute_initial() = 0;
/// Assigs new credit to the source after a cycle ends.
/// @param cycle Duration of a cycle.
/// @param max_downstream_credit Maximum downstream capacity as reported by
/// the downstream manager. Controllers may use
/// this capacity as an upper bound.
virtual assignment compute(timespan cycle, int32_t max_downstream_credit) = 0;
// -- virtual functions ------------------------------------------------------
/// Computes a credit assignment to the source after crossing the
/// low-threshold. May assign zero credit.
virtual assignment compute_bridge();
/// Returns the threshold for when we may give extra credit to a source
/// during a cycle.
/// @returns Zero or a negative value if the controller never grants bridge
/// credit, otherwise the threshold for calling `compute_bridge` to
/// generate additional credit.
virtual int32_t threshold() const noexcept;
private:
// -- member variables -------------------------------------------------------
/// Points to the parent system.
scheduled_actor* self_;
};
} // namespace caf
...@@ -37,6 +37,14 @@ namespace stream { ...@@ -37,6 +37,14 @@ namespace stream {
extern CAF_CORE_EXPORT const timespan desired_batch_complexity; extern CAF_CORE_EXPORT const timespan desired_batch_complexity;
extern CAF_CORE_EXPORT const timespan max_batch_delay; extern CAF_CORE_EXPORT const timespan max_batch_delay;
extern CAF_CORE_EXPORT const timespan credit_round_interval; extern CAF_CORE_EXPORT const timespan credit_round_interval;
extern CAF_CORE_EXPORT const atom_value credit_policy;
namespace size_policy {
extern CAF_CORE_EXPORT const int32_t bytes_per_batch;
extern CAF_CORE_EXPORT const int32_t buffer_capacity;
} // namespace size_policy
} // namespace stream } // namespace stream
......
...@@ -106,15 +106,6 @@ public: ...@@ -106,15 +106,6 @@ public:
/// @copydoc apply /// @copydoc apply
virtual result_type apply(long double&) = 0; virtual result_type apply(long double&) = 0;
/// @copydoc apply
virtual result_type apply(timespan x) = 0;
/// @copydoc apply
virtual result_type apply(timestamp x) = 0;
/// @copydoc apply
virtual result_type apply(atom_value x) = 0;
/// @copydoc apply /// @copydoc apply
virtual result_type apply(std::string&) = 0; virtual result_type apply(std::string&) = 0;
...@@ -133,7 +124,7 @@ public: ...@@ -133,7 +124,7 @@ public:
/// Reads a byte sequence from the input. /// Reads a byte sequence from the input.
/// @param x The byte sequence. /// @param x The byte sequence.
/// @returns A non-zero error code on failure, `sec::success` otherwise. /// @returns A non-zero error code on failure, `sec::success` otherwise.
virtual result_type apply_raw(span<byte> x) = 0; virtual result_type apply(span<byte> x) = 0;
/// Adds each boolean in `xs` to the output. Derived classes can override this /// Adds each boolean in `xs` to the output. Derived classes can override this
/// member function to pack the booleans, for example to avoid using one byte /// member function to pack the booleans, for example to avoid using one byte
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#include "caf/credit_controller.hpp"
namespace caf {
namespace detail {
/// Computes credit for an attached source based on measuring the complexity of
/// incoming batches.
class complexity_based_credit_controller : public credit_controller {
public:
// -- member types -----------------------------------------------------------
using super = credit_controller;
// -- constants --------------------------------------------------------------
/// Stores how many elements we buffer at most after the handshake.
int32_t initial_buffer_size = 50;
/// Stores how many elements we allow per batch after the handshake.
int32_t initial_batch_size = 10;
// -- constructors, destructors, and assignment operators --------------------
explicit complexity_based_credit_controller(scheduled_actor* self);
~complexity_based_credit_controller() override;
// -- overrides --------------------------------------------------------------
void before_processing(downstream_msg::batch& x) override;
void after_processing(downstream_msg::batch& x) override;
assignment compute_initial() override;
assignment compute(timespan cycle, int32_t) override;
private:
// -- member variables -------------------------------------------------------
/// Total number of elements in all processed batches in the current cycle.
int64_t num_elements_ = 0;
/// Elapsed time for processing all elements of all batches in the current
/// cycle.
timespan processing_time_;
/// Timestamp of the last call to `before_processing`.
timestamp processing_begin_;
/// Stores the desired per-batch complexity.
timespan complexity_;
};
} // namespace detail
} // namespace caf
...@@ -55,11 +55,15 @@ protected: ...@@ -55,11 +55,15 @@ protected:
}; };
/// Wraps a user-defined function and gives it a uniform signature. /// Wraps a user-defined function and gives it a uniform signature.
template <class Base, class F, class ArgsPtr, bool ReturnsBehavior, template <class Base, class F, class Tuple, bool ReturnsBehavior,
bool HasSelfPtr> bool HasSelfPtr>
class init_fun_factory_helper final : public init_fun_factory_helper_base { class init_fun_factory_helper final : public init_fun_factory_helper_base {
public: public:
init_fun_factory_helper(F fun, ArgsPtr args) using args_pointer = std::shared_ptr<Tuple>;
static constexpr bool args_empty = std::tuple_size<Tuple>::value == 0;
init_fun_factory_helper(F fun, args_pointer args)
: fun_(std::move(fun)), args_(std::move(args)) { : fun_(std::move(fun)), args_(std::move(args)) {
// nop // nop
} }
...@@ -71,40 +75,52 @@ public: ...@@ -71,40 +75,52 @@ public:
behavior operator()(local_actor* self) final { behavior operator()(local_actor* self) final {
if (hook_ != nullptr) if (hook_ != nullptr)
hook_(self); hook_(self);
bool_token<ReturnsBehavior> returns_behavior_token; auto dptr = static_cast<Base*>(self);
bool_token<HasSelfPtr> captures_self_token; if constexpr (ReturnsBehavior) {
return apply(returns_behavior_token, captures_self_token, self); auto unbox = [](auto x) -> behavior { return std::move(x.unbox()); };
} if constexpr (args_empty) {
if constexpr (HasSelfPtr) {
private: // behavior (pointer)
// behavior (pointer) return unbox(fun_(dptr));
behavior apply(std::true_type, std::true_type, local_actor* ptr) { } else {
auto res = apply_moved_args_prefixed(fun_, get_indices(*args_), *args_, // behavior ()
static_cast<Base*>(ptr)); return unbox(fun_());
return std::move(res.unbox()); }
} } else {
if constexpr (HasSelfPtr) {
// void (pointer) // behavior (pointer, args...)
behavior apply(std::false_type, std::true_type, local_actor* ptr) { auto res = apply_moved_args_prefixed(fun_, get_indices(*args_),
apply_moved_args_prefixed(fun_, get_indices(*args_), *args_, *args_, dptr);
static_cast<Base*>(ptr)); return unbox(std::move(res));
return behavior{}; } else {
} // behavior (args...)
return unbox(apply_args(fun_, get_indices(*args_), *args_));
// behavior () }
behavior apply(std::true_type, std::false_type, local_actor*) { }
auto res = apply_args(fun_, get_indices(*args_), *args_); } else {
return std::move(res.unbox()); if constexpr (args_empty) {
} if constexpr (HasSelfPtr) {
// void (pointer)
// void () fun_(dptr);
behavior apply(std::false_type, std::false_type, local_actor*) { } else {
apply_args(fun_, get_indices(*args_), *args_); // void ()
return behavior{}; fun_();
}
} else {
if constexpr (HasSelfPtr) {
// void (pointer, args...)
apply_moved_args_prefixed(fun_, get_indices(*args_), *args_, dptr);
} else {
// void (args...)
apply_args(fun_, get_indices(*args_), *args_);
}
}
return {};
}
} }
F fun_; F fun_;
ArgsPtr args_; args_pointer args_;
}; };
template <class Base, class F> template <class Base, class F>
...@@ -125,8 +141,7 @@ public: ...@@ -125,8 +141,7 @@ public:
constexpr bool selfptr = std::is_pointer<first_arg>::value; constexpr bool selfptr = std::is_pointer<first_arg>::value;
constexpr bool rets = std::is_convertible<res_type, behavior>::value; constexpr bool rets = std::is_convertible<res_type, behavior>::value;
using tuple_type = decltype(std::make_tuple(detail::spawn_fwd<Ts>(xs)...)); using tuple_type = decltype(std::make_tuple(detail::spawn_fwd<Ts>(xs)...));
using tuple_ptr = std::shared_ptr<tuple_type>; using helper = init_fun_factory_helper<Base, F, tuple_type, rets, selfptr>;
using helper = init_fun_factory_helper<Base, F, tuple_ptr, rets, selfptr>;
return ptr_type{new helper{std::move(f), sizeof...(Ts) > 0 return ptr_type{new helper{std::move(f), sizeof...(Ts) > 0
? std::make_shared<tuple_type>( ? std::make_shared<tuple_type>(
detail::spawn_fwd<Ts>(xs)...) detail::spawn_fwd<Ts>(xs)...)
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#include "caf/error.hpp"
#include "caf/serializer.hpp"
namespace caf::detail {
class serialized_size_inspector final : public serializer {
public:
using super = serializer;
using super::super;
size_t result() const noexcept {
return result_;
}
result_type begin_object(uint16_t typenr, string_view type_name) override;
result_type end_object() override;
result_type begin_sequence(size_t num) override;
result_type end_sequence() override;
result_type apply(bool x) override;
result_type apply(int8_t x) override;
result_type apply(uint8_t x) override;
result_type apply(int16_t x) override;
result_type apply(uint16_t x) override;
result_type apply(int32_t x) override;
result_type apply(uint32_t x) override;
result_type apply(int64_t x) override;
result_type apply(uint64_t x) override;
result_type apply(float x) override;
result_type apply(double x) override;
result_type apply(long double x) override;
result_type apply(string_view x) override;
result_type apply(const std::u16string& x) override;
result_type apply(const std::u32string& x) override;
result_type apply(span<const byte> x) override;
result_type apply(const std::vector<bool>& xs) override;
private:
size_t result_ = 0;
};
template <class T>
size_t serialized_size(actor_system& sys, const T& x) {
serialized_size_inspector f{sys};
auto err = f(x);
static_cast<void>(err);
return f.result();
}
} // namespace caf::detail
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#include "caf/credit_controller.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 size_based_credit_controller : public credit_controller {
public:
// -- member types -----------------------------------------------------------
using super = credit_controller;
// -- constants --------------------------------------------------------------
/// Configures at what buffer level we grant bridge credit (0 to 1).
static constexpr float buffer_threshold = 0.75f;
/// Configures how many samples we require for recalculating buffer sizes.
static constexpr int32_t min_samples = 10;
/// 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(scheduled_actor* self);
~size_based_credit_controller() override;
// -- overrides --------------------------------------------------------------
void before_processing(downstream_msg::batch& x) override;
void after_processing(downstream_msg::batch& x) override;
assignment compute_initial() override;
assignment compute(timespan cycle, int32_t) override;
assignment compute_bridge() override;
int32_t threshold() const noexcept override;
private:
// -- member variables -------------------------------------------------------
/// Total number of elements in all processed batches in the current cycle.
int64_t num_batches_ = 0;
/// Stores how many elements the buffer should hold at most.
int32_t buffer_size_ = initial_buffer_size;
/// Stores how many elements each batch should contain.
int32_t batch_size_ = initial_batch_size;
/// Configures how many bytes we store in total.
int32_t buffer_capacity_;
/// Configures how many bytes we transfer per batch.
int32_t bytes_per_batch_;
/// Stores how many elements we have sampled during the current cycle.
int32_t sampled_elements_ = 0;
/// Stores approximately how many bytes the sampled elements require when
/// serialized.
int32_t sampled_total_size_ = 0;
/// Counter for keeping track of when to sample a batch.
int32_t sample_counter_ = 0;
/// Configured how many batches we skip for the size sampling.
int32_t sample_rate_ = 1;
};
} // namespace caf::detail
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#include "caf/credit_controller.hpp"
namespace caf {
namespace detail {
/// Computes predictable credit in unit tests.
class test_credit_controller : public credit_controller {
public:
// -- member types -----------------------------------------------------------
using super = credit_controller;
// -- constructors, destructors, and assignment operators --------------------
using super::super;
~test_credit_controller() override;
// -- overrides --------------------------------------------------------------
void before_processing(downstream_msg::batch& x) override;
void after_processing(downstream_msg::batch& x) override;
assignment compute_initial() override;
assignment compute(timespan cycle, int32_t) override;
private:
/// Total number of elements in all processed batches in the current cycle.
int64_t num_elements_ = 0;
};
} // namespace detail
} // namespace caf
...@@ -695,8 +695,8 @@ template <class T, class To> ...@@ -695,8 +695,8 @@ template <class T, class To>
class has_convertible_data_member { class has_convertible_data_member {
private: private:
template <class U> template <class U>
static auto sfinae(U* x) static auto sfinae(U* x) -> std::integral_constant<
-> decltype(std::declval<To*>() = x->data(), std::true_type()); bool, std::is_convertible<decltype(x->data()), To*>::value>;
template <class U> template <class U>
static auto sfinae(...) -> std::false_type; static auto sfinae(...) -> std::false_type;
......
...@@ -137,6 +137,8 @@ class scoped_actor; ...@@ -137,6 +137,8 @@ class scoped_actor;
class serializer; class serializer;
class stream_manager; class stream_manager;
class string_view; class string_view;
class tracing_data;
class tracing_data_factory;
class type_erased_tuple; class type_erased_tuple;
class type_erased_value; class type_erased_value;
class uniform_type_info_map; class uniform_type_info_map;
...@@ -307,7 +309,8 @@ using stream_manager_ptr = intrusive_ptr<stream_manager>; ...@@ -307,7 +309,8 @@ using stream_manager_ptr = intrusive_ptr<stream_manager>;
// -- unique pointer aliases --------------------------------------------------- // -- unique pointer aliases ---------------------------------------------------
using type_erased_value_ptr = std::unique_ptr<type_erased_value>;
using mailbox_element_ptr = std::unique_ptr<mailbox_element, detail::disposer>; using mailbox_element_ptr = std::unique_ptr<mailbox_element, detail::disposer>;
using tracing_data_ptr = std::unique_ptr<tracing_data>;
using type_erased_value_ptr = std::unique_ptr<type_erased_value>;
} // namespace caf } // namespace caf
...@@ -23,6 +23,7 @@ ...@@ -23,6 +23,7 @@
#include "caf/actor_clock.hpp" #include "caf/actor_clock.hpp"
#include "caf/actor_control_block.hpp" #include "caf/actor_control_block.hpp"
#include "caf/credit_controller.hpp"
#include "caf/detail/core_export.hpp" #include "caf/detail/core_export.hpp"
#include "caf/downstream_msg.hpp" #include "caf/downstream_msg.hpp"
#include "caf/meta/type_name.hpp" #include "caf/meta/type_name.hpp"
...@@ -55,63 +56,22 @@ public: ...@@ -55,63 +56,22 @@ public:
stream_slots slots; stream_slots slots;
/// Stores the last computed desired batch size. /// Stores the last computed desired batch size.
int32_t desired_batch_size; int32_t desired_batch_size = 0;
/// Amount of credit we have signaled upstream. /// Amount of credit we have signaled upstream.
int32_t assigned_credit; int32_t assigned_credit = 0;
/// Priority of incoming batches from this source. /// Priority of incoming batches from this source.
stream_priority prio; stream_priority prio = stream_priority::normal;
/// ID of the last acknowledged batch ID. /// ID of the last acknowledged batch ID.
int64_t last_acked_batch_id; int64_t last_acked_batch_id = 0;
/// ID of the last received batch. /// ID of the last received batch.
int64_t last_batch_id; int64_t last_batch_id = 0;
/// Amount of credit we assign sources after receiving `open`. /// Controller for assigning credit to the source.
static constexpr int initial_credit = 50; std::unique_ptr<credit_controller> controller_;
/// Stores statistics for measuring complexity of incoming batches.
struct CAF_CORE_EXPORT stats_t {
/// Wraps a time measurement for a single processed batch.
struct measurement {
/// Number of items in the batch.
int32_t batch_size;
/// Elapsed time for processing all elements of the batch.
timespan calculation_time;
};
/// Wraps the resulf of `stats_t::calculate()`.
struct CAF_CORE_EXPORT calculation_result {
/// Number of items per credit cycle.
int32_t max_throughput;
/// Number of items per batch to reach the desired batch complexity.
int32_t items_per_batch;
};
/// Total number of elements in all processed batches.
int64_t num_elements;
/// Elapsed time for processing all elements of all batches.
timespan processing_time;
stats_t();
/// Returns the maximum number of items this actor could handle for given
/// cycle length with a minimum of 1.
calculation_result calculate(timespan cycle, timespan desired_complexity);
/// Adds a measurement to this statistic.
void store(measurement x);
/// Resets this statistic.
void reset();
};
/// Summarizes how many elements we processed during the last cycle and how
/// much time we spent processing those elements.
stats_t stats;
/// Stores the time point of the last credit decision for this source. /// Stores the time point of the last credit decision for this source.
actor_clock::time_point last_credit_decision; actor_clock::time_point last_credit_decision;
...@@ -147,10 +107,8 @@ public: ...@@ -147,10 +107,8 @@ public:
/// waiting in the mailbox. /// waiting in the mailbox.
/// @param now Current timestamp. /// @param now Current timestamp.
/// @param cycle Time between credit rounds. /// @param cycle Time between credit rounds.
/// @param desired_batch_complexity Desired processing time per batch.
void emit_ack_batch(local_actor* self, int32_t queued_items, void emit_ack_batch(local_actor* self, int32_t queued_items,
actor_clock::time_point now, timespan cycle, actor_clock::time_point now, timespan cycle);
timespan desired_batch_complexity);
/// Returns whether the path received no input since last emitting /// Returns whether the path received no input since last emitting
/// `ack_batch`, i.e., `last_acked_batch_id == last_batch_id`. /// `ack_batch`, i.e., `last_acked_batch_id == last_batch_id`.
...@@ -167,8 +125,8 @@ public: ...@@ -167,8 +125,8 @@ public:
emit_irregular_shutdown(local_actor* self, stream_slots slots, emit_irregular_shutdown(local_actor* self, stream_slots slots,
const strong_actor_ptr& hdl, error reason); const strong_actor_ptr& hdl, error reason);
private: /// Returns a pointer to the parent actor.
actor_clock& clock(); scheduled_actor* self();
}; };
/// @relates inbound_path /// @relates inbound_path
......
...@@ -81,13 +81,13 @@ public: ...@@ -81,13 +81,13 @@ public:
// -- operators -------------------------------------------------------------- // -- operators --------------------------------------------------------------
forward_iterator& operator++() { forward_iterator& operator++() {
ptr = promote(ptr->next); ptr = ptr->next;
return *this; return *this;
} }
forward_iterator operator++(int) { forward_iterator operator++(int) {
forward_iterator res = *this; forward_iterator res = *this;
ptr = promote(ptr->next); ptr = ptr->next;
return res; return res;
} }
......
...@@ -171,10 +171,8 @@ public: ...@@ -171,10 +171,8 @@ public:
/// @private /// @private
task_size_type next_task_size() const noexcept { task_size_type next_task_size() const noexcept {
if (head_.next == nullptr) auto ptr = head_.next;
return 0; return ptr != &tail_ ? policy_.task_size(*promote(ptr)) : 0;
auto ptr = promote(head_.next);
return policy_.task_size(*ptr);
} }
/// @private /// @private
...@@ -201,16 +199,6 @@ public: ...@@ -201,16 +199,6 @@ public:
// -- iterator access -------------------------------------------------------- // -- iterator access --------------------------------------------------------
/// Returns an iterator to the dummy before the first element.
iterator before_begin() noexcept {
return &head_;
}
/// Returns an iterator to the dummy before the first element.
const_iterator before_begin() const noexcept {
return &head_;
}
/// Returns an iterator to the dummy before the first element. /// Returns an iterator to the dummy before the first element.
iterator begin() noexcept { iterator begin() noexcept {
return head_.next; return head_.next;
...@@ -250,6 +238,7 @@ public: ...@@ -250,6 +238,7 @@ public:
/// Returns a pointer to the last element. /// Returns a pointer to the last element.
pointer back() noexcept { pointer back() noexcept {
CAF_ASSERT(head_.next != &tail_);
return promote(tail_.next); return promote(tail_.next);
} }
...@@ -317,7 +306,7 @@ public: ...@@ -317,7 +306,7 @@ public:
/// @private /// @private
void lifo_append(node_pointer ptr) { void lifo_append(node_pointer ptr) {
if (old_last_ == nullptr) { if (old_last_ == nullptr) {
old_last_ = back(); old_last_ = tail_.next;
push_back(promote(ptr)); push_back(promote(ptr));
} else { } else {
ptr->next = new_head_; ptr->next = new_head_;
......
...@@ -19,8 +19,10 @@ ...@@ -19,8 +19,10 @@
#pragma once #pragma once
#include <cstddef> #include <cstddef>
#include <memory>
#include "caf/actor_control_block.hpp" #include "caf/actor_control_block.hpp"
#include "caf/config.hpp"
#include "caf/detail/core_export.hpp" #include "caf/detail/core_export.hpp"
#include "caf/detail/disposer.hpp" #include "caf/detail/disposer.hpp"
#include "caf/detail/tuple_vals.hpp" #include "caf/detail/tuple_vals.hpp"
...@@ -35,6 +37,7 @@ ...@@ -35,6 +37,7 @@
#include "caf/meta/omittable_if_empty.hpp" #include "caf/meta/omittable_if_empty.hpp"
#include "caf/meta/type_name.hpp" #include "caf/meta/type_name.hpp"
#include "caf/ref_counted.hpp" #include "caf/ref_counted.hpp"
#include "caf/tracing_data.hpp"
#include "caf/type_erased_tuple.hpp" #include "caf/type_erased_tuple.hpp"
namespace caf { namespace caf {
...@@ -56,6 +59,13 @@ public: ...@@ -56,6 +59,13 @@ public:
/// if this is empty then the original sender receives the response. /// if this is empty then the original sender receives the response.
forwarding_stack stages; forwarding_stack stages;
#ifdef CAF_ENABLE_ACTOR_PROFILER
/// Optional tracing information. This field is unused by default, but an
/// @ref actor_profiler can make use of it to inject application-specific
/// instrumentation.
tracing_data_ptr tracing_id;
#endif // CAF_ENABLE_ACTOR_PROFILER
mailbox_element(); mailbox_element();
mailbox_element(strong_actor_ptr&& x, message_id y, forwarding_stack&& z); mailbox_element(strong_actor_ptr&& x, message_id y, forwarding_stack&& z);
...@@ -99,7 +109,11 @@ struct mailbox_category_corrector<upstream_msg> { ...@@ -99,7 +109,11 @@ struct mailbox_category_corrector<upstream_msg> {
template <class Inspector> template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, mailbox_element& x) { typename Inspector::result_type inspect(Inspector& f, mailbox_element& x) {
return f(meta::type_name("mailbox_element"), x.sender, x.mid, return f(meta::type_name("mailbox_element"), x.sender, x.mid,
meta::omittable_if_empty(), x.stages, x.content()); meta::omittable_if_empty(), x.stages,
#ifdef CAF_ENABLE_ACTOR_PROFILER
x.tracing_id,
#endif // CAF_ENABLE_ACTOR_PROFILER
x.content());
} }
/// Encapsulates arbitrary data in a message element. /// Encapsulates arbitrary data in a message element.
......
...@@ -54,7 +54,7 @@ public: ...@@ -54,7 +54,7 @@ public:
// -- overridden functions of monitorable_actor ------------------------------ // -- overridden functions of monitorable_actor ------------------------------
bool cleanup(error&& fail_state, execution_unit* ptr) override { bool cleanup(error&& fail_state, execution_unit* ptr) override {
auto me = dptr()->ctrl(); auto me = this->ctrl();
for (auto& subscription : subscriptions_) for (auto& subscription : subscriptions_)
subscription->unsubscribe(me); subscription->unsubscribe(me);
subscriptions_.clear(); subscriptions_.clear();
...@@ -69,7 +69,7 @@ public: ...@@ -69,7 +69,7 @@ public:
CAF_LOG_TRACE(CAF_ARG(what)); CAF_LOG_TRACE(CAF_ARG(what));
if (what == invalid_group) if (what == invalid_group)
return; return;
if (what->subscribe(dptr()->ctrl())) if (what->subscribe(this->ctrl()))
subscriptions_.emplace(what); subscriptions_.emplace(what);
} }
...@@ -77,7 +77,7 @@ public: ...@@ -77,7 +77,7 @@ public:
void leave(const group& what) { void leave(const group& what) {
CAF_LOG_TRACE(CAF_ARG(what)); CAF_LOG_TRACE(CAF_ARG(what));
if (subscriptions_.erase(what) > 0) if (subscriptions_.erase(what) > 0)
what->unsubscribe(dptr()->ctrl()); what->unsubscribe(this->ctrl());
} }
/// Returns all subscribed groups. /// Returns all subscribed groups.
...@@ -86,10 +86,6 @@ public: ...@@ -86,10 +86,6 @@ public:
} }
private: private:
Subtype* dptr() {
return static_cast<Subtype*>(this);
}
// -- data members ----------------------------------------------------------- // -- data members -----------------------------------------------------------
/// Stores all subscribed groups. /// Stores all subscribed groups.
......
...@@ -84,12 +84,12 @@ private: ...@@ -84,12 +84,12 @@ private:
if constexpr (std::is_empty<T>::value if constexpr (std::is_empty<T>::value
|| is_allowed_unsafe_message_type_v<T>) { || is_allowed_unsafe_message_type_v<T>) {
// skip element // skip element
} else if constexpr (detail::can_apply_v<Subtype, decltype(x)>) {
CAF_READ_INSPECTOR_TRY(dref.apply(x))
} else if constexpr (std::is_integral<T>::value) { } else if constexpr (std::is_integral<T>::value) {
using squashed_type = detail::squashed_int_t<T>; using squashed_type = detail::squashed_int_t<T>;
auto squashed_x = static_cast<squashed_type>(x); auto squashed_x = static_cast<squashed_type>(x);
CAF_READ_INSPECTOR_TRY(dref.apply(squashed_x)) CAF_READ_INSPECTOR_TRY(dref.apply(squashed_x))
} else if constexpr (detail::can_apply_v<Subtype, decltype(x)>) {
CAF_READ_INSPECTOR_TRY(dref.apply(x))
} else if constexpr (std::is_array<T>::value) { } else if constexpr (std::is_array<T>::value) {
std::make_index_sequence<std::extent<T>::value> seq; std::make_index_sequence<std::extent<T>::value> seq;
CAF_READ_INSPECTOR_TRY(apply_array(dref, x, seq)) CAF_READ_INSPECTOR_TRY(apply_array(dref, x, seq))
......
...@@ -54,7 +54,7 @@ public: ...@@ -54,7 +54,7 @@ public:
auto this_worker = this; auto this_worker = this;
this_thread_ = std::thread{[this_worker] { this_thread_ = std::thread{[this_worker] {
CAF_SET_LOGGER_SYS(&this_worker->system()); CAF_SET_LOGGER_SYS(&this_worker->system());
detail::set_thread_name("caf.multiplexer"); detail::set_thread_name("caf.worker");
this_worker->system().thread_started(); this_worker->system().thread_started();
this_worker->run(); this_worker->run();
this_worker->system().thread_terminates(); this_worker->system().thread_terminates();
......
...@@ -127,6 +127,8 @@ enum class sec : uint8_t { ...@@ -127,6 +127,8 @@ enum class sec : uint8_t {
unavailable_or_would_block, unavailable_or_would_block,
/// Resolving a path on a remote node failed. /// Resolving a path on a remote node failed.
remote_lookup_failed, remote_lookup_failed,
/// Serialization failed because actor_system::tracing_context is null.
no_tracing_context,
}; };
/// @relates sec /// @relates sec
......
...@@ -113,15 +113,6 @@ public: ...@@ -113,15 +113,6 @@ public:
/// @copydoc apply /// @copydoc apply
virtual result_type apply(long double x) = 0; virtual result_type apply(long double x) = 0;
/// @copydoc apply
virtual result_type apply(timespan x) = 0;
/// @copydoc apply
virtual result_type apply(timestamp x) = 0;
/// @copydoc apply
virtual result_type apply(atom_value x) = 0;
/// @copydoc apply /// @copydoc apply
virtual result_type apply(string_view x) = 0; virtual result_type apply(string_view x) = 0;
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#include <memory>
#include "caf/detail/core_export.hpp"
#include "caf/fwd.hpp"
namespace caf {
/// Marker interface for application-specific tracing data. This interface
/// enables users to inject application-specific instrumentation into CAF's
/// messaging layer. CAF provides no default implementation for this
/// customization point.
class CAF_CORE_EXPORT tracing_data {
public:
virtual ~tracing_data();
/// Writes the content of this object to `sink`.
virtual error serialize(serializer& sink) const = 0;
/// @copydoc serialize
virtual error_code<sec> serialize(binary_serializer& sink) const = 0;
};
/// @relates tracing_data
using tracing_data_ptr = std::unique_ptr<tracing_data>;
/// @relates tracing_data
CAF_CORE_EXPORT error inspect(serializer& sink, const tracing_data_ptr& x);
/// @relates tracing_data
CAF_CORE_EXPORT error_code<sec>
inspect(binary_serializer& sink, const tracing_data_ptr& x);
/// @relates tracing_data
CAF_CORE_EXPORT error inspect(deserializer& source, tracing_data_ptr& x);
/// @relates tracing_data
CAF_CORE_EXPORT error_code<sec>
inspect(binary_deserializer& source, tracing_data_ptr& x);
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#include "caf/fwd.hpp"
namespace caf {
/// Creates instances of @ref tracing_data.
class tracing_data_factory {
public:
virtual ~tracing_data_factory();
/// Deserializes tracing data from `source` and either overrides the content
/// of `dst` or allocates a new object if `dst` is null.
/// @returns the result of `source`.
virtual error deserialize(deserializer& source,
std::unique_ptr<tracing_data>& dst) const = 0;
/// @copydoc deserialize
virtual error_code<sec>
deserialize(binary_deserializer& source,
std::unique_ptr<tracing_data>& dst) const = 0;
};
} // namespace caf
...@@ -170,7 +170,7 @@ template <class Inspector> ...@@ -170,7 +170,7 @@ template <class Inspector>
typename Inspector::result_type typename Inspector::result_type
inspect(Inspector& f, upstream_msg::ack_batch& x) { inspect(Inspector& f, upstream_msg::ack_batch& x) {
return f(meta::type_name("ack_batch"), x.new_capacity, x.desired_batch_size, return f(meta::type_name("ack_batch"), x.new_capacity, x.desired_batch_size,
x.acknowledged_id); x.acknowledged_id, x.max_capacity);
} }
/// @relates upstream_msg::drop /// @relates upstream_msg::drop
......
...@@ -92,12 +92,12 @@ private: ...@@ -92,12 +92,12 @@ private:
if constexpr (std::is_empty<T>::value if constexpr (std::is_empty<T>::value
|| is_allowed_unsafe_message_type_v<T>) { || is_allowed_unsafe_message_type_v<T>) {
// skip element // skip element
} else if constexpr (detail::can_apply_v<Subtype, decltype(x)>) {
CAF_WRITE_INSPECTOR_TRY(dref.apply(x))
} else if constexpr (std::is_integral<T>::value) { } else if constexpr (std::is_integral<T>::value) {
using squashed_type = detail::squashed_int_t<T>; using squashed_type = detail::squashed_int_t<T>;
auto& squashed_x = reinterpret_cast<squashed_type&>(x); auto& squashed_x = reinterpret_cast<squashed_type&>(x);
CAF_WRITE_INSPECTOR_TRY(dref.apply(squashed_x)) CAF_WRITE_INSPECTOR_TRY(dref.apply(squashed_x))
} else if constexpr (detail::can_apply_v<Subtype, decltype(x)>) {
CAF_WRITE_INSPECTOR_TRY(dref.apply(x))
} else if constexpr (std::is_array<T>::value) { } else if constexpr (std::is_array<T>::value) {
std::make_index_sequence<std::extent<T>::value> seq; std::make_index_sequence<std::extent<T>::value> seq;
CAF_WRITE_INSPECTOR_TRY(apply_array(dref, x, seq)) CAF_WRITE_INSPECTOR_TRY(apply_array(dref, x, seq))
......
...@@ -226,7 +226,8 @@ actor_system::actor_system(actor_system_config& cfg) ...@@ -226,7 +226,8 @@ actor_system::actor_system(actor_system_config& cfg)
await_actors_before_shutdown_(true), await_actors_before_shutdown_(true),
detached_(0), detached_(0),
cfg_(cfg), cfg_(cfg),
logger_dtor_done_(false) { logger_dtor_done_(false),
tracing_context_(cfg.tracing_context) {
CAF_SET_LOGGER_SYS(this); CAF_SET_LOGGER_SYS(this);
for (auto& hook : cfg.thread_hooks_) for (auto& hook : cfg.thread_hooks_)
hook->init(*this); hook->init(*this);
......
...@@ -82,7 +82,9 @@ actor_system_config::actor_system_config() ...@@ -82,7 +82,9 @@ actor_system_config::actor_system_config()
.add<timespan>(stream_max_batch_delay, "max-batch-delay", .add<timespan>(stream_max_batch_delay, "max-batch-delay",
"maximum delay for partial batches") "maximum delay for partial batches")
.add<timespan>(stream_credit_round_interval, "credit-round-interval", .add<timespan>(stream_credit_round_interval, "credit-round-interval",
"time between emitting credit"); "time between emitting credit")
.add<atom_value>("credit-policy",
"selects an algorithm for credit computation");
opt_group{custom_options_, "scheduler"} opt_group{custom_options_, "scheduler"}
.add<atom_value>("policy", "'stealing' (default) or 'sharing'") .add<atom_value>("policy", "'stealing' (default) or 'sharing'")
.add<size_t>("max-threads", "maximum number of worker threads") .add<size_t>("max-threads", "maximum number of worker threads")
...@@ -158,6 +160,11 @@ settings actor_system_config::dump_content() const { ...@@ -158,6 +160,11 @@ settings actor_system_config::dump_content() const {
defaults::stream::max_batch_delay); defaults::stream::max_batch_delay);
put_missing(stream_group, "credit-round-interval", put_missing(stream_group, "credit-round-interval",
defaults::stream::credit_round_interval); defaults::stream::credit_round_interval);
put_missing(stream_group, "credit-policy", defaults::stream::credit_policy);
put_missing(stream_group, "size-policy.buffer-capacity",
defaults::stream::size_policy::buffer_capacity);
put_missing(stream_group, "size-policy.bytes-per-batch",
defaults::stream::size_policy::bytes_per_batch);
// -- scheduler parameters // -- scheduler parameters
auto& scheduler_group = result["scheduler"].as_dictionary(); auto& scheduler_group = result["scheduler"].as_dictionary();
put_missing(scheduler_group, "policy", defaults::scheduler::policy); put_missing(scheduler_group, "policy", defaults::scheduler::policy);
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/detail/complexity_based_credit_controller.hpp"
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/scheduled_actor.hpp"
// Safe us some typing and very ugly formatting.
#define impl complexity_based_credit_controller
namespace caf::detail {
impl::impl(scheduled_actor* self) : super(self) {
auto& cfg = self->system().config();
complexity_ = cfg.stream_desired_batch_complexity;
}
impl::~impl() {
// nop
}
void impl::before_processing(downstream_msg::batch& x) {
num_elements_ += x.xs_size;
processing_begin_ = make_timestamp();
}
void impl::after_processing(downstream_msg::batch&) {
processing_time_ += make_timestamp() - processing_begin_;
}
credit_controller::assignment impl::compute_initial() {
return {initial_buffer_size, initial_batch_size};
}
credit_controller::assignment
impl::compute(timespan cycle, int32_t downstream_capacity) {
// Max throughput = C * (N / t), where C = cycle length, N = measured items,
// and t = measured time. Desired batch size is the same formula with D
// (desired complexity) instead of C. We compute our values in 64-bit for
// more precision before truncating to a 32-bit integer type at the end.
int64_t total_ns = processing_time_.count();
if (total_ns == 0)
return {1, 1};
// Helper for truncating a 64-bit integer to a 32-bit integer with a minimum
// value of 1.
auto clamp = [](int64_t x) -> int32_t {
static constexpr auto upper_bound = std::numeric_limits<int32_t>::max();
if (x > upper_bound)
return upper_bound;
if (x <= 0)
return 1;
return static_cast<int32_t>(x);
};
// Instead of C * (N / t) we calculate (C * N) / t to avoid double conversion
// and rounding errors.
assignment result;
// Give enough credit to last 2 cycles, but don't exceed downstream capacity.
result.credit = 2 * clamp((cycle.count() * num_elements_) / total_ns);
result.credit = std::min(result.credit, downstream_capacity);
result.batch_size = clamp((complexity_.count() * num_elements_) / total_ns);
// Reset state and return.
num_elements_ = 0;
processing_time_ = timespan{0};
return result;
}
} // namespace caf::detail
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/credit_controller.hpp"
namespace caf {
credit_controller::credit_controller(scheduled_actor* self) : self_(self) {
// nop
}
credit_controller::~credit_controller() {
// nop
}
credit_controller::assignment credit_controller::compute_bridge() {
return {0, 0};
}
int32_t credit_controller::threshold() const noexcept {
return -1;
}
} // namespace caf
...@@ -49,6 +49,14 @@ namespace stream { ...@@ -49,6 +49,14 @@ namespace stream {
const timespan desired_batch_complexity = us(50); const timespan desired_batch_complexity = us(50);
const timespan max_batch_delay = ms(5); const timespan max_batch_delay = ms(5);
const timespan credit_round_interval = ms(10); const timespan credit_round_interval = ms(10);
const atom_value credit_policy = atom("complexity");
namespace size_policy {
const int32_t bytes_per_batch = 2048; // 2 KB
const int32_t buffer_capacity = 64 * 1024; // 64 KB
} // namespace size_policy
} // namespace stream } // namespace stream
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/detail/serialized_size.hpp"
#include <iomanip>
#include <sstream>
#include "caf/error.hpp"
#include "caf/string_view.hpp"
namespace caf::detail {
error serialized_size_inspector::begin_object(uint16_t nr, string_view name) {
if (nr != 0)
return apply(nr);
apply(nr);
return apply(name);
}
error serialized_size_inspector::end_object() {
return none;
}
error serialized_size_inspector::begin_sequence(size_t list_size) {
// Use varbyte encoding to compress sequence size on the wire.
// For 64-bit values, the encoded representation cannot get larger than 10
// bytes. A scratch space of 16 bytes suffices as upper bound.
uint8_t buf[16];
auto i = buf;
auto x = static_cast<uint32_t>(list_size);
while (x > 0x7f) {
*i++ = (static_cast<uint8_t>(x) & 0x7f) | 0x80;
x >>= 7;
}
*i++ = static_cast<uint8_t>(x) & 0x7f;
result_ += static_cast<size_t>(i - buf);
return none;
}
error serialized_size_inspector::end_sequence() {
return none;
}
error serialized_size_inspector::apply(bool) {
result_ += sizeof(uint8_t);
return none;
}
error serialized_size_inspector::apply(int8_t x) {
result_ += sizeof(x);
return none;
}
error serialized_size_inspector::apply(uint8_t x) {
result_ += sizeof(x);
return none;
}
error serialized_size_inspector::apply(int16_t x) {
result_ += sizeof(x);
return none;
}
error serialized_size_inspector::apply(uint16_t x) {
result_ += sizeof(x);
return none;
}
error serialized_size_inspector::apply(int32_t x) {
result_ += sizeof(x);
return none;
}
error serialized_size_inspector::apply(uint32_t x) {
result_ += sizeof(x);
return none;
}
error serialized_size_inspector::apply(int64_t x) {
result_ += sizeof(x);
return none;
}
error serialized_size_inspector::apply(uint64_t x) {
result_ += sizeof(x);
return none;
}
error serialized_size_inspector::apply(float x) {
result_ += sizeof(x);
return none;
}
error serialized_size_inspector::apply(double x) {
result_ += sizeof(x);
return none;
}
error serialized_size_inspector::apply(long double x) {
// The IEEE-754 conversion does not work for long double
// => fall back to string serialization (event though it sucks).
std::ostringstream oss;
oss << std::setprecision(std::numeric_limits<long double>::digits) << x;
auto tmp = oss.str();
return apply(tmp);
}
error serialized_size_inspector::apply(string_view x) {
begin_sequence(x.size());
result_ += x.size();
return end_sequence();
}
error serialized_size_inspector::apply(const std::u16string& x) {
begin_sequence(x.size());
result_ += x.size() * sizeof(uint16_t);
return end_sequence();
}
error serialized_size_inspector::apply(const std::u32string& x) {
begin_sequence(x.size());
result_ += x.size() * sizeof(uint32_t);
return end_sequence();
}
error serialized_size_inspector::apply(span<const byte> x) {
result_ += x.size();
return none;
}
error serialized_size_inspector::apply(const std::vector<bool>& xs) {
begin_sequence(xs.size());
result_ += (xs.size() + static_cast<size_t>(xs.size() % 8 != 0)) / 8;
return end_sequence();
}
} // namespace caf::detail
This diff is collapsed.
...@@ -100,7 +100,8 @@ bool node_id::default_data::valid() const noexcept { ...@@ -100,7 +100,8 @@ bool node_id::default_data::valid() const noexcept {
size_t node_id::default_data::hash_code() const noexcept { size_t node_id::default_data::hash_code() const noexcept {
// XOR the first few bytes from the node ID and the process ID. // XOR the first few bytes from the node ID and the process ID.
auto x = static_cast<size_t>(pid_); auto x = static_cast<size_t>(pid_);
auto y = *reinterpret_cast<const size_t*>(host_.data()); size_t y;
memcpy(&y, host_.data(), sizeof(size_t));
return x ^ y; return x ^ y;
} }
......
...@@ -1158,12 +1158,13 @@ scheduled_actor::advance_streams(actor_clock::time_point now) { ...@@ -1158,12 +1158,13 @@ scheduled_actor::advance_streams(actor_clock::time_point now) {
CAF_LOG_DEBUG("new credit round"); CAF_LOG_DEBUG("new credit round");
auto cycle = stream_ticks_.interval(); auto cycle = stream_ticks_.interval();
cycle *= static_cast<decltype(cycle)::rep>(credit_round_ticks_); cycle *= static_cast<decltype(cycle)::rep>(credit_round_ticks_);
auto bc = home_system().config().stream_desired_batch_complexity;
auto& qs = get_downstream_queue().queues(); auto& qs = get_downstream_queue().queues();
for (auto& kvp : qs) { for (auto& kvp : qs) {
auto inptr = kvp.second.policy().handler.get(); auto inptr = kvp.second.policy().handler.get();
auto bs = static_cast<int32_t>(kvp.second.total_task_size()); if (inptr != nullptr) {
inptr->emit_ack_batch(this, bs, now, cycle, bc); auto tts = static_cast<int32_t>(kvp.second.total_task_size());
inptr->emit_ack_batch(this, tts, now, cycle);
}
} }
} }
return stream_ticks_.next_timeout(now, {max_batch_delay_ticks_, return stream_ticks_.next_timeout(now, {max_batch_delay_ticks_,
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/detail/size_based_credit_controller.hpp"
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/defaults.hpp"
#include "caf/detail/serialized_size.hpp"
#include "caf/scheduled_actor.hpp"
// Safe us some typing and very ugly formatting.
#define impl size_based_credit_controller
namespace caf::detail {
impl::impl(scheduled_actor* self) : super(self) {
auto& cfg = self->system().config();
buffer_capacity_ = get_or(cfg, "stream.size-policy.buffer-capacity",
defaults::stream::size_policy::buffer_capacity);
bytes_per_batch_ = get_or(cfg, "stream.size-policy.bytes-per-batch",
defaults::stream::size_policy::bytes_per_batch);
}
impl::~impl() {
// nop
}
void impl::before_processing(downstream_msg::batch& x) {
if (++sample_counter_ == sample_rate_) {
sampled_elements_ += x.xs_size;
sampled_total_size_ += serialized_size(self()->system(), x.xs);
sample_counter_ = 0;
}
++num_batches_;
}
void impl::after_processing(downstream_msg::batch&) {
// nop
}
credit_controller::assignment impl::compute_initial() {
return {buffer_size_, batch_size_};
}
credit_controller::assignment impl::compute(timespan, int32_t) {
if (sampled_elements_ >= min_samples) {
// Helper for truncating a 64-bit integer to a 32-bit integer with a
// minimum value of 1.
auto clamp_i32 = [](int64_t x) -> int32_t {
static constexpr auto upper_bound = std::numeric_limits<int32_t>::max();
if (x > upper_bound)
return upper_bound;
if (x <= 0)
return 1;
return static_cast<int32_t>(x);
};
// Calculate ideal batch size by size.
auto bytes_per_element = clamp_i32(sampled_total_size_ / sampled_elements_);
batch_size_ = clamp_i32(bytes_per_batch_ / bytes_per_element);
buffer_size_ = clamp_i32(buffer_capacity_ / bytes_per_element);
// Reset bookkeeping state.
sampled_elements_ = 0;
sampled_total_size_ = 0;
// Adjust the sample rate to reach min_samples in the next cycle.
sample_rate_ = clamp_i32(num_batches_ / min_samples);
if (sample_counter_ >= sample_rate_)
sample_counter_ = 0;
num_batches_ = 0;
}
return {buffer_size_, batch_size_};
}
credit_controller::assignment impl::compute_bridge() {
CAF_ASSERT(batch_size_ > 0);
CAF_ASSERT(buffer_size_ > batch_size_);
return {buffer_size_, batch_size_};
}
int32_t impl::threshold() const noexcept {
return static_cast<int32_t>(buffer_size_ * buffer_threshold);
}
} // namespace caf::detail
...@@ -155,7 +155,6 @@ void stream_manager::advance() { ...@@ -155,7 +155,6 @@ void stream_manager::advance() {
if (!inbound_paths_.empty()) { if (!inbound_paths_.empty()) {
auto now = self_->clock().now(); auto now = self_->clock().now();
auto& cfg = self_->system().config(); auto& cfg = self_->system().config();
auto bc = cfg.stream_desired_batch_complexity;
auto interval = cfg.stream_credit_round_interval; auto interval = cfg.stream_credit_round_interval;
auto& qs = self_->get_downstream_queue().queues(); auto& qs = self_->get_downstream_queue().queues();
// Iterate all queues for inbound traffic. // Iterate all queues for inbound traffic.
...@@ -163,8 +162,8 @@ void stream_manager::advance() { ...@@ -163,8 +162,8 @@ void stream_manager::advance() {
auto inptr = kvp.second.policy().handler.get(); auto inptr = kvp.second.policy().handler.get();
// Ignore inbound paths of other managers. // Ignore inbound paths of other managers.
if (inptr->mgr.get() == this) { if (inptr->mgr.get() == this) {
auto bs = static_cast<int32_t>(kvp.second.total_task_size()); auto tts = static_cast<int32_t>(kvp.second.total_task_size());
inptr->emit_ack_batch(self_, bs, now, interval, bc); inptr->emit_ack_batch(self_, tts, now, interval);
} }
} }
} }
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/detail/test_credit_controller.hpp"
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/scheduled_actor.hpp"
namespace caf {
namespace detail {
test_credit_controller::~test_credit_controller() {
// nop
}
void test_credit_controller::before_processing(downstream_msg::batch& x) {
num_elements_ += x.xs_size;
}
void test_credit_controller::after_processing(downstream_msg::batch&) {
// nop
}
credit_controller::assignment test_credit_controller::compute_initial() {
return {50, 50};
}
credit_controller::assignment
test_credit_controller::compute(timespan cycle, int32_t) {
auto& cfg = self()->system().config();
auto complexity = cfg.stream_desired_batch_complexity;
// Max throughput = C * (N / t), where C = cycle length, N = measured items,
// and t = measured time. Desired batch size is the same formula with D
// (desired complexity) instead of C. We compute our values in 64-bit for
// more precision before truncating to a 32-bit integer type at the end.
int64_t total_ns = num_elements_ * 1000; // calculate with 1us per element
if (total_ns == 0)
return {1, 1};
// Helper for truncating a 64-bit integer to a 32-bit integer with a minimum
// value of 1.
auto clamp = [](int64_t x) -> int32_t {
static constexpr auto upper_bound = std::numeric_limits<int32_t>::max();
if (x > upper_bound)
return upper_bound;
if (x <= 0)
return 1;
return static_cast<int32_t>(x);
};
// Instead of C * (N / t) we calculate (C * N) / t to avoid double conversion
// and rounding errors.
assignment result;
result.credit = clamp((cycle.count() * num_elements_) / total_ns);
result.batch_size = clamp((complexity.count() * num_elements_) / total_ns);
// Reset state and return.
num_elements_ = 0;
return result;
}
} // namespace detail
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/tracing_data.hpp"
#include <cstdint>
#include "caf/actor_system.hpp"
#include "caf/deserializer.hpp"
#include "caf/error.hpp"
#include "caf/logger.hpp"
#include "caf/sec.hpp"
#include "caf/serializer.hpp"
#include "caf/tracing_data_factory.hpp"
namespace caf {
tracing_data::~tracing_data() {
// nop
}
namespace {
template <class Serializer>
auto inspect_impl(Serializer& sink, const tracing_data_ptr& x) {
if (x == nullptr) {
uint8_t dummy = 0;
return sink(dummy);
}
uint8_t dummy = 1;
if (auto err = sink(dummy))
return err;
return x->serialize(sink);
}
template <class Deserializer>
typename Deserializer::result_type
inspect_impl(Deserializer& source, tracing_data_ptr& x) {
uint8_t dummy = 0;
if (auto err = source(dummy))
return err;
if (dummy == 0) {
x.reset();
return {};
}
auto ctx = source.context();
if (ctx == nullptr)
return sec::no_context;
auto tc = ctx->system().tracing_context();
if (tc == nullptr)
return sec::no_tracing_context;
return tc->deserialize(source, x);
}
} // namespace
error inspect(serializer& sink, const tracing_data_ptr& x) {
return inspect_impl(sink, x);
}
error_code<sec> inspect(binary_serializer& sink, const tracing_data_ptr& x) {
return inspect_impl(sink, x);
}
error inspect(deserializer& source, tracing_data_ptr& x) {
return inspect_impl(source, x);
}
error_code<sec> inspect(binary_deserializer& source, tracing_data_ptr& x) {
return inspect_impl(source, x);
}
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/tracing_data_factory.hpp"
namespace caf {
tracing_data_factory::~tracing_data_factory() {
// nop
}
} // namespace caf
...@@ -5,7 +5,7 @@ ...@@ -5,7 +5,7 @@
* | |___ / ___ \| _| Framework * * | |___ / ___ \| _| Framework *
* \____/_/ \_|_| * * \____/_/ \_|_| *
* * * *
* Copyright 2011-2018 Dominik Charousset * * Copyright 2011-2019 Dominik Charousset *
* * * *
* Distributed under the terms and conditions of the BSD 3-Clause License or * * Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software * * (at your option) under the terms and conditions of the Boost Software *
...@@ -16,75 +16,65 @@ ...@@ -16,75 +16,65 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#include "caf/config.hpp" #define CAF_SUITE detail.serialized_size
#define CAF_SUITE inbound_path #include "caf/detail/serialized_size.hpp"
#include "caf/test/unit_test.hpp"
#include <string> #include "caf/test/dsl.hpp"
#include "caf/inbound_path.hpp" #include <vector>
using namespace std; #include "caf/binary_serializer.hpp"
using namespace caf; #include "caf/byte.hpp"
#include "caf/byte_buffer.hpp"
namespace { using namespace caf;
#define PRINT(format, ...) \ using caf::detail::serialized_size;
{ \
char buf[200]; \
snprintf(buf, 200, format, __VA_ARGS__); \
CAF_MESSAGE(buf); \
}
struct fixture { namespace {
inbound_path::stats_t x;
void calculate(int32_t total_items, int32_t total_time) {
int32_t c = 1000;
int32_t d = 100;
int32_t n = total_items;
int32_t t = total_time;
int32_t m = t > 0 ? std::max((c * n) / t, 1) : 1;
int32_t b = t > 0 ? std::max((d * n) / t, 1) : 1;
PRINT("with a cycle C = %dns, desied complexity D = %d,", c, d);
PRINT("number of items N = %d, and time delta t = %d:", n, t);
PRINT("- throughput M = max(C * N / t, 1) = max(%d * %d / %d, 1) = %d",
c, n, t, m);
PRINT("- items/batch B = max(D * N / t, 1) = max(%d * %d / %d, 1) = %d",
d, n, t, b);
auto cr = x.calculate(timespan(c), timespan(d));
CAF_CHECK_EQUAL(cr.items_per_batch, b);
CAF_CHECK_EQUAL(cr.max_throughput, m);
}
void store(int32_t batch_size, int32_t calculation_time_ns) { struct fixture : test_coordinator_fixture<> {
inbound_path::stats_t::measurement m{batch_size, template <class... Ts>
timespan{calculation_time_ns}}; size_t actual_size(const Ts&... xs) {
x.store(m); byte_buffer buf;
binary_serializer sink{sys, buf};
if (auto err = sink(xs...))
CAF_FAIL("failed to serialize data: " << sys.render(err));
return buf.size();
} }
}; };
} // namespace } // namespace
CAF_TEST_FIXTURE_SCOPE(inbound_path_tests, fixture) #define CHECK_SAME_SIZE(...) \
CAF_CHECK_EQUAL(serialized_size(sys, __VA_ARGS__), actual_size(__VA_ARGS__))
CAF_TEST(default_constructed) {
calculate(0, 0); CAF_TEST_FIXTURE_SCOPE(serialized_size_tests, fixture)
CAF_TEST(numbers) {
CHECK_SAME_SIZE(int8_t{42});
CHECK_SAME_SIZE(int16_t{42});
CHECK_SAME_SIZE(int32_t{42});
CHECK_SAME_SIZE(int64_t{42});
CHECK_SAME_SIZE(uint8_t{42});
CHECK_SAME_SIZE(uint16_t{42});
CHECK_SAME_SIZE(uint32_t{42});
CHECK_SAME_SIZE(uint64_t{42});
CHECK_SAME_SIZE(4.2f);
CHECK_SAME_SIZE(4.2);
} }
CAF_TEST(one_store) { CAF_TEST(containers) {
CAF_MESSAGE("store a measurement for 500ns with batch size of 50"); CHECK_SAME_SIZE(std::string{"foobar"});
store(50, 500); CHECK_SAME_SIZE(std::vector<char>({'a', 'b', 'c'}));
calculate(50, 500); CHECK_SAME_SIZE(std::vector<std::string>({"hello", "world"}));
} }
CAF_TEST(multiple_stores) { CAF_TEST(messages) {
CAF_MESSAGE("store measurements: (50, 500ns), (60, 400ns), (40, 600ns)"); CHECK_SAME_SIZE(make_message(42));
store(50, 500); CHECK_SAME_SIZE(make_message(1, 2, 3));
store(40, 600); CHECK_SAME_SIZE(make_message("hello", "world"));
store(60, 400);
calculate(150, 1500);
} }
CAF_TEST_FIXTURE_SCOPE_END() CAF_TEST_FIXTURE_SCOPE_END()
...@@ -87,7 +87,6 @@ CAF_TEST(default_constructed) { ...@@ -87,7 +87,6 @@ CAF_TEST(default_constructed) {
CAF_REQUIRE_EQUAL(queue.peek(), nullptr); CAF_REQUIRE_EQUAL(queue.peek(), nullptr);
CAF_REQUIRE_EQUAL(queue.next(), nullptr); CAF_REQUIRE_EQUAL(queue.next(), nullptr);
CAF_REQUIRE_EQUAL(queue.begin(), queue.end()); CAF_REQUIRE_EQUAL(queue.begin(), queue.end());
CAF_REQUIRE_EQUAL(queue.before_begin()->next, queue.end().ptr);
} }
CAF_TEST(inc_deficit) { CAF_TEST(inc_deficit) {
......
...@@ -83,7 +83,6 @@ CAF_TEST(default_constructed) { ...@@ -83,7 +83,6 @@ CAF_TEST(default_constructed) {
CAF_REQUIRE_EQUAL(queue.total_task_size(), 0); CAF_REQUIRE_EQUAL(queue.total_task_size(), 0);
CAF_REQUIRE_EQUAL(queue.peek(), nullptr); CAF_REQUIRE_EQUAL(queue.peek(), nullptr);
CAF_REQUIRE_EQUAL(queue.begin(), queue.end()); CAF_REQUIRE_EQUAL(queue.begin(), queue.end());
CAF_REQUIRE_EQUAL(queue.before_begin()->next, queue.end().ptr);
} }
CAF_TEST(push_back) { CAF_TEST(push_back) {
......
...@@ -329,7 +329,6 @@ public: ...@@ -329,7 +329,6 @@ public:
void advance_time() { void advance_time() {
auto cycle = std::chrono::milliseconds(100); auto cycle = std::chrono::milliseconds(100);
auto desired_batch_complexity = std::chrono::microseconds(50);
auto f = [&](tick_type x) { auto f = [&](tick_type x) {
if (x % ticks_per_force_batches_interval == 0) { if (x % ticks_per_force_batches_interval == 0) {
// Force batches on all output paths. // Force batches on all output paths.
...@@ -341,9 +340,8 @@ public: ...@@ -341,9 +340,8 @@ public:
auto& qs = get<dmsg_id::value>(mbox.queues()).queues(); auto& qs = get<dmsg_id::value>(mbox.queues()).queues();
for (auto& kvp : qs) { for (auto& kvp : qs) {
auto inptr = kvp.second.policy().handler.get(); auto inptr = kvp.second.policy().handler.get();
auto bs = static_cast<int32_t>(kvp.second.total_task_size()); auto tts = static_cast<int32_t>(kvp.second.total_task_size());
inptr->emit_ack_batch(this, bs, now(), cycle, inptr->emit_ack_batch(this, tts, now(), cycle);
desired_batch_complexity);
} }
} }
}; };
......
...@@ -114,4 +114,10 @@ CAF_TEST(make_span) { ...@@ -114,4 +114,10 @@ CAF_TEST(make_span) {
CAF_CHECK_EQUAL(begin(ys), begin(zs)); CAF_CHECK_EQUAL(begin(ys), begin(zs));
} }
CAF_TEST(spans are convertible from compatible containers) {
std::vector<int> xs{1, 2, 3};
span<const int> ys{xs};
CAF_CHECK(std::equal(xs.begin(), xs.end(), ys.begin()));
}
CAF_TEST_FIXTURE_SCOPE_END() CAF_TEST_FIXTURE_SCOPE_END()
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#define CAF_SUITE tracing_data
#include "caf/tracing_data.hpp"
#include "caf/test/dsl.hpp"
#include <vector>
#include "caf/actor_profiler.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/config.hpp"
#include "caf/tracing_data_factory.hpp"
#ifdef CAF_ENABLE_ACTOR_PROFILER
using std::string;
using namespace caf;
namespace {
class dummy_tracing_data : public tracing_data {
public:
string value;
dummy_tracing_data(string value) : value(std::move(value)) {
// nop
}
error serialize(serializer& sink) const override {
return sink(value);
}
error_code<sec> serialize(binary_serializer& sink) const override {
return sink(value);
}
};
class dummy_tracing_data_factory : public tracing_data_factory {
public:
error deserialize(deserializer& source,
std::unique_ptr<tracing_data>& dst) const override {
return deserialize_impl(source, dst);
}
error_code<sec>
deserialize(binary_deserializer& source,
std::unique_ptr<tracing_data>& dst) const override {
return deserialize_impl(source, dst);
}
private:
template <class Deserializer>
typename Deserializer::result_type
deserialize_impl(Deserializer& source,
std::unique_ptr<tracing_data>& dst) const {
string value;
if (auto err = source(value))
return err;
dst.reset(new dummy_tracing_data(std::move(value)));
return {};
}
};
class dummy_profiler : public actor_profiler {
public:
void add_actor(const local_actor&, const local_actor*) override {
// nop
}
void remove_actor(const local_actor&) override {
// nop
}
void before_processing(const local_actor&, const mailbox_element&) override {
// nop
}
void after_processing(const local_actor&, invoke_message_result) override {
// nop
}
void
before_sending(const local_actor& self, mailbox_element& element) override {
element.tracing_id.reset(new dummy_tracing_data(self.name()));
}
void
before_sending_scheduled(const local_actor& self, actor_clock::time_point,
mailbox_element& element) override {
element.tracing_id.reset(new dummy_tracing_data(self.name()));
}
};
actor_system_config& init(actor_system_config& cfg, actor_profiler& profiler,
tracing_data_factory& factory) {
test_coordinator_fixture<>::init_config(cfg);
cfg.profiler = &profiler;
cfg.tracing_context = &factory;
return cfg;
}
struct fixture {
using scheduler_type = caf::scheduler::test_coordinator;
fixture()
: sys(init(cfg, profiler, factory)),
sched(dynamic_cast<scheduler_type&>(sys.scheduler())) {
run();
}
void run() {
sched.run();
}
dummy_profiler profiler;
dummy_tracing_data_factory factory;
actor_system_config cfg;
actor_system sys;
scheduler_type& sched;
};
const std::string& tracing_id(local_actor* self) {
auto element = self->current_mailbox_element();
if (element == nullptr)
CAF_FAIL("current_mailbox_element == null");
auto tid = element->tracing_id.get();
if (tid == nullptr)
CAF_FAIL("tracing_id == null");
auto dummy_tid = dynamic_cast<dummy_tracing_data*>(tid);
if (dummy_tid == nullptr)
CAF_FAIL("dummy_tracing_id == null");
return dummy_tid->value;
}
# define NAMED_ACTOR_STATE(type) \
struct type##_state { \
const char* name = #type; \
}
NAMED_ACTOR_STATE(alice);
NAMED_ACTOR_STATE(bob);
NAMED_ACTOR_STATE(carl);
} // namespace
CAF_TEST_FIXTURE_SCOPE(actor_profiler_tests, fixture)
CAF_TEST(profilers inject tracing data into asynchronous messages) {
CAF_MESSAGE("spawn a foo and a bar");
auto carl_fun = [](stateful_actor<carl_state>* self) -> behavior {
return {
[=](const string& str) {
CAF_CHECK_EQUAL(str, "hello carl");
CAF_CHECK_EQUAL(tracing_id(self), "bob");
},
};
};
auto bob_fun = [](stateful_actor<bob_state>* self, actor carl) -> behavior {
return {
[=](const string& str) {
CAF_CHECK_EQUAL(str, "hello bob");
CAF_CHECK_EQUAL(tracing_id(self), "alice");
self->send(carl, "hello carl");
},
};
};
auto alice_fun = [](stateful_actor<alice_state>* self, actor bob) {
self->send(bob, "hello bob");
};
sys.spawn(alice_fun, sys.spawn(bob_fun, sys.spawn(carl_fun)));
run();
}
CAF_TEST(tracing data is serializable) {
byte_buffer buf;
binary_serializer sink{sys, buf};
tracing_data_ptr data;
tracing_data_ptr copy;
data.reset(new dummy_tracing_data("iTrace"));
CAF_CHECK_EQUAL(sink(data), none);
binary_deserializer source{sys, buf};
CAF_CHECK_EQUAL(source(copy), none);
CAF_REQUIRE_NOT_EQUAL(copy.get(), nullptr);
CAF_CHECK_EQUAL(dynamic_cast<dummy_tracing_data&>(*copy).value, "iTrace");
}
CAF_TEST_FIXTURE_SCOPE_END()
#endif // CAF_ENABLE_ACTOR_PROFILER
...@@ -693,6 +693,7 @@ public: ...@@ -693,6 +693,7 @@ public:
cfg.set("middleman.network-backend", caf::atom("testing")); cfg.set("middleman.network-backend", caf::atom("testing"));
cfg.set("middleman.manual-multiplexing", true); cfg.set("middleman.manual-multiplexing", true);
cfg.set("middleman.workers", size_t{0}); cfg.set("middleman.workers", size_t{0});
cfg.set("stream.credit-policy", caf::atom("testing"));
return cfg; 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