Commit 25765733 authored by Dominik Charousset's avatar Dominik Charousset

Rename stream scatterer to downstream manager

parent 1629ebc4
...@@ -55,7 +55,6 @@ set (LIBCAF_CORE_SRCS ...@@ -55,7 +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_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
...@@ -95,12 +94,11 @@ set (LIBCAF_CORE_SRCS ...@@ -95,12 +94,11 @@ set (LIBCAF_CORE_SRCS
src/stream_aborter.cpp src/stream_aborter.cpp
src/stream_manager.cpp src/stream_manager.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
......
...@@ -78,7 +78,6 @@ ...@@ -78,7 +78,6 @@
#include "caf/proxy_registry.hpp" #include "caf/proxy_registry.hpp"
#include "caf/downstream_msg.hpp" #include "caf/downstream_msg.hpp"
#include "caf/behavior_policy.hpp" #include "caf/behavior_policy.hpp"
#include "caf/fused_scatterer.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"
...@@ -99,6 +98,7 @@ ...@@ -99,6 +98,7 @@
#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/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"
......
...@@ -16,27 +16,28 @@ ...@@ -16,27 +16,28 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_BROADCAST_SCATTERER_HPP #ifndef CAF_BROADCAST_DOWNSTREAM_MANAGER_HPP
#define CAF_BROADCAST_SCATTERER_HPP #define CAF_BROADCAST_DOWNSTREAM_MANAGER_HPP
#include <algorithm> #include <algorithm>
#include "caf/buffered_scatterer.hpp" #include "caf/buffered_downstream_manager.hpp"
#include "caf/outbound_path.hpp" #include "caf/outbound_path.hpp"
#include "caf/detail/algorithms.hpp" #include "caf/detail/algorithms.hpp"
#include "caf/detail/path_state.hpp" #include "caf/detail/path_state.hpp"
#include "caf/detail/select_all.hpp" #include "caf/detail/select_all.hpp"
#include "caf/detail/unordered_flat_map.hpp"
namespace caf { namespace caf {
template <class T, class Filter = unit_t, class Select = detail::select_all> template <class T, class Filter = unit_t, class Select = detail::select_all>
class broadcast_scatterer : public buffered_scatterer<T> { class broadcast_downstream_manager : public buffered_downstream_manager<T> {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
/// Base type. /// Base type.
using super = buffered_scatterer<T>; using super = buffered_downstream_manager<T>;
/// Type of `paths_`. /// Type of `paths_`.
using typename super::map_type; using typename super::map_type;
...@@ -58,7 +59,7 @@ public: ...@@ -58,7 +59,7 @@ public:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
broadcast_scatterer(scheduled_actor* selfptr) : super(selfptr) { broadcast_downstream_manager(scheduled_actor* selfptr) : super(selfptr) {
// nop // nop
} }
...@@ -240,4 +241,4 @@ private: ...@@ -240,4 +241,4 @@ private:
} // namespace caf } // namespace caf
#endif // CAF_BROADCAST_SCATTERER_HPP #endif // CAF_BROADCAST_DOWNSTREAM_MANAGER_HPP
...@@ -16,37 +16,37 @@ ...@@ -16,37 +16,37 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_BUFFERED_SCATTERER_HPP #ifndef CAF_BUFFERED_DOWNSTREAM_MANAGER_HPP
#define CAF_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/downstream_manager_base.hpp"
#include "caf/logger.hpp" #include "caf/logger.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:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
using super = stream_scatterer_impl; using super = downstream_manager_base;
using value_type = T; using output_type = T;
using buffer_type = std::deque<value_type>; using buffer_type = std::deque<output_type>;
using chunk_type = std::vector<value_type>; using chunk_type = std::vector<output_type>;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
buffered_scatterer(scheduled_actor* self) : super(self) { buffered_downstream_manager(scheduled_actor* self) : super(self) {
// nop // nop
} }
...@@ -78,6 +78,10 @@ public: ...@@ -78,6 +78,10 @@ public:
return get_chunk(buf_, n); return get_chunk(buf_, n);
} }
bool terminal() const noexcept override {
return false;
}
size_t capacity() const noexcept override { size_t capacity() const noexcept override {
// TODO: get rid of magic number // TODO: get rid of magic number
static constexpr size_t max_buf_size = 100; static constexpr size_t max_buf_size = 100;
...@@ -102,4 +106,4 @@ protected: ...@@ -102,4 +106,4 @@ protected:
} // namespace caf } // namespace caf
#endif // CAF_BUFFERED_SCATTERER_HPP #endif // CAF_BUFFERED_DOWNSTREAM_MANAGER_HPP
...@@ -16,46 +16,46 @@ ...@@ -16,46 +16,46 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_INVALID_STREAM_SCATTERER_HPP #ifndef CAF_DEFAULT_DOWNSTREAM_MANAGER_HPP
#define CAF_INVALID_STREAM_SCATTERER_HPP #define CAF_DEFAULT_DOWNSTREAM_MANAGER_HPP
#include "caf/stream_scatterer.hpp" #include "caf/broadcast_downstream_manager.hpp"
#include "caf/stream_source_trait.hpp"
#include "caf/stream_stage_trait.hpp"
namespace caf { #include "caf/detail/type_traits.hpp"
/// Type-erased policy for dispatching data to sinks.
class invalid_stream_scatterer : public stream_scatterer {
public:
invalid_stream_scatterer(scheduled_actor* self);
~invalid_stream_scatterer() override;
size_t num_paths() const noexcept override;
bool remove_path(stream_slot slots, error reason,
bool silent) noexcept override;
path_ptr path(stream_slot slots) noexcept override;
void emit_batches() override;
void force_emit_batches() override; namespace caf {
size_t capacity() const noexcept override;
size_t buffered() const noexcept override;
protected:
bool insert_path(unique_path_ptr) override;
void for_each_path_impl(path_visitor& f) override;
bool check_paths_impl(path_algorithm algo,
path_predicate& pred) const noexcept override;
void clear_paths() override; /// Selects a downstream manager implementation based on the signature of
/// various handlers.
template <class F>
struct default_downstream_manager {
/// The function signature of `F`.
using fun_sig = typename detail::get_callable_trait<F>::fun_sig;
/// The source trait for `F`.
using source_trait = stream_source_trait<fun_sig>;
/// The stage trait for `F`.
using stage_trait = stream_stage_trait<fun_sig>;
/// The output type as returned by the source or stage trait.
using output_type =
typename std::conditional<
source_trait::valid,
typename source_trait::output,
typename stage_trait::output
>::type;
/// The default downstream manager deduced by this trait.
using type = broadcast_downstream_manager<output_type>;
}; };
template <class F>
using default_downstream_manager_t =
typename default_downstream_manager<F>::type;
} // namespace caf } // namespace caf
#endif // CAF_INVALID_STREAM_SCATTERER_HPP #endif // CAF_DEFAULT_DOWNSTREAM_MANAGER_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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_PUSH5_SCATTERER_HPP
#define CAF_PUSH5_SCATTERER_HPP
#include "caf/broadcast_scatterer.hpp"
namespace caf {
namespace detail {
/// Always pushs exactly 5 elements to sinks. Used in unit tests only.
template <class T>
class push5_scatterer : public broadcast_scatterer<T> {
public:
using super = broadcast_scatterer<T>;
push5_scatterer(local_actor* self) : super(self) {
// nop
}
long min_buffer_size() const override {
return 5;
}
};
} // namespace detail
} // namespace caf
#endif // CAF_PUSH5_SCATTERER_HPP
...@@ -29,12 +29,13 @@ namespace caf { ...@@ -29,12 +29,13 @@ namespace caf {
namespace detail { namespace detail {
/// Identifies an unbound sequence of messages. /// Identifies an unbound sequence of messages.
template <class Scatterer, class Pull, class Done, class Finalize> template <class DownstreamManager, class Pull, class Done, class Finalize>
class stream_source_driver_impl final : public stream_source_driver<Scatterer> { class stream_source_driver_impl final
: public stream_source_driver<DownstreamManager> {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
using super = stream_source_driver<Scatterer>; using super = stream_source_driver<DownstreamManager>;
using output_type = typename super::output_type; using output_type = typename super::output_type;
......
...@@ -31,13 +31,13 @@ namespace detail { ...@@ -31,13 +31,13 @@ namespace detail {
/// Default implementation for a `stream_stage_driver` that hardwires `message` /// Default implementation for a `stream_stage_driver` that hardwires `message`
/// as result type and implements `process` and `finalize` using user-provided /// as result type and implements `process` and `finalize` using user-provided
/// function objects (usually lambdas). /// function objects (usually lambdas).
template <class Input, class Scatterer, class Process, class Finalize> template <class Input, class DownstreamManager, class Process, class Finalize>
class stream_stage_driver_impl final class stream_stage_driver_impl final
: public stream_stage_driver<Input, Scatterer> { : public stream_stage_driver<Input, DownstreamManager> {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
using super = stream_stage_driver<Input, Scatterer>; using super = stream_stage_driver<Input, DownstreamManager>;
using typename super::input_type; using typename super::input_type;
......
...@@ -40,7 +40,7 @@ public: ...@@ -40,7 +40,7 @@ public:
using driver_type = Driver; using driver_type = Driver;
using scatterer_type = typename Driver::scatterer_type; using downstream_manager_type = typename Driver::downstream_manager_type;
using input_type = typename driver_type::input_type; using input_type = typename driver_type::input_type;
......
...@@ -16,8 +16,8 @@ ...@@ -16,8 +16,8 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_STREAM_SCATTERER_HPP #ifndef CAFDOWNSTREAM_MANAGER_HPP
#define CAF_STREAM_SCATTERER_HPP #define CAFDOWNSTREAM_MANAGER_HPP
#include <memory> #include <memory>
...@@ -25,8 +25,12 @@ ...@@ -25,8 +25,12 @@
namespace caf { namespace caf {
/// Type-erased policy for dispatching data to sinks. /// Manages downstream communication for a `stream_manager`. The downstream
class stream_scatterer { /// 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: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
...@@ -60,9 +64,9 @@ public: ...@@ -60,9 +64,9 @@ public:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
explicit stream_scatterer(scheduled_actor* self); explicit downstream_manager(scheduled_actor* self);
virtual ~stream_scatterer(); virtual ~downstream_manager();
// -- properties ------------------------------------------------------------- // -- properties -------------------------------------------------------------
...@@ -70,9 +74,7 @@ public: ...@@ -70,9 +74,7 @@ public:
return self_; return self_;
} }
// -- meta information ------------------------------------------------------- /// Returns `true` if this manager belongs to a sink, i.e., terminates the
/// Returns `true` if thie scatterer belongs to a sink, i.e., terminates the
/// stream and never has outbound paths. /// stream and never has outbound paths.
virtual bool terminal() const noexcept; virtual bool terminal() const noexcept;
...@@ -113,18 +115,18 @@ public: ...@@ -113,18 +115,18 @@ public:
} }
/// Returns the current number of paths. /// Returns the current number of paths.
virtual size_t num_paths() const noexcept = 0; virtual size_t num_paths() const noexcept;
/// Adds a pending path to `target` to the scatterer. /// Adds a pending path to `target` to the manager.
/// @returns The added path on success, `nullptr` otherwise. /// @returns The added path on success, `nullptr` otherwise.
path_ptr add_path(stream_slot slot, strong_actor_ptr target); path_ptr add_path(stream_slot slot, strong_actor_ptr target);
/// Removes a path from the scatterer and returns it. /// Removes a path from the manager and returns it.
virtual bool remove_path(stream_slot slot, error reason, virtual bool remove_path(stream_slot slot, error reason,
bool silent) noexcept = 0; bool silent) noexcept;
/// Returns the path associated to `slots` or `nullptr`. /// Returns the path associated to `slot` or `nullptr`.
virtual path_ptr path(stream_slot slots) noexcept = 0; virtual path_ptr path(stream_slot slot) noexcept;
/// Returns `true` if there is no data pending and no unacknowledged batch on /// Returns `true` if there is no data pending and no unacknowledged batch on
/// any path. /// any path.
...@@ -152,34 +154,34 @@ public: ...@@ -152,34 +154,34 @@ public:
size_t total_credit() const; size_t total_credit() const;
/// Sends batches to sinks. /// Sends batches to sinks.
virtual void emit_batches() = 0; virtual void emit_batches();
/// Sends batches to sinks regardless of whether or not the batches reach the /// Sends batches to sinks regardless of whether or not the batches reach the
/// desired batch size. /// desired batch size.
virtual void force_emit_batches() = 0; virtual void force_emit_batches();
/// Returns the currently available capacity for the output buffer. /// Returns the currently available capacity for the output buffer.
virtual size_t capacity() const noexcept = 0; virtual size_t capacity() const noexcept;
/// Returns the size of the output buffer. /// Returns the size of the output buffer.
virtual size_t buffered() const noexcept = 0; virtual size_t buffered() const noexcept;
/// Silently removes all paths. /// Silently removes all paths.
virtual void clear_paths() = 0; virtual void clear_paths();
protected: protected:
// -- customization points --------------------------------------------------- // -- customization points ---------------------------------------------------
/// Inserts `ptr` to the implementation-specific container. /// Inserts `ptr` to the implementation-specific container.
virtual bool insert_path(unique_path_ptr ptr) = 0; virtual bool insert_path(unique_path_ptr ptr);
/// Applies `f` to each path. /// Applies `f` to each path.
virtual void for_each_path_impl(path_visitor& f) = 0; virtual void for_each_path_impl(path_visitor& f);
/// Dispatches the predicate to `std::all_of`, `std::any_of`, or /// Dispatches the predicate to `std::all_of`, `std::any_of`, or
/// `std::none_of`. /// `std::none_of`.
virtual bool check_paths_impl(path_algorithm algo, virtual bool check_paths_impl(path_algorithm algo,
path_predicate& pred) const noexcept = 0; path_predicate& pred) const noexcept;
/// Emits a regular (`reason == nullptr`) or irregular (`reason != nullptr`) /// Emits a regular (`reason == nullptr`) or irregular (`reason != nullptr`)
/// shutdown if `silent == false`. /// shutdown if `silent == false`.
...@@ -212,4 +214,4 @@ protected: ...@@ -212,4 +214,4 @@ protected:
} // namespace caf } // namespace caf
#endif // CAF_STREAM_SCATTERER_HPP #endif // CAFDOWNSTREAM_MANAGER_HPP
...@@ -16,34 +16,35 @@ ...@@ -16,34 +16,35 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_STREAM_SCATTERER_IMPL_HPP #ifndef CAFDOWNSTREAM_MANAGER_BASE_HPP
#define CAF_STREAM_SCATTERER_IMPL_HPP #define CAFDOWNSTREAM_MANAGER_BASE_HPP
#include <cstddef> #include <cstddef>
#include <memory> #include <memory>
#include "caf/stream_scatterer.hpp" #include "caf/downstream_manager.hpp"
#include "caf/detail/unordered_flat_map.hpp" #include "caf/detail/unordered_flat_map.hpp"
namespace caf { namespace caf {
/// Type-erased policy for dispatching data to sinks. /// The default downstream manager base stores outbound paths in an unordered
class stream_scatterer_impl : public stream_scatterer { /// map. It always takes ownership of the pahts by using unique pointers.
class downstream_manager_base : public downstream_manager {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
/// Base type. /// Base type.
using super = stream_scatterer; using super = downstream_manager;
/// Maps slots to paths. /// Maps slots to paths.
using map_type = detail::unordered_flat_map<stream_slot, unique_path_ptr>; using map_type = detail::unordered_flat_map<stream_slot, unique_path_ptr>;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
explicit stream_scatterer_impl(scheduled_actor* self); explicit downstream_manager_base(scheduled_actor* self);
virtual ~stream_scatterer_impl(); virtual ~downstream_manager_base();
// -- properties ------------------------------------------------------------- // -- properties -------------------------------------------------------------
...@@ -81,4 +82,4 @@ protected: ...@@ -81,4 +82,4 @@ protected:
} // namespace caf } // namespace caf
#endif // CAF_STREAM_SCATTERER_IMPL_HPP #endif // CAFDOWNSTREAM_MANAGER_BASE_HPP
...@@ -16,17 +16,17 @@ ...@@ -16,17 +16,17 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_FUSED_SCATTERER #ifndef CAF_FUSED_DOWNSTREAM_MANAGER_HPP
#define CAF_FUSED_SCATTERER #define CAF_FUSED_DOWNSTREAM_MANAGER_HPP
#include <tuple> #include <tuple>
#include <cstddef> #include <cstddef>
#include "caf/downstream_manager.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/outbound_path.hpp" #include "caf/outbound_path.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/stream.hpp" #include "caf/stream.hpp"
#include "caf/stream_scatterer.hpp"
#include "caf/stream_slot.hpp" #include "caf/stream_slot.hpp"
#include "caf/detail/type_list.hpp" #include "caf/detail/type_list.hpp"
...@@ -63,13 +63,13 @@ private: ...@@ -63,13 +63,13 @@ private:
Iter i_; Iter i_;
}; };
struct scatterer_selector { struct downstream_manager_selector {
inline stream_scatterer* operator()(const message&) { inline downstream_manager* operator()(const message&) {
return nullptr; return nullptr;
} }
template <class T, class... Ts> template <class T, class... Ts>
stream_scatterer* operator()(const message& msg, T& x, Ts&... xs) { downstream_manager* operator()(const message& msg, T& x, Ts&... xs) {
if (msg.match_element<stream<typename T::value_type>>(0)) if (msg.match_element<stream<typename T::value_type>>(0))
return &x; return &x;
return (*this)(msg, xs...); return (*this)(msg, xs...);
...@@ -79,7 +79,7 @@ struct scatterer_selector { ...@@ -79,7 +79,7 @@ struct scatterer_selector {
template <size_t I, size_t E> template <size_t I, size_t E>
struct init_ptr_array { struct init_ptr_array {
template <class... Ts> template <class... Ts>
static void apply(stream_scatterer* (&xs)[E], std::tuple<Ts...>& ys) { static void apply(downstream_manager* (&xs)[E], std::tuple<Ts...>& ys) {
xs[I] = &std::get<I>(ys); xs[I] = &std::get<I>(ys);
init_ptr_array<I + 1, E>::apply(xs, ys); init_ptr_array<I + 1, E>::apply(xs, ys);
} }
...@@ -88,26 +88,24 @@ struct init_ptr_array { ...@@ -88,26 +88,24 @@ struct init_ptr_array {
template <size_t I> template <size_t I>
struct init_ptr_array<I, I> { struct init_ptr_array<I, I> {
template <class... Ts> template <class... Ts>
static void apply(stream_scatterer* (&)[I], std::tuple<Ts...>&) { static void apply(downstream_manager* (&)[I], std::tuple<Ts...>&) {
// nop // nop
} }
}; };
} // namespace detail } // namespace detail
/// A scatterer that delegates to any number of sub-scatterers. Data is only /// A downstream manager that delegates to any number of sub-managers.
/// pushed to the main scatterer `T` per default.
template <class T, class... Ts> template <class T, class... Ts>
class fused_scatterer : public stream_scatterer { class fused_downstream_manager : public downstream_manager {
public: public:
// -- member and nested types ------------------------------------------------ // -- member and nested types ------------------------------------------------
/// Base type. /// Base type.
using super = stream_scatterer; using super = downstream_manager;
/// A tuple holding all nested scatterers. /// A tuple holding all nested managers.
using nested_scatterers = std::tuple<T, Ts...>; using nested_managers = std::tuple<T, Ts...>;
/// Pointer to an outbound path. /// Pointer to an outbound path.
using typename super::path_ptr; using typename super::path_ptr;
...@@ -121,19 +119,19 @@ public: ...@@ -121,19 +119,19 @@ public:
/// State held for each slot. /// State held for each slot.
struct non_owning_ptr { struct non_owning_ptr {
path_ptr ptr; path_ptr ptr;
stream_scatterer* owner; downstream_manager* owner;
}; };
/// Maps slots to path and nested scatterer. /// Maps slots to path and nested managers.
using map_type = detail::unordered_flat_map<stream_slot, non_owning_ptr>; using map_type = detail::unordered_flat_map<stream_slot, non_owning_ptr>;
/// Maps slots to paths that haven't a scatterer assigned yet. /// Maps slots to paths that haven't a managers assigned yet.
using unassigned_map_type = detail::unordered_flat_map<stream_slot, using unassigned_map_type = detail::unordered_flat_map<stream_slot,
unique_path_ptr>; unique_path_ptr>;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
fused_scatterer(scheduled_actor* self) fused_downstream_manager(scheduled_actor* self)
: super(self), : super(self),
nested_(self, detail::pack_repeat<Ts...>(self)) { nested_(self, detail::pack_repeat<Ts...>(self)) {
detail::init_ptr_array<0, sizeof...(Ts) + 1>::apply(ptrs_, nested_); detail::init_ptr_array<0, sizeof...(Ts) + 1>::apply(ptrs_, nested_);
...@@ -163,14 +161,14 @@ public: ...@@ -163,14 +161,14 @@ public:
// Fetch pointer from the unassigned paths. // Fetch pointer from the unassigned paths.
auto i = unassigned_paths_.find(slot); auto i = unassigned_paths_.find(slot);
if (i == unassigned_paths_.end()) { if (i == unassigned_paths_.end()) {
CAF_LOG_ERROR("cannot assign nested scatterer to unknown slot"); CAF_LOG_ERROR("cannot assign nested manager to unknown slot");
return; return;
} }
// Error or not, remove entry from unassigned_paths_ before leaving. // Error or not, remove entry from unassigned_paths_ before leaving.
auto cleanup = detail::make_scope_guard([&] { auto cleanup = detail::make_scope_guard([&] {
unassigned_paths_.erase(i); unassigned_paths_.erase(i);
}); });
// Transfer ownership to nested scatterer. // Transfer ownership to nested manager.
auto ptr = i->second.get(); auto ptr = i->second.get();
CAF_ASSERT(ptr != nullptr); CAF_ASSERT(ptr != nullptr);
auto owner = &get<U>(); auto owner = &get<U>();
...@@ -189,6 +187,10 @@ public: ...@@ -189,6 +187,10 @@ public:
// -- overridden functions --------------------------------------------------- // -- overridden functions ---------------------------------------------------
bool terminal() const noexcept override {
return false;
}
size_t num_paths() const noexcept override { size_t num_paths() const noexcept override {
return paths_.size(); return paths_.size();
} }
...@@ -289,13 +291,12 @@ protected: ...@@ -289,13 +291,12 @@ protected:
} }
private: private:
nested_scatterers nested_; nested_managers nested_;
stream_scatterer* ptrs_[sizeof...(Ts) + 1]; downstream_manager* ptrs_[sizeof...(Ts) + 1];
map_type paths_; map_type paths_;
unassigned_map_type unassigned_paths_; unassigned_map_type unassigned_paths_;
}; };
} // namespace caf } // namespace caf
#endif // CAF_FUSED_SCATTERER #endif // CAF_FUSED_DOWNSTREAM_MANAGER_HPP
...@@ -48,9 +48,7 @@ template <class, class> class stream_source; ...@@ -48,9 +48,7 @@ template <class, class> class stream_source;
template <class, class, int> class actor_cast_access; template <class, class, int> class actor_cast_access;
template <class, class, class> class broadcast_scatterer; template <class, class, class> class broadcast_downstream_manager;
template <class, class, class> class broadcast_topic_scatterer;
template <class, class, class> class random_topic_scatterer;
template <class, class, class> class stream_stage; template <class, class, class> class stream_stage;
// -- variadic templates ------------------------------------------------------- // -- variadic templates -------------------------------------------------------
...@@ -102,15 +100,16 @@ class mailbox_element; ...@@ -102,15 +100,16 @@ class mailbox_element;
class message_builder; 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 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 ------------------------------------------------------------------
......
...@@ -29,26 +29,26 @@ namespace caf { ...@@ -29,26 +29,26 @@ namespace caf {
/// Helper trait for deducing an `output_stream` from the arguments to /// Helper trait for deducing an `output_stream` from the arguments to
/// `scheduled_actor::make_source`. /// `scheduled_actor::make_source`.
template <class Scatterer, class... Ts> template <class DownstreamManager, class... Ts>
struct make_source_result { struct make_source_result {
/// Type of a single element. /// Type of a single element.
using value_type = typename Scatterer::value_type; using output_type = typename DownstreamManager::output_type;
/// Fully typed stream manager as returned by `make_source`. /// Fully typed stream manager as returned by `make_source`.
using source_type = stream_source<value_type, Scatterer>; using source_type = stream_source<output_type, DownstreamManager>;
/// Pointer to a fully typed stream manager. /// Pointer to a fully typed stream manager.
using source_ptr_type = intrusive_ptr<source_type>; using source_ptr_type = intrusive_ptr<source_type>;
/// The return type for `scheduled_actor::make_source`. /// The return type for `scheduled_actor::make_source`.
using type = output_stream<value_type, std::tuple<Ts...>, source_ptr_type>; using type = output_stream<output_type, std::tuple<Ts...>, source_ptr_type>;
}; };
/// Helper type for defining an `output_stream` from a `Scatterer` plus /// Helper type for defining an `output_stream` from a downstream manager plus
/// the types of the handshake arguments. /// the types of the handshake arguments.
template <class Scatterer, class... Ts> template <class DownstreamManager, class... Ts>
using make_source_result_t = using make_source_result_t =
typename make_source_result<Scatterer, typename make_source_result<DownstreamManager,
detail::strip_and_convert_t<Ts>...>::type; detail::strip_and_convert_t<Ts>...>::type;
} // namespace caf } // namespace caf
......
...@@ -30,27 +30,27 @@ namespace caf { ...@@ -30,27 +30,27 @@ namespace caf {
/// Helper trait for deducing an `output_stream` from the arguments to /// Helper trait for deducing an `output_stream` from the arguments to
/// `scheduled_actor::make_stage`. /// `scheduled_actor::make_stage`.
template <class In, class Scatterer, class... Ts> template <class In, class DownstreamManager, class... Ts>
class make_stage_result { class make_stage_result {
public: public:
/// Type of a single element. /// Type of a single element.
using value_type = typename Scatterer::value_type; using output_type = typename DownstreamManager::output_type;
/// Fully typed stream manager as returned by `make_stage`. /// Fully typed stream manager as returned by `make_stage`.
using stage_type = stream_stage<In, value_type, Scatterer>; using stage_type = stream_stage<In, output_type, DownstreamManager>;
/// Pointer to a fully typed stream manager. /// Pointer to a fully typed stream manager.
using stage_ptr_type = intrusive_ptr<stage_type>; using stage_ptr_type = intrusive_ptr<stage_type>;
/// The return type for `scheduled_actor::make_sink`. /// The return type for `scheduled_actor::make_sink`.
using type = output_stream<value_type, std::tuple<Ts...>, stage_ptr_type>; using type = output_stream<output_type, std::tuple<Ts...>, stage_ptr_type>;
}; };
/// Helper type for defining a `make_stage_result` from a `Scatterer` plus /// Helper type for defining a `make_stage_result` from a downstream manager
/// additional handshake types. Hardwires `message` as result type. /// plus additional handshake types. Hardwires `message` as result type.
template <class In, class Scatterer, class... Ts> template <class In, class DownstreamManager, class... Ts>
using make_stage_result_t = using make_stage_result_t =
typename make_stage_result<In, Scatterer, typename make_stage_result<In, DownstreamManager,
detail::strip_and_convert_t<Ts>...>::type; detail::strip_and_convert_t<Ts>...>::type;
} // namespace caf } // namespace caf
......
...@@ -36,7 +36,7 @@ ...@@ -36,7 +36,7 @@
namespace caf { namespace caf {
/// State for a single path to a sink on a `stream_scatterer`. /// State for a single path to a sink of a `downstream_manager`.
class outbound_path { class outbound_path {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
......
This diff is collapsed.
...@@ -25,6 +25,7 @@ ...@@ -25,6 +25,7 @@
#include "caf/actor.hpp" #include "caf/actor.hpp"
#include "caf/actor_cast.hpp" #include "caf/actor_cast.hpp"
#include "caf/downstream_manager.hpp"
#include "caf/downstream_msg.hpp" #include "caf/downstream_msg.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/mailbox_element.hpp" #include "caf/mailbox_element.hpp"
...@@ -33,7 +34,6 @@ ...@@ -33,7 +34,6 @@
#include "caf/output_stream.hpp" #include "caf/output_stream.hpp"
#include "caf/ref_counted.hpp" #include "caf/ref_counted.hpp"
#include "caf/stream_result.hpp" #include "caf/stream_result.hpp"
#include "caf/stream_scatterer.hpp"
#include "caf/stream_slot.hpp" #include "caf/stream_slot.hpp"
#include "caf/upstream_msg.hpp" #include "caf/upstream_msg.hpp"
...@@ -98,8 +98,8 @@ public: ...@@ -98,8 +98,8 @@ public:
// -- pure virtual member functions ------------------------------------------ // -- pure virtual member functions ------------------------------------------
/// Returns the scatterer for outgoing data. /// Returns the manager for downstream communication.
virtual stream_scatterer& out() = 0; virtual downstream_manager& out() = 0;
/// Returns whether the stream has reached the end and can be discarded /// Returns whether the stream has reached the end and can be discarded
/// safely. /// safely.
......
...@@ -24,7 +24,6 @@ ...@@ -24,7 +24,6 @@
#include "caf/intrusive_ptr.hpp" #include "caf/intrusive_ptr.hpp"
#include "caf/stream_manager.hpp" #include "caf/stream_manager.hpp"
#include "caf/stream_result.hpp" #include "caf/stream_result.hpp"
#include "caf/terminal_stream_scatterer.hpp"
#include "caf/detail/type_traits.hpp" #include "caf/detail/type_traits.hpp"
...@@ -49,7 +48,7 @@ public: ...@@ -49,7 +48,7 @@ public:
return !this->continuous() && this->inbound_paths_.empty(); return !this->continuous() && this->inbound_paths_.empty();
} }
stream_scatterer& out() override { downstream_manager& out() override {
return dummy_out_; return dummy_out_;
} }
...@@ -62,7 +61,7 @@ public: ...@@ -62,7 +61,7 @@ public:
} }
private: private:
terminal_stream_scatterer dummy_out_; downstream_manager dummy_out_;
}; };
template <class In> template <class In>
......
...@@ -30,7 +30,7 @@ ...@@ -30,7 +30,7 @@
namespace caf { namespace caf {
template <class Out, class Scatterer> template <class Out, class DownstreamManager>
class stream_source : public virtual stream_manager { class stream_source : public virtual stream_manager {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
...@@ -43,7 +43,7 @@ public: ...@@ -43,7 +43,7 @@ public:
// nop // nop
} }
Scatterer& out() override { DownstreamManager& out() override {
return out_; return out_;
} }
...@@ -82,15 +82,15 @@ public: ...@@ -82,15 +82,15 @@ public:
} }
protected: protected:
Scatterer out_; DownstreamManager out_;
}; };
template <class Out, class Scatterer> template <class Out, class DownstreamManager>
using stream_source_ptr = intrusive_ptr<stream_source<Out, Scatterer>>; using stream_source_ptr = intrusive_ptr<stream_source<Out, DownstreamManager>>;
template <class Scatterer> template <class DownstreamManager>
using stream_source_ptr_t = using stream_source_ptr_t =
stream_source_ptr<typename Scatterer::value_type, Scatterer>; stream_source_ptr<typename DownstreamManager::output_type, DownstreamManager>;
} // namespace caf } // namespace caf
......
...@@ -27,25 +27,23 @@ ...@@ -27,25 +27,23 @@
namespace caf { namespace caf {
/// Identifies an unbound sequence of messages. /// Identifies an unbound sequence of messages.
template <class Scatterer> template <class DownstreamManager>
class stream_source_driver { class stream_source_driver {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
/// Policy for distributing data to outbound paths. /// Type of the downstream manager, i.e., the type returned by
using scatterer_type = Scatterer; /// `stream_manager::out()`.
using downstream_manager_type = DownstreamManager;
/// Element type of the output stream. /// Element type of the output stream.
using output_type = typename scatterer_type::value_type; using output_type = typename downstream_manager_type::output_type;
/// Type of the output stream. /// Type of the output stream.
using stream_type = stream<output_type>; using stream_type = stream<output_type>;
/// Tuple for creating the `open_stream_msg` handshake.
using handshake_tuple_type = std::tuple<stream_type>;
/// Implemented `stream_source` interface. /// Implemented `stream_source` interface.
using source_type = stream_source<output_type, scatterer_type>; using source_type = stream_source<output_type, downstream_manager_type>;
/// Smart pointer to the interface type. /// Smart pointer to the interface type.
using source_ptr_type = intrusive_ptr<source_type>; using source_ptr_type = intrusive_ptr<source_type>;
...@@ -65,8 +63,8 @@ public: ...@@ -65,8 +63,8 @@ public:
// -- pure virtual functions ------------------------------------------------- // -- pure virtual functions -------------------------------------------------
/// Generates more stream elements. /// Generates more elements for `dst`.
virtual void pull(downstream<output_type>& out, size_t num) = 0; virtual void pull(downstream<output_type>& dst, size_t num) = 0;
/// Returns `true` if the source is done sending, otherwise `false`. /// Returns `true` if the source is done sending, otherwise `false`.
virtual bool done() const noexcept = 0; virtual bool done() const noexcept = 0;
......
...@@ -19,6 +19,8 @@ ...@@ -19,6 +19,8 @@
#ifndef CAF_STREAM_SOURCE_TRAIT_HPP #ifndef CAF_STREAM_SOURCE_TRAIT_HPP
#define CAF_STREAM_SOURCE_TRAIT_HPP #define CAF_STREAM_SOURCE_TRAIT_HPP
#include "caf/fwd.hpp"
#include "caf/detail/type_traits.hpp" #include "caf/detail/type_traits.hpp"
namespace caf { namespace caf {
...@@ -26,10 +28,15 @@ namespace caf { ...@@ -26,10 +28,15 @@ namespace caf {
/// Deduces the output type and the state type for a stream source from its /// Deduces the output type and the state type for a stream source from its
/// `pull` implementation. /// `pull` implementation.
template <class Pull> template <class Pull>
struct stream_source_trait; struct stream_source_trait {
static constexpr bool valid = false;
using output = unit_t;
using state = unit_t;
};
template <class State, class T> template <class State, class T>
struct stream_source_trait<void (State&, downstream<T>&, size_t)> { struct stream_source_trait<void (State&, downstream<T>&, size_t)> {
static constexpr bool valid = true;
using output = T; using output = T;
using state = State; using state = State;
}; };
......
...@@ -28,13 +28,13 @@ ...@@ -28,13 +28,13 @@
namespace caf { namespace caf {
template <class In, class Out, class Scatterer> template <class In, class Out, class DownstreamManager>
class stream_stage : public stream_source<Out, Scatterer>, class stream_stage : public stream_source<Out, DownstreamManager>,
public stream_sink<In> { public stream_sink<In> {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
using left_super = stream_source<Out, Scatterer>; using left_super = stream_source<Out, DownstreamManager>;
using right_super = stream_sink<In>; using right_super = stream_sink<In>;
...@@ -47,14 +47,14 @@ public: ...@@ -47,14 +47,14 @@ public:
// nop // nop
} }
Scatterer& out() override { DownstreamManager& out() override {
return left_super::out(); return left_super::out();
} }
}; };
template <class In, class Out, class Scatterer> template <class In, class Out, class DownstreamManager>
using stream_stage_ptr = using stream_stage_ptr =
intrusive_ptr<stream_stage<In, Out, Scatterer>>; intrusive_ptr<stream_stage<In, Out, DownstreamManager>>;
} // namespace caf } // namespace caf
......
...@@ -29,7 +29,7 @@ ...@@ -29,7 +29,7 @@
namespace caf { namespace caf {
/// Encapsulates user-provided functionality for generating a stream stage. /// Encapsulates user-provided functionality for generating a stream stage.
template <class Input, class Scatterer> template <class Input, class DownstreamManager>
class stream_stage_driver { class stream_stage_driver {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
...@@ -38,16 +38,16 @@ public: ...@@ -38,16 +38,16 @@ public:
using input_type = Input; using input_type = Input;
/// Policy for distributing data to outbound paths. /// Policy for distributing data to outbound paths.
using scatterer_type = Scatterer; using downstream_manager_type = DownstreamManager;
/// Element type of the output stream. /// Element type of the output stream.
using output_type = typename scatterer_type::value_type; using output_type = typename downstream_manager_type::output_type;
/// Type of the output stream. /// Type of the output stream.
using stream_type = stream<output_type>; using stream_type = stream<output_type>;
/// Implemented `stream_stage` interface. /// Implemented `stream_stage` interface.
using stage_type = stream_stage<input_type, output_type, Scatterer>; using stage_type = stream_stage<input_type, output_type, DownstreamManager>;
/// Smart pointer to the interface type. /// Smart pointer to the interface type.
using stage_ptr_type = intrusive_ptr<stage_type>; using stage_ptr_type = intrusive_ptr<stage_type>;
......
...@@ -53,12 +53,16 @@ struct stream_stage_trait_invoke_all { ...@@ -53,12 +53,16 @@ struct stream_stage_trait_invoke_all {
// -- trait implementation ----------------------------------------------------- // -- trait implementation -----------------------------------------------------
template <class F> template <class F>
struct stream_stage_trait; struct stream_stage_trait {
static constexpr bool valid = false;
using output = unit_t;
};
/// Deduces the input type, output type and the state type for a stream stage /// Deduces the input type, output type and the state type for a stream stage
/// from its `process` implementation. /// from its `process` implementation.
template <class State, class In, class Out> template <class State, class In, class Out>
struct stream_stage_trait<void (State&, downstream<Out>&, In)> { struct stream_stage_trait<void (State&, downstream<Out>&, In)> {
static constexpr bool valid = true;
using state = State; using state = State;
using input = In; using input = In;
using output = Out; using output = Out;
...@@ -67,6 +71,7 @@ struct stream_stage_trait<void (State&, downstream<Out>&, In)> { ...@@ -67,6 +71,7 @@ struct stream_stage_trait<void (State&, downstream<Out>&, In)> {
template <class State, class In, class Out> template <class State, class In, class Out>
struct stream_stage_trait<void (State&, downstream<Out>&, std::vector<In>&)> { struct stream_stage_trait<void (State&, downstream<Out>&, std::vector<In>&)> {
static constexpr bool valid = true;
using state = State; using state = State;
using input = In; using input = In;
using output = Out; using output = Out;
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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_TERMINAL_STREAM_SCATTERER_HPP
#define CAF_TERMINAL_STREAM_SCATTERER_HPP
#include "caf/invalid_stream_scatterer.hpp"
namespace caf {
/// Special-purpose scatterer for sinks that terminate a stream. A terminal
/// stream scatterer generates infinite credit.
class terminal_stream_scatterer : public invalid_stream_scatterer {
public:
using super = invalid_stream_scatterer;
terminal_stream_scatterer(scheduled_actor* self);
~terminal_stream_scatterer() override;
size_t capacity() const noexcept override;
bool terminal() const noexcept override;
};
} // namespace caf
#endif // CAF_TERMINAL_STREAM_SCATTERER_HPP
...@@ -16,9 +16,10 @@ ...@@ -16,9 +16,10 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#include "caf/stream_scatterer.hpp" #include "caf/downstream_manager.hpp"
#include <functional> #include <functional>
#include <limits>
#include "caf/actor_addr.hpp" #include "caf/actor_addr.hpp"
#include "caf/actor_cast.hpp" #include "caf/actor_cast.hpp"
...@@ -28,36 +29,66 @@ ...@@ -28,36 +29,66 @@
namespace caf { namespace caf {
stream_scatterer::stream_scatterer::path_visitor::~path_visitor() { // -- constructors, destructors, and assignment operators ----------------------
downstream_manager::downstream_manager::path_visitor::~path_visitor() {
// nop // nop
} }
stream_scatterer::stream_scatterer::path_predicate::~path_predicate() { downstream_manager::downstream_manager::path_predicate::~path_predicate() {
// nop // nop
} }
stream_scatterer::stream_scatterer(scheduled_actor* self) : self_(self) { downstream_manager::downstream_manager(scheduled_actor* self) : self_(self) {
// nop // nop
} }
stream_scatterer::~stream_scatterer() { downstream_manager::~downstream_manager() {
// nop // nop
} }
bool stream_scatterer::clean() const noexcept { // -- properties ---------------------------------------------------------------
bool downstream_manager::terminal() const noexcept {
return true;
}
// -- path management ----------------------------------------------------------
size_t downstream_manager::num_paths() const noexcept {
return 0;
}
downstream_manager::path_ptr
downstream_manager::add_path(stream_slot slot, strong_actor_ptr target) {
CAF_LOG_TRACE(CAF_ARG(slot) << CAF_ARG(target));
unique_path_ptr ptr{new outbound_path(slot, std::move(target))};
auto result = ptr.get();
return insert_path(std::move(ptr)) ? result : nullptr;
}
bool downstream_manager::remove_path(stream_slot, error, bool) noexcept {
return false;
}
auto downstream_manager::path(stream_slot) noexcept -> path_ptr {
return nullptr;
}
bool downstream_manager::clean() const noexcept {
auto pred = [](const outbound_path& x) { auto pred = [](const outbound_path& x) {
return x.next_batch_id == x.next_ack_id; return x.next_batch_id == x.next_ack_id;
}; };
return buffered() == 0 && all_paths(pred); return buffered() == 0 && all_paths(pred);
} }
void stream_scatterer::close() { void downstream_manager::close() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
for_each_path([&](outbound_path& x) { about_to_erase(&x, false, nullptr); }); for_each_path([&](outbound_path& x) { about_to_erase(&x, false, nullptr); });
clear_paths(); clear_paths();
} }
void stream_scatterer::abort(error reason) { void downstream_manager::abort(error reason) {
CAF_LOG_TRACE(CAF_ARG(reason)); CAF_LOG_TRACE(CAF_ARG(reason));
for_each_path([&](outbound_path& x) { for_each_path([&](outbound_path& x) {
auto tmp = reason; auto tmp = reason;
...@@ -66,45 +97,75 @@ void stream_scatterer::abort(error reason) { ...@@ -66,45 +97,75 @@ void stream_scatterer::abort(error reason) {
clear_paths(); clear_paths();
} }
size_t stream_scatterer::min_credit() const { size_t downstream_manager::min_credit() const {
if (empty()) if (empty())
return 0u; return 0u;
auto result = std::numeric_limits<size_t>::max(); auto result = std::numeric_limits<size_t>::max();
const_cast<stream_scatterer*>(this)->for_each_path([&](outbound_path& x) { const_cast<downstream_manager*>(this)->for_each_path([&](outbound_path& x) {
result = std::min(result, static_cast<size_t>(x.open_credit)); result = std::min(result, static_cast<size_t>(x.open_credit));
}); });
return result; return result;
} }
size_t stream_scatterer::max_credit() const { size_t downstream_manager::max_credit() const {
size_t result = 0; size_t result = 0;
const_cast<stream_scatterer*>(this)->for_each_path([&](outbound_path& x) { const_cast<downstream_manager*>(this)->for_each_path([&](outbound_path& x) {
result = std::max(result, static_cast<size_t>(x.open_credit)); result = std::max(result, static_cast<size_t>(x.open_credit));
}); });
return result; return result;
} }
size_t stream_scatterer::total_credit() const { size_t downstream_manager::total_credit() const {
size_t result = 0; size_t result = 0;
const_cast<stream_scatterer*>(this)->for_each_path([&](outbound_path& x) { const_cast<downstream_manager*>(this)->for_each_path([&](outbound_path& x) {
result += static_cast<size_t>(x.open_credit); result += static_cast<size_t>(x.open_credit);
}); });
return result; return result;
} }
bool stream_scatterer::terminal() const noexcept { void downstream_manager::emit_batches() {
// nop
}
void downstream_manager::force_emit_batches() {
// nop
}
size_t downstream_manager::capacity() const noexcept {
return std::numeric_limits<size_t>::max();
}
size_t downstream_manager::buffered() const noexcept {
return 0;
}
void downstream_manager::clear_paths() {
// nop
}
bool downstream_manager::insert_path(unique_path_ptr) {
return false; return false;
} }
stream_scatterer::path_ptr stream_scatterer::add_path(stream_slot slot, void downstream_manager::for_each_path_impl(path_visitor&) {
strong_actor_ptr target) { // nop
CAF_LOG_TRACE(CAF_ARG(slot) << CAF_ARG(target)); }
unique_path_ptr ptr{new outbound_path(slot, std::move(target))};
auto result = ptr.get(); bool downstream_manager::check_paths_impl(path_algorithm algo,
return insert_path(std::move(ptr)) ? result : nullptr; path_predicate&) const noexcept {
// Return the result for empty ranges as specified by the C++ standard.
switch (algo) {
default:
CAF_ASSERT(algo == path_algorithm::all_of);
return true;
case path_algorithm::any_of:
return false;
case path_algorithm::none_of:
return true;
}
} }
void stream_scatterer::about_to_erase(outbound_path* ptr, bool silent, void downstream_manager::about_to_erase(outbound_path* ptr, bool silent,
error* reason) { error* reason) {
CAF_LOG_TRACE(CAF_ARG(ptr) << CAF_ARG(silent) << CAF_ARG(reason)); CAF_LOG_TRACE(CAF_ARG(ptr) << CAF_ARG(silent) << CAF_ARG(reason));
if (!silent) { if (!silent) {
......
...@@ -16,7 +16,7 @@ ...@@ -16,7 +16,7 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#include "caf/stream_scatterer_impl.hpp" #include "caf/downstream_manager_base.hpp"
#include <functional> #include <functional>
...@@ -25,21 +25,21 @@ ...@@ -25,21 +25,21 @@
namespace caf { namespace caf {
stream_scatterer_impl::stream_scatterer_impl(scheduled_actor* self) downstream_manager_base::downstream_manager_base(scheduled_actor* self)
: super(self) { : super(self) {
// nop // nop
} }
stream_scatterer_impl::~stream_scatterer_impl() { downstream_manager_base::~downstream_manager_base() {
// nop // nop
} }
size_t stream_scatterer_impl::num_paths() const noexcept { size_t downstream_manager_base::num_paths() const noexcept {
return paths_.size(); return paths_.size();
} }
bool stream_scatterer_impl::remove_path(stream_slot slot, error reason, bool downstream_manager_base::remove_path(stream_slot slot, error reason,
bool silent) noexcept { bool silent) noexcept {
CAF_LOG_TRACE(CAF_ARG(slot) << CAF_ARG(reason) << CAF_ARG(silent)); CAF_LOG_TRACE(CAF_ARG(slot) << CAF_ARG(reason) << CAF_ARG(silent));
auto i = paths_.find(slot); auto i = paths_.find(slot);
if (i != paths_.end()) { if (i != paths_.end()) {
...@@ -50,16 +50,16 @@ bool stream_scatterer_impl::remove_path(stream_slot slot, error reason, ...@@ -50,16 +50,16 @@ bool stream_scatterer_impl::remove_path(stream_slot slot, error reason,
return false; return false;
} }
auto stream_scatterer_impl::path(stream_slot slot) noexcept -> path_ptr { auto downstream_manager_base::path(stream_slot slot) noexcept -> path_ptr {
auto i = paths_.find(slot); auto i = paths_.find(slot);
return i != paths_.end() ? i->second.get() : nullptr; return i != paths_.end() ? i->second.get() : nullptr;
} }
void stream_scatterer_impl::clear_paths() { void downstream_manager_base::clear_paths() {
paths_.clear(); paths_.clear();
} }
bool stream_scatterer_impl::insert_path(unique_path_ptr ptr) { bool downstream_manager_base::insert_path(unique_path_ptr ptr) {
CAF_LOG_TRACE(CAF_ARG(ptr)); CAF_LOG_TRACE(CAF_ARG(ptr));
CAF_ASSERT(ptr != nullptr); CAF_ASSERT(ptr != nullptr);
auto slot = ptr->slots.sender; auto slot = ptr->slots.sender;
...@@ -67,13 +67,14 @@ bool stream_scatterer_impl::insert_path(unique_path_ptr ptr) { ...@@ -67,13 +67,14 @@ bool stream_scatterer_impl::insert_path(unique_path_ptr ptr) {
return paths_.emplace(slot, std::move(ptr)).second; return paths_.emplace(slot, std::move(ptr)).second;
} }
void stream_scatterer_impl::for_each_path_impl(path_visitor& f) { void downstream_manager_base::for_each_path_impl(path_visitor& f) {
for (auto& kvp : paths_) for (auto& kvp : paths_)
f(*kvp.second); f(*kvp.second);
} }
bool stream_scatterer_impl::check_paths_impl(path_algorithm algo, bool downstream_manager_base::check_paths_impl(path_algorithm algo,
path_predicate& pred) const noexcept { path_predicate& pred)
const noexcept {
auto f = [&](const map_type::value_type& x) { auto f = [&](const map_type::value_type& x) {
return pred(*x.second); return pred(*x.second);
}; };
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#include "caf/invalid_stream_scatterer.hpp"
#include "caf/logger.hpp"
#include "caf/message_builder.hpp"
#include "caf/outbound_path.hpp"
#include "caf/stream.hpp"
#include "caf/stream_slot.hpp"
namespace caf {
invalid_stream_scatterer::invalid_stream_scatterer(scheduled_actor* self)
: stream_scatterer(self) {
// nop
}
invalid_stream_scatterer::~invalid_stream_scatterer() {
// nop
}
size_t invalid_stream_scatterer::num_paths() const noexcept {
return 0;
}
bool invalid_stream_scatterer::remove_path(stream_slot, error, bool) noexcept {
return false;
}
auto invalid_stream_scatterer::path(stream_slot) noexcept -> path_ptr {
return nullptr;
}
void invalid_stream_scatterer::emit_batches() {
// nop
}
void invalid_stream_scatterer::force_emit_batches() {
// nop
}
size_t invalid_stream_scatterer::capacity() const noexcept {
return 0u;
}
size_t invalid_stream_scatterer::buffered() const noexcept {
return 0u;
}
bool invalid_stream_scatterer::insert_path(unique_path_ptr) {
return false;
}
void invalid_stream_scatterer::for_each_path_impl(path_visitor&) {
// nop
}
bool invalid_stream_scatterer::check_paths_impl(path_algorithm algo,
path_predicate&)
const noexcept {
// Return the result for empty ranges as specified by the C++ standard.
switch (algo) {
default: // all_of
return true;
case path_algorithm::any_of:
return false;
case path_algorithm::none_of:
return true;
}
}
void invalid_stream_scatterer::clear_paths() {
// nop
}
} // namespace caf
...@@ -30,7 +30,6 @@ ...@@ -30,7 +30,6 @@
#include "caf/outbound_path.hpp" #include "caf/outbound_path.hpp"
#include "caf/response_promise.hpp" #include "caf/response_promise.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/stream_scatterer.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. *
******************************************************************************/
#include "caf/terminal_stream_scatterer.hpp"
#include <limits>
namespace caf {
terminal_stream_scatterer::terminal_stream_scatterer(scheduled_actor* self)
: super(self) {
// nop
}
terminal_stream_scatterer::~terminal_stream_scatterer() {
// nop
}
size_t terminal_stream_scatterer::capacity() const noexcept {
return std::numeric_limits<size_t>::max();
}
bool terminal_stream_scatterer::terminal() const noexcept {
return true;
}
} // namespace caf
...@@ -17,13 +17,13 @@ ...@@ -17,13 +17,13 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#define CAF_SUITE broadcast_scatterer #define CAF_SUITE broadcast_downstream_manager
#include <memory> #include <memory>
#include "caf/actor_system.hpp" #include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp" #include "caf/actor_system_config.hpp"
#include "caf/broadcast_scatterer.hpp" #include "caf/broadcast_downstream_manager.hpp"
#include "caf/scheduled_actor.hpp" #include "caf/scheduled_actor.hpp"
#include "caf/mixin/sender.hpp" #include "caf/mixin/sender.hpp"
...@@ -119,8 +119,8 @@ public: ...@@ -119,8 +119,8 @@ public:
const char* cstr_name; const char* cstr_name;
/// Scatterer-under-test. /// Manager-under-test.
broadcast_scatterer<int> bs; broadcast_downstream_manager<int> bs;
/// Dummy mailbox. /// Dummy mailbox.
std::vector<message> mbox; std::vector<message> mbox;
...@@ -317,7 +317,7 @@ receive_checker<F> operator<<(receive_checker<F> xs, not_empty_t) { ...@@ -317,7 +317,7 @@ receive_checker<F> operator<<(receive_checker<F> xs, not_empty_t) {
// -- unit tests --------------------------------------------------------------- // -- unit tests ---------------------------------------------------------------
CAF_TEST_FIXTURE_SCOPE(broadcast_scatterer, fixture) CAF_TEST_FIXTURE_SCOPE(broadcast_downstream_manager, fixture)
CAF_TEST(one_path_force) { CAF_TEST(one_path_force) {
// Give alice 100 elements to send and a path to bob with desired batch size // Give alice 100 elements to send and a path to bob with desired batch size
......
...@@ -104,7 +104,7 @@ TESTEE(sum_up) { ...@@ -104,7 +104,7 @@ TESTEE(sum_up) {
} }
TESTEE_STATE(stream_multiplexer) { TESTEE_STATE(stream_multiplexer) {
stream_stage_ptr<int, int, broadcast_scatterer<int>> stage; stream_stage_ptr<int, int, broadcast_downstream_manager<int>> stage;
}; };
TESTEE(stream_multiplexer) { TESTEE(stream_multiplexer) {
......
...@@ -26,7 +26,7 @@ ...@@ -26,7 +26,7 @@
#include "caf/actor_system.hpp" #include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp" #include "caf/actor_system_config.hpp"
#include "caf/event_based_actor.hpp" #include "caf/event_based_actor.hpp"
#include "caf/fused_scatterer.hpp" #include "caf/fused_downstream_manager.hpp"
#include "caf/stateful_actor.hpp" #include "caf/stateful_actor.hpp"
using std::string; using std::string;
...@@ -37,9 +37,9 @@ namespace { ...@@ -37,9 +37,9 @@ namespace {
TESTEE_SETUP(); TESTEE_SETUP();
using int_scatterer = broadcast_scatterer<int>; using int_downstream_manager = broadcast_downstream_manager<int>;
using string_scatterer = broadcast_scatterer<string>; using string_downstream_manager = broadcast_downstream_manager<string>;
using ints_atom = atom_constant<atom("ints")>; using ints_atom = atom_constant<atom("ints")>;
...@@ -163,11 +163,11 @@ TESTEE(collect) { ...@@ -163,11 +163,11 @@ TESTEE(collect) {
}; };
} }
using int_scatterer = broadcast_scatterer<int>; using int_downstream_manager = broadcast_downstream_manager<int>;
using string_scatterer = broadcast_scatterer<string>; using string_downstream_manager = broadcast_downstream_manager<string>;
using scatterer = fused_scatterer<int_scatterer, string_scatterer>; using downstream_manager = fused_downstream_manager<int_downstream_manager, string_downstream_manager>;
class fused_stage : public stream_manager { class fused_stage : public stream_manager {
public: public:
...@@ -189,14 +189,14 @@ public: ...@@ -189,14 +189,14 @@ public:
if (batch.xs.match_elements<int_vec>()) { if (batch.xs.match_elements<int_vec>()) {
CAF_MESSAGE("handle an integer batch"); CAF_MESSAGE("handle an integer batch");
auto& xs = batch.xs.get_mutable_as<int_vec>(0); auto& xs = batch.xs.get_mutable_as<int_vec>(0);
auto& buf = out_.get<int_scatterer>().buf(); auto& buf = out_.get<int_downstream_manager>().buf();
buf.insert(buf.end(), xs.begin(), xs.end()); buf.insert(buf.end(), xs.begin(), xs.end());
return; return;
} }
if (batch.xs.match_elements<string_vec>()) { if (batch.xs.match_elements<string_vec>()) {
CAF_MESSAGE("handle a string batch"); CAF_MESSAGE("handle a string batch");
auto& xs = batch.xs.get_mutable_as<string_vec>(0); auto& xs = batch.xs.get_mutable_as<string_vec>(0);
auto& buf = out_.get<string_scatterer>().buf(); auto& buf = out_.get<string_downstream_manager>().buf();
buf.insert(buf.end(), xs.begin(), xs.end()); buf.insert(buf.end(), xs.begin(), xs.end());
return; return;
} }
...@@ -207,12 +207,12 @@ public: ...@@ -207,12 +207,12 @@ public:
return out_.capacity() == 0; return out_.capacity() == 0;
} }
scatterer& out() noexcept override { downstream_manager& out() noexcept override {
return out_; return out_;
} }
private: private:
scatterer out_; downstream_manager out_;
}; };
TESTEE_STATE(stream_multiplexer) { TESTEE_STATE(stream_multiplexer) {
...@@ -226,14 +226,14 @@ TESTEE(stream_multiplexer) { ...@@ -226,14 +226,14 @@ TESTEE(stream_multiplexer) {
auto& stg = self->state.stage; auto& stg = self->state.stage;
CAF_MESSAGE("received 'join' request for integers"); CAF_MESSAGE("received 'join' request for integers");
auto result = stg->add_unchecked_outbound_path<int>(); auto result = stg->add_unchecked_outbound_path<int>();
stg->out().assign<int_scatterer>(result.out()); stg->out().assign<int_downstream_manager>(result.out());
return result; return result;
}, },
[=](join_atom, strings_atom) { [=](join_atom, strings_atom) {
auto& stg = self->state.stage; auto& stg = self->state.stage;
CAF_MESSAGE("received 'join' request for strings"); CAF_MESSAGE("received 'join' request for strings");
auto result = stg->add_unchecked_outbound_path<string>(); auto result = stg->add_unchecked_outbound_path<string>();
stg->out().assign<string_scatterer>(result.out()); stg->out().assign<string_downstream_manager>(result.out());
return result; return result;
}, },
[=](const stream<int>& in) { [=](const stream<int>& in) {
......
...@@ -36,8 +36,9 @@ ...@@ -36,8 +36,9 @@
#include "caf/actor_system.hpp" #include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp" #include "caf/actor_system_config.hpp"
#include "caf/broadcast_scatterer.hpp" #include "caf/broadcast_downstream_manager.hpp"
#include "caf/buffered_scatterer.hpp" #include "caf/buffered_downstream_manager.hpp"
#include "caf/downstream_manager.hpp"
#include "caf/downstream_msg.hpp" #include "caf/downstream_msg.hpp"
#include "caf/inbound_path.hpp" #include "caf/inbound_path.hpp"
#include "caf/mailbox_element.hpp" #include "caf/mailbox_element.hpp"
...@@ -45,7 +46,6 @@ ...@@ -45,7 +46,6 @@
#include "caf/outbound_path.hpp" #include "caf/outbound_path.hpp"
#include "caf/send.hpp" #include "caf/send.hpp"
#include "caf/stream_manager.hpp" #include "caf/stream_manager.hpp"
#include "caf/stream_scatterer.hpp"
#include "caf/stream_sink_driver.hpp" #include "caf/stream_sink_driver.hpp"
#include "caf/stream_slot.hpp" #include "caf/stream_slot.hpp"
#include "caf/stream_source_driver.hpp" #include "caf/stream_source_driver.hpp"
...@@ -235,8 +235,8 @@ public: ...@@ -235,8 +235,8 @@ public:
void start_streaming(entity& ref, int num_messages) { void start_streaming(entity& ref, int num_messages) {
CAF_REQUIRE_NOT_EQUAL(num_messages, 0); CAF_REQUIRE_NOT_EQUAL(num_messages, 0);
using scatterer = broadcast_scatterer<int>; using downstream_manager = broadcast_downstream_manager<int>;
struct driver final : public stream_source_driver<scatterer> { struct driver final : public stream_source_driver<downstream_manager> {
public: public:
driver(int sentinel) : x_(0), sentinel_(sentinel) { driver(int sentinel) : x_(0), sentinel_(sentinel) {
// nop // nop
...@@ -262,8 +262,8 @@ public: ...@@ -262,8 +262,8 @@ public:
} }
void forward_to(entity& ref) { void forward_to(entity& ref) {
using scatterer = broadcast_scatterer<int>; using downstream_manager = broadcast_downstream_manager<int>;
struct driver final : public stream_stage_driver<int, scatterer> { struct driver final : public stream_stage_driver<int, downstream_manager> {
public: public:
driver(vector<int>* log) : log_(log) { driver(vector<int>* log) : log_(log) {
// nop // nop
...@@ -385,7 +385,7 @@ public: ...@@ -385,7 +385,7 @@ public:
mboxqueue mbox; mboxqueue mbox;
const char* name_; const char* name_;
vector<int> data; // Keeps track of all received data from all batches. vector<int> data; // Keeps track of all received data from all batches.
stream_stage_ptr<int, int, broadcast_scatterer<int>> forwarder; stream_stage_ptr<int, int, broadcast_downstream_manager<int>> forwarder;
tick_type ticks_per_force_batches_interval; tick_type ticks_per_force_batches_interval;
tick_type ticks_per_credit_interval; tick_type ticks_per_credit_interval;
......
...@@ -26,7 +26,7 @@ ...@@ -26,7 +26,7 @@
#include "caf/actor_system.hpp" #include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp" #include "caf/actor_system_config.hpp"
#include "caf/atom.hpp" #include "caf/atom.hpp"
#include "caf/broadcast_scatterer.hpp" #include "caf/broadcast_downstream_manager.hpp"
#include "caf/event_based_actor.hpp" #include "caf/event_based_actor.hpp"
#include "caf/stateful_actor.hpp" #include "caf/stateful_actor.hpp"
...@@ -55,7 +55,7 @@ struct select { ...@@ -55,7 +55,7 @@ struct select {
} }
}; };
using scatterer = broadcast_scatterer<value_type, level, select>; using downstream_manager = broadcast_downstream_manager<value_type, level, select>;
using buf = std::vector<value_type>; using buf = std::vector<value_type>;
...@@ -101,11 +101,11 @@ TESTEE(log_producer) { ...@@ -101,11 +101,11 @@ TESTEE(log_producer) {
return false; return false;
}, },
unit, unit,
policy::arg<scatterer>::value policy::arg<downstream_manager>::value
); );
auto& out = res.ptr()->out(); auto& out = res.ptr()->out();
static_assert(std::is_same<decltype(out), scatterer&>::value, static_assert(std::is_same<decltype(out), downstream_manager&>::value,
"source has wrong scatterer type"); "source has wrong downstream_manager type");
out.set_filter(res.out(), lvl); out.set_filter(res.out(), lvl);
return res; return res;
} }
...@@ -113,7 +113,7 @@ TESTEE(log_producer) { ...@@ -113,7 +113,7 @@ TESTEE(log_producer) {
} }
TESTEE_STATE(log_dispatcher) { TESTEE_STATE(log_dispatcher) {
stream_stage_ptr<value_type, value_type, scatterer> stage; stream_stage_ptr<value_type, value_type, downstream_manager> stage;
}; };
TESTEE(log_dispatcher) { TESTEE(log_dispatcher) {
...@@ -130,7 +130,7 @@ TESTEE(log_dispatcher) { ...@@ -130,7 +130,7 @@ TESTEE(log_dispatcher) {
[=](unit_t&, const error&) { [=](unit_t&, const error&) {
CAF_MESSAGE(self->name() << " is done"); CAF_MESSAGE(self->name() << " is done");
}, },
policy::arg<scatterer>::value policy::arg<downstream_manager>::value
); );
return { return {
[=](join_atom, level lvl) { [=](join_atom, level lvl) {
......
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