Commit 11298e99 authored by Dominik Charousset's avatar Dominik Charousset

Merge branch 'topic/streaming'

parents ec0bc837 9828d2b9
...@@ -73,6 +73,8 @@ Usage: $0 [OPTION]... [VAR=VALUE]... ...@@ -73,6 +73,8 @@ Usage: $0 [OPTION]... [VAR=VALUE]...
--no-io build without I/O module --no-io build without I/O module
--no-python build without python binding --no-python build without python binding
--no-summary do not print configuration before building --no-summary do not print configuration before building
--core-only sets no-examples, no-opencl, no-openssl,
no-tools, no-python, and no-io
Testing: Testing:
--with-asio use ASIO multiplexer in unit tests --with-asio use ASIO multiplexer in unit tests
...@@ -86,6 +88,7 @@ Usage: $0 [OPTION]... [VAR=VALUE]... ...@@ -86,6 +88,7 @@ Usage: $0 [OPTION]... [VAR=VALUE]...
- DEBUG - DEBUG
- TRACE - TRACE
--with-address-sanitizer build with address sanitizer if available --with-address-sanitizer build with address sanitizer if available
--with-asan alias for --with-address-sanitier
--with-gcov build with gcov coverage enabled --with-gcov build with gcov coverage enabled
Influential Environment Variables (only on first invocation): Influential Environment Variables (only on first invocation):
...@@ -242,6 +245,9 @@ while [ $# -ne 0 ]; do ...@@ -242,6 +245,9 @@ while [ $# -ne 0 ]; do
--with-address-sanitizer) --with-address-sanitizer)
append_cache_entry CAF_ENABLE_ADDRESS_SANITIZER BOOL yes append_cache_entry CAF_ENABLE_ADDRESS_SANITIZER BOOL yes
;; ;;
--with-asan)
append_cache_entry CAF_ENABLE_ADDRESS_SANITIZER BOOL yes
;;
--with-gcov) --with-gcov)
append_cache_entry CAF_ENABLE_GCOV BOOL yes append_cache_entry CAF_ENABLE_GCOV BOOL yes
;; ;;
...@@ -397,6 +403,14 @@ while [ $# -ne 0 ]; do ...@@ -397,6 +403,14 @@ while [ $# -ne 0 ]; do
--no-summary) --no-summary)
append_cache_entry CAF_NO_SUMMARY BOOL yes append_cache_entry CAF_NO_SUMMARY BOOL yes
;; ;;
--core-only)
append_cache_entry CAF_NO_TOOLS BOOL yes
append_cache_entry CAF_NO_IO BOOL yes
append_cache_entry CAF_NO_PYTHON BOOL yes
append_cache_entry CAF_NO_EXAMPLES BOOL yes
append_cache_entry CAF_NO_OPENCL BOOL yes
append_cache_entry CAF_NO_OPENSSL BOOL yes
;;
*) *)
echo "Invalid option '$1'. Try $0 --help to see available options." echo "Invalid option '$1'. Try $0 --help to see available options."
exit 1 exit 1
......
...@@ -64,34 +64,33 @@ chopstick::behavior_type taken_chopstick(chopstick::pointer self, ...@@ -64,34 +64,33 @@ chopstick::behavior_type taken_chopstick(chopstick::pointer self,
}; };
} }
/* Based on: http://www.dalnefre.com/wp/2010/08/dining-philosophers-in-humus/ // Based on: http://www.dalnefre.com/wp/2010/08/dining-philosophers-in-humus/
* //
* //
* +-------------+ {busy|taken} // +-------------+ {busy|taken}
* /-------->| thinking |<------------------\ // /-------->| thinking |<------------------\
* | +-------------+ | // | +-------------+ |
* | | | // | | |
* | | {eat} | // | | {eat} |
* | | | // | | |
* | V | // | V |
* | +-------------+ {busy} +-------------+ // | +-------------+ {busy} +-------------+
* | | hungry |----------->| denied | // | | hungry |----------->| denied |
* | +-------------+ +-------------+ // | +-------------+ +-------------+
* | | // | |
* | | {taken} // | | {taken}
* | | // | |
* | V // | V
* | +-------------+ // | +-------------+
* | | granted | // | | granted |
* | +-------------+ // | +-------------+
* | | | // | | |
* | {busy} | | {taken} // | {busy} | | {taken}
* \-----------/ | // \-----------/ |
* | V // | V
* | {think} +-------------+ // | {think} +-------------+
* \---------| eating | // \---------| eating |
* +-------------+ // +-------------+
*/
class philosopher : public event_based_actor { class philosopher : public event_based_actor {
public: public:
......
...@@ -48,31 +48,30 @@ behavior calculator_fun() { ...@@ -48,31 +48,30 @@ behavior calculator_fun() {
}; };
} }
/* State transition of the client for connecting to the server: // State transition of the client for connecting to the server:
* //
* +-------------+ // +-------------+
* | init | // | init |
* +-------------+ // +-------------+
* | // |
* V // V
* +-------------+ // +-------------+
* | unconnected |<------------------\ // | unconnected |<------------------\
* +-------------+ | // +-------------+ |
* | | // | |
* | {connect Host Port} | // | {connect Host Port} |
* | | // | |
* V | // V |
* +-------------+ {error} | // +-------------+ {error} |
* /-------------->| connecting |------------------>| // /-------------->| connecting |------------------>|
* | +-------------+ ^ // | +-------------+ ^
* | | | // | | |
* | | {ok, Calculator} | // | | {ok, Calculator} |
* |{connect Host Port} | | // |{connect Host Port} | |
* | V | // | V |
* | +-------------+ {DOWN server} | // | +-------------+ {DOWN server} |
* \---------------| running |-------------------/ // \---------------| running |-------------------/
* +-------------+ // +-------------+
*/
namespace client { namespace client {
......
...@@ -40,6 +40,7 @@ set (LIBCAF_CORE_SRCS ...@@ -40,6 +40,7 @@ set (LIBCAF_CORE_SRCS
src/decorated_tuple.cpp src/decorated_tuple.cpp
src/default_attachable.cpp src/default_attachable.cpp
src/deserializer.cpp src/deserializer.cpp
src/downstream_messages.cpp
src/duration.cpp src/duration.cpp
src/dynamic_message_data.cpp src/dynamic_message_data.cpp
src/error.cpp src/error.cpp
...@@ -54,8 +55,6 @@ set (LIBCAF_CORE_SRCS ...@@ -54,8 +55,6 @@ set (LIBCAF_CORE_SRCS
src/group_manager.cpp src/group_manager.cpp
src/group_module.cpp src/group_module.cpp
src/inbound_path.cpp src/inbound_path.cpp
src/invalid_stream_gatherer.cpp
src/invalid_stream_scatterer.cpp
src/invoke_result_visitor.cpp src/invoke_result_visitor.cpp
src/local_actor.cpp src/local_actor.cpp
src/logger.cpp src/logger.cpp
...@@ -75,8 +74,6 @@ set (LIBCAF_CORE_SRCS ...@@ -75,8 +74,6 @@ set (LIBCAF_CORE_SRCS
src/pretty_type_name.cpp src/pretty_type_name.cpp
src/private_thread.cpp src/private_thread.cpp
src/proxy_registry.cpp src/proxy_registry.cpp
src/pull5_gatherer.cpp
src/random_gatherer.cpp
src/raw_event_based_actor.cpp src/raw_event_based_actor.cpp
src/ref_counted.cpp src/ref_counted.cpp
src/replies_to.cpp src/replies_to.cpp
...@@ -95,22 +92,17 @@ set (LIBCAF_CORE_SRCS ...@@ -95,22 +92,17 @@ set (LIBCAF_CORE_SRCS
src/splitter.cpp src/splitter.cpp
src/stream.cpp src/stream.cpp
src/stream_aborter.cpp src/stream_aborter.cpp
src/stream_gatherer.cpp
src/stream_gatherer.cpp
src/stream_gatherer_impl.cpp
src/stream_id.cpp
src/stream_manager.cpp src/stream_manager.cpp
src/stream_msg_visitor.cpp
src/stream_priority.cpp src/stream_priority.cpp
src/stream_scatterer.cpp src/downstream_manager.cpp
src/stream_scatterer_impl.cpp src/downstream_manager_base.cpp
src/stringification_inspector.cpp src/stringification_inspector.cpp
src/sync_request_bouncer.cpp src/sync_request_bouncer.cpp
src/term.cpp src/term.cpp
src/terminal_stream_scatterer.cpp
src/test_actor_clock.cpp src/test_actor_clock.cpp
src/test_coordinator.cpp src/test_coordinator.cpp
src/thread_safe_actor_clock.cpp src/thread_safe_actor_clock.cpp
src/tick_emitter.cpp
src/timestamp.cpp src/timestamp.cpp
src/try_match.cpp src/try_match.cpp
src/type_erased_tuple.cpp src/type_erased_tuple.cpp
......
...@@ -121,18 +121,17 @@ public: ...@@ -121,18 +121,17 @@ public:
ctx); ctx);
} }
// flags storing runtime information used by ... // flags storing runtime information used by ...
static constexpr int has_timeout_flag = 0x0004; // single_timeout static constexpr int has_timeout_flag = 0x0004; // single_timeout
static constexpr int is_registered_flag = 0x0008; // (several actors) static constexpr int is_registered_flag = 0x0008; // (several actors)
static constexpr int is_initialized_flag = 0x0010; // event-based actors static constexpr int is_initialized_flag = 0x0010; // event-based actors
static constexpr int is_blocking_flag = 0x0020; // blocking_actor static constexpr int is_blocking_flag = 0x0020; // blocking_actor
static constexpr int is_detached_flag = 0x0040; // local_actor static constexpr int is_detached_flag = 0x0040; // local_actor
static constexpr int is_priority_aware_flag = 0x0080; // local_actor static constexpr int is_serializable_flag = 0x0100; // local_actor
static constexpr int is_serializable_flag = 0x0100; // local_actor static constexpr int is_migrated_from_flag = 0x0200; // local_actor
static constexpr int is_migrated_from_flag = 0x0200; // local_actor static constexpr int has_used_aout_flag = 0x0400; // local_actor
static constexpr int has_used_aout_flag = 0x0400; // local_actor static constexpr int is_terminated_flag = 0x0800; // local_actor
static constexpr int is_terminated_flag = 0x0800; // local_actor static constexpr int is_cleaned_up_flag = 0x1000; // monitorable_actor
static constexpr int is_cleaned_up_flag = 0x1000; // monitorable_actor
inline void setf(int flag) { inline void setf(int flag) {
auto x = flags(); auto x = flags();
......
...@@ -49,21 +49,23 @@ public: ...@@ -49,21 +49,23 @@ public:
virtual time_point now() const noexcept; virtual time_point now() const noexcept;
/// Returns the difference between `t0` and `t1`, allowing the clock to /// Returns the difference between `t0` and `t1`, allowing the clock to
/// return any arbitrary value depending on the measurement that took place. /// return an arbitrary value depending on the measurement that took place
virtual duration_type difference(atom_value measurement, time_point t0, /// and the units measured.
time_point t1) const noexcept; virtual duration_type difference(atom_value measurement, long units,
time_point t0, time_point t1) const noexcept;
/// Schedules a `timeout_msg` for `self` at time point `t`, overriding any /// Schedules a `timeout_msg` for `self` at time point `t`, overriding any
/// previous receive timeout. /// previous receive timeout.
virtual void set_receive_timeout(time_point t, abstract_actor* self, virtual void set_ordinary_timeout(time_point t, abstract_actor* self,
uint32_t id) = 0; atom_value type, uint64_t id) = 0;
/// Schedules a `sec::request_timeout` for `self` at time point `t`. /// Schedules a `sec::request_timeout` for `self` at time point `t`.
virtual void set_request_timeout(time_point t, abstract_actor* self, virtual void set_request_timeout(time_point t, abstract_actor* self,
message_id id) = 0; message_id id) = 0;
/// Cancels a pending receive timeout. /// Cancels a pending receive timeout.
virtual void cancel_receive_timeout(abstract_actor* self) = 0; virtual void cancel_ordinary_timeout(abstract_actor* self,
atom_value type) = 0;
/// Cancels the pending request timeout for `id`. /// Cancels the pending request timeout for `id`.
virtual void cancel_request_timeout(abstract_actor* self, message_id id) = 0; virtual void cancel_request_timeout(abstract_actor* self, message_id id) = 0;
...@@ -78,6 +80,9 @@ public: ...@@ -78,6 +80,9 @@ public:
/// Schedules an arbitrary message to `target` for time point `t`. /// Schedules an arbitrary message to `target` for time point `t`.
virtual void schedule_message(time_point t, group target, virtual void schedule_message(time_point t, group target,
strong_actor_ptr sender, message content) = 0; strong_actor_ptr sender, message content) = 0;
/// Cancels all timeouts and scheduled messages.
virtual void cancel_all() = 0;
}; };
} // namespace caf } // namespace caf
......
...@@ -156,7 +156,8 @@ actor_factory make_actor_factory(F fun) { ...@@ -156,7 +156,8 @@ actor_factory make_actor_factory(F fun) {
typename ctrait::arg_types>; typename ctrait::arg_types>;
fd f{fun, static_cast<impl*>(x)}; fd f{fun, static_cast<impl*>(x)};
empty_type_erased_tuple dummy_; empty_type_erased_tuple dummy_;
auto& ct = msg.empty() ? dummy_ : const_cast<message&>(msg).content(); auto& ct = msg.empty() ? dummy_
: const_cast<message&>(msg).content();
auto opt = ct.apply(f); auto opt = ct.apply(f);
if (!opt) if (!opt)
return {}; return {};
...@@ -191,11 +192,6 @@ actor_factory_result dyn_spawn_class(actor_config& cfg, message& msg) { ...@@ -191,11 +192,6 @@ actor_factory_result dyn_spawn_class(actor_config& cfg, message& msg) {
template <class T, class... Ts> template <class T, class... Ts>
actor_factory make_actor_factory() { actor_factory make_actor_factory() {
/*
static_assert(std::is_same<T*, decltype(new T(std::declval<actor_config&>(),
std::declval<Ts>()...))>::value,
"no constructor for T(Ts...) exists");
*/
static_assert(detail::conjunction< static_assert(detail::conjunction<
std::is_lvalue_reference<Ts>::value... std::is_lvalue_reference<Ts>::value...
>::value, >::value,
......
...@@ -46,8 +46,8 @@ ...@@ -46,8 +46,8 @@
#include "caf/composable_behavior_based_actor.hpp" #include "caf/composable_behavior_based_actor.hpp"
#include "caf/prohibit_top_level_spawn_marker.hpp" #include "caf/prohibit_top_level_spawn_marker.hpp"
#include "caf/detail/spawn_fwd.hpp"
#include "caf/detail/init_fun_factory.hpp" #include "caf/detail/init_fun_factory.hpp"
#include "caf/detail/spawn_fwd.hpp"
namespace caf { namespace caf {
...@@ -502,35 +502,17 @@ public: ...@@ -502,35 +502,17 @@ public:
/// @warning must be called by thread which is about to terminate /// @warning must be called by thread which is about to terminate
void thread_terminates(); void thread_terminates();
/// @endcond
private:
template <class T>
void check_invariants() {
static_assert(!std::is_base_of<prohibit_top_level_spawn_marker, T>::value,
"This actor type cannot be spawned throught an actor system. "
"Probably you have tried to spawn a broker or opencl actor.");
}
expected<strong_actor_ptr> dyn_spawn_impl(const std::string& name,
message& args,
execution_unit* ctx,
bool check_interface,
optional<const mpi&> expected_ifs);
template <class C, spawn_options Os, class... Ts> template <class C, spawn_options Os, class... Ts>
infer_handle_from_class_t<C> infer_handle_from_class_t<C>
spawn_impl(actor_config& cfg, Ts&&... xs) { spawn_impl(actor_config& cfg, Ts&&... xs) {
static_assert(is_unbound(Os), static_assert(is_unbound(Os),
"top-level spawns cannot have monitor or link flag"); "top-level spawns cannot have monitor or link flag");
cfg.flags = has_priority_aware_flag(Os) // TODO: use `if constexpr` when switching to C++17
? abstract_actor::is_priority_aware_flag
: 0;
if (has_detach_flag(Os) || std::is_base_of<blocking_actor, C>::value) if (has_detach_flag(Os) || std::is_base_of<blocking_actor, C>::value)
cfg.flags |= abstract_actor::is_detached_flag; cfg.flags |= abstract_actor::is_detached_flag;
if (has_hide_flag(Os)) if (has_hide_flag(Os))
cfg.flags |= abstract_actor::is_hidden_flag; cfg.flags |= abstract_actor::is_hidden_flag;
if (!cfg.host) if (cfg.host == nullptr)
cfg.host = dummy_execution_unit(); cfg.host = dummy_execution_unit();
CAF_SET_LOGGER_SYS(this); CAF_SET_LOGGER_SYS(this);
auto res = make_actor<C>(next_actor_id(), node(), this, auto res = make_actor<C>(next_actor_id(), node(), this,
...@@ -540,6 +522,22 @@ private: ...@@ -540,6 +522,22 @@ private:
return res; return res;
} }
/// @endcond
private:
template <class T>
void check_invariants() {
static_assert(!std::is_base_of<prohibit_top_level_spawn_marker, T>::value,
"This actor type cannot be spawned throught an actor system. "
"Probably you have tried to spawn a broker or opencl actor.");
}
expected<strong_actor_ptr> dyn_spawn_impl(const std::string& name,
message& args,
execution_unit* ctx,
bool check_interface,
optional<const mpi&> expected_ifs);
/// Sets the internal actor for dynamic spawn operations. /// Sets the internal actor for dynamic spawn operations.
inline void spawn_serv(strong_actor_ptr x) { inline void spawn_serv(strong_actor_ptr x) {
internal_actors_[internal_actor_id(atom("SpawnServ"))] = std::move(x); internal_actors_[internal_actor_id(atom("SpawnServ"))] = std::move(x);
......
...@@ -233,29 +233,46 @@ public: ...@@ -233,29 +233,46 @@ public:
return *this; return *this;
} }
/// Stores whether the help text for this config object was /// Sets a config by using its INI name `config_name` to `config_value`.
/// printed. If set to `true`, the application should not use actor_system_config& set(const char* cn, config_value cv);
/// this config object to initialize an `actor_system` and
/// return from `main` immediately. // -- parser and CLI state ---------------------------------------------------
/// Stores whether the help text was printed. If set to `true`, the
/// application should not use this config to initialize an `actor_system`
/// and instead return from `main` immediately.
bool cli_helptext_printed; bool cli_helptext_printed;
/// Stores CLI arguments that were not consumed by CAF.
message args_remainder;
// -- caf-run parameters -----------------------------------------------------
/// Stores whether this node was started in slave mode. /// Stores whether this node was started in slave mode.
bool slave_mode; bool slave_mode;
/// Stores the name of this node when started in slave mode. /// Name of this node when started in slave mode.
std::string slave_name; std::string slave_name;
/// Stores credentials for connecting to the bootstrap node /// Credentials for connecting to the bootstrap node.
/// when using the caf-run launcher.
std::string bootstrap_node; std::string bootstrap_node;
/// Stores CLI arguments that were not consumed by CAF. // -- streaming parameters ---------------------------------------------------
message args_remainder;
/// Sets a config by using its INI name `config_name` to `config_value`. /// Maximum delay for sending underfull batches.
actor_system_config& set(const char* cn, config_value cv); size_t streaming_desired_batch_complexity_us;
/// Maximum delay for sending underfull batches.
size_t streaming_max_batch_delay_us;
/// Timespan between two credit rounds.
size_t streaming_credit_round_interval_us;
/// Greatest common divisor of `streaming_max_batch_delay_us` and
/// `streaming_credit_round_interval_us`.
size_t streaming_tick_duration_us() const noexcept;
// -- config parameters of the scheduler ------------------------------------- // -- scheduling parameters --------------------------------------------------
atom_value scheduler_policy; atom_value scheduler_policy;
size_t scheduler_max_threads; size_t scheduler_max_threads;
...@@ -264,7 +281,7 @@ public: ...@@ -264,7 +281,7 @@ public:
size_t scheduler_profiling_ms_resolution; size_t scheduler_profiling_ms_resolution;
std::string scheduler_profiling_output_file; std::string scheduler_profiling_output_file;
// -- config parameters for work-stealing ------------------------------------ // -- work-stealing parameters -----------------------------------------------
size_t work_stealing_aggressive_poll_attempts; size_t work_stealing_aggressive_poll_attempts;
size_t work_stealing_aggressive_steal_interval; size_t work_stealing_aggressive_steal_interval;
...@@ -274,7 +291,7 @@ public: ...@@ -274,7 +291,7 @@ public:
size_t work_stealing_relaxed_steal_interval; size_t work_stealing_relaxed_steal_interval;
size_t work_stealing_relaxed_sleep_duration_us; size_t work_stealing_relaxed_sleep_duration_us;
// -- config parameters for the logger --------------------------------------- // -- logger parameters ------------------------------------------------------
std::string logger_file_name; std::string logger_file_name;
std::string logger_file_format; std::string logger_file_format;
...@@ -289,7 +306,7 @@ public: ...@@ -289,7 +306,7 @@ public:
std::string& logger_filename CAF_DEPRECATED; std::string& logger_filename CAF_DEPRECATED;
std::string& logger_filter CAF_DEPRECATED; std::string& logger_filter CAF_DEPRECATED;
// -- config parameters of the middleman ------------------------------------- // -- middleman parameters ---------------------------------------------------
atom_value middleman_network_backend; atom_value middleman_network_backend;
std::string middleman_app_identifier; std::string middleman_app_identifier;
...@@ -303,11 +320,11 @@ public: ...@@ -303,11 +320,11 @@ public:
size_t middleman_cached_udp_buffers; size_t middleman_cached_udp_buffers;
size_t middleman_max_pending_msgs; size_t middleman_max_pending_msgs;
// -- config parameters of the OpenCL module --------------------------------- // -- OpenCL parameters ------------------------------------------------------
std::string opencl_device_ids; std::string opencl_device_ids;
// -- config parameters of the OpenSSL module --------------------------------- // -- OpenSSL parameters -----------------------------------------------------
std::string openssl_certificate; std::string openssl_certificate;
std::string openssl_key; std::string openssl_key;
......
...@@ -58,9 +58,11 @@ ...@@ -58,9 +58,11 @@
#include "caf/ref_counted.hpp" #include "caf/ref_counted.hpp"
#include "caf/thread_hook.hpp" #include "caf/thread_hook.hpp"
#include "caf/typed_actor.hpp" #include "caf/typed_actor.hpp"
#include "caf/stream_slot.hpp"
#include "caf/actor_system.hpp" #include "caf/actor_system.hpp"
#include "caf/deserializer.hpp" #include "caf/deserializer.hpp"
#include "caf/scoped_actor.hpp" #include "caf/scoped_actor.hpp"
#include "caf/upstream_msg.hpp"
#include "caf/actor_ostream.hpp" #include "caf/actor_ostream.hpp"
#include "caf/function_view.hpp" #include "caf/function_view.hpp"
#include "caf/index_mapping.hpp" #include "caf/index_mapping.hpp"
...@@ -74,12 +76,11 @@ ...@@ -74,12 +76,11 @@
#include "caf/stateful_actor.hpp" #include "caf/stateful_actor.hpp"
#include "caf/typed_behavior.hpp" #include "caf/typed_behavior.hpp"
#include "caf/proxy_registry.hpp" #include "caf/proxy_registry.hpp"
#include "caf/downstream_msg.hpp"
#include "caf/behavior_policy.hpp" #include "caf/behavior_policy.hpp"
#include "caf/message_builder.hpp" #include "caf/message_builder.hpp"
#include "caf/message_handler.hpp" #include "caf/message_handler.hpp"
#include "caf/response_handle.hpp" #include "caf/response_handle.hpp"
#include "caf/fused_scatterer.hpp"
#include "caf/random_gatherer.hpp"
#include "caf/system_messages.hpp" #include "caf/system_messages.hpp"
#include "caf/abstract_channel.hpp" #include "caf/abstract_channel.hpp"
#include "caf/may_have_timeout.hpp" #include "caf/may_have_timeout.hpp"
...@@ -90,16 +91,14 @@ ...@@ -90,16 +91,14 @@
#include "caf/event_based_actor.hpp" #include "caf/event_based_actor.hpp"
#include "caf/primitive_variant.hpp" #include "caf/primitive_variant.hpp"
#include "caf/timeout_definition.hpp" #include "caf/timeout_definition.hpp"
#include "caf/broadcast_scatterer.hpp"
#include "caf/actor_system_config.hpp" #include "caf/actor_system_config.hpp"
#include "caf/binary_deserializer.hpp" #include "caf/binary_deserializer.hpp"
#include "caf/composable_behavior.hpp" #include "caf/composable_behavior.hpp"
#include "caf/typed_actor_pointer.hpp" #include "caf/typed_actor_pointer.hpp"
#include "caf/scoped_execution_unit.hpp" #include "caf/scoped_execution_unit.hpp"
#include "caf/typed_response_promise.hpp" #include "caf/typed_response_promise.hpp"
#include "caf/random_topic_scatterer.hpp"
#include "caf/broadcast_topic_scatterer.hpp"
#include "caf/typed_event_based_actor.hpp" #include "caf/typed_event_based_actor.hpp"
#include "caf/fused_downstream_manager.hpp"
#include "caf/abstract_composable_behavior.hpp" #include "caf/abstract_composable_behavior.hpp"
#include "caf/decorator/sequencer.hpp" #include "caf/decorator/sequencer.hpp"
......
...@@ -179,6 +179,12 @@ using tick_atom = atom_constant<atom("tick")>; ...@@ -179,6 +179,12 @@ using tick_atom = atom_constant<atom("tick")>;
/// Used for pending out of order messages. /// Used for pending out of order messages.
using pending_atom = atom_constant<atom("pending")>; using pending_atom = atom_constant<atom("pending")>;
/// Used as timeout type for `timeout_msg`.
using receive_atom = atom_constant<atom("receive")>;
/// Used as timeout type for `timeout_msg`.
using stream_atom = atom_constant<atom("stream")>;
} // namespace caf } // namespace caf
namespace std { namespace std {
......
...@@ -23,26 +23,39 @@ ...@@ -23,26 +23,39 @@
#include <mutex> #include <mutex>
#include <condition_variable> #include <condition_variable>
#include "caf/fwd.hpp" #include "caf/actor_config.hpp"
#include "caf/send.hpp" #include "caf/actor_marker.hpp"
#include "caf/none.hpp"
#include "caf/after.hpp" #include "caf/after.hpp"
#include "caf/extend.hpp"
#include "caf/behavior.hpp" #include "caf/behavior.hpp"
#include "caf/extend.hpp"
#include "caf/fwd.hpp"
#include "caf/is_timeout_or_catch_all.hpp"
#include "caf/local_actor.hpp" #include "caf/local_actor.hpp"
#include "caf/typed_actor.hpp"
#include "caf/actor_config.hpp"
#include "caf/actor_marker.hpp"
#include "caf/mailbox_element.hpp" #include "caf/mailbox_element.hpp"
#include "caf/is_timeout_or_catch_all.hpp" #include "caf/none.hpp"
#include "caf/send.hpp"
#include "caf/typed_actor.hpp"
#include "caf/policy/arg.hpp"
#include "caf/policy/priority_aware.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/detail/type_list.hpp"
#include "caf/detail/apply_args.hpp" #include "caf/detail/apply_args.hpp"
#include "caf/detail/type_traits.hpp"
#include "caf/detail/blocking_behavior.hpp" #include "caf/detail/blocking_behavior.hpp"
#include "caf/detail/type_list.hpp"
#include "caf/detail/type_traits.hpp"
#include "caf/intrusive/drr_cached_queue.hpp"
#include "caf/intrusive/drr_queue.hpp"
#include "caf/intrusive/fifo_inbox.hpp"
#include "caf/intrusive/wdrr_dynamic_multiplexed_queue.hpp"
#include "caf/intrusive/wdrr_fixed_multiplexed_queue.hpp"
#include "caf/mixin/sender.hpp"
#include "caf/mixin/requester.hpp" #include "caf/mixin/requester.hpp"
#include "caf/mixin/sender.hpp"
#include "caf/mixin/subscriber.hpp" #include "caf/mixin/subscriber.hpp"
namespace caf { namespace caf {
...@@ -66,7 +79,39 @@ class blocking_actor ...@@ -66,7 +79,39 @@ class blocking_actor
mixin::subscriber>, mixin::subscriber>,
public dynamically_typed_actor_base { public dynamically_typed_actor_base {
public: public:
// -- member types ----------------------------------------------------------- // -- nested and member types ------------------------------------------------
/// Base type.
using super = extended_base;
/// Stores asynchronous messages with default priority.
using default_queue = intrusive::drr_cached_queue<policy::normal_messages>;
/// Stores asynchronous messages with hifh priority.
using urgent_queue = intrusive::drr_cached_queue<policy::urgent_messages>;
/// Configures the FIFO inbox with two nested queues:
///
/// 1. Default asynchronous messages
/// 2. High-priority asynchronous messages
struct mailbox_policy {
using deficit_type = size_t;
using mapped_type = mailbox_element;
using unique_pointer = mailbox_element_ptr;
using queue_type =
intrusive::wdrr_fixed_multiplexed_queue<policy::priority_aware,
default_queue, urgent_queue>;
static constexpr size_t default_queue_index = 0;
static constexpr size_t urgent_queue_index = 1;
};
/// A queue optimized for single-reader-many-writers.
using mailbox_type = intrusive::fifo_inbox<mailbox_policy>;
/// Absolute timeout type. /// Absolute timeout type.
using timeout_type = std::chrono::high_resolution_clock::time_point; using timeout_type = std::chrono::high_resolution_clock::time_point;
...@@ -174,6 +219,23 @@ public: ...@@ -174,6 +219,23 @@ public:
} }
}; };
struct mailbox_visitor {
blocking_actor* self;
bool& done;
receive_cond& rcc;
message_id mid;
detail::blocking_behavior& bhvr;
// Dispatches messages with high and normal priority to the same handler.
template <class Queue>
intrusive::task_result operator()(size_t, Queue&, mailbox_element& x) {
return (*this)(x);
}
// Consumes `x`.
intrusive::task_result operator()(mailbox_element& x);
};
// -- constructors and destructors ------------------------------------------- // -- constructors and destructors -------------------------------------------
blocking_actor(actor_config& cfg); blocking_actor(actor_config& cfg);
...@@ -305,9 +367,12 @@ public: ...@@ -305,9 +367,12 @@ public:
virtual bool await_data(timeout_type timeout); virtual bool await_data(timeout_type timeout);
/// Returns the next element from the mailbox or `nullptr`. /// Returns the next element from the mailbox or `nullptr`.
/// The default implementation simply returns `next_message()`.
virtual mailbox_element_ptr dequeue(); virtual mailbox_element_ptr dequeue();
/// Returns the queue for storing incoming messages.
inline mailbox_type& mailbox() {
return mailbox_;
}
/// @cond PRIVATE /// @cond PRIVATE
/// Receives messages until either a pre- or postcheck of `rcc` fails. /// Receives messages until either a pre- or postcheck of `rcc` fails.
...@@ -348,6 +413,20 @@ public: ...@@ -348,6 +413,20 @@ public:
void receive_impl(receive_cond& rcc, message_id mid, void receive_impl(receive_cond& rcc, message_id mid,
detail::blocking_behavior& bhvr); detail::blocking_behavior& bhvr);
bool cleanup(error&& fail_state, execution_unit* host) override;
sec build_pipeline(stream_slot in, stream_slot out, stream_manager_ptr mgr);
// -- backwards compatibility ------------------------------------------------
inline mailbox_element_ptr next_message() {
return dequeue();
}
inline bool has_next_message() {
return !mailbox_.empty();
}
/// @endcond /// @endcond
private: private:
...@@ -369,6 +448,11 @@ private: ...@@ -369,6 +448,11 @@ private:
res += attach_functor(x); res += attach_functor(x);
return res; return res;
} }
// -- member variables -------------------------------------------------------
// used by both event-based and blocking actors
mailbox_type mailbox_;
}; };
} // namespace caf } // namespace caf
......
This diff is collapsed.
...@@ -16,35 +16,37 @@ ...@@ -16,35 +16,37 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_MIXIN_BUFFERED_SCATTERER_HPP #ifndef CAF_BUFFERED_DOWNSTREAM_MANAGER_HPP
#define CAF_MIXIN_BUFFERED_SCATTERER_HPP #define CAF_BUFFERED_DOWNSTREAM_MANAGER_HPP
#include <deque> #include <deque>
#include <vector> #include <vector>
#include <cstddef> #include <cstddef>
#include <iterator> #include <iterator>
#include "caf/sec.hpp" #include "caf/downstream_manager_base.hpp"
#include "caf/stream_edge_impl.hpp" #include "caf/logger.hpp"
#include "caf/actor_control_block.hpp"
#include "caf/stream_scatterer_impl.hpp"
namespace caf { namespace caf {
/// Mixin for streams with any number of downstreams. `Subtype` must provide a /// Mixin for streams with any number of downstreams. `Subtype` must provide a
/// member function `buf()` returning a queue with `std::deque`-like interface. /// member function `buf()` returning a queue with `std::deque`-like interface.
template <class T> template <class T>
class buffered_scatterer : public stream_scatterer_impl { class buffered_downstream_manager : public downstream_manager_base {
public: public:
using super = stream_scatterer_impl; // -- member types -----------------------------------------------------------
using value_type = T; using super = downstream_manager_base;
using buffer_type = std::deque<value_type>; using output_type = T;
using chunk_type = std::vector<value_type>; using buffer_type = std::deque<output_type>;
buffered_scatterer(local_actor* selfptr) : super(selfptr) { using chunk_type = std::vector<output_type>;
// -- constructors, destructors, and assignment operators --------------------
buffered_downstream_manager(scheduled_actor* self) : super(self) {
// nop // nop
} }
...@@ -57,8 +59,8 @@ public: ...@@ -57,8 +59,8 @@ public:
static chunk_type get_chunk(buffer_type& buf, long n) { static chunk_type get_chunk(buffer_type& buf, long n) {
CAF_LOG_TRACE(CAF_ARG(buf) << CAF_ARG(n)); CAF_LOG_TRACE(CAF_ARG(buf) << CAF_ARG(n));
chunk_type xs; chunk_type xs;
if (n > 0) { if (!buf.empty() && n > 0) {
xs.reserve(static_cast<size_t>(n)); xs.reserve(std::min(static_cast<size_t>(n), buf.size()));
if (static_cast<size_t>(n) < buf.size()) { if (static_cast<size_t>(n) < buf.size()) {
auto first = buf.begin(); auto first = buf.begin();
auto last = first + static_cast<ptrdiff_t>(n); auto last = first + static_cast<ptrdiff_t>(n);
...@@ -76,8 +78,24 @@ public: ...@@ -76,8 +78,24 @@ public:
return get_chunk(buf_, n); return get_chunk(buf_, n);
} }
long buffered() const override { bool terminal() const noexcept override {
return static_cast<long>(buf_.size()); 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 stored < desired ? desired - stored : 0u;
}
size_t buffered() const noexcept override {
return buf_.size();
} }
buffer_type& buf() { buffer_type& buf() {
...@@ -94,4 +112,4 @@ protected: ...@@ -94,4 +112,4 @@ protected:
} // namespace caf } // namespace caf
#endif // CAF_MIXIN_BUFFERED_SCATTERER_HPP #endif // CAF_BUFFERED_DOWNSTREAM_MANAGER_HPP
...@@ -23,7 +23,6 @@ ...@@ -23,7 +23,6 @@
#include "caf/replies_to.hpp" #include "caf/replies_to.hpp"
#include "caf/detail/type_list.hpp" #include "caf/detail/type_list.hpp"
#include "caf/detail/typed_actor_util.hpp"
namespace caf { namespace caf {
......
...@@ -76,46 +76,49 @@ struct composed_type<detail::type_list<typed_mpi<detail::type_list<In...>, ...@@ -76,46 +76,49 @@ struct composed_type<detail::type_list<typed_mpi<detail::type_list<In...>,
// case #2 // case #2
template <class... In, class... Out, class... Xs, class Ys, template <class... In, class... Out, class... Xs, class Ys,
class... MapsTo, class... Zs, class... Rs> class M, class... Ms, class... Zs, class... Rs>
struct composed_type<detail::type_list<typed_mpi<detail::type_list<In...>, struct composed_type<detail::type_list<typed_mpi<detail::type_list<In...>,
output_tuple<Out...>>, Xs...>, output_tuple<Out...>>, Xs...>,
Ys, Ys,
detail::type_list<typed_mpi<detail::type_list<Out...>, detail::type_list<typed_mpi<detail::type_list<Out...>,
output_stream<MapsTo...>>, output_stream<M, Ms...>>,
Zs...>, Zs...>,
detail::type_list<Rs...>> detail::type_list<Rs...>>
: composed_type<detail::type_list<Xs...>, Ys, Ys, : composed_type<detail::type_list<Xs...>, Ys, Ys,
detail::type_list<Rs..., detail::type_list<Rs...,
typed_mpi<detail::type_list<In...>, typed_mpi<detail::type_list<In...>,
output_stream<MapsTo...>>>> {}; output_stream<M, Ms...>>>> {
};
// case #3 // case #3
template <class... In, class... Out, class... Xs, class Ys, template <class... In, class O, class... Out, class... Xs, class Ys,
class... MapsTo, class... Zs, class... Rs> class M, class... Ms, class... Zs, class... Rs>
struct composed_type<detail::type_list<typed_mpi<detail::type_list<In...>, struct composed_type<detail::type_list<typed_mpi<detail::type_list<In...>,
output_stream<Out...>>, Xs...>, output_stream<O, Out...>>, Xs...>,
Ys, Ys,
detail::type_list<typed_mpi<detail::type_list<Out...>, detail::type_list<typed_mpi<detail::type_list<O, Out...>,
output_tuple<MapsTo...>>, Zs...>, output_tuple<M, Ms...>>, Zs...>,
detail::type_list<Rs...>> detail::type_list<Rs...>>
: composed_type<detail::type_list<Xs...>, Ys, Ys, : composed_type<detail::type_list<Xs...>, Ys, Ys,
detail::type_list<Rs..., detail::type_list<Rs...,
typed_mpi<detail::type_list<In...>, typed_mpi<detail::type_list<In...>,
output_stream<MapsTo...>>>> {}; output_stream<M, Ms...>>>> {
};
// case #4 // case #4
template <class... In, class... Out, class... Xs, class Ys, template <class... In, class O, class... Out, class... Xs, class Ys,
class... MapsTo, class... Zs, class... Rs> class M, class... Ms, class... Zs, class... Rs>
struct composed_type<detail::type_list<typed_mpi<detail::type_list<In...>, struct composed_type<detail::type_list<typed_mpi<detail::type_list<In...>,
output_stream<Out...>>, Xs...>, output_stream<O, Out...>>, Xs...>,
Ys, Ys,
detail::type_list<typed_mpi<detail::type_list<Out...>, detail::type_list<typed_mpi<detail::type_list<O, Out...>,
output_stream<MapsTo...>>, Zs...>, output_stream<M, Ms...>>, Zs...>,
detail::type_list<Rs...>> detail::type_list<Rs...>>
: composed_type<detail::type_list<Xs...>, Ys, Ys, : composed_type<detail::type_list<Xs...>, Ys, Ys,
detail::type_list<Rs..., detail::type_list<Rs...,
typed_mpi<detail::type_list<In...>, typed_mpi<detail::type_list<In...>,
output_stream<MapsTo...>>>> {}; output_stream<M, Ms...>>>> {
};
// default case (recurse over Zs) // default case (recurse over Zs)
template <class In, class Out, class... Xs, class Ys, template <class In, class Out, class... Xs, class Ys,
......
...@@ -62,6 +62,8 @@ ...@@ -62,6 +62,8 @@
// CAF_PUSH_WARNINGS and CAF_POP_WARNINGS // CAF_PUSH_WARNINGS and CAF_POP_WARNINGS
#if defined(__clang__) #if defined(__clang__)
# define CAF_CLANG # define CAF_CLANG
# define CAF_LIKELY(x) __builtin_expect((x), 1)
# define CAF_UNLIKELY(x) __builtin_expect((x), 0)
# define CAF_DEPRECATED __attribute__((deprecated)) # define CAF_DEPRECATED __attribute__((deprecated))
# define CAF_DEPRECATED_MSG(msg) __attribute__((deprecated(msg))) # define CAF_DEPRECATED_MSG(msg) __attribute__((deprecated(msg)))
# define CAF_PUSH_WARNINGS \ # define CAF_PUSH_WARNINGS \
...@@ -104,6 +106,8 @@ ...@@ -104,6 +106,8 @@
# endif # endif
#elif defined(__GNUC__) #elif defined(__GNUC__)
# define CAF_GCC # define CAF_GCC
# define CAF_LIKELY(x) __builtin_expect((x), 1)
# define CAF_UNLIKELY(x) __builtin_expect((x), 0)
# define CAF_DEPRECATED __attribute__((deprecated)) # define CAF_DEPRECATED __attribute__((deprecated))
# define CAF_DEPRECATED_MSG(msg) __attribute__((deprecated(msg))) # define CAF_DEPRECATED_MSG(msg) __attribute__((deprecated(msg)))
# define CAF_PUSH_WARNINGS # define CAF_PUSH_WARNINGS
...@@ -130,6 +134,8 @@ ...@@ -130,6 +134,8 @@
# endif # endif
#elif defined(_MSC_VER) #elif defined(_MSC_VER)
# define CAF_MSVC # define CAF_MSVC
# define CAF_LIKELY(x) x
# define CAF_UNLIKELY(x) x
# define CAF_DEPRECATED # define CAF_DEPRECATED
# define CAF_DEPRECATED_MSG(msg) # define CAF_DEPRECATED_MSG(msg)
# define CAF_PUSH_WARNINGS # define CAF_PUSH_WARNINGS
...@@ -144,6 +150,8 @@ ...@@ -144,6 +150,8 @@
# define NOMINMAX # define NOMINMAX
# endif // NOMINMAX # endif // NOMINMAX
#else #else
# define CAF_LIKELY(x) x
# define CAF_UNLIKELY(x) x
# define CAF_DEPRECATED # define CAF_DEPRECATED
# define CAF_PUSH_WARNINGS # define CAF_PUSH_WARNINGS
# define CAF_POP_WARNINGS # define CAF_POP_WARNINGS
......
...@@ -26,7 +26,6 @@ ...@@ -26,7 +26,6 @@
#include "caf/expected.hpp" #include "caf/expected.hpp"
#include "caf/optional.hpp" #include "caf/optional.hpp"
#include "caf/replies_to.hpp" #include "caf/replies_to.hpp"
#include "caf/stream_result.hpp"
#include "caf/detail/implicit_conversions.hpp" #include "caf/detail/implicit_conversions.hpp"
...@@ -52,21 +51,14 @@ struct dmi<result<Ys...> (Xs...)> { ...@@ -52,21 +51,14 @@ struct dmi<result<Ys...> (Xs...)> {
output_tuple<implicit_conversions_t<Ys>...>>; output_tuple<implicit_conversions_t<Ys>...>>;
}; };
// case #2b: function returning a stream_result<...> // case #2b: function returning a std::tuple<...>
template <class Y, class... Xs>
struct dmi<stream_result<Y> (Xs...)> {
using type = typed_mpi<type_list<typename param_decay<Xs>::type...>,
output_tuple<implicit_conversions_t<Y>>>;
};
// case #2c: function returning a std::tuple<...>
template <class... Ys, class... Xs> template <class... Ys, class... Xs>
struct dmi<std::tuple<Ys...> (Xs...)> { struct dmi<std::tuple<Ys...> (Xs...)> {
using type = typed_mpi<type_list<typename param_decay<Xs>::type...>, using type = typed_mpi<type_list<typename param_decay<Xs>::type...>,
output_tuple<implicit_conversions_t<Ys>...>>; output_tuple<implicit_conversions_t<Ys>...>>;
}; };
// case #2d: function returning a std::tuple<...> // case #2c: function returning a std::tuple<...>
template <class... Ys, class... Xs> template <class... Ys, class... Xs>
struct dmi<delegated<Ys...> (Xs...)> { struct dmi<delegated<Ys...> (Xs...)> {
using type = typed_mpi<type_list<typename param_decay<Xs>::type...>, using type = typed_mpi<type_list<typename param_decay<Xs>::type...>,
...@@ -88,9 +80,9 @@ struct dmi<optional<Y> (Xs...)> : dmi<Y (Xs...)> {}; ...@@ -88,9 +80,9 @@ struct dmi<optional<Y> (Xs...)> : dmi<Y (Xs...)> {};
template <class Y, class... Xs> template <class Y, class... Xs>
struct dmi<expected<Y> (Xs...)> : dmi<Y (Xs...)> {}; struct dmi<expected<Y> (Xs...)> : dmi<Y (Xs...)> {};
// case #5: function returning an annotated_stream<> // case #5: function returning an output_stream<>
template <class Y, class... Ys, class... Xs> template <class Y, class... Ys, class P, class... Xs>
struct dmi<annotated_stream<Y, Ys...> (Xs...)> : dmi<Y (Xs...)> { struct dmi<output_stream<Y, std::tuple<Ys...>, P> (Xs...)> : dmi<Y (Xs...)> {
using type = using type =
typed_mpi<type_list<typename param_decay<Xs>::type...>, typed_mpi<type_list<typename param_decay<Xs>::type...>,
output_tuple<stream<Y>, strip_and_convert_t<Ys>...>>; output_tuple<stream<Y>, strip_and_convert_t<Ys>...>>;
......
...@@ -12,59 +12,50 @@ ...@@ -12,59 +12,50 @@
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. * * License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* * * *
* If you did not receive a copy of the license files, see * * If you did not receive a copy of the license files, see *
* http://openresult.org/licenses/BSD-3-Clause and * * http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_STREAM_RESULT_TRAIT_HPP #ifndef CAF_DEFAULT_DOWNSTREAM_MANAGER_HPP
#define CAF_STREAM_RESULT_TRAIT_HPP #define CAF_DEFAULT_DOWNSTREAM_MANAGER_HPP
#include "caf/unit.hpp" #include "caf/broadcast_downstream_manager.hpp"
#include "caf/behavior.hpp" #include "caf/stream_source_trait.hpp"
#include "caf/stream_stage_trait.hpp"
#include "caf/detail/type_traits.hpp" #include "caf/detail/type_traits.hpp"
namespace caf { namespace caf {
/// Selects a downstream manager implementation based on the signature of
/// various handlers.
template <class F> template <class F>
struct stream_result_trait; struct default_downstream_manager {
/// The function signature of `F`.
template <class T> using fun_sig = typename detail::get_callable_trait<F>::fun_sig;
struct stream_result_trait<void (expected<T>)> {
using type = T; /// The source trait for `F`.
template <class OnResult> using source_trait = stream_source_trait<fun_sig>;
static behavior make_result_handler(OnResult f) {
return { /// The stage trait for `F`.
[=](T& res) { using stage_trait = stream_stage_trait<fun_sig>;
f(std::move(res));
}, /// The output type as returned by the source or stage trait.
[=](error& err) { using output_type =
f(std::move(err)); typename std::conditional<
} source_trait::valid,
}; typename source_trait::output,
} typename stage_trait::output
}; >::type;
template <> /// The default downstream manager deduced by this trait.
struct stream_result_trait<void (expected<void>)> { using type = broadcast_downstream_manager<output_type>;
using type = void;
template <class OnResult>
static behavior make_result_handler(OnResult f) {
return {
[=]() {
f(unit);
},
[=](error& err) {
f(std::move(err));
}
};
}
}; };
template <class F> template <class F>
using stream_result_trait_t = using default_downstream_manager_t =
stream_result_trait<typename detail::get_callable_trait<F>::fun_sig>; typename default_downstream_manager<F>::type;
} // namespace caf } // namespace caf
#endif // CAF_STREAM_RESULT_TRAIT_HPP #endif // CAF_DEFAULT_DOWNSTREAM_MANAGER_HPP
...@@ -16,55 +16,55 @@ ...@@ -16,55 +16,55 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_BROADCAST_TOPIC_SCATTERER_HPP #ifndef CAF_DETAIL_ALGORITHMS_HPP
#define CAF_BROADCAST_TOPIC_SCATTERER_HPP #define CAF_DETAIL_ALGORITHMS_HPP
#include <map>
#include <tuple>
#include <deque>
#include <vector> #include <vector>
#include <functional>
#include "caf/topic_scatterer.hpp" #include "caf/detail/type_traits.hpp"
namespace caf { namespace caf {
namespace detail {
/// A topic scatterer that delivers data in broadcast fashion to all sinks. /// Like `std::for_each`, but for multiple containers.
template <class T, class Filter, class Select> /// @pre `x.size() <= y.size()` for each `y` in `xs`
class broadcast_topic_scatterer template <class F, class Container, class... Containers>
: public topic_scatterer<T, Filter, Select> { void zip_foreach(F f, Container&& x, Containers&&... xs) {
public: for (size_t i = 0; i < x.size(); ++i)
/// Base type. f(x[i], xs[i]...);
using super = topic_scatterer<T, Filter, Select>; }
broadcast_topic_scatterer(local_actor* selfptr) : super(selfptr) { /// Like `std::accumulate`, but for multiple containers.
// nop /// @pre `x.size() <= y.size()` for each `y` in `xs`
} template <class F, class T, class Container, class... Containers>
T zip_fold(F f, T init, Container&& x, Containers&&... xs) {
for (size_t i = 0; i < x.size(); ++i)
init = f(init, x[i], xs[i]...);
return init;
}
long credit() const override { /// Decorates a container of type `T` to appear as container of type `U`.
// We receive messages until we have exhausted all downstream credit and template <class F, class Container>
// have filled our buffer to its minimum size. struct container_view {
return this->min_credit() + this->min_buffer_size(); Container& x;
using value_type = typename detail::get_callable_trait<F>::result_type;
inline size_t size() const {
return x.size();
} }
value_type operator[](size_t i) {
void emit_batches() override { F f;
CAF_LOG_TRACE(""); return f(x[i]);
this->fan_out();
for (auto& kvp : this->lanes_) {
auto& l = kvp.second;
auto chunk = super::get_chunk(l.buf, super::min_credit(l.paths));
auto csize = static_cast<long>(chunk.size());
if (csize == 0)
continue;
auto wrapped_chunk = make_message(std::move(chunk));
for (auto& x : l.paths) {
CAF_ASSERT(x->open_credit >= csize);
x->emit_batch(csize, wrapped_chunk);
}
}
} }
}; };
/// Returns a container view for `x`.
/// @relates container_view
template <class F, class Container>
container_view<F, Container> make_container_view(Container& x) {
return {x};
}
} // namespace detail
} // namespace caf } // namespace caf
#endif // CAF_BROADCAST_TOPIC_SCATTERER_HPP #endif // CAF_DETAIL_ALGORITHMS_HPP
...@@ -19,14 +19,6 @@ ...@@ -19,14 +19,6 @@
#ifndef CAF_ARG_WRAPPER_HPP #ifndef CAF_ARG_WRAPPER_HPP
#define CAF_ARG_WRAPPER_HPP #define CAF_ARG_WRAPPER_HPP
#include <tuple>
#include <string>
#include "caf/detail/int_list.hpp"
#include "caf/detail/type_list.hpp"
#include "caf/detail/apply_args.hpp"
#include "caf/detail/type_traits.hpp"
namespace caf { namespace caf {
namespace detail { namespace detail {
...@@ -40,50 +32,12 @@ struct arg_wrapper { ...@@ -40,50 +32,12 @@ struct arg_wrapper {
} }
}; };
template <class... Ts>
struct named_args_tuple {
const std::array<const char*, sizeof...(Ts)>& names;
std::tuple<const Ts&...> xs;
};
template <class... Ts>
named_args_tuple<Ts...>
make_named_args_tuple(std::array<const char*, sizeof...(Ts)>& names,
const Ts&... xs) {
return {names, std::forward_as_tuple(xs...)};
};
template <size_t I, class... Ts>
auto get(const named_args_tuple<Ts...>& x)
-> arg_wrapper<typename type_at<I, Ts...>::type>{
CAF_ASSERT(x.names[I] != nullptr);
return {x.names[I], std::get<I>(x.xs)};
}
/// Used to implement `CAF_ARG`. /// Used to implement `CAF_ARG`.
template <class T> template <class T>
static arg_wrapper<T> make_arg_wrapper(const char* name, const T& value) { static arg_wrapper<T> make_arg_wrapper(const char* name, const T& value) {
return {name, value}; return {name, value};
} }
struct arg_wrapper_maker {
template <class... Ts>
auto operator()(const Ts&... xs) const -> decltype(std::make_tuple(xs...)) {
return std::make_tuple(xs...);
}
};
/// Used to implement `CAF_ARGS`.
template <class... Ts>
static std::tuple<arg_wrapper<Ts>...>
make_args_wrapper(std::array<const char*, sizeof...(Ts)> names,
const Ts&... xs) {
arg_wrapper_maker f;
typename il_range<0, sizeof...(Ts)>::type indices;
auto tup = make_named_args_tuple(names, xs...);
return apply_args(f, indices, tup);
}
} // namespace detail } // namespace detail
} // namespace caf } // namespace caf
......
...@@ -28,13 +28,13 @@ namespace detail { ...@@ -28,13 +28,13 @@ namespace detail {
class disposer { class disposer {
public: public:
inline void operator()(memory_managed* ptr) const { inline void operator()(memory_managed* ptr) const noexcept {
ptr->request_deletion(false); ptr->request_deletion(false);
} }
template <class T> template <class T>
typename std::enable_if<!std::is_base_of<memory_managed, T>::value>::type typename std::enable_if<!std::is_base_of<memory_managed, T>::value>::type
operator()(T* ptr) const { operator()(T* ptr) const noexcept {
delete ptr; delete ptr;
} }
}; };
......
...@@ -5,7 +5,8 @@ ...@@ -5,7 +5,8 @@
* | |___ / ___ \| _| Framework * * | |___ / ___ \| _| Framework *
* \____/_/ \_|_| * * \____/_/ \_|_| *
* * * *
* Copyright 2011-2018 Dominik Charousset * * Copyright (C) 2011 - 2017 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* * * *
* 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,28 +17,18 @@ ...@@ -16,28 +17,18 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#include "caf/stream_scatterer.hpp" #ifndef CAF_DETAIL_ENQUEUE_RESULT_HPP
#define CAF_DETAIL_ENQUEUE_RESULT_HPP
#include "caf/logger.hpp" #include "caf/intrusive/inbox_result.hpp"
#include "caf/actor_addr.hpp"
#include "caf/actor_cast.hpp"
#include "caf/outbound_path.hpp"
namespace caf { namespace caf {
namespace detail {
stream_scatterer::~stream_scatterer() { /// Alias for backwards compatibility.
// nop using enqueue_result = intrusive::inbox_result;
}
bool stream_scatterer::remove_path(const stream_id& sid,
const strong_actor_ptr& x, error reason,
bool silent) {
return remove_path(sid, actor_cast<actor_addr>(x), std::move(reason), silent);
}
stream_scatterer::path_type* stream_scatterer::find(const stream_id& sid,
const strong_actor_ptr& x) {
return find(sid, actor_cast<actor_addr>(x));
}
} // namespace intrusive
} // namespace caf } // namespace caf
#endif // CAF_DETAIL_ENQUEUE_RESULT_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2017 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* 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. *
******************************************************************************/
#ifndef CAF_DETAIL_GCD_HPP
#define CAF_DETAIL_GCD_HPP
namespace caf {
namespace detail {
template <class T>
static T gcd(T a, T b) {
T r;
while (b != 0) {
r = a % b;
a = b;
b = r;
}
return a;
}
} // namespace detail
} // namespace caf
#endif // CAF_DETAIL_GCD_HPP
...@@ -25,6 +25,7 @@ ...@@ -25,6 +25,7 @@
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/detail/apply_args.hpp" #include "caf/detail/apply_args.hpp"
#include "caf/detail/spawn_fwd.hpp"
#include "caf/detail/type_traits.hpp" #include "caf/detail/type_traits.hpp"
namespace caf { namespace caf {
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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. *
******************************************************************************/
#ifndef CAF_DETAIL_INTRUSIVE_PARTITIONED_LIST_HPP
#define CAF_DETAIL_INTRUSIVE_PARTITIONED_LIST_HPP
#include "caf/config.hpp"
#include <memory>
#include <iterator>
#include <algorithm>
#include "caf/behavior.hpp"
#include "caf/invoke_message_result.hpp"
namespace caf {
namespace detail {
/// Describes a partitioned list of elements. The first part
/// of the list is described by the iterator pair `[begin, separator)`.
/// The second part by `[continuation, end)`. Actors use the second half
/// of the list to store previously skipped elements. Priority-aware actors
/// also use the first half of the list to sort messages by priority.
template <class T, class Delete = std::default_delete<T>>
class intrusive_partitioned_list {
public:
using value_type = T;
using pointer = value_type*;
using deleter_type = Delete;
struct iterator : std::iterator<std::bidirectional_iterator_tag, value_type> {
pointer ptr;
iterator(pointer init = nullptr) : ptr(init) {
// nop
}
iterator(const iterator&) = default;
iterator& operator=(const iterator&) = default;
iterator& operator++() {
ptr = ptr->next;
return *this;
}
iterator operator++(int) {
iterator res = *this;
ptr = ptr->next;
return res;
}
iterator& operator--() {
ptr = ptr->prev;
return *this;
}
iterator operator--(int) {
iterator res = *this;
ptr = ptr->prev;
return res;
}
const value_type& operator*() const {
return *ptr;
}
value_type& operator*() {
return *ptr;
}
pointer operator->() {
return ptr;
}
bool operator==(const iterator& other) const {
return ptr == other.ptr;
}
bool operator!=(const iterator& other) const {
return ptr != other.ptr;
}
iterator next() const {
return ptr->next;
}
};
intrusive_partitioned_list() {
head_.next = &separator_;
separator_.prev = &head_;
separator_.next = &tail_;
tail_.prev = &separator_;
}
~intrusive_partitioned_list() {
clear();
}
iterator begin() {
return head_.next;
}
iterator separator() {
return &separator_;
}
iterator continuation() {
return separator_.next;
}
iterator end() {
return &tail_;
}
using range = std::pair<iterator, iterator>;
/// Returns the two iterator pairs describing the first and second part
/// of the partitioned list.
std::array<range, 2> ranges() {
return {{range{begin(), separator()}, range{continuation(), end()}}};
}
template <class F>
void clear(F f) {
for (auto& range : ranges()) {
auto i = range.first;
auto e = range.second;
while (i != e) {
auto ptr = i.ptr;
++i;
f(*ptr);
delete_(ptr);
}
}
if (head_.next != &separator_) {
head_.next = &separator_;
separator_.prev = &head_;
}
if (separator_.next != &tail_) {
separator_.next = &tail_;
tail_.prev = &separator_;
}
}
void clear() {
auto nop = [](value_type&) {};
clear(nop);
}
iterator insert(iterator next, pointer val) {
auto prev = next->prev;
val->prev = prev;
val->next = next.ptr;
prev->next = val;
next->prev = val;
return val;
}
bool empty() const {
return head_.next == &separator_ && separator_.next == &tail_;
}
pointer take(iterator pos) {
auto res = pos.ptr;
auto next = res->next;
auto prev = res->prev;
prev->next = next;
next->prev = prev;
return res;
}
iterator erase(iterator pos) {
auto next = pos->next;
delete_(take(pos));
return next;
}
size_t count(size_t max_count = std::numeric_limits<size_t>::max()) {
size_t result = 0;
for (auto& range : ranges())
for (auto i = range.first; i != range.second; ++i)
if (++result == max_count)
return max_count;
return result;
}
private:
value_type head_;
value_type separator_;
value_type tail_;
deleter_type delete_;
};
} // namespace detail
} // namespace caf
#endif // CAF_DETAIL_INTRUSIVE_PARTITIONED_LIST_HPP
...@@ -21,22 +21,28 @@ ...@@ -21,22 +21,28 @@
#include <tuple> #include <tuple>
#include "caf/expected.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/none.hpp" #include "caf/make_message.hpp"
#include "caf/unit.hpp" #include "caf/make_sink_result.hpp"
#include "caf/skip.hpp" #include "caf/make_source_result.hpp"
#include "caf/result.hpp" #include "caf/make_stage_result.hpp"
#include "caf/message.hpp" #include "caf/message.hpp"
#include "caf/expected.hpp" #include "caf/none.hpp"
#include "caf/optional.hpp" #include "caf/optional.hpp"
#include "caf/make_message.hpp" #include "caf/result.hpp"
#include "caf/skip.hpp"
#include "caf/unit.hpp"
#include "caf/detail/int_list.hpp"
#include "caf/detail/apply_args.hpp" #include "caf/detail/apply_args.hpp"
#include "caf/detail/int_list.hpp"
#include "caf/detail/type_traits.hpp"
namespace caf { namespace caf {
namespace detail { namespace detail {
/// Inspects the result of message handlers and triggers type-depended actions
/// such as generating result messages.
class invoke_result_visitor { class invoke_result_visitor {
public: public:
virtual ~invoke_result_visitor(); virtual ~invoke_result_visitor();
...@@ -45,26 +51,32 @@ public: ...@@ -45,26 +51,32 @@ public:
// nop // nop
} }
// severs as catch-all case for all values producing // -- virtual handlers -------------------------------------------------------
// no result such as response_promise
/// Called whenever no result messages gets produced, e.g., when returning a
/// `response_promise`.
virtual void operator()() = 0; virtual void operator()() = 0;
// called if the message handler explicitly returned an error /// Called if the message handler returned an error.
virtual void operator()(error&) = 0; virtual void operator()(error&) = 0;
// called if the message handler returned any "ordinary" value /// Called if the message handler returned any "ordinary" value.
virtual void operator()(message&) = 0; virtual void operator()(message&) = 0;
// called if the message handler returns explictly none or unit /// Called if the message handler returns "nothing", for example a
/// default-constructed `optional<T>`.
virtual void operator()(const none_t&) = 0; virtual void operator()(const none_t&) = 0;
// map unit to an empty message // -- on-the-fly type conversions --------------------------------------------
/// Called if the message handler returns `void` or `unit_t`.
inline void operator()(const unit_t&) { inline void operator()(const unit_t&) {
message empty_msg; message empty_msg;
(*this)(empty_msg); (*this)(empty_msg);
} }
// unwrap optionals /// Unwraps an `optional<T>` by recursively calling the visitor with either
/// `none_t` or `T`.
template <class T> template <class T>
void operator()(optional<T>& x) { void operator()(optional<T>& x) {
if (x) if (x)
...@@ -73,7 +85,8 @@ public: ...@@ -73,7 +85,8 @@ public:
(*this)(none); (*this)(none);
} }
// unwrap expecteds /// Unwraps an `expected<T>` by recursively calling the visitor with either
/// `error` or `T`.
template <class T> template <class T>
void operator()(expected<T>& x) { void operator()(expected<T>& x) {
if (x) if (x)
...@@ -82,71 +95,109 @@ public: ...@@ -82,71 +95,109 @@ public:
(*this)(x.error()); (*this)(x.error());
} }
// convert values to messages /// Wraps arbitrary values into a `message` and calls the visitor recursively.
template <class... Ts> template <class... Ts>
void operator()(Ts&... xs) { 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)...); auto tmp = make_message(std::move(xs)...);
(*this)(tmp); (*this)(tmp);
} }
// unwrap tuples /// Wraps the tuple into a `message` and calls the visitor recursively with
/// its contents.
template <class... Ts> template <class... Ts>
void operator()(std::tuple<Ts...>& xs) { void operator()(std::tuple<Ts...>& xs) {
apply_args(*this, get_indices(xs), xs); apply_args(*this, get_indices(xs), xs);
} }
// disambiguations /// Disambiguates the variadic `operator<Ts...>()`.
inline void operator()(none_t& x) { inline void operator()(none_t& x) {
(*this)(const_cast<const none_t&>(x)); (*this)(const_cast<const none_t&>(x));
} }
/// Disambiguates the variadic `operator<Ts...>()`.
inline void operator()(unit_t& x) { inline void operator()(unit_t& x) {
(*this)(const_cast<const unit_t&>(x)); (*this)(const_cast<const unit_t&>(x));
} }
// special purpose handler that don't procude results // -- special-purpose handlers that don't procude results --------------------
/// Calls `(*this)()`.
inline void operator()(response_promise&) { inline void operator()(response_promise&) {
(*this)(); (*this)();
} }
/// Calls `(*this)()`.
template <class... Ts> template <class... Ts>
void operator()(typed_response_promise<Ts...>&) { void operator()(typed_response_promise<Ts...>&) {
(*this)(); (*this)();
} }
/// Calls `(*this)()`.
template <class... Ts> template <class... Ts>
void operator()(delegated<Ts...>&) { void operator()(delegated<Ts...>&) {
(*this)(); (*this)();
} }
template <class T> /// Calls `(*this)()`.
void operator()(stream<T>&) { template <class Out, class... Ts>
void operator()(output_stream<Out, Ts...>&) {
(*this)(); (*this)();
} }
// visit API - returns true if T was visited, false if T was skipped /// Calls `(*this)()`.
template <class Out, class... Ts>
void operator()(outbound_stream_slot<Out, Ts...>&) {
(*this)();
}
/// Calls `(*this)()`.
template <class In>
void operator()(inbound_stream_slot<In>&) {
(*this)();
}
/// Calls `(*this)()`.
template <class In>
void operator()(make_sink_result<In>&) {
(*this)();
}
/// Calls `(*this)()`.
template <class DownstreamManager, class... Ts>
void operator()(make_source_result<DownstreamManager, Ts...>&) {
(*this)();
}
/// Calls `(*this)()`.
template <class In, class DownstreamManager, class... Ts>
void operator()(make_stage_result<In, DownstreamManager, Ts...>&) {
(*this)();
}
// -- visit API: return true if T was visited, false if T was skipped --------
/// Delegates `x` to the appropriate handler and returns `true`.
template <class T> template <class T>
bool visit(T& x) { bool visit(T& x) {
(*this)(x); (*this)(x);
return true; return true;
} }
/// Returns `false`.
inline bool visit(skip_t&) { inline bool visit(skip_t&) {
return false; return false;
} }
/// Returns `false`.
inline bool visit(const skip_t&) { inline bool visit(const skip_t&) {
return false; return false;
} }
template <class T> /// Returns `false` if `x != none`, otherwise calls the void handler and
bool visit(stream<T>&) { /// returns `true`..
return true;
}
inline bool visit(optional<skip_t>& x) { inline bool visit(optional<skip_t>& x) {
if (x) if (x)
return false; return false;
...@@ -154,6 +205,7 @@ public: ...@@ -154,6 +205,7 @@ public:
return true; return true;
} }
/// Dispatches on the runtime-type of `x`.
template <class... Ts> template <class... Ts>
bool visit(result<Ts...>& x) { bool visit(result<Ts...>& x) {
switch (x.flag) { switch (x.flag) {
......
...@@ -122,7 +122,7 @@ public: ...@@ -122,7 +122,7 @@ public:
} }
inline explicit operator bool() const noexcept { inline explicit operator bool() const noexcept {
return static_cast<bool>(ptr_); return ptr_ != nullptr;
} }
inline message_data* get() const noexcept { inline message_data* get() const noexcept {
...@@ -134,6 +134,22 @@ private: ...@@ -134,6 +134,22 @@ private:
intrusive_ptr<message_data> ptr_; intrusive_ptr<message_data> ptr_;
}; };
inline bool operator==(const message_data::cow_ptr& x, std::nullptr_t) {
return x.get() == nullptr;
}
inline bool operator==(std::nullptr_t, const message_data::cow_ptr& x) {
return x.get() == nullptr;
}
inline bool operator!=(const message_data::cow_ptr& x, std::nullptr_t) {
return x.get() != nullptr;
}
inline bool operator!=(std::nullptr_t, const message_data::cow_ptr& x) {
return x.get() != nullptr;
}
} // namespace detail } // namespace detail
} // namespace caf } // namespace caf
......
...@@ -29,30 +29,6 @@ ...@@ -29,30 +29,6 @@
namespace caf { namespace caf {
namespace detail { namespace detail {
/*
template <class Input, class X, class... Ts>
struct mpi_splice_one;
template <class Input, class X>
struct mpi_splice_one<Input, X> {
using type = X;
};
template <class Input, class X, class... Ts>
struct mpi_splice_one<Input, X, none_t, Ts...> {
using type = none_t;
};
template <class Input, class... Ys, class... Zs, class... Ts>
struct mpi_splice_one<Input,
typed_mpi<Input, type_list<Ys...>>,
typed_mpi<Input, type_list<Zs...>>,
Ts...>
: mpi_splice_one<typed_mpi<type_list<Xs...>, type_list<Ys..., Zs...>>, Ts...> {
// combine signatures with same input
};
*/
template <class T, class... Lists> template <class T, class... Lists>
struct mpi_splice_by_input; struct mpi_splice_by_input;
......
...@@ -5,7 +5,8 @@ ...@@ -5,7 +5,8 @@
* | |___ / ___ \| _| Framework * * | |___ / ___ \| _| Framework *
* \____/_/ \_|_| * * \____/_/ \_|_| *
* * * *
* Copyright 2011-2018 Dominik Charousset * * Copyright (C) 2011 - 2017 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* * * *
* 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,42 +17,38 @@ ...@@ -16,42 +17,38 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_PUSH5_SCATTERER_HPP #ifndef CAF_DETAIL_OVERLOAD_HPP
#define CAF_PUSH5_SCATTERER_HPP #define CAF_DETAIL_OVERLOAD_HPP
#include "caf/broadcast_scatterer.hpp"
namespace caf { namespace caf {
namespace detail { namespace detail {
/// Always pushs exactly 5 elements to sinks. Used in unit tests only. template <class... Fs>
template <class T> struct overload;
class push5_scatterer : public broadcast_scatterer<T> {
public:
using super = broadcast_scatterer<T>;
push5_scatterer(local_actor* self) : super(self) { template <class F>
struct overload<F> : F {
using F::operator();
overload(F f) : F(f) {
// nop // nop
} }
};
using super::min_batch_size; template <class F, class... Fs>
using super::max_batch_size; struct overload<F, Fs...> : F, overload<Fs...> {
using super::min_buffer_size; using F::operator();
using overload<Fs...>::operator();
long min_batch_size() const override { overload(F f, Fs... fs) : F(f), overload<Fs...>(fs...) {
return 1; // nop
}
long max_batch_size() const override {
return 5;
}
long min_buffer_size() const override {
return 5;
} }
}; };
template <class... Fs>
overload<Fs...> make_overload(Fs... fs) {
return {fs...};
}
} // namespace detail } // namespace detail
} // namespace caf } // namespace caf
#endif // CAF_PUSH5_SCATTERER_HPP #endif // CAF_DETAIL_OVERLOAD_HPP
...@@ -16,43 +16,62 @@ ...@@ -16,43 +16,62 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_BROADCAST_SCATTERER_HPP #ifndef CAF_DETAIL_PATH_STATE_HPP
#define CAF_BROADCAST_SCATTERER_HPP #define CAF_DETAIL_PATH_STATE_HPP
#include "caf/buffered_scatterer.hpp" #include <new>
#include <vector>
#include "caf/unit.hpp"
namespace caf { namespace caf {
namespace detail {
/// Bundles a filter and a buffer.
template <class Filter, class T>
struct path_state {
Filter filter;
std::vector<T> buf;
};
/// Compress path_state if `Filter` is `unit`.
template <class T> template <class T>
class broadcast_scatterer : public buffered_scatterer<T> { struct path_state<unit_t, T> {
public: using buffer_type = std::vector<T>;
using super = buffered_scatterer<T>;
union {
unit_t filter;
buffer_type buf;
};
path_state() {
new (&buf) buffer_type();
}
path_state(path_state&& other) {
new (&buf) buffer_type(std::move(other.buf));
}
path_state(const path_state& other) {
new (&buf) buffer_type(other.buf);
}
broadcast_scatterer(local_actor* selfptr) : super(selfptr) { path_state& operator=(path_state&& other) {
// nop buf = std::move(other.buf);
return *this;
} }
long credit() const override { path_state& operator=(const path_state& other) {
// We receive messages until we have exhausted all downstream credit and buf = other.buf;
// have filled our buffer to its minimum size. return *this;
return this->min_credit() + this->min_buffer_size();
} }
void emit_batches() override { ~path_state() {
CAF_LOG_TRACE(""); buf.~buffer_type();
auto chunk = this->get_chunk(this->min_credit());
auto csize = static_cast<long>(chunk.size());
CAF_LOG_TRACE(CAF_ARG(chunk));
if (csize == 0)
return;
auto wrapped_chunk = make_message(std::move(chunk));
for (auto& x : this->paths_) {
CAF_ASSERT(x->open_credit >= csize);
x->emit_batch(csize, wrapped_chunk);
}
} }
}; };
} // namespace detail
} // namespace caf } // namespace caf
#endif // CAF_BROADCAST_SCATTERER_HPP #endif // CAF_DETAIL_PATH_STATE_HPP
...@@ -16,27 +16,20 @@ ...@@ -16,27 +16,20 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_PULL5_GATHERER_HPP #ifndef CAF_DETAIL_SELECT_ALL_HPP
#define CAF_PULL5_GATHERER_HPP #define CAF_DETAIL_SELECT_ALL_HPP
#include "caf/random_gatherer.hpp"
namespace caf { namespace caf {
namespace detail { namespace detail {
/// Always pulls exactly 5 elements from sources. Used in unit tests only. struct select_all {
class pull5_gatherer : public random_gatherer { template <class T, class U>
public: constexpr bool operator()(const T&, const U&) const noexcept {
using super = random_gatherer; return true;
}
pull5_gatherer(local_actor* selfptr);
void assign_credit(long available) override;
long initial_credit(long, inbound_path*) override;
}; };
} // namespace detail } // namespace detail
} // namespace caf } // namespace caf
#endif // CAF_PULL5_GATHERER_HPP #endif // CAF_DETAIL_SELECT_ALL_HPP
...@@ -37,9 +37,10 @@ public: ...@@ -37,9 +37,10 @@ public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
/// Request for a `timeout_msg`. /// Request for a `timeout_msg`.
struct receive_timeout { struct ordinary_timeout {
strong_actor_ptr self; strong_actor_ptr self;
uint32_t id; atom_value type;
uint64_t id;
}; };
/// Request for a `sec::request_timeout` error. /// Request for a `sec::request_timeout` error.
...@@ -61,14 +62,15 @@ public: ...@@ -61,14 +62,15 @@ public:
message content; message content;
}; };
using value_type = variant<receive_timeout, request_timeout, using value_type = variant<ordinary_timeout, request_timeout,
actor_msg, group_msg>; actor_msg, group_msg>;
using map_type = std::multimap<time_point, value_type>; using map_type = std::multimap<time_point, value_type>;
using secondary_map = std::multimap<abstract_actor*, map_type::iterator>; using secondary_map = std::multimap<abstract_actor*, map_type::iterator>;
struct receive_predicate { struct ordinary_predicate {
atom_value type;
bool operator()(const secondary_map::value_type& x) const noexcept; bool operator()(const secondary_map::value_type& x) const noexcept;
}; };
...@@ -80,7 +82,7 @@ public: ...@@ -80,7 +82,7 @@ public:
struct visitor { struct visitor {
simple_actor_clock* thisptr; simple_actor_clock* thisptr;
void operator()(receive_timeout& x); void operator()(ordinary_timeout& x);
void operator()(request_timeout& x); void operator()(request_timeout& x);
...@@ -89,13 +91,13 @@ public: ...@@ -89,13 +91,13 @@ public:
void operator()(group_msg& x); void operator()(group_msg& x);
}; };
void set_receive_timeout(time_point t, abstract_actor* self, void set_ordinary_timeout(time_point t, abstract_actor* self,
uint32_t id) override; atom_value type, uint64_t id) override;
void set_request_timeout(time_point t, abstract_actor* self, void set_request_timeout(time_point t, abstract_actor* self,
message_id id) override; message_id id) override;
void cancel_receive_timeout(abstract_actor* self) override; void cancel_ordinary_timeout(abstract_actor* self, atom_value type) override;
void cancel_request_timeout(abstract_actor* self, message_id id) override; void cancel_request_timeout(abstract_actor* self, message_id id) override;
...@@ -107,6 +109,8 @@ public: ...@@ -107,6 +109,8 @@ public:
void schedule_message(time_point t, group target, strong_actor_ptr sender, void schedule_message(time_point t, group target, strong_actor_ptr sender,
message content) override; message content) override;
void cancel_all() override;
inline const map_type& schedule() const { inline const map_type& schedule() const {
return schedule_; return schedule_;
} }
...@@ -117,7 +121,8 @@ public: ...@@ -117,7 +121,8 @@ public:
protected: protected:
template <class Predicate> template <class Predicate>
secondary_map::iterator lookup(abstract_actor* self, Predicate pred) { secondary_map::iterator lookup(abstract_actor* self,
Predicate pred) {
auto e = actor_lookup_.end(); auto e = actor_lookup_.end();
auto range = actor_lookup_.equal_range(self); auto range = actor_lookup_.equal_range(self);
if (range.first == range.second) if (range.first == range.second)
...@@ -142,8 +147,10 @@ protected: ...@@ -142,8 +147,10 @@ protected:
actor_lookup_.erase(i); actor_lookup_.erase(i);
} }
/// Timeout schedule.
map_type schedule_; map_type schedule_;
/// Secondary index for accessing timeouts by actor.
secondary_map actor_lookup_; secondary_map actor_lookup_;
}; };
......
This diff is collapsed.
...@@ -22,11 +22,10 @@ ...@@ -22,11 +22,10 @@
#include <memory> #include <memory>
#include <unordered_map> #include <unordered_map>
#include "caf/logger.hpp"
#include "caf/ref_counted.hpp" #include "caf/ref_counted.hpp"
#include "caf/stream_manager.hpp"
#include "caf/random_gatherer.hpp"
#include "caf/scheduled_actor.hpp" #include "caf/scheduled_actor.hpp"
#include "caf/broadcast_scatterer.hpp" #include "caf/stream_manager.hpp"
namespace caf { namespace caf {
namespace detail { namespace detail {
...@@ -44,80 +43,26 @@ namespace detail { ...@@ -44,80 +43,26 @@ namespace detail {
/// Policies need to provide the following member types and functions: /// Policies need to provide the following member types and functions:
/// ///
/// ~~~{.cpp} /// ~~~{.cpp}
/// struct policy { /// TODO
/// /// Any number of user-defined arguments.
/// policy(...);
///
/// /// Tuple of available substream scatterers to subscribers. Each
/// /// element of the tuple is a subtype of `topic_scatterer<T>` (or
/// /// provides a similar interface).
/// using substream_scatterers = ...;
///
/// /// A compile-time index type identifying individual substreams.
/// using substream_index_type = ...;
///
/// /// Represent a single topic for filtering stream data.
/// using topic_type = ...;
///
/// /// Groups multiple topics into a single selection filter.
/// using filter_type = ...;
///
/// /// Policy for gathering data from peers.
/// using gatherer_type = ...;
///
/// /// Policy for scattering data to peers.
/// using scatterer_type = ...;
///
/// /// Decides whether a filter applies to a given message.
/// bool selected(const filter_type& sieve, const message& sand) const;
///
/// // + one overload to `selected` for each substream data type.
///
/// /// Accesses individual substream scatterers by index.
/// auto& substream(substream_scatterers&, substream_index_type);
///
/// /// Accesses individual substream scatterers by stream ID.
/// stream_scatterer* substream(substream_scatterers&, const stream_id&);
///
/// /// Returns true if the substreams have no data pending and the policy
/// /// decides to shutdown this peer.
/// bool at_end() const;
///
/// /// Returns true if outgoing data originating on this peer is
/// /// forwarded to all substreams. Otherwise, the substreams receive data
/// /// produced on other peers exclusively.
/// bool substream_local_data() const;
///
/// /// Handles a batch if matches the data types managed by any of the
/// /// substreams. Returns `none` if the batch is a peer message, otherwise
/// /// the error resulting from handling the batch.
/// optional<error> batch(const stream_id& sid, const actor_addr& hdl,
/// long xs_size, message& xs, int64_t xs_id);
///
/// ///
/// void push_to_substreams(message msg);
/// }; /// };
/// ~~~ /// ~~~
template <class Policy> template <class Policy>
class stream_distribution_tree : public stream_manager { class stream_distribution_tree : public stream_manager {
public: public:
// -- nested types ----------------------------------------------------------- // -- nested types -----------------------------------------------------------
using super = stream_manager;
using gatherer_type = typename Policy::gatherer_type; using super = stream_manager;
using scatterer_type = typename Policy::scatterer_type; using downstream_manager_type = typename Policy::downstream_manager_type;
// --- constructors and destructors ------------------------------------------ // --- constructors and destructors ------------------------------------------
template <class... Ts> template <class... Ts>
stream_distribution_tree(scheduled_actor* selfptr, Ts&&... xs) stream_distribution_tree(scheduled_actor* selfptr, Ts&&... xs)
: self_(selfptr), : super(selfptr),
in_(selfptr),
out_(selfptr), out_(selfptr),
policy_(this, std::forward<Ts>(xs)...) { policy_(this, std::forward<Ts>(xs)...) {
// nop continuous(true);
} }
~stream_distribution_tree() override { ~stream_distribution_tree() override {
...@@ -125,7 +70,7 @@ public: ...@@ -125,7 +70,7 @@ public:
} }
// -- Accessors -------------------------------------------------------------- // -- Accessors --------------------------------------------------------------
inline Policy& policy() { inline Policy& policy() {
return policy_; return policy_;
} }
...@@ -134,109 +79,71 @@ public: ...@@ -134,109 +79,71 @@ public:
return policy_; return policy_;
} }
void close_remote_input();
// -- Overridden member functions of `stream_manager` ------------------------
/// Terminates the core actor with log output `log_message` if `at_end`
/// returns `true`.
void shutdown_if_at_end(const char* log_message) {
CAF_IGNORE_UNUSED(log_message);
if (policy_.at_end()) {
CAF_LOG_DEBUG(log_message);
self_->quit(caf::exit_reason::user_shutdown);
}
}
// -- overridden member functions of `stream_manager` ------------------------ // -- overridden member functions of `stream_manager` ------------------------
error ack_open(const stream_id& sid, const actor_addr& rebind_from, void handle(inbound_path* path, downstream_msg::batch& x) override {
strong_actor_ptr rebind_to, long initial_demand, CAF_LOG_TRACE(CAF_ARG(path) << CAF_ARG(x));
bool redeployable) override { auto slot = path->slots.receiver;
CAF_LOG_TRACE(CAF_ARG(sid) << CAF_ARG(rebind_from) << CAF_ARG(rebind_to) policy_.before_handle_batch(slot, path->hdl);
<< CAF_ARG(initial_demand) << CAF_ARG(redeployable)); policy_.handle_batch(slot, path->hdl, x.xs);
auto res = super::ack_open(sid, rebind_from, rebind_to, initial_demand, policy_.after_handle_batch(slot, path->hdl);
redeployable);
if (res == none)
policy_.ack_open_success(sid, rebind_from, rebind_to);
else
policy_.ack_open_failure(sid, rebind_from, rebind_to, res);
return res;
} }
error process_batch(message& xs) override { void handle(inbound_path* path, downstream_msg::close& x) override {
CAF_LOG_TRACE(CAF_ARG(xs)); CAF_LOG_TRACE(CAF_ARG(path) << CAF_ARG(x));
policy_.handle_batch(xs); CAF_IGNORE_UNUSED(x);
return none; policy_.path_closed(path->slots.receiver);
} }
error batch(const stream_id& sid, const actor_addr& hdl, long xs_size, void handle(inbound_path* path, downstream_msg::forced_close& x) override {
message& xs, int64_t xs_id) override { CAF_LOG_TRACE(CAF_ARG(path) << CAF_ARG(x));
CAF_LOG_TRACE(CAF_ARG(sid) << CAF_ARG(hdl) << CAF_ARG(xs_size) policy_.path_force_closed(path->slots.receiver, x.reason);
<< CAF_ARG(xs) << CAF_ARG(xs_id));
policy_.before_handle_batch(sid, hdl, xs_size, xs, xs_id);
auto res = super::batch(sid, hdl, xs_size, xs, xs_id);
policy_.after_handle_batch(sid, hdl, xs_id);
push();
return res;
} }
bool done() const override { 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; return false;
} }
gatherer_type& in() override { void handle(stream_slots slots, upstream_msg::drop& x) override {
return in_; CAF_LOG_TRACE(CAF_ARG(slots) << CAF_ARG(x));
CAF_IGNORE_UNUSED(x);
auto slot = slots.receiver;
if (out().remove_path(slots.receiver, none, true))
policy_.path_dropped(slot);
} }
scatterer_type& out() override { void handle(stream_slots slots, upstream_msg::forced_drop& x) override {
return out_; CAF_LOG_TRACE(CAF_ARG(slots) << CAF_ARG(x));
} CAF_IGNORE_UNUSED(x);
auto slot = slots.receiver;
void downstream_demand(outbound_path*, long) override { if (out().remove_path(slots.receiver, x.reason, true))
CAF_LOG_TRACE(""); policy_.path_force_dropped(slot, x.reason);
push();
in_.assign_credit(out_.credit());
}
error close(const stream_id& sid, const actor_addr& hdl) override {
CAF_LOG_TRACE(CAF_ARG(sid) << CAF_ARG(hdl));
if (in_.remove_path(sid, hdl, none, true))
return policy_.path_closed(sid, hdl);
return none;
} }
error drop(const stream_id& sid, const actor_addr& hdl) override { bool done() const override {
CAF_LOG_TRACE(CAF_ARG(sid) << CAF_ARG(hdl)); return !continuous() && pending_handshakes_ == 0
if (out_.remove_path(sid, hdl, none, true)) && inbound_paths_.empty() && out_.clean();
return policy_.path_dropped(sid, hdl);
return none;
}
error forced_close(const stream_id& sid, const actor_addr& hdl,
error reason) override {
CAF_LOG_TRACE(CAF_ARG(sid) << CAF_ARG(hdl) << CAF_ARG(reason));
if (in_.remove_path(sid, hdl, reason, true))
return policy_.path_force_closed(sid, hdl, std::move(reason));
return none;
} }
error forced_drop(const stream_id& sid, const actor_addr& hdl, bool idle() const noexcept override {
error reason) override { // Same as `stream_stage<...>`::idle().
if (out_.remove_path(sid, hdl, reason, true)) return out_.stalled() || (out_.clean() && this->inbound_paths_up_to_date());
return policy_.path_force_dropped(sid, hdl, std::move(reason));
return none;
} }
scheduled_actor* self() { downstream_manager_type& out() override {
return self_; return out_;
} }
private: private:
scheduled_actor* self_; downstream_manager_type out_;
gatherer_type in_;
scatterer_type out_;
Policy policy_; Policy policy_;
}; };
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2017 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* 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. *
******************************************************************************/
#ifndef CAF_STREAM_SINK_DRIVER_IMPL_HPP
#define CAF_STREAM_SINK_DRIVER_IMPL_HPP
#include "caf/none.hpp"
#include "caf/stream_finalize_trait.hpp"
#include "caf/stream_sink_driver.hpp"
#include "caf/stream_sink_trait.hpp"
namespace caf {
namespace 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 detail
} // namespace caf
#endif // CAF_STREAM_SINK_DRIVER_IMPL_HPP
...@@ -16,79 +16,66 @@ ...@@ -16,79 +16,66 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_STREAM_SINK_IMPL_HPP #ifndef CAF_DETAIL_STREAM_SINK_IMPL_HPP
#define CAF_STREAM_SINK_IMPL_HPP #define CAF_DETAIL_STREAM_SINK_IMPL_HPP
#include "caf/sec.hpp" #include "caf/config.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/make_counted.hpp"
#include "caf/message_id.hpp" #include "caf/message_id.hpp"
#include "caf/sec.hpp"
#include "caf/stream_manager.hpp" #include "caf/stream_manager.hpp"
#include "caf/stream_sink_trait.hpp" #include "caf/stream_sink.hpp"
#include "caf/policy/arg.hpp"
namespace caf { namespace caf {
namespace detail {
template <class Fun, class Finalize, class Gatherer, class Scatterer> template <class Driver>
class stream_sink_impl : public stream_manager { class stream_sink_impl : public Driver::sink_type {
public: public:
using super = stream_manager; using super = typename Driver::sink_type;
using trait = stream_sink_trait_t<Fun, Finalize>;
using state_type = typename trait::state; using driver_type = Driver;
using input_type = typename trait::input; using input_type = typename driver_type::input_type;
using output_type = typename trait::output; template <class... Ts>
stream_sink_impl(scheduled_actor* self, Ts&&... xs)
stream_sink_impl(local_actor* self, Fun fun, Finalize fin) : stream_manager(self),
: fun_(std::move(fun)), super(self),
fin_(std::move(fin)), driver_(std::forward<Ts>(xs)...) {
in_(self) {
// nop // nop
} }
state_type& state() { void handle(inbound_path*, downstream_msg::batch& x) override {
return state_; CAF_LOG_TRACE(CAF_ARG(x));
}
Gatherer& in() override {
return in_;
}
Scatterer& out() override {
return out_;
}
bool done() const override {
return in_.closed();
}
protected:
error process_batch(message& msg) override {
CAF_LOG_TRACE(CAF_ARG(msg));
using vec_type = std::vector<input_type>; using vec_type = std::vector<input_type>;
if (msg.match_elements<vec_type>()) { if (x.xs.match_elements<vec_type>()) {
auto& xs = msg.get_as<vec_type>(0); driver_.process(x.xs.get_mutable_as<vec_type>(0));
for (auto& x : xs) return;
fun_(state_, x);
return none;
} }
CAF_LOG_ERROR("received unexpected batch type"); CAF_LOG_ERROR("received unexpected batch type (dropped)");
return sec::unexpected_message;
} }
message make_final_result() override { protected:
return trait::make_result(state_, fin_); void finalize(const error& reason) override {
driver_.finalize(reason);
} }
private: private:
state_type state_; driver_type driver_;
Fun fun_;
Finalize fin_;
Gatherer in_;
Scatterer out_;
}; };
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 detail
} // namespace caf } // namespace caf
#endif // CAF_STREAM_SINK_IMPL_HPP #endif // CAF_DETAIL_STREAM_SINK_IMPL_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2017 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* 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. *
******************************************************************************/
#ifndef CAF_STREAM_SOURCE_DRIVER_IMPL_HPP
#define CAF_STREAM_SOURCE_DRIVER_IMPL_HPP
#include "caf/none.hpp"
#include "caf/stream_finalize_trait.hpp"
#include "caf/stream_source_driver.hpp"
#include "caf/stream_source_trait.hpp"
namespace caf {
namespace 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 detail
} // namespace caf
#endif // CAF_STREAM_SOURCE_DRIVER_IMPL_HPP
...@@ -16,87 +16,80 @@ ...@@ -16,87 +16,80 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_STREAM_SOURCE_IMPL_HPP #ifndef CAF_DETAIL_STREAM_SOURCE_IMPL_HPP
#define CAF_STREAM_SOURCE_IMPL_HPP #define CAF_DETAIL_STREAM_SOURCE_IMPL_HPP
#include "caf/logger.hpp"
#include "caf/downstream.hpp" #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/outbound_path.hpp"
#include "caf/stream_manager.hpp" #include "caf/stream_source.hpp"
#include "caf/stream_source_trait.hpp" #include "caf/stream_source_trait.hpp"
#include "caf/invalid_stream_gatherer.hpp"
namespace caf { namespace caf {
namespace detail {
template <class Fun, class Predicate, class DownstreamPolicy> template <class Driver>
class stream_source_impl : public stream_manager { class stream_source_impl : public Driver::source_type {
public: public:
using trait = stream_source_trait_t<Fun>; // -- member types -----------------------------------------------------------
using state_type = typename trait::state; using super = typename Driver::source_type;
using output_type = typename trait::output; using driver_type = Driver;
stream_source_impl(local_actor* self, Fun fun, Predicate pred) // -- constructors, destructors, and assignment operators --------------------
: fun_(std::move(fun)),
pred_(std::move(pred)),
out_(self) {
// nop
}
bool done() const override { template <class... Ts>
return at_end() && out_.paths_clean(); stream_source_impl(scheduled_actor* self, Ts&&... xs)
: stream_manager(self),
super(self),
at_end_(false),
driver_(std::forward<Ts>(xs)...) {
// nop
} }
invalid_stream_gatherer& in() override { // -- implementation of virtual functions ------------------------------------
return in_;
}
DownstreamPolicy& out() override { bool done() const override {
return out_; return this->pending_handshakes_ == 0 && at_end_ && this->out_.clean();
} }
bool generate_messages() override { bool generate_messages() override {
// produce new elements CAF_LOG_TRACE("");
auto capacity = out_.credit() - out_.buffered(); if (at_end_)
if (capacity <= 0)
return false; return false;
downstream<typename DownstreamPolicy::value_type> ds{out_.buf()}; auto hint = this->out_.capacity();
fun_(state_, ds, static_cast<size_t>(capacity)); CAF_LOG_DEBUG(CAF_ARG(hint));
return true; if (hint == 0)
} return false;
downstream<typename Driver::output_type> ds{this->out_.buf()};
state_type& state() { driver_.pull(ds, hint);
return state_; if (driver_.done())
at_end_ = true;
return hint != this->out_.capacity();
} }
protected: protected:
bool at_end() const { void finalize(const error& reason) override {
return pred_(state_); driver_.finalize(reason);
}
void downstream_demand(outbound_path* path, long) override {
CAF_LOG_TRACE(CAF_ARG(path));
if (!at_end()) {
generate_messages();
push();
} else if (out_.buffered() > 0) {
push();
} else {
auto sid = path->sid;
auto hdl = path->hdl;
out().remove_path(sid, hdl, none, false);
}
} }
private: private:
state_type state_; bool at_end_;
Fun fun_; Driver driver_;
Predicate pred_;
DownstreamPolicy out_;
invalid_stream_gatherer in_;
}; };
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 detail
} // namespace caf } // namespace caf
#endif // CAF_STREAM_SOURCE_IMPL_HPP #endif // CAF_DETAIL_STREAM_SOURCE_IMPL_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2017 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* 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. *
******************************************************************************/
#ifndef CAF_STREAM_STAGE_DRIVER_IMPL_HPP
#define CAF_STREAM_STAGE_DRIVER_IMPL_HPP
#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 {
namespace 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(Init init, Process f, Finalize fin)
: 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_;
message result_;
};
} // namespace detail
} // namespace caf
#endif // CAF_STREAM_STAGE_DRIVER_IMPL_HPP
...@@ -16,107 +16,85 @@ ...@@ -16,107 +16,85 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_STREAM_STAGE_IMPL_HPP #ifndef CAF_DETAIL_STREAM_STAGE_IMPL_HPP
#define CAF_STREAM_STAGE_IMPL_HPP #define CAF_DETAIL_STREAM_STAGE_IMPL_HPP
#include "caf/sec.hpp"
#include "caf/logger.hpp"
#include "caf/downstream.hpp" #include "caf/downstream.hpp"
#include "caf/logger.hpp"
#include "caf/make_counted.hpp"
#include "caf/outbound_path.hpp" #include "caf/outbound_path.hpp"
#include "caf/sec.hpp"
#include "caf/stream_manager.hpp" #include "caf/stream_manager.hpp"
#include "caf/stream_stage.hpp"
#include "caf/stream_stage_trait.hpp" #include "caf/stream_stage_trait.hpp"
namespace caf { namespace caf {
namespace detail {
template <class Fun, class Cleanup, template <class Driver>
class UpstreamPolicy, class DownstreamPolicy> class stream_stage_impl : public Driver::stage_type {
class stream_stage_impl : public stream_manager {
public: public:
using trait = stream_stage_trait_t<Fun>; // -- member types -----------------------------------------------------------
using state_type = typename trait::state; using super = typename Driver::stage_type;
using input_type = typename trait::input; using driver_type = Driver;
using output_type = typename trait::output; using downstream_manager_type = typename Driver::downstream_manager_type;
stream_stage_impl(local_actor* self, const stream_id&, using input_type = typename driver_type::input_type;
Fun fun, Cleanup cleanup)
: fun_(std::move(fun)),
cleanup_(std::move(cleanup)),
in_(self),
out_(self) {
// nop
}
state_type& state() { using output_type = typename driver_type::output_type;
return state_;
}
UpstreamPolicy& in() override { // -- constructors, destructors, and assignment operators --------------------
return in_;
}
DownstreamPolicy& out() override { template <class... Ts>
return out_; stream_stage_impl(scheduled_actor* self, Ts&&... xs)
: stream_manager(self),
super(self),
driver_(std::forward<Ts>(xs)...) {
// nop
} }
bool done() const override { // -- implementation of virtual functions ------------------------------------
return in_.closed() && out_.closed();
}
protected: bool done() const override {
void input_closed(error reason) override { return !this->continuous() && this->pending_handshakes_ == 0
if (reason == none) { && this->inbound_paths_.empty() && this->out_.clean();
if (out_.buffered() == 0)
out_.close();
} else {
out_.abort(std::move(reason));
}
} }
error process_batch(message& msg) override { void handle(inbound_path*, downstream_msg::batch& x) override {
CAF_LOG_TRACE(CAF_ARG(msg)); CAF_LOG_TRACE(CAF_ARG(x));
using vec_type = std::vector<output_type>; using vec_type = std::vector<input_type>;
if (msg.match_elements<vec_type>()) { if (x.xs.match_elements<vec_type>()) {
auto& xs = msg.get_as<vec_type>(0); downstream<output_type> ds{this->out_.buf()};
downstream<typename DownstreamPolicy::value_type> ds{out_.buf()}; driver_.process(ds, x.xs.get_mutable_as<vec_type>(0));
for (auto& x : xs) return;
fun_(state_, ds, x);
return none;
} }
CAF_LOG_ERROR("received unexpected batch type"); CAF_LOG_ERROR("received unexpected batch type (dropped)");
return sec::unexpected_message;
} }
message make_output_token(const stream_id& x) const override { bool congested() const noexcept override {
return make_message(stream<output_type>{x}); return this->out_.capacity() == 0;
} }
void downstream_demand(outbound_path* path, long) override { protected:
CAF_LOG_TRACE(CAF_ARG(path)); void finalize(const error& reason) override {
auto hdl = path->hdl; driver_.finalize(reason);
if(out_.buffered() > 0)
push();
else if (in_.closed()) {
// don't pass path->hdl: path can become invalid
auto sid = path->sid;
out_.remove_path(sid, hdl, none, false);
}
auto current_size = out_.buffered();
auto desired_size = out_.credit();
if (current_size < desired_size)
in_.assign_credit(desired_size - current_size);
} }
private: private:
state_type state_; driver_type driver_;
Fun fun_;
Cleanup cleanup_;
UpstreamPolicy in_;
DownstreamPolicy out_;
}; };
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 detail
} // namespace caf } // namespace caf
#endif // CAF_STREAM_STAGE_IMPL_HPP #endif // CAF_DETAIL_STREAM_STAGE_IMPL_HPP
...@@ -19,10 +19,11 @@ ...@@ -19,10 +19,11 @@
#ifndef CAF_DETAIL_STRINGIFICATION_INSPECTOR_HPP #ifndef CAF_DETAIL_STRINGIFICATION_INSPECTOR_HPP
#define CAF_DETAIL_STRINGIFICATION_INSPECTOR_HPP #define CAF_DETAIL_STRINGIFICATION_INSPECTOR_HPP
#include <string> #include <chrono>
#include <vector>
#include <functional> #include <functional>
#include <string>
#include <type_traits> #include <type_traits>
#include <vector>
#include "caf/atom.hpp" #include "caf/atom.hpp"
#include "caf/none.hpp" #include "caf/none.hpp"
...@@ -145,6 +146,17 @@ public: ...@@ -145,6 +146,17 @@ public:
result_ += ']'; result_ += ']';
} }
template <class T>
enable_if_t<has_peek_all<T>::value
&& !is_iterable<T>::value // pick begin()/end() over peek_all
&& !is_inspectable<stringification_inspector, T>::value
&& !has_to_string<T>::value>
consume(T& xs) {
result_ += '[';
xs.peek_all(*this);
result_ += ']';
}
template <class T> template <class T>
void consume(T* xs, size_t n) { void consume(T* xs, size_t n) {
result_ += '('; result_ += '(';
...@@ -175,10 +187,53 @@ public: ...@@ -175,10 +187,53 @@ public:
} }
} }
/// Print duration types with nanosecond resolution.
template <class Rep>
void consume(std::chrono::duration<Rep, std::nano>& x) {
result_ += std::to_string(x.count());
result_ += "ns";
}
/// Print duration types with microsecond resolution.
template <class Rep>
void consume(std::chrono::duration<Rep, std::micro>& x) {
result_ += std::to_string(x.count());
result_ += "us";
}
/// Print duration types with millisecond resolution.
template <class Rep>
void consume(std::chrono::duration<Rep, std::milli>& x) {
result_ += std::to_string(x.count());
result_ += "ms";
}
/// Print duration types with second resolution.
template <class Rep>
void consume(std::chrono::duration<Rep, std::ratio<1>>& x) {
result_ += std::to_string(x.count());
result_ += "s";
}
/// Print duration types with minute resolution.
template <class Rep>
void consume(std::chrono::duration<Rep, std::ratio<60>>& x) {
result_ += std::to_string(x.count());
result_ += "min";
}
/// Print duration types with hour resolution.
template <class Rep>
void consume(std::chrono::duration<Rep, std::ratio<3600>>& x) {
result_ += std::to_string(x.count());
result_ += "h";
}
/// Fallback printing `<unprintable>`. /// Fallback printing `<unprintable>`.
template <class T> template <class T>
enable_if_t< enable_if_t<
!is_iterable<T>::value !is_iterable<T>::value
&& !has_peek_all<T>::value
&& !std::is_pointer<T>::value && !std::is_pointer<T>::value
&& !is_inspectable<stringification_inspector, T>::value && !is_inspectable<stringification_inspector, T>::value
&& !std::is_arithmetic<T>::value && !std::is_arithmetic<T>::value
......
...@@ -21,26 +21,29 @@ ...@@ -21,26 +21,29 @@
#include <cstdint> #include <cstdint>
#include "caf/error.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/exit_reason.hpp"
namespace caf { #include "caf/intrusive/task_result.hpp"
class actor_addr;
class message_id;
class local_actor;
class mailbox_element;
} // namespace caf
namespace caf { namespace caf {
namespace detail { namespace detail {
/// Drains a mailbox and sends an error message to each unhandled request.
struct sync_request_bouncer { struct sync_request_bouncer {
error rsn; error rsn;
explicit sync_request_bouncer(error r); explicit sync_request_bouncer(error r);
void operator()(const strong_actor_ptr& sender, const message_id& mid) const; void operator()(const strong_actor_ptr& sender, const message_id& mid) const;
void operator()(const mailbox_element& e) const; void operator()(const mailbox_element& e) const;
/// Unwrap WDRR queues. Nesting WDRR queues results in a Key/Queue prefix for
/// each layer of nesting.
template <class Key, class Queue, class... Ts>
intrusive::task_result operator()(const Key&, const Queue&,
const Ts&... xs) const {
(*this)(xs...);
return intrusive::task_result::resume;
}
}; };
} // namespace detail } // namespace detail
......
...@@ -28,8 +28,13 @@ class test_actor_clock : public simple_actor_clock { ...@@ -28,8 +28,13 @@ class test_actor_clock : public simple_actor_clock {
public: public:
time_point current_time; time_point current_time;
test_actor_clock();
time_point now() const noexcept override; time_point now() const noexcept override;
duration_type difference(atom_value measurement, long units, time_point t0,
time_point t1) const noexcept override;
/// Tries to dispatch the next timeout or delayed message regardless of its /// Tries to dispatch the next timeout or delayed message regardless of its
/// timestamp. Returns `false` if `schedule().empty()`, otherwise `true`. /// timestamp. Returns `false` if `schedule().empty()`, otherwise `true`.
bool dispatch_once(); bool dispatch_once();
...@@ -40,6 +45,12 @@ public: ...@@ -40,6 +45,12 @@ public:
/// Advances the time by `x` and dispatches timeouts and delayed messages. /// Advances the time by `x` and dispatches timeouts and delayed messages.
void advance_time(duration_type x); void advance_time(duration_type x);
/// Configures the returned value for `difference`. For example, inserting
/// `('foo', 120ns)` causes the clock to return `units * 120ns` for any call
/// to `difference` with `measurement == 'foo'` regardless of the time points
/// passed to the function.
std::map<atom_value, duration_type> time_per_unit;
}; };
} // namespace detail } // namespace detail
......
...@@ -34,13 +34,13 @@ public: ...@@ -34,13 +34,13 @@ public:
thread_safe_actor_clock(); thread_safe_actor_clock();
void set_receive_timeout(time_point t, abstract_actor* self, void set_ordinary_timeout(time_point t, abstract_actor* self,
uint32_t id) override; atom_value type, uint64_t id) override;
void set_request_timeout(time_point t, abstract_actor* self, void set_request_timeout(time_point t, abstract_actor* self,
message_id id) override; message_id id) override;
void cancel_receive_timeout(abstract_actor* self) override; void cancel_ordinary_timeout(abstract_actor* self, atom_value type) override;
void cancel_request_timeout(abstract_actor* self, message_id id) override; void cancel_request_timeout(abstract_actor* self, message_id id) override;
...@@ -52,6 +52,8 @@ public: ...@@ -52,6 +52,8 @@ public:
void schedule_message(time_point t, group target, strong_actor_ptr sender, void schedule_message(time_point t, group target, strong_actor_ptr sender,
message content) override; message content) override;
void cancel_all() override;
void run_dispatch_loop(); void run_dispatch_loop();
void cancel_dispatch_loop(); void cancel_dispatch_loop();
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2017 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* 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. *
******************************************************************************/
#ifndef CAF_DETAIL_TICK_EMITTER_HPP
#define CAF_DETAIL_TICK_EMITTER_HPP
#include <chrono>
#include <initializer_list>
#include "caf/config.hpp"
namespace caf {
namespace 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 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.
inline 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 diff = now - start_;
auto current_tick_id = diff.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.
long timeouts(time_point t, std::initializer_list<long> 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<long> periods);
private:
time_point start_;
duration_type interval_;
long last_tick_id_;
};
} // namespace detail
} // namespace caf
#endif // CAF_DETAIL_TICK_EMITTER_HPP
...@@ -490,6 +490,20 @@ public: ...@@ -490,6 +490,20 @@ public:
static constexpr bool value = std::is_same<bool, result_type>::value; static constexpr bool value = std::is_same<bool, result_type>::value;
}; };
/// Checks wheter `F` is callable with arguments of types `Ts...`.
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());
template <class U>
static auto sfinae(...) -> std::false_type;
using type = decltype(sfinae<F>(nullptr));
static constexpr bool value = type::value;
};
/// Checks wheter `F` takes mutable references. /// Checks wheter `F` takes mutable references.
/// ///
/// A manipulator is a functor that manipulates its arguments via /// A manipulator is a functor that manipulates its arguments via
...@@ -545,6 +559,19 @@ public: ...@@ -545,6 +559,19 @@ public:
static constexpr bool value = false; static constexpr bool value = false;
}; };
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 char fun(const void*);
public:
static constexpr bool value = sizeof(fun(static_cast<T*>(nullptr))) > 1;
};
CAF_HAS_MEMBER_TRAIT(size); CAF_HAS_MEMBER_TRAIT(size);
CAF_HAS_MEMBER_TRAIT(data); CAF_HAS_MEMBER_TRAIT(data);
...@@ -627,6 +654,34 @@ struct transfer_const<const T, U> { ...@@ -627,6 +654,34 @@ struct transfer_const<const T, U> {
template <class T, class U> template <class T, class U>
using transfer_const_t = typename transfer_const<T, U>::type; using transfer_const_t = typename transfer_const<T, U>::type;
/// Checks whether `T` is an `actor` or a `typed_actor<...>`.
template <class T>
struct is_actor_handle : std::false_type {};
template <>
struct is_actor_handle<actor> : std::true_type {};
template <class... Ts>
struct is_actor_handle<typed_actor<Ts...>> : std::true_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 {};
template <class T>
struct is_result<result<T>> : std::true_type {};
template <class T>
struct is_expected : std::false_type {};
template <class T>
struct is_expected<expected<T>> : std::true_type {};
} // namespace detail } // namespace detail
} // namespace caf } // namespace caf
......
...@@ -23,7 +23,9 @@ ...@@ -23,7 +23,9 @@
#include "caf/delegated.hpp" #include "caf/delegated.hpp"
#include "caf/replies_to.hpp" #include "caf/replies_to.hpp"
#include "caf/response_promise.hpp"
#include "caf/system_messages.hpp" #include "caf/system_messages.hpp"
#include "caf/typed_response_promise.hpp"
#include "caf/detail/type_list.hpp" #include "caf/detail/type_list.hpp"
...@@ -49,6 +51,20 @@ struct input_is { ...@@ -49,6 +51,20 @@ struct input_is {
struct eval : input_is_eval_impl<Arguments, Signature> { }; struct eval : input_is_eval_impl<Arguments, Signature> { };
}; };
template <class... Ts>
struct make_response_promise_helper {
using type = typed_response_promise<Ts...>;
};
template <class... Ts>
struct make_response_promise_helper<typed_response_promise<Ts...>>
: make_response_promise_helper<Ts...> {};
template <>
struct make_response_promise_helper<response_promise> {
using type = response_promise;
};
template <class Output, class F> template <class Output, class F>
struct type_checker { struct type_checker {
static void check() { static void check() {
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2017 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* 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. *
******************************************************************************/
#ifndef CAF_DETAIL_UNORDERED_FLAT_MAP_HPP
#define CAF_DETAIL_UNORDERED_FLAT_MAP_HPP
#include <vector>
#include <algorithm>
#include <stdexcept>
#include <functional>
#include "caf/config.hpp"
#include "caf/detail/comparable.hpp"
namespace caf {
namespace detail {
/// A map abstraction with an unsorted `std::vector` providing `O(n)` lookup.
template <class Key, class T,
class Allocator = std::allocator<std::pair<Key, T>>>
class unordered_flat_map {
public:
// -- member types ----------------------------------------------------------
using key_type = Key;
using mapped_type = T;
using value_type = std::pair<Key, T>;
using vector_type = std::vector<value_type, Allocator>;
using allocator_type = typename vector_type::allocator_type;
using size_type = typename vector_type::size_type;
using difference_type = typename vector_type::difference_type;
using reference = typename vector_type::reference;
using const_reference = typename vector_type::const_reference;
using pointer = typename vector_type::pointer;
using const_pointer = typename vector_type::const_pointer;
using iterator = typename vector_type::iterator;
using const_iterator = typename vector_type::const_iterator;
using reverse_iterator = typename vector_type::reverse_iterator;
using const_reverse_iterator = typename vector_type::const_reverse_iterator;
// -- constructors, destructors, and assignment operators -------------------
unordered_flat_map() = default;
unordered_flat_map(std::initializer_list<value_type> l) : xs_(l) {
// nop
}
template <class InputIterator>
unordered_flat_map(InputIterator first, InputIterator last)
: xs_(first, last) {
// nop
}
// -- iterator access --------------------------------------------------------
iterator begin() noexcept {
return xs_.begin();
}
const_iterator begin() const noexcept {
return xs_.begin();
}
const_iterator cbegin() const noexcept {
return xs_.cbegin();
}
reverse_iterator rbegin() noexcept {
return xs_.rbegin();
}
const_reverse_iterator rbegin() const noexcept {
return xs_.rbegin();
}
iterator end() noexcept {
return xs_.end();
}
const_iterator end() const noexcept {
return xs_.end();
}
const_iterator cend() const noexcept {
return xs_.end();
}
reverse_iterator rend() noexcept {
return xs_.rend();
}
const_reverse_iterator rend() const noexcept {
return xs_.rend();
}
// -- size and capacity ------------------------------------------------------
bool empty() const noexcept {
return xs_.empty();
}
size_type size() const noexcept {
return xs_.size();
}
void reserve(size_type count) {
xs_.reserve(count);
}
void shrink_to_fit() {
xs_.shrink_to_fit();
}
// -- access to members ------------------------------------------------------
/// Gives raw access to the underlying container.
vector_type& container() noexcept {
return xs_;
}
/// Gives raw access to the underlying container.
const vector_type& container() const noexcept {
return xs_;
}
// -- modifiers -------------------------------------------------------------
void clear() noexcept {
return xs_.clear();
}
void swap(unordered_flat_map& other) {
xs_.swap(other);
}
// -- insertion -------------------------------------------------------------
std::pair<iterator, bool> insert(value_type x) {
return insert(end(), std::move(x));
}
std::pair<iterator, bool> insert(iterator hint, value_type x) {
return insert(static_cast<const_iterator>(hint), std::move(x));
}
std::pair<iterator, bool> insert(const_iterator hint, value_type x) {
auto i = find(x.first);
if (i == end())
return {xs_.insert(hint, std::move(x)), true};
return {i, false};
}
template <class InputIterator>
void insert(InputIterator first, InputIterator last) {
while (first != last)
insert(*first++);
}
template <class... Ts>
std::pair<iterator, bool> emplace(Ts&&... xs) {
return emplace_hint(end(), std::forward<Ts>(xs)...);
}
template <class... Ts>
std::pair<iterator, bool> emplace_hint(const_iterator hint, Ts&&... xs) {
return insert(hint, value_type(std::forward<Ts>(xs)...));
}
// -- removal ----------------------------------------------------------------
iterator erase(const_iterator i) {
return xs_.erase(i);
}
iterator erase(const_iterator first, const_iterator last) {
return xs_.erase(first, last);
}
size_type erase(const key_type& x) {
auto pred = [&](const value_type& y) {
return x == y.first;
};
auto i = std::remove_if(begin(), end(), pred);
if (i == end())
return 0;
erase(i);
return 1;
}
// -- lookup ----------------------------------------------------------------
template <class K>
mapped_type& at(const K& key) {
auto i = find(key);
if (i == end())
#ifdef CAF_NO_EXCEPTIONS
CAF_CRITICAL("caf::detail::unordered_flat_map::at out of range");
#else
throw std::out_of_range{"caf::detail::unordered_flat_map::at"};
#endif
return i->second;
}
template <class K>
const mapped_type& at(const K& key) const {
/// We call the non-const version in order to avoid code duplication but
/// restore the const-ness when returning from the function.
return const_cast<unordered_flat_map&>(*this).at(key);
}
mapped_type& operator[](const key_type& key) {
auto i = find(key);
if (i != end())
return i->second;
return xs_.insert(i, value_type{key, mapped_type{}})->second;
}
template <class K>
iterator find(const K& key) {
auto pred = [&](const value_type& y) {
return key == y.first;
};
return std::find_if(xs_.begin(), xs_.end(), pred);
}
template <class K>
const_iterator find(const K& key) const {
/// We call the non-const version in order to avoid code duplication but
/// restore the const-ness when returning from the function.
return const_cast<unordered_flat_map&>(*this).find(key);
}
template <class K>
size_type count(const K& key) const {
return find(key) == end() ? 0 : 1;
}
private:
vector_type xs_;
};
/// @relates unordered_flat_map
template <class K, class T, class A>
bool operator==(const unordered_flat_map<K, T, A>& xs,
const unordered_flat_map<K, T, A>& ys) {
return xs.size() == ys.size() && std::equal(xs.begin(), xs.end(), ys.begin());
}
/// @relates unordered_flat_map
template <class K, class T, class A>
bool operator!=(const unordered_flat_map<K, T, A>& xs,
const unordered_flat_map<K, T, A>& ys) {
return !(xs == ys);
}
/// @relates unordered_flat_map
template <class K, class T, class A>
bool operator<(const unordered_flat_map<K, T, A>& xs,
const unordered_flat_map<K, T, A>& ys) {
return std::lexicographical_compare(xs.begin(), xs.end(),
ys.begin(), ys.end());
}
/// @relates unordered_flat_map
template <class K, class T, class A>
bool operator>=(const unordered_flat_map<K, T, A>& xs,
const unordered_flat_map<K, T, A>& ys) {
return !(xs < ys);
}
} // namespace detail
} // namespace caf
#endif // CAF_DETAIL_UNORDERED_FLAT_MAP_HPP
...@@ -48,6 +48,11 @@ public: ...@@ -48,6 +48,11 @@ public:
buf_.emplace_back(std::forward<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 // @private
queue_type& buf() { queue_type& buf() {
return buf_; return buf_;
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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. *
******************************************************************************/
#ifndef CAFDOWNSTREAM_MANAGER_HPP
#define CAFDOWNSTREAM_MANAGER_HPP
#include <memory>
#include "caf/fwd.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 pahts.
class downstream_manager {
public:
// -- member types -----------------------------------------------------------
/// Outbound path.
using path_type = outbound_path;
/// Pointer to an outbound path.
using path_ptr = path_type*;
/// Unique pointer to an outbound path.
using unique_path_ptr = std::unique_ptr<path_type>;
/// Function object for iterating over all paths.
struct path_visitor {
virtual ~path_visitor();
virtual void operator()(outbound_path& x) = 0;
};
/// Predicate object for paths.
struct 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(scheduled_actor* self);
virtual ~downstream_manager();
// -- properties -------------------------------------------------------------
scheduled_actor* self() const {
return self_;
}
/// Returns `true` if this manager belongs to a sink, i.e., terminates the
/// stream and never has outbound paths.
virtual bool terminal() const noexcept;
// -- 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);
}
/// 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 and returns it.
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 `true` if there is no data pending and no unacknowledged batch on
/// any path.
bool clean() const noexcept;
/// Removes all paths gracefully.
virtual void close();
/// Removes all paths with an error message.
virtual void abort(error reason);
/// Returns `num_paths() == 0`.
inline 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 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 -------------------------------------------------------
scheduled_actor* self_;
};
} // namespace caf
#endif // CAFDOWNSTREAM_MANAGER_HPP
...@@ -16,63 +16,70 @@ ...@@ -16,63 +16,70 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_STREAM_ID_HPP #ifndef CAFDOWNSTREAM_MANAGER_BASE_HPP
#define CAF_STREAM_ID_HPP #define CAFDOWNSTREAM_MANAGER_BASE_HPP
#include <cstdint> #include <cstddef>
#include <memory>
#include "caf/actor_addr.hpp" #include "caf/downstream_manager.hpp"
#include "caf/meta/type_name.hpp" #include "caf/detail/unordered_flat_map.hpp"
#include "caf/detail/comparable.hpp"
namespace caf { namespace caf {
class stream_id : detail::comparable<stream_id> { /// The default downstream manager base stores outbound paths in an unordered
/// map. It always takes ownership of the pahts by using unique pointers.
class downstream_manager_base : public downstream_manager {
public: public:
stream_id(stream_id&&) = default; // -- member types -----------------------------------------------------------
stream_id(const stream_id&) = default;
stream_id& operator=(stream_id&&) = default;
stream_id& operator=(const stream_id&) = default;
stream_id(); /// Base type.
using super = downstream_manager;
stream_id(none_t); /// Maps slots to paths.
using map_type = detail::unordered_flat_map<stream_slot, unique_path_ptr>;
stream_id(actor_addr origin_actor, uint64_t origin_nr); // -- constructors, destructors, and assignment operators --------------------
stream_id(actor_control_block* origin_actor, uint64_t origin_nr); explicit downstream_manager_base(scheduled_actor* self);
stream_id(const strong_actor_ptr& origin_actor, uint64_t origin_nr); virtual ~downstream_manager_base();
int64_t compare(const stream_id& other) const; // -- properties -------------------------------------------------------------
actor_addr origin; inline const map_type& paths() const {
uint64_t nr; return paths_;
}
inline bool valid() const { inline map_type& paths() {
return origin != nullptr; return paths_;
} }
};
template <class Inspector> // -- path management --------------------------------------------------------
typename Inspector::result_type inspect(Inspector& f, stream_id& x) {
return f(meta::type_name("stream_id"), x.origin, x.nr);
}
} // namespace caf size_t num_paths() const noexcept override;
namespace std { bool remove_path(stream_slot slots, error reason,
template <> bool silent) noexcept override;
struct hash<caf::stream_id> {
size_t operator()(const caf::stream_id& x) const { path_ptr path(stream_slot slots) noexcept override;
auto tmp = reinterpret_cast<ptrdiff_t>(x.origin.get())
^ static_cast<ptrdiff_t>(x.nr); void clear_paths() override;
return static_cast<size_t>(tmp);
} 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_;
}; };
} // namespace std
} // namespace caf
#endif // CAF_STREAM_ID_HPP #endif // CAFDOWNSTREAM_MANAGER_BASE_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2017 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* 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. *
******************************************************************************/
#ifndef CAF_DOWNSTREAM_MSG_HPP
#define CAF_DOWNSTREAM_MSG_HPP
#include <utility>
#include <vector>
#include <cstdint>
#include "caf/actor_addr.hpp"
#include "caf/actor_control_block.hpp"
#include "caf/atom.hpp"
#include "caf/message.hpp"
#include "caf/stream_priority.hpp"
#include "caf/stream_slot.hpp"
#include "caf/variant.hpp"
#include "caf/tag/boxing_type.hpp"
#include "caf/detail/type_list.hpp"
namespace caf {
/// Stream messages that travel downstream, i.e., batches and close messages.
struct downstream_msg : tag::boxing_type {
// -- nested types -----------------------------------------------------------
/// Transmits stream data.
struct 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 close {
/// Allows the testing DSL to unbox this type automagically.
using outer_type = downstream_msg;
};
/// Propagates a fatal error from sources to sinks.
struct forced_close {
/// Allows the testing DSL to unbox this type automagically.
using outer_type = downstream_msg;
/// Reason for shutting down the stream.
error reason;
};
// -- 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>
typename Inspector::result_type inspect(Inspector& f,
downstream_msg::batch& x) {
return f(meta::type_name("batch"), meta::omittable(), x.xs_size, x.xs, x.id);
}
/// @relates downstream_msg::close
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, downstream_msg::close&) {
return f(meta::type_name("close"));
}
/// @relates downstream_msg::forced_close
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f,
downstream_msg::forced_close& x) {
return f(meta::type_name("forced_close"), x.reason);
}
/// @relates downstream_msg
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, downstream_msg& x) {
return f(meta::type_name("downstream_msg"), x.slots, x.sender, x.content);
}
} // namespace caf
#endif // CAF_DOWNSTREAM_MSG_HPP
This diff is collapsed.
This diff is collapsed.
...@@ -19,46 +19,52 @@ ...@@ -19,46 +19,52 @@
#ifndef CAF_FWD_HPP #ifndef CAF_FWD_HPP
#define CAF_FWD_HPP #define CAF_FWD_HPP
#include <memory>
#include <cstdint> #include <cstdint>
#include <memory>
#include <tuple>
namespace caf { namespace caf {
// -- 1 param templates -------------------------------------------------------- // -- 1 param templates --------------------------------------------------------
template <class> class param; template <class> class behavior_type_of;
template <class> class stream;
template <class> class optional;
template <class> class expected;
template <class> class downstream; template <class> class downstream;
template <class> class expected;
template <class> class intrusive_ptr; template <class> class intrusive_ptr;
template <class> class behavior_type_of; template <class> class optional;
template <class> class param;
template <class> class stream;
template <class> class stream_sink;
template <class> class stream_source;
template <class> class trivial_match_case; template <class> class trivial_match_case;
template <class> class weak_intrusive_ptr; template <class> class weak_intrusive_ptr;
template <class> struct timeout_definition; template <class> struct timeout_definition;
// -- 2 param templates --------------------------------------------------------
template <class, class> class stream_stage;
// -- 3 param templates -------------------------------------------------------- // -- 3 param templates --------------------------------------------------------
template <class, class, int> class actor_cast_access; template <class, class, int> class actor_cast_access;
template <class, class, class> class broadcast_topic_scatterer; template <class, class, class> class broadcast_downstream_manager;
template <class, class, class> class random_topic_scatterer;
// -- variadic templates ------------------------------------------------------- // -- variadic templates -------------------------------------------------------
template <class...> class result; template <class...> class result;
template <class...> class variant; template <class...> class variant;
template <class...> class delegated; template <class...> class delegated;
template <class...> class result;
template <class...> class typed_actor; template <class...> class typed_actor;
template <class...> class typed_actor_pointer; template <class...> class typed_actor_pointer;
template <class...> class typed_response_promise;
template <class...> class typed_event_based_actor; template <class...> class typed_event_based_actor;
template <class...> class typed_response_promise;
// -- variadic templates with 1 fixed argument --------------------------------- // -- variadic templates with fixed arguments ----------------------------------
//
template <class, class...> class fused_scatterer; template <class, class...> class output_stream;
template <class, class...> class annotated_stream;
// -- classes ------------------------------------------------------------------ // -- classes ------------------------------------------------------------------
...@@ -70,7 +76,6 @@ class node_id; ...@@ -70,7 +76,6 @@ class node_id;
class behavior; class behavior;
class duration; class duration;
class resumable; class resumable;
class stream_id;
class actor_addr; class actor_addr;
class actor_pool; class actor_pool;
class message_id; class message_id;
...@@ -94,31 +99,33 @@ class blocking_actor; ...@@ -94,31 +99,33 @@ class blocking_actor;
class execution_unit; class execution_unit;
class proxy_registry; class proxy_registry;
class stream_manager; class stream_manager;
class random_gatherer;
class stream_gatherer;
class actor_companion; class actor_companion;
class mailbox_element; class mailbox_element;
class message_builder;
class message_handler; class message_handler;
class scheduled_actor; class scheduled_actor;
class stream_scatterer;
class response_promise; class response_promise;
class event_based_actor; class event_based_actor;
class type_erased_tuple; class type_erased_tuple;
class type_erased_value; class type_erased_value;
class stream_msg_visitor; class downstream_manager;
class actor_control_block; class actor_control_block;
class actor_system_config; class actor_system_config;
class uniform_type_info_map; class uniform_type_info_map;
class forwarding_actor_proxy; class forwarding_actor_proxy;
class downstream_manager_base;
// -- structs ------------------------------------------------------------------ // -- structs ------------------------------------------------------------------
struct unit_t; struct unit_t;
struct exit_msg; struct exit_msg;
struct down_msg; struct down_msg;
struct stream_msg;
struct timeout_msg; struct timeout_msg;
struct stream_slots;
struct upstream_msg;
struct group_down_msg; struct group_down_msg;
struct downstream_msg;
struct open_stream_msg;
struct invalid_actor_t; struct invalid_actor_t;
struct invalid_actor_addr_t; struct invalid_actor_addr_t;
struct illegal_message_element; struct illegal_message_element;
...@@ -132,6 +139,15 @@ enum class atom_value : uint64_t; ...@@ -132,6 +139,15 @@ enum class atom_value : uint64_t;
// -- aliases ------------------------------------------------------------------ // -- aliases ------------------------------------------------------------------
using actor_id = uint64_t; using actor_id = uint64_t;
using stream_slot = uint16_t;
// -- intrusive containers -----------------------------------------------------
namespace intrusive {
enum class task_result;
} // namespace intrusive
// -- marker classes for mixins ------------------------------------------------ // -- marker classes for mixins ------------------------------------------------
......
This diff is collapsed.
...@@ -113,9 +113,7 @@ private: ...@@ -113,9 +113,7 @@ private:
I last_; I last_;
}; };
/** /// @relates input_range
* @relates input_range
*/
template <class I> template <class I>
input_range_impl<I> make_input_range(I first, I last) { input_range_impl<I> make_input_range(I first, I last) {
return {first, last}; return {first, last};
......
This diff is collapsed.
This diff is collapsed.
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2017 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* 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. *
******************************************************************************/
#ifndef CAF_INTRUSIVE_FORWARD_ITERATOR_HPP
#define CAF_INTRUSIVE_FORWARD_ITERATOR_HPP
#include <cstddef>
#include <iterator>
#include <type_traits>
namespace caf {
namespace intrusive {
// A forward iterator for intrusive lists.
template <class T>
class forward_iterator {
public:
// -- member types -----------------------------------------------------------
using difference_type = std::ptrdiff_t;
using value_type = T;
using pointer = value_type*;
using const_pointer = const value_type*;
using reference = value_type&;
using const_reference = const value_type&;
using node_type =
typename std::conditional<
std::is_const<T>::value,
const typename T::node_type,
typename T::node_type
>::type;
using node_pointer = node_type*;
using iterator_category = std::forward_iterator_tag;
// -- member variables -------------------------------------------------------
node_pointer ptr;
// -- constructors, destructors, and assignment operators --------------------
constexpr forward_iterator(node_pointer init = nullptr) : ptr(init) {
// nop
}
forward_iterator(const forward_iterator&) = default;
forward_iterator& operator=(const forward_iterator&) = default;
// -- convenience functions --------------------------------------------------
forward_iterator next() {
return ptr->next;
}
// -- operators --------------------------------------------------------------
forward_iterator& operator++() {
ptr = promote(ptr->next);
return *this;
}
forward_iterator operator++(int) {
forward_iterator res = *this;
ptr = promote(ptr->next);
return res;
}
reference operator*() {
return *promote(ptr);
}
const_reference operator*() const {
return *promote(ptr);
}
pointer operator->() {
return promote(ptr);
}
const pointer operator->() const {
return promote(ptr);
}
};
/// @relates forward_iterator
template <class T>
bool operator==(const forward_iterator<T>& x,
const forward_iterator<T>& y) {
return x.ptr == y.ptr;
}
/// @relates forward_iterator
template <class T>
bool operator!=(const forward_iterator<T>& x,
const forward_iterator<T>& y) {
return x.ptr != y.ptr;
}
} // namespace intrusive
} // namespace caf
#endif // CAF_INTRUSIVE_FORWARD_ITERATOR_HPP
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
...@@ -19,6 +19,8 @@ ...@@ -19,6 +19,8 @@
#ifndef CAF_INVOKE_MESSAGE_RESULT_HPP #ifndef CAF_INVOKE_MESSAGE_RESULT_HPP
#define CAF_INVOKE_MESSAGE_RESULT_HPP #define CAF_INVOKE_MESSAGE_RESULT_HPP
#include <string>
namespace caf { namespace caf {
enum invoke_message_result { enum invoke_message_result {
......
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
...@@ -29,11 +29,12 @@ ...@@ -29,11 +29,12 @@
#include "caf/none.hpp" #include "caf/none.hpp"
#include "caf/intrusive_ptr.hpp" #include "caf/intrusive_ptr.hpp"
#include "caf/message.hpp"
#include "caf/duration.hpp"
#include "caf/behavior.hpp" #include "caf/behavior.hpp"
#include "caf/ref_counted.hpp" #include "caf/duration.hpp"
#include "caf/match_case.hpp"
#include "caf/may_have_timeout.hpp" #include "caf/may_have_timeout.hpp"
#include "caf/message.hpp"
#include "caf/ref_counted.hpp"
#include "caf/timeout_definition.hpp" #include "caf/timeout_definition.hpp"
#include "caf/detail/behavior_impl.hpp" #include "caf/detail/behavior_impl.hpp"
......
This diff is collapsed.
This diff is collapsed.
...@@ -31,6 +31,8 @@ public: ...@@ -31,6 +31,8 @@ public:
virtual type_erased_tuple& content() = 0; virtual type_erased_tuple& content() = 0;
virtual const type_erased_tuple& content() const = 0;
virtual message move_content_to_message() = 0; virtual message move_content_to_message() = 0;
virtual message copy_content_to_message() const = 0; virtual message copy_content_to_message() const = 0;
......
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment