Commit 7787c1b8 authored by Dominik Charousset's avatar Dominik Charousset

Implement new class design for stream management

parent 9b3c07ad
...@@ -39,8 +39,6 @@ set (LIBCAF_CORE_SRCS ...@@ -39,8 +39,6 @@ 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_path.cpp
src/downstream_policy.cpp
src/duration.cpp src/duration.cpp
src/dynamic_message_data.cpp src/dynamic_message_data.cpp
src/error.cpp src/error.cpp
...@@ -51,10 +49,12 @@ set (LIBCAF_CORE_SRCS ...@@ -51,10 +49,12 @@ set (LIBCAF_CORE_SRCS
src/get_mac_addresses.cpp src/get_mac_addresses.cpp
src/get_process_id.cpp src/get_process_id.cpp
src/get_root_uuid.cpp src/get_root_uuid.cpp
src/greedy.cpp
src/group.cpp src/group.cpp
src/group_manager.cpp src/group_manager.cpp
src/group_module.cpp src/group_module.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
...@@ -69,11 +69,13 @@ set (LIBCAF_CORE_SRCS ...@@ -69,11 +69,13 @@ set (LIBCAF_CORE_SRCS
src/message_view.cpp src/message_view.cpp
src/monitorable_actor.cpp src/monitorable_actor.cpp
src/node_id.cpp src/node_id.cpp
src/outbound_path.cpp
src/parse_ini.cpp src/parse_ini.cpp
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.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
...@@ -91,16 +93,19 @@ set (LIBCAF_CORE_SRCS ...@@ -91,16 +93,19 @@ 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_handler.cpp src/stream_gatherer.cpp
src/stream_gatherer.cpp
src/stream_gatherer_impl.cpp
src/stream_id.cpp src/stream_id.cpp
src/stream_manager.cpp
src/stream_msg_visitor.cpp src/stream_msg_visitor.cpp
src/stream_priority.cpp src/stream_priority.cpp
src/stream_sink.cpp src/stream_scatterer.cpp
src/stream_source.cpp src/stream_scatterer_impl.cpp
src/stream_stage.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_coordinator.cpp src/test_coordinator.cpp
src/timestamp.cpp src/timestamp.cpp
src/try_match.cpp src/try_match.cpp
...@@ -108,8 +113,6 @@ set (LIBCAF_CORE_SRCS ...@@ -108,8 +113,6 @@ set (LIBCAF_CORE_SRCS
src/type_erased_value.cpp src/type_erased_value.cpp
src/uniform_type_info_map.cpp src/uniform_type_info_map.cpp
src/unprofiled.cpp src/unprofiled.cpp
src/upstream_path.cpp
src/upstream_policy.cpp
src/work_sharing.cpp src/work_sharing.cpp
src/work_stealing.cpp) src/work_stealing.cpp)
......
...@@ -17,26 +17,20 @@ ...@@ -17,26 +17,20 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_POLICY_BROADCAST_HPP #ifndef CAF_BROADCAST_SCATTERER_HPP
#define CAF_POLICY_BROADCAST_HPP #define CAF_BROADCAST_SCATTERER_HPP
#include "caf/downstream_policy.hpp" #include "caf/buffered_scatterer.hpp"
#include "caf/mixin/buffered_policy.hpp"
namespace caf { namespace caf {
namespace policy {
template <class T, class Base = mixin::buffered_policy<T, downstream_policy>> template <class T>
class broadcast : public Base { class broadcast_scatterer : public buffered_scatterer<T> {
public: public:
template <class... Ts> using super = buffered_scatterer<T>;
broadcast(Ts&&... xs) : Base(std::forward<Ts>(xs)...) {
// nop
}
void emit_batches() override { broadcast_scatterer(local_actor* selfptr) : super(selfptr) {
this->emit_broadcast(); // nop
} }
long credit() const override { long credit() const override {
...@@ -44,9 +38,21 @@ public: ...@@ -44,9 +38,21 @@ public:
// have filled our buffer to its minimum size. // have filled our buffer to its minimum size.
return this->min_credit() + this->min_buffer_size(); return this->min_credit() + this->min_buffer_size();
} }
void emit_batches() override {
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 policy
} // namespace caf } // namespace caf
#endif // CAF_POLICY_BROADCAST_HPP #endif // CAF_BROADCAST_SCATTERER_HPP
...@@ -17,55 +17,56 @@ ...@@ -17,55 +17,56 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_DOWNSTREAM_PATH_HPP #ifndef CAF_TOPIC_SCATTERER_HPP
#define CAF_DOWNSTREAM_PATH_HPP #define CAF_TOPIC_SCATTERER_HPP
#include <map>
#include <tuple>
#include <deque> #include <deque>
#include <vector> #include <vector>
#include <cstdint> #include <functional>
#include <cstddef>
#include "caf/fwd.hpp" #include "caf/buffered_scatterer.hpp"
#include "caf/stream_msg.hpp"
#include "caf/meta/type_name.hpp"
namespace caf { namespace caf {
/// Denotes a downstream actor in a stream topology. All downstream actors use /// A topic scatterer that delivers data in broadcast fashion to all sinks.
/// the stream ID registered with the hosting downstream object. template <class T, class Filter,
class downstream_path { class KeyCompare = std::equal_to<typename Filter::value_type>,
long KeyIndex = 0>
class broadcast_topic_scatterer
: public topic_scatterer<T, Filter, KeyCompare, KeyIndex> {
public: public:
/// Handle to the downstream actor. /// Base type.
strong_actor_ptr hdl; using super = buffered_scatterer<T>;
/// Next expected batch ID.
int64_t next_batch_id;
/// Currently available credit for this path.
long open_credit;
/// Stores whether the downstream actor is failsafe, i.e., allows the runtime broadcast_topic_scatterer(local_actor* selfptr) : super(selfptr) {
/// to redeploy it on failure. If this field is set to `false` then // nop
/// `unacknowledged_batches` is unused. }
bool redeployable;
/// Next expected batch ID to be acknowledged. Actors can receive a more long credit() const override {
/// advanced batch ID in an ACK message, since CAF uses accumulative ACKs. // We receive messages until we have exhausted all downstream credit and
int64_t next_ack_id; // have filled our buffer to its minimum size.
return this->min_credit() + this->min_buffer_size();
}
/// Caches batches until receiving an ACK. void emit_batches() override {
std::deque<std::pair<int64_t, stream_msg::batch>> unacknowledged_batches; this->fan_out();
for (auto& kvp : this->lanes_) {
downstream_path(strong_actor_ptr p, bool redeploy); 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) {
x->open_credit -= csize;
this->emit_batch(*x, static_cast<size_t>(csize), wrapped_chunk);
}
}
}
}; };
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, downstream_path& x) {
return f(meta::type_name("downstream_path"), x.hdl, x.next_batch_id,
x.open_credit, x.redeployable, x.unacknowledged_batches);
}
} // namespace caf } // namespace caf
#endif // CAF_DOWNSTREAM_PATH_HPP #endif // CAF_TOPIC_SCATTERER_HPP
...@@ -17,8 +17,8 @@ ...@@ -17,8 +17,8 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_MIXIN_BUFFERED_POLICYS_HPP #ifndef CAF_MIXIN_BUFFERED_SCATTERER_HPP
#define CAF_MIXIN_BUFFERED_POLICYS_HPP #define CAF_MIXIN_BUFFERED_SCATTERER_HPP
#include <deque> #include <deque>
#include <vector> #include <vector>
...@@ -26,25 +26,26 @@ ...@@ -26,25 +26,26 @@
#include <iterator> #include <iterator>
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/logger.hpp" #include "caf/stream_edge_impl.hpp"
#include "caf/actor_control_block.hpp" #include "caf/actor_control_block.hpp"
#include "caf/stream_scatterer_impl.hpp"
namespace caf { namespace caf {
namespace mixin {
/// 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, class Base, class Subtype = Base> template <class T>
class buffered_policy : public Base { class buffered_scatterer : public stream_scatterer_impl {
public: public:
using super = stream_scatterer_impl;
using value_type = T; using value_type = T;
using buffer_type = std::deque<value_type>; using buffer_type = std::deque<value_type>;
using chunk_type = std::vector<value_type>; using chunk_type = std::vector<value_type>;
template <class... Ts> buffered_scatterer(local_actor* selfptr) : super(selfptr) {
buffered_policy(Ts&&... xs) : Base(std::forward<Ts>(xs)...) {
// nop // nop
} }
...@@ -55,6 +56,7 @@ public: ...@@ -55,6 +56,7 @@ public:
/// @pre `n <= buf_.size()` /// @pre `n <= buf_.size()`
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));
chunk_type xs; chunk_type xs;
if (n > 0) { if (n > 0) {
xs.reserve(static_cast<size_t>(n)); xs.reserve(static_cast<size_t>(n));
...@@ -75,36 +77,10 @@ public: ...@@ -75,36 +77,10 @@ public:
return get_chunk(buf_, n); return get_chunk(buf_, n);
} }
long buf_size() const override { long buffered() const override {
return static_cast<long>(buf_.size()); return static_cast<long>(buf_.size());
} }
void emit_broadcast() override {
auto chunk = 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->open_credit -= csize;
this->emit_batch(*x, static_cast<size_t>(csize), wrapped_chunk);
}
}
void emit_anycast() override {
this->sort_paths_by_credit();
for (auto& x : this->paths_) {
auto chunk = get_chunk(x->open_credit);
auto csize = chunk.size();
if (csize == 0)
return;
x->open_credit -= csize;
this->emit_batch(*x, csize, std::move(make_message(std::move(chunk))));
}
}
buffer_type& buf() { buffer_type& buf() {
return buf_; return buf_;
} }
...@@ -117,7 +93,6 @@ protected: ...@@ -117,7 +93,6 @@ protected:
buffer_type buf_; buffer_type buf_;
}; };
} // namespace mixin
} // namespace caf } // namespace caf
#endif // CAF_MIXIN_BUFFERED_POLICYS_HPP #endif // CAF_MIXIN_BUFFERED_SCATTERER_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_APPLY_ARGS_HPP
#define CAF_DETAIL_APPLY_ARGS_HPP
#include <utility>
#include "caf/detail/int_list.hpp"
#include "caf/detail/type_list.hpp"
namespace caf {
namespace detail {
// this utterly useless function works around a bug in Clang that causes
// the compiler to reject the trailing return type of apply_args because
// "get" is not defined (it's found during ADL)
template<long Pos, class... Ts>
typename tl_at<type_list<Ts...>, Pos>::type get(const type_list<Ts...>&);
template <class F, long... Is, class Tuple>
auto apply_args(F& f, detail::int_list<Is...>, Tuple& tup)
-> decltype(f(get<Is>(tup)...)) {
return f(get<Is>(tup)...);
}
template <class F, long... Is, class Tuple>
auto apply_moved_args(F& f, detail::int_list<Is...>, Tuple& tup)
-> decltype(f(std::move(get<Is>(tup))...)) {
return f(std::move(get<Is>(tup))...);
}
template <class F, class Tuple, class... Ts>
auto apply_args_prefixed(F& f, detail::int_list<>, Tuple&, Ts&&... xs)
-> decltype(f(std::forward<Ts>(xs)...)) {
return f(std::forward<Ts>(xs)...);
}
template <class F, long... Is, class Tuple, class... Ts>
auto apply_args_prefixed(F& f, detail::int_list<Is...>, Tuple& tup, Ts&&... xs)
-> decltype(f(std::forward<Ts>(xs)..., get<Is>(tup)...)) {
return f(std::forward<Ts>(xs)..., get<Is>(tup)...);
}
template <class F, class Tuple, class... Ts>
auto apply_moved_args_prefixed(F& f, detail::int_list<>, Tuple&, Ts&&... xs)
-> decltype(f(std::forward<Ts>(xs)...)) {
return f(std::forward<Ts>(xs)...);
}
template <class F, long... Is, class Tuple, class... Ts>
auto apply_moved_args_prefixed(F& f, detail::int_list<Is...>, Tuple& tup, Ts&&... xs)
-> decltype(f(std::forward<Ts>(xs)..., std::move(get<Is>(tup))...)) {
return f(std::forward<Ts>(xs)..., std::move(get<Is>(tup))...);
}
template <class F, long... Is, class Tuple, class... Ts>
auto apply_args_suffxied(F& f, detail::int_list<Is...>, Tuple& tup, Ts&&... xs)
-> decltype(f(get<Is>(tup)..., std::forward<Ts>(xs)...)) {
return f(get<Is>(tup)..., std::forward<Ts>(xs)...);
}
} // namespace detail
} // namespace caf
#endif // CAF_DETAIL_APPLY_ARGS_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_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 detail {
/// Enables automagical string conversion for `CAF_ARG`.
template <class T>
struct arg_wrapper {
const char* name;
const T& value;
arg_wrapper(const char* x, const T& y) : name(x), value(y) {
// nop
}
};
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`.
template <class T>
static arg_wrapper<T> make_arg_wrapper(const char* name, const T& 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 caf
#endif // CAF_ARG_WRAPPER_HPP
...@@ -17,28 +17,27 @@ ...@@ -17,28 +17,27 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_POLICY_GREEDY_HPP #ifndef CAF_PULL5_GATHERER_HPP
#define CAF_POLICY_GREEDY_HPP #define CAF_PULL5_GATHERER_HPP
#include "caf/upstream_policy.hpp" #include "caf/random_gatherer.hpp"
namespace caf { namespace caf {
namespace policy { namespace detail {
/// Sends ACKs as early and often as possible. /// Always pulls exactly 5 elements from sources. Used in unit tests only.
class greedy : public upstream_policy { class pull5_gatherer : public random_gatherer {
public: public:
template <class... Ts> using super = random_gatherer;
greedy(Ts&&... xs) : upstream_policy(std::forward<Ts>(xs)...) {
// nop
}
~greedy() override; pull5_gatherer(local_actor* selfptr);
void fill_assignment_vec(long downstream_credit) override; void assign_credit(long available) override;
long initial_credit(long, inbound_path*) override;
}; };
} // namespace policy } // namespace detail
} // namespace caf } // namespace caf
#endif // CAF_POLICY_GREEDY_HPP #endif // CAF_PULL5_GATHERER_HPP
...@@ -17,31 +17,38 @@ ...@@ -17,31 +17,38 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_POLICY_ANYCAST_HPP #ifndef CAF_PUSH5_SCATTERER_HPP
#define CAF_POLICY_ANYCAST_HPP #define CAF_PUSH5_SCATTERER_HPP
#include "caf/downstream_policy.hpp" #include "caf/broadcast_scatterer.hpp"
#include "caf/mixin/buffered_policy.hpp"
namespace caf { namespace caf {
namespace policy { namespace detail {
template <class T, class Base = mixin::buffered_policy<T, downstream_policy>> /// Always pushs exactly 5 elements to sinks. Used in unit tests only.
class anycast : public Base { template <class T>
class push5_scatterer : public broadcast_scatterer<T> {
public: public:
void emit_batches() override { using super = broadcast_scatterer<T>;
this->emit_anycast();
push5_scatterer(local_actor* self) : super(self) {
// nop
}
long min_batch_size() const override {
return 1;
}
long max_batch_size() const override {
return 5;
} }
long credit() const override { long min_buffer_size() const override {
// We receive messages until we have exhausted all downstream credit and return 5;
// have filled our buffer to its minimum size.
return this->total_credit() + this->min_buffer_size();
} }
}; };
} // namespace policy } // namespace detail
} // namespace caf } // namespace caf
#endif // CAF_POLICY_ANYCAST_HPP #endif // CAF_PUSH5_SCATTERER_HPP
...@@ -23,9 +23,7 @@ ...@@ -23,9 +23,7 @@
#include <deque> #include <deque>
#include <vector> #include <vector>
#include "caf/stream_msg.hpp"
#include "caf/make_message.hpp" #include "caf/make_message.hpp"
#include "caf/downstream_path.hpp"
namespace caf { namespace caf {
......
...@@ -31,7 +31,6 @@ template <class> class param; ...@@ -31,7 +31,6 @@ template <class> class param;
template <class> class stream; template <class> class stream;
template <class> class optional; template <class> class optional;
template <class> class expected; template <class> class expected;
template <class> class upstream;
template <class> class downstream; template <class> class downstream;
template <class> class intrusive_ptr; template <class> class intrusive_ptr;
template <class> class behavior_type_of; template <class> class behavior_type_of;
...@@ -82,24 +81,23 @@ class deserializer; ...@@ -82,24 +81,23 @@ class deserializer;
class group_module; class group_module;
class message_view; class message_view;
class scoped_actor; class scoped_actor;
class stream_stage; class inbound_path;
class outbound_path;
class stream_source; class stream_source;
class upstream_path;
class abstract_actor; class abstract_actor;
class abstract_group; class abstract_group;
class actor_registry; class actor_registry;
class blocking_actor; class blocking_actor;
class execution_unit; class execution_unit;
class proxy_registry; class proxy_registry;
class stream_handler; class stream_manager;
class upstream_policy; class stream_gatherer;
class actor_companion; class actor_companion;
class downstream_path;
class mailbox_element; class mailbox_element;
class message_handler; class message_handler;
class scheduled_actor; class scheduled_actor;
class stream_scatterer;
class response_promise; class response_promise;
class downstream_policy;
class event_based_actor; class event_based_actor;
class type_erased_tuple; class type_erased_tuple;
class type_erased_value; class type_erased_value;
...@@ -201,8 +199,8 @@ using weak_actor_ptr = weak_intrusive_ptr<actor_control_block>; ...@@ -201,8 +199,8 @@ using weak_actor_ptr = weak_intrusive_ptr<actor_control_block>;
// -- intrusive pointer aliases ------------------------------------------------ // -- intrusive pointer aliases ------------------------------------------------
using stream_handler_ptr = intrusive_ptr<stream_handler>;
using strong_actor_ptr = intrusive_ptr<actor_control_block>; using strong_actor_ptr = intrusive_ptr<actor_control_block>;
using stream_manager_ptr = intrusive_ptr<stream_manager>;
// -- unique pointer aliases --------------------------------------------------- // -- unique pointer aliases ---------------------------------------------------
......
...@@ -17,13 +17,15 @@ ...@@ -17,13 +17,15 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_UPSTREAM_PATH_HPP #ifndef CAF_INBOUND_PATH_HPP
#define CAF_UPSTREAM_PATH_HPP #define CAF_INBOUND_PATH_HPP
#include <cstddef> #include <cstddef>
#include <cstdint> #include <cstdint>
#include "caf/stream_id.hpp" #include "caf/stream_id.hpp"
#include "caf/stream_msg.hpp"
#include "caf/stream_aborter.hpp"
#include "caf/stream_priority.hpp" #include "caf/stream_priority.hpp"
#include "caf/actor_control_block.hpp" #include "caf/actor_control_block.hpp"
...@@ -31,20 +33,31 @@ ...@@ -31,20 +33,31 @@
namespace caf { namespace caf {
/// Denotes an upstream actor in a stream topology. Each upstream actor can /// State for a path to an upstream actor (source).
/// refer to the stream using a different stream ID. class inbound_path {
class upstream_path {
public: public:
/// Handle to the upstream actor. /// Stream aborter flag to monitor a path.
strong_actor_ptr hdl; static constexpr const auto aborter_type = stream_aborter::source_aborter;
/// Message type for propagating graceful shutdowns.
using regular_shutdown = stream_msg::drop;
/// Message type for propagating errors.
using irregular_shutdown = stream_msg::forced_drop;
/// Pointer to the parent actor.
local_actor* self;
/// Stream ID used on this upstream path. /// Stream ID used on this source.
stream_id sid; stream_id sid;
/// Priority of this input channel. /// Handle to the source.
strong_actor_ptr hdl;
/// Priority of incoming batches from this source.
stream_priority prio; stream_priority prio;
/// ID of the last received batch we have acknowledged. /// ID of the last acknowledged batch ID.
int64_t last_acked_batch_id; int64_t last_acked_batch_id;
/// ID of the last received batch. /// ID of the last received batch.
...@@ -53,15 +66,40 @@ public: ...@@ -53,15 +66,40 @@ public:
/// Amount of credit we have signaled upstream. /// Amount of credit we have signaled upstream.
long assigned_credit; long assigned_credit;
upstream_path(strong_actor_ptr ptr, stream_id id, stream_priority p); /// Stores whether the source actor is failsafe, i.e., allows the runtime to
/// redeploy it on failure.
bool redeployable;
/// Stores whether an error occurred during stream processing. Configures
/// whether the destructor sends `close` or `forced_close` messages.
error shutdown_reason;
/// Constructs a path for given handle and stream ID.
inbound_path(local_actor* selfptr, const stream_id& id, strong_actor_ptr ptr);
~inbound_path();
/// Updates `last_batch_id` and `assigned_credit`.
void handle_batch(long batch_size, int64_t batch_id);
/// Emits a `stream_msg::ack_batch` on this path and sets `assigned_credit`
/// to `initial_demand`.
void emit_ack_open(actor_addr rebind_from, long initial_demand,
bool redeployable);
void emit_ack_batch(long new_demand);
static void emit_irregular_shutdown(local_actor* self, const stream_id& sid,
const strong_actor_ptr& hdl,
error reason);
}; };
template <class Inspector> template <class Inspector>
typename Inspector::return_type inspect(Inspector& f, upstream_path& x) { typename Inspector::return_type inspect(Inspector& f, inbound_path& x) {
return f(meta::type_name("upstream_path"), x.hdl, x.sid, x.prio, return f(meta::type_name("inbound_path"), x.hdl, x.sid, x.prio,
x.last_acked_batch_id, x.last_batch_id, x.assigned_credit); x.last_acked_batch_id, x.last_batch_id, x.assigned_credit);
} }
} // namespace caf } // namespace caf
#endif // CAF_UPSTREAM_PATH_HPP #endif // CAF_INBOUND_PATH_HPP
...@@ -17,56 +17,61 @@ ...@@ -17,56 +17,61 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_STREAM_SOURCE_HPP #ifndef CAF_INVALID_STREAM_GATHERER_HPP
#define CAF_STREAM_SOURCE_HPP #define CAF_INVALID_STREAM_GATHERER_HPP
#include <memory> #include "caf/stream_gatherer.hpp"
#include "caf/extend.hpp"
#include "caf/stream_handler.hpp"
#include "caf/downstream_policy.hpp"
#include "caf/mixin/has_downstreams.hpp"
namespace caf { namespace caf {
class stream_source : public extend<stream_handler, stream_source>:: /// Type-erased policy for receiving data from sources.
with<mixin::has_downstreams> { class invalid_stream_gatherer : public stream_gatherer {
public: public:
stream_source(downstream_policy* out_ptr); invalid_stream_gatherer() = default;
~invalid_stream_gatherer() override;
path_ptr add_path(const stream_id& sid, strong_actor_ptr x,
strong_actor_ptr original_stage, stream_priority prio,
long available_credit, bool redeployable,
response_promise result_cb) override;
bool remove_path(const stream_id& sid, const actor_addr& x, error reason,
bool silent) override;
void close(message reason) override;
void abort(error reason) override;
long num_paths() const override;
bool closed() const override;
bool continuous() const override;
~stream_source() override; void continuous(bool value) override;
bool done() const override; path_ptr path_at(size_t index) override;
error downstream_ack(strong_actor_ptr& hdl, int64_t bid, long value) override; path_ptr find(const stream_id& sid, const actor_addr& x) override;
void abort(strong_actor_ptr& cause, const error& reason) override; long high_watermark() const override;
inline downstream_policy& out() { long min_credit_assignment() const override;
return *out_ptr_;
}
/// Convenience function to trigger generation of new elements. long max_credit() const override;
void generate();
protected: void high_watermark(long x) override;
void downstream_demand(downstream_path* path, long demand) override;
/// Queries the current amount of elements in the output buffer. void min_credit_assignment(long x) override;
virtual long buf_size() const = 0;
/// Generate new elements for the output buffer. The size hint `n` indicates void max_credit(long x) override;
/// how much elements can be shipped immediately.
virtual void generate(size_t n) = 0;
/// Queries whether the source is exhausted. void assign_credit(long downstream_capacity) override;
virtual bool at_end() const = 0;
private: long initial_credit(long downstream_capacity, path_type* x) override;
downstream_policy* out_ptr_;
}; };
} // namespace caf } // namespace caf
#endif // CAF_STREAM_SOURCE_HPP #endif // CAF_INVALID_STREAM_GATHERER_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_INVALID_STREAM_SCATTERER_HPP
#define CAF_INVALID_STREAM_SCATTERER_HPP
#include "caf/stream_scatterer.hpp"
namespace caf {
/// Type-erased policy for dispatching data to sinks.
class invalid_stream_scatterer : public stream_scatterer {
public:
invalid_stream_scatterer() = default;
~invalid_stream_scatterer() override;
path_ptr add_path(const stream_id& sid, strong_actor_ptr origin,
strong_actor_ptr sink_ptr,
mailbox_element::forwarding_stack stages,
message_id handshake_mid, message handshake_data,
stream_priority prio, bool redeployable) override;
path_ptr confirm_path(const stream_id& sid, const actor_addr& from,
strong_actor_ptr to, long initial_demand,
bool redeployable) override;
bool remove_path(const stream_id& sid, const actor_addr& x,
error reason, bool silent) override;
void close() override;
void abort(error reason) override;
long num_paths() const override;
bool closed() const override;
bool continuous() const override;
void continuous(bool value) override;
path_type* path_at(size_t index) override;
void emit_batches() override;
path_type* find(const stream_id& sid, const actor_addr& x) override;
long credit() const override;
long buffered() const override;
long min_batch_size() const override;
long max_batch_size() const override;
long min_buffer_size() const override;
duration max_batch_delay() const override;
void min_batch_size(long x) override;
void max_batch_size(long x) override;
void min_buffer_size(long x) override;
void max_batch_delay(duration x) override;
};
} // namespace caf
#endif // CAF_INVALID_STREAM_SCATTERER_HPP
...@@ -220,6 +220,63 @@ public: ...@@ -220,6 +220,63 @@ public:
return current_element_->sender; return current_element_->sender;
} }
/// Returns the ID of the current message.
inline message_id current_message_id() {
CAF_ASSERT(current_element_);
return current_element_->mid;
}
/// Returns the ID of the current message and marks the ID stored in the
/// current mailbox element as answered.
inline message_id take_current_message_id() {
CAF_ASSERT(current_element_);
auto result = current_element_->mid;
current_element_->mid.mark_as_answered();
return result;
}
/// Marks the current message ID as answered.
inline void drop_current_message_id() {
CAF_ASSERT(current_element_);
current_element_->mid.mark_as_answered();
}
/// Returns a pointer to the next stage from the forwarding path of the
/// current message or `nullptr` if the path is empty.
inline strong_actor_ptr current_next_stage() {
CAF_ASSERT(current_element_);
auto& stages = current_element_->stages;
if (!stages.empty())
stages.back();
return nullptr;
}
/// Returns a pointer to the next stage from the forwarding path of the
/// current message and removes it from the path. Returns `nullptr` if the
/// path is empty.
inline strong_actor_ptr take_current_next_stage() {
CAF_ASSERT(current_element_);
auto& stages = current_element_->stages;
if (!stages.empty()) {
auto result = stages.back();
stages.pop_back();
return result;
}
return nullptr;
}
/// Returns the forwarding stack from the current mailbox element.
const mailbox_element::forwarding_stack& current_forwarding_stack() {
CAF_ASSERT(current_element_);
return current_element_->stages;
}
/// Moves the forwarding stack from the current mailbox element.
mailbox_element::forwarding_stack take_current_forwarding_stack() {
CAF_ASSERT(current_element_);
return std::move(current_element_->stages);
}
/// Returns a pointer to the currently processed mailbox element. /// Returns a pointer to the currently processed mailbox element.
inline mailbox_element* current_mailbox_element() { inline mailbox_element* current_mailbox_element() {
return current_element_; return current_element_;
......
...@@ -306,6 +306,8 @@ bool operator==(const logger::field& x, const logger::field& y); ...@@ -306,6 +306,8 @@ bool operator==(const logger::field& x, const logger::field& y);
#define CAF_ARG(argument) caf::logger::make_arg_wrapper(#argument, argument) #define CAF_ARG(argument) caf::logger::make_arg_wrapper(#argument, argument)
#define CAF_ARG2(argname, argval) caf::logger::make_arg_wrapper(argname, argval)
#ifdef CAF_MSVC #ifdef CAF_MSVC
#define CAF_PRETTY_FUN __FUNCSIG__ #define CAF_PRETTY_FUN __FUNCSIG__
#else // CAF_MSVC #else // CAF_MSVC
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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_MIXIN_HAS_UPSTREAMS_HPP
#define CAF_MIXIN_HAS_UPSTREAMS_HPP
#include "caf/sec.hpp"
#include "caf/expected.hpp"
#include "caf/stream_id.hpp"
#include "caf/stream_priority.hpp"
#include "caf/actor_control_block.hpp"
namespace caf {
namespace mixin {
/// Mixin for streams with has number of upstream actors.
template <class Base, class Subtype>
class has_upstreams : public Base {
public:
error close_upstream(strong_actor_ptr& ptr) final {
if (in().remove_path(ptr)) {
if (in().closed())
dptr()->last_upstream_closed();
return none;
}
return sec::invalid_upstream;
}
private:
Subtype* dptr() {
return static_cast<Subtype*>(this);
}
upstream_policy& in() {
return dptr()->in();
}
};
} // namespace mixin
} // namespace caf
#endif // CAF_MIXIN_HAS_UPSTREAMS_HPP
...@@ -17,28 +17,27 @@ ...@@ -17,28 +17,27 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_POLICY_PULL5_HPP #ifndef CAF_NO_STAGES_HPP
#define CAF_POLICY_PULL5_HPP #define CAF_NO_STAGES_HPP
#include "caf/upstream_policy.hpp" #include "caf/mailbox_element.hpp"
namespace caf { namespace caf {
namespace policy {
/// Sends ACKs as early and often as possible. /// Convenience tag type for producing empty forwarding stacks.
class pull5 : public upstream_policy { struct no_stages_t {
public: constexpr no_stages_t() {
template <class... Ts>
pull5(Ts&&... xs) : upstream_policy(std::forward<Ts>(xs)...) {
// nop // nop
} }
~pull5() override; inline operator mailbox_element::forwarding_stack() const {
return {};
void fill_assignment_vec(long downstream_credit) override; }
}; };
} // namespace policy /// Convenience tag for producing empty forwarding stacks.
constexpr no_stages_t no_stages = no_stages_t{};
} // namespace caf } // namespace caf
#endif // CAF_POLICY_PULL5_HPP #endif // CAF_NO_STAGES_HPP
...@@ -17,91 +17,107 @@ ...@@ -17,91 +17,107 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_STREAM_HANDLER_HPP #ifndef CAF_OUTBOUND_PATH_HPP
#define CAF_STREAM_HANDLER_HPP #define CAF_OUTBOUND_PATH_HPP
#include <deque>
#include <vector> #include <vector>
#include <cstdint> #include <cstdint>
#include <cstddef> #include <cstddef>
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/ref_counted.hpp" #include "caf/stream_id.hpp"
#include "caf/stream_msg.hpp"
#include "caf/stream_aborter.hpp"
#include "caf/actor_control_block.hpp"
#include "caf/meta/type_name.hpp"
namespace caf { namespace caf {
/// Manages a single stream with any number of down- and upstream actors. /// State for a single path to a sink on a `stream_scatterer`.
class stream_handler : public ref_counted { class outbound_path {
public: public:
~stream_handler() override; /// Stream aborter flag to monitor a path.
static constexpr const auto aborter_type = stream_aborter::sink_aborter;
/// Message type for propagating graceful shutdowns.
using regular_shutdown = stream_msg::close;
/// Message type for propagating errors.
using irregular_shutdown = stream_msg::forced_close;
/// Stores information about the initiator of the steam.
struct client_data {
strong_actor_ptr hdl;
message_id mid;
};
// -- handler for downstream events ------------------------------------------ /// Pointer to the parent actor.
local_actor* self;
/// Add a new downstream actor to the stream with in-flight /// Stream ID used by the sink.
/// `stream_msg::open` message. stream_id sid;
/// @param hdl Handle to the new downstream actor.
/// @pre `hdl != nullptr`
virtual error add_downstream(strong_actor_ptr& hdl);
/// Confirms a downstream actor after receiving its `stream_msg::ack_open`. /// Handle to the sink.
/// @param hdl Handle to the new downstream actor. strong_actor_ptr hdl;
/// @param initial_demand Credit received with `ack_open`.
/// @param redeployable Denotes whether the runtime can redeploy
/// the downstream actor on failure.
/// @pre `hdl != nullptr`
virtual error confirm_downstream(const strong_actor_ptr& rebind_from,
strong_actor_ptr& hdl, long initial_demand,
bool redeployable);
/// Handles cumulative ACKs with new demand from a downstream actor. /// Next expected batch ID.
/// @pre `hdl != nullptr` int64_t next_batch_id;
/// @pre `new_demand > 0`
virtual error downstream_ack(strong_actor_ptr& hdl, int64_t batch_id,
long new_demand);
/// Push new data to downstream actors by sending batches. The amount of /// Currently available credit for this path.
/// pushed data is limited by `hint` or the available credit if long open_credit;
/// `hint == nullptr`.
virtual error push();
// -- handler for upstream events -------------------------------------------- /// Stores whether the downstream actor is failsafe, i.e., allows the runtime
/// to redeploy it on failure. If this field is set to `false` then
/// `unacknowledged_batches` is unused.
bool redeployable;
/// Add a new upstream actor to the stream and return an initial credit. /// Next expected batch ID to be acknowledged. Actors can receive a more
virtual expected<long> add_upstream(strong_actor_ptr& hdl, /// advanced batch ID in an ACK message, since CAF uses accumulative ACKs.
const stream_id& sid, int64_t next_ack_id;
stream_priority prio);
/// Handles data from an upstream actor. /// Caches batches until receiving an ACK.
virtual error upstream_batch(strong_actor_ptr& hdl, int64_t xs_id, std::deque<std::pair<int64_t, stream_msg::batch>> unacknowledged_batches;
long xs_size, message& xs);
/// Closes an upstream. /// Caches the initiator of the stream (client) with the original request ID
virtual error close_upstream(strong_actor_ptr& hdl); /// until the stream handshake is either confirmed or aborted. Once
/// confirmed, the next stage takes responsibility for answering to the
/// client.
client_data cd;
// -- handler for stream-wide events ----------------------------------------- /// Stores whether an error occurred during stream processing.
error shutdown_reason;
/// Signals an error at the up- or downstream actor `hdl`. This function is /// Constructs a path for given handle and stream ID.
/// called with `hdl == nullptr` if the parent actor shuts down. outbound_path(local_actor* selfptr, const stream_id& id,
virtual void abort(strong_actor_ptr& hdl, const error& reason) = 0; strong_actor_ptr ptr);
// -- accessors -------------------------------------------------------------- ~outbound_path();
virtual bool done() const = 0; /// Sets `open_credit` to `initial_credit` and clears `cached_handshake`.
void handle_ack_open(long initial_credit);
/// Returns a type-erased `stream<T>` as handshake token for downstream void emit_open(strong_actor_ptr origin,
/// actors. Returns an empty message for sinks. mailbox_element::forwarding_stack stages, message_id mid,
virtual message make_output_token(const stream_id&) const; message handshake_data, stream_priority prio,
bool redeployable);
/// Returns the downstream policy if this handler is a sink or stage. /// Emits a `stream_msg::batch` on this path, decrements `open_credit` by
virtual optional<downstream_policy&> dp(); /// `xs_size` and increments `next_batch_id` by 1.
void emit_batch(long xs_size, message xs);
/// Returns the upstream policy if this handler is a source or stage. static void emit_irregular_shutdown(local_actor* self, const stream_id& sid,
virtual optional<upstream_policy&> up(); const strong_actor_ptr& hdl,
error reason);
}; };
/// A reference counting pointer to a `stream_handler`. template <class Inspector>
/// @relates stream_handler typename Inspector::result_type inspect(Inspector& f, outbound_path& x) {
using stream_handler_ptr = intrusive_ptr<stream_handler>; return f(meta::type_name("outbound_path"), x.hdl, x.sid, x.next_batch_id,
x.open_credit, x.redeployable, x.unacknowledged_batches);
}
} // namespace caf } // namespace caf
#endif // CAF_STREAM_HANDLER_HPP #endif // CAF_OUTBOUND_PATH_HPP
...@@ -20,14 +20,10 @@ ...@@ -20,14 +20,10 @@
#ifndef CAF_POLICY_ARG_HPP #ifndef CAF_POLICY_ARG_HPP
#define CAF_POLICY_ARG_HPP #define CAF_POLICY_ARG_HPP
#include "caf/downstream_policy.hpp"
#include "caf/mixin/buffered_policy.hpp"
namespace caf { namespace caf {
namespace policy { namespace policy {
/// Provides a wrapper to pass policy arguments to functions. /// Provides a wrapper to pass policy types as values to functions.
template <class... Ts> template <class... Ts>
struct arg { struct arg {
public: public:
......
...@@ -17,17 +17,27 @@ ...@@ -17,17 +17,27 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#include "caf/downstream_path.hpp" #ifndef CAF_RANDOM_GATHERER_HPP
#define CAF_RANDOM_GATHERER_HPP
#include "caf/stream_gatherer_impl.hpp"
namespace caf { namespace caf {
downstream_path::downstream_path(strong_actor_ptr p, bool redeploy) /// Pulls data from sources in arbitrary order.
: hdl(std::move(p)), class random_gatherer : public stream_gatherer_impl {
next_batch_id(0), public:
open_credit(0), using super = stream_gatherer_impl;
redeployable(redeploy),
next_ack_id(0) { random_gatherer(local_actor* selfptr);
// nop
} ~random_gatherer() override;
void assign_credit(long downstream_capacity) override;
long initial_credit(long downstream_capacity, path_type* x) override;
};
} // namespace caf } // namespace caf
#endif // CAF_RANDOM_GATHERER_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_RANDOM_TOPIC_SCATTERER_HPP
#define CAF_RANDOM_TOPIC_SCATTERER_HPP
#include <map>
#include <tuple>
#include <deque>
#include <vector>
#include <functional>
#include "caf/topic_scatterer.hpp"
namespace caf {
/// A topic scatterer that delivers data to sinks in random order.
template <class T, class Filter,
class KeyCompare = std::equal_to<typename Filter::value_type>,
long KeyIndex = 0>
class random_topic_scatterer
: public topic_scatterer<T, Filter, KeyCompare, KeyIndex> {
public:
using super = topic_scatterer<T, Filter, KeyCompare, KeyIndex>;
random_topic_scatterer(local_actor* selfptr) : super(selfptr) {
// nop
}
long credit() const override {
// We receive messages until we have exhausted all downstream credit and
// have filled our buffer to its minimum size.
return this->total_credit() + this->min_buffer_size();
}
void emit_batches() override {
this->fan_out();
for (auto& kvp : this->lanes_) {
auto& l = kvp.second;
super::sort_by_credit(l.paths);
for (auto& x : l.paths) {
auto chunk = super::get_chunk(l.buf, x->open_credit);
auto csize = static_cast<long>(chunk.size());
if (csize == 0)
break;
x->emit_batch(csize, make_message(std::move(chunk)));
}
}
}
};
} // namespace caf
#endif // CAF_RANDOM_TOPIC_SCATTERER_HPP
This diff is collapsed.
...@@ -108,8 +108,10 @@ enum class sec : uint8_t { ...@@ -108,8 +108,10 @@ enum class sec : uint8_t {
stream_init_failed, stream_init_failed,
/// Unable to process a stream since due to missing state. /// Unable to process a stream since due to missing state.
invalid_stream_state, invalid_stream_state,
/// Stream aborted due to unexpected error.
unhandled_stream_error,
/// A function view was called without assigning an actor first. /// A function view was called without assigning an actor first.
bad_function_call bad_function_call = 40
}; };
/// @relates sec /// @relates sec
......
...@@ -22,7 +22,8 @@ ...@@ -22,7 +22,8 @@
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/stream_id.hpp" #include "caf/stream_id.hpp"
#include "caf/stream_handler.hpp" #include "caf/stream_manager.hpp"
#include "caf/meta/type_name.hpp" #include "caf/meta/type_name.hpp"
namespace caf { namespace caf {
...@@ -46,13 +47,17 @@ public: ...@@ -46,13 +47,17 @@ public:
stream& operator=(stream&&) = default; stream& operator=(stream&&) = default;
stream& operator=(const stream&) = default; stream& operator=(const stream&) = default;
stream(none_t) : stream() {
// nop
}
stream(stream_id sid) : id_(std::move(sid)) { stream(stream_id sid) : id_(std::move(sid)) {
// nop // nop
} }
/// Convenience constructor for returning the result of `self->new_stream` /// Convenience constructor for returning the result of `self->new_stream`
/// and similar functions. /// and similar functions.
stream(stream_id sid, stream_handler_ptr sptr) stream(stream_id sid, stream_manager_ptr sptr)
: id_(std::move(sid)), : id_(std::move(sid)),
ptr_(std::move(sptr)) { ptr_(std::move(sptr)) {
// nop // nop
...@@ -60,7 +65,7 @@ public: ...@@ -60,7 +65,7 @@ public:
/// Convenience constructor for returning the result of `self->new_stream` /// Convenience constructor for returning the result of `self->new_stream`
/// and similar functions. /// and similar functions.
stream(stream other, stream_handler_ptr sptr) stream(stream other, stream_manager_ptr sptr)
: id_(std::move(other.id_)), : id_(std::move(other.id_)),
ptr_(std::move(sptr)) { ptr_(std::move(sptr)) {
// nop // nop
...@@ -78,7 +83,7 @@ public: ...@@ -78,7 +83,7 @@ public:
} }
/// Returns the handler assigned to this stream on this actor. /// Returns the handler assigned to this stream on this actor.
inline const stream_handler_ptr& ptr() const { inline const stream_manager_ptr& ptr() const {
return ptr_; return ptr_;
} }
...@@ -93,7 +98,7 @@ private: ...@@ -93,7 +98,7 @@ private:
// -- member variables ------------------------------------------------------- // -- member variables -------------------------------------------------------
stream_id id_; stream_id id_;
stream_handler_ptr ptr_; stream_manager_ptr ptr_;
}; };
/// @relates stream /// @relates stream
......
...@@ -29,9 +29,15 @@ namespace caf { ...@@ -29,9 +29,15 @@ namespace caf {
class stream_aborter : public attachable { class stream_aborter : public attachable {
public: public:
enum mode {
source_aborter,
sink_aborter
};
struct token { struct token {
const actor_addr& observer; const actor_addr& observer;
const stream_id& sid; const stream_id& sid;
mode m;
static constexpr size_t token_type = attachable::token::stream_aborter; static constexpr size_t token_type = attachable::token::stream_aborter;
}; };
...@@ -42,25 +48,27 @@ public: ...@@ -42,25 +48,27 @@ public:
bool matches(const attachable::token& what) override; bool matches(const attachable::token& what) override;
inline static attachable_ptr make(actor_addr observed, actor_addr observer, inline static attachable_ptr make(actor_addr observed, actor_addr observer,
const stream_id& sid) { const stream_id& sid, mode m) {
return attachable_ptr{ return attachable_ptr{
new stream_aborter(std::move(observed), std::move(observer), sid)}; new stream_aborter(std::move(observed), std::move(observer), sid, m)};
} }
/// Adds a stream aborter to `observed`. /// Adds a stream aborter to `observed`.
static void add(strong_actor_ptr observed, actor_addr observer, static void add(strong_actor_ptr observed, actor_addr observer,
const stream_id& sid); const stream_id& sid, mode m);
/// Removes a stream aborter from `observed`. /// Removes a stream aborter from `observed`.
static void del(strong_actor_ptr observed, const actor_addr& observer, static void del(strong_actor_ptr observed, const actor_addr& observer,
const stream_id& sid); const stream_id& sid, mode m);
private: private:
stream_aborter(actor_addr&& observed, actor_addr&& observer, stream_aborter(actor_addr&& observed, actor_addr&& observer,
const stream_id& type); const stream_id& type, mode m);
actor_addr observed_; actor_addr observed_;
actor_addr observer_; actor_addr observer_;
stream_id sid_; stream_id sid_;
mode mode_;
}; };
} // namespace caf } // namespace caf
......
...@@ -17,156 +17,120 @@ ...@@ -17,156 +17,120 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_UPSTREAM_POLICY_HPP #ifndef CAF_STREAM_GATHERER_HPP
#define CAF_UPSTREAM_POLICY_HPP #define CAF_STREAM_GATHERER_HPP
#include <vector> #include <vector>
#include <cstdint> #include <cstdint>
#include <utility> #include <utility>
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/none.hpp"
#include "caf/error.hpp"
namespace caf { namespace caf {
class upstream_policy { /// Type-erased policy for receiving data from sources.
class stream_gatherer {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
/// A raw pointer to a downstream path. /// Type of a single path to a data source.
using path_ptr = upstream_path*; using path_type = inbound_path;
/// A unique pointer to a upstream path. /// Pointer to a single path to a data source.
using path_uptr = std::unique_ptr<upstream_path>; using path_ptr = path_type*;
/// Stores all available paths.
using path_uptr_vec = std::vector<path_uptr>;
/// List of views to paths.
using path_ptr_vec = std::vector<path_ptr>;
/// Describes an assignment of credit to an upstream actor.
using assignment_pair = std::pair<upstream_path*, long>;
/// Describes an assignment of credit to all upstream actors.
using assignment_vec = std::vector<assignment_pair>;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
upstream_policy(local_actor* selfptr); stream_gatherer() = default;
virtual ~upstream_policy();
// -- path management -------------------------------------------------------- virtual ~stream_gatherer();
/// Returns `true` if all upstream paths are closed and this upstream is not // -- pure virtual memeber functions -----------------------------------------
/// flagged as `continuous`, `false` otherwise.
inline bool closed() const {
return paths_.empty() && !continuous_;
}
/// Returns whether this upstream remains open even if no more upstream path /// Adds a path to the edge and emits `ack_open` to the source.
/// exists. /// @param sid Stream ID used by the source.
inline bool continuous() const { /// @param x Handle to the source.
return continuous_; /// @param original_stage Actor that received the stream handshake initially.
} /// @param prio Priority of data on this path.
/// @param available_credit Maximum credit for granting to the source.
/// @param redeployable Stores whether the source can re-appear after aborts.
/// @param result_cb Callback for the listener of the final stream result.
/// The gatherer must ignore the promise when returning
/// `nullptr`, because the previous stage is responsible for
/// it until the gatherer acknowledges the handshake. The
/// callback is invalid if the stream has a next stage.
/// @returns The added path on success, `nullptr` otherwise.
virtual path_ptr add_path(const stream_id& sid, strong_actor_ptr x,
strong_actor_ptr original_stage,
stream_priority prio, long available_credit,
bool redeployable, response_promise result_cb) = 0;
/// Sets whether this upstream remains open even if no more upstream path /// Removes a path from the gatherer.
/// exists. virtual bool remove_path(const stream_id& sid, const actor_addr& x,
inline void continuous(bool value) { error reason, bool silent) = 0;
continuous_ = value;
}
/// Sends an abort message to all upstream actors and closes the stream. /// Removes all paths gracefully.
void abort(strong_actor_ptr& cause, const error& reason); virtual void close(message result) = 0;
/// Removes all paths with an error message.
virtual void abort(error reason) = 0;
/// Assigns credit to upstream actors according to the current capacity of /// Returns the number of paths managed on this edge.
/// all downstream actors (and a minimum buffer size) combined. virtual long num_paths() const = 0;
void assign_credit(long downstream_capacity);
/// Adds a new upstream actor and returns the initial credit. /// Returns `true` if no downstream exists, `false` otherwise.
expected<long> add_path(strong_actor_ptr hdl, const stream_id& sid, virtual bool closed() const = 0;
stream_priority prio,
long downstream_credit);
/// Removes `hdl` as upstream actor. The actor is removed without any further /// Returns whether this edge remains open after the last path is removed.
/// signaling if `err == none`, otherwise the upstream actor receives an virtual bool continuous() const = 0;
/// abort message.
bool remove_path(const strong_actor_ptr& hdl, error err = none);
upstream_path* find(const strong_actor_ptr& x) const; /// Sets whether this edge remains open after the last path is removed.
virtual void continuous(bool value) = 0;
inline const path_uptr_vec& paths() const { /// Returns the stored state for `x` if `x` is a known path and associated to
return paths_; /// `sid`, otherwise `nullptr`.
} virtual path_ptr find(const stream_id& sid, const actor_addr& x) = 0;
// -- required state --------------------------------------------------------- /// Returns the stored state for `x` if `x` is a known path and associated to
/// `sid`, otherwise `nullptr`.
inline local_actor* self() const { virtual path_ptr path_at(size_t index) = 0;
return self_;
}
// -- configuration parameters -----------------------------------------------
/// Returns the point at which an actor stops sending out demand immediately /// Returns the point at which an actor stops sending out demand immediately
/// (waiting for the available credit to first drop below the watermark). /// (waiting for the available credit to first drop below the watermark).
long high_watermark() const { virtual long high_watermark() const = 0;
return high_watermark_;
}
/// Sets the point at which an actor stops sending out demand immediately
/// (waiting for the available credit to first drop below the watermark).
void high_watermark(long x) {
high_watermark_ = x;
}
/// Returns the minimum amount of credit required to send a `demand` message. /// Returns the minimum amount of credit required to send a `demand` message.
long min_credit_assignment() const { virtual long min_credit_assignment() const = 0;
return min_credit_assignment_;
}
/// Sets the minimum amount of credit required to send a `demand` message.
void min_credit_assignment(long x) {
min_credit_assignment_ = x;
}
/// Returns the maximum credit assigned to a single upstream actors. /// Returns the maximum credit assigned to a single upstream actors.
long max_credit() const { virtual long max_credit() const = 0;
return max_credit_;
}
/// Sets the maximum credit assigned to a single upstream actors. /// Sets the point at which an actor stops sending out demand immediately
void max_credit(long x) { /// (waiting for the available credit to first drop below the watermark).
max_credit_ = x; virtual void high_watermark(long x) = 0;
}
protected: /// Sets the minimum amount of credit required to send a `demand` message.
/// Assigns new credit to upstream actors by filling `assignment_vec_`. virtual void min_credit_assignment(long x) = 0;
virtual void fill_assignment_vec(long downstream_credit) = 0;
/// Creates initial credit for `ptr`. /// Sets the maximum credit assigned to a single upstream actors.
virtual long initial_credit(upstream_path* ptr, long downstream_credit); virtual void max_credit(long x) = 0;
/// Pointer to the parent actor. /// Assigns new credit to all sources.
local_actor* self_; virtual void assign_credit(long downstream_capacity) = 0;
/// List of all known paths. /// Calculates initial credit for `x` after adding it to the gatherer.
path_uptr_vec paths_; virtual long initial_credit(long downstream_capacity, path_ptr x) = 0;
/// An assignment vector that's re-used whenever calling the policy. // -- convenience functions --------------------------------------------------
assignment_vec assignment_vec_;
/// Stores whether this stream remains open even if all paths have been /// Removes a path from the gatherer.
/// closed. bool remove_path(const stream_id& sid, const strong_actor_ptr& x,
bool continuous_; error reason, bool silent);
long high_watermark_; /// Convenience function for calling `find(x, actor_cast<actor_addr>(x))`.
long min_credit_assignment_; path_ptr find(const stream_id& sid, const strong_actor_ptr& x);
long max_credit_;
}; };
} // namespace caf } // namespace caf
#endif // CAF_UPSTREAM_POLICY_HPP #endif // CAF_STREAM_GATHERER_HPP
...@@ -17,56 +17,67 @@ ...@@ -17,56 +17,67 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_STREAM_STAGE_HPP #ifndef CAF_STREAM_GATHERER_IMPL_HPP
#define CAF_STREAM_STAGE_HPP #define CAF_STREAM_GATHERER_IMPL_HPP
#include "caf/fwd.hpp" #include <vector>
#include "caf/extend.hpp" #include <cstdint>
#include "caf/stream_handler.hpp" #include <utility>
#include "caf/mixin/has_upstreams.hpp" #include "caf/fwd.hpp"
#include "caf/mixin/has_downstreams.hpp" #include "caf/stream_gatherer.hpp"
#include "caf/stream_edge_impl.hpp"
#include "caf/response_promise.hpp"
namespace caf { namespace caf {
/// Models a stream stage with up- and downstreams. /// Type-erased policy for receiving data from sources.
class stream_stage : public extend<stream_handler, stream_stage>:: class stream_gatherer_impl : public stream_edge_impl<stream_gatherer> {
with<mixin::has_upstreams, mixin::has_downstreams> {
public: public:
// -- constructors, destructors, and assignment operators -------------------- using super = stream_edge_impl<stream_gatherer>;
using assignment_pair = std::pair<path_type*, long>;
stream_gatherer_impl(local_actor* selfptr);
~stream_gatherer_impl();
path_ptr add_path(const stream_id& sid, strong_actor_ptr x,
strong_actor_ptr original_stage, stream_priority prio,
long available_credit, bool redeployable,
response_promise result_cb) override;
stream_stage(upstream_policy* in_ptr, downstream_policy* out_ptr); bool remove_path(const stream_id& sid, const actor_addr& x, error reason,
bool silent) override;
// -- overrides -------------------------------------------------------------- void close(message result) override;
bool done() const override; void abort(error reason) override;
void abort(strong_actor_ptr&, const error&) override; long high_watermark() const override;
error downstream_ack(strong_actor_ptr&, int64_t, long) override; long min_credit_assignment() const override;
error upstream_batch(strong_actor_ptr&, int64_t, long, message&) override; long max_credit() const override;
void last_upstream_closed(); void high_watermark(long x) override;
inline upstream_policy& in() { void min_credit_assignment(long x) override;
return *in_ptr_;
}
inline downstream_policy& out() { void max_credit(long x) override;
return *out_ptr_;
}
protected: protected:
void downstream_demand(downstream_path*, long) override; void emit_credits();
virtual error process_batch(message& msg) = 0; long high_watermark_;
long min_credit_assignment_;
long max_credit_;
std::vector<assignment_pair> assignment_vec_;
private: /// Listeners for the final result.
upstream_policy* in_ptr_; std::vector<response_promise> listeners_;
downstream_policy* out_ptr_;
}; };
} // namespace caf } // namespace caf
#endif // CAF_STREAM_STAGE_HPP #endif // CAF_STREAM_GATHERER_IMPL_HPP
...@@ -39,6 +39,8 @@ public: ...@@ -39,6 +39,8 @@ public:
stream_id(); stream_id();
stream_id(none_t);
stream_id(actor_addr origin_actor, uint64_t origin_nr); stream_id(actor_addr origin_actor, uint64_t origin_nr);
stream_id(actor_control_block* origin_actor, uint64_t origin_nr); stream_id(actor_control_block* origin_actor, uint64_t origin_nr);
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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_MANAGER_HPP
#define CAF_STREAM_MANAGER_HPP
#include <vector>
#include <cstdint>
#include <cstddef>
#include "caf/fwd.hpp"
#include "caf/ref_counted.hpp"
#include "caf/mailbox_element.hpp"
namespace caf {
/// Manages a single stream with any number of down- and upstream actors.
/// @relates stream_msg
class stream_manager : public ref_counted {
public:
~stream_manager() override;
/// Handles `stream_msg::open` messages.
/// @returns Initial credit to the source.
/// @param hdl Handle to the sender.
/// @param original_stage Handle to the initial receiver of the handshake.
/// @param priority Affects credit assignment and maximum bandwidth.
/// @param redeployable Configures whether `hdl` could get redeployed, i.e.,
/// can resume after an abort.
/// @param result_cb Callback for the listener of the final stream result.
/// Ignored when returning `nullptr`, because the previous
/// stage is responsible for it until this manager
/// acknowledges the handshake.
/// @pre `hdl != nullptr`
virtual error open(const stream_id& sid, strong_actor_ptr hdl,
strong_actor_ptr original_stage, stream_priority priority,
bool redeployable, response_promise result_cb);
/// Handles `stream_msg::ack_open` messages.
/// @param hdl Handle to the sender.
/// @param initial_demand Credit received with `ack_open`.
/// @param redeployable Denotes whether the runtime can redeploy
/// the downstream actor on failure.
/// @pre `hdl != nullptr`
virtual error ack_open(const stream_id& sid, const actor_addr& rebind_from,
strong_actor_ptr rebind_to, long initial_demand,
bool redeployable);
/// Handles `stream_msg::batch` messages.
/// @param hdl Handle to the sender.
/// @param xs_size Size of the vector stored in `xs`.
/// @param xs A type-erased vector of size `xs_size`.
/// @param xs_id ID of this batch (must be ACKed).
/// @pre `hdl != nullptr`
/// @pre `xs_size > 0`
virtual error batch(const stream_id& sid, const actor_addr& hdl, long xs_size,
message& xs, int64_t xs_id);
/// Handles `stream_msg::ack_batch` messages.
/// @param hdl Handle to the sender.
/// @param new_demand New credit for sending data.
/// @param cumulative_batch_id Id of last handled batch.
/// @pre `hdl != nullptr`
virtual error ack_batch(const stream_id& sid, const actor_addr& hdl,
long new_demand, int64_t cumulative_batch_id);
/// Handles `stream_msg::close` messages.
/// @param hdl Handle to the sender.
/// @pre `hdl != nullptr`
virtual error close(const stream_id& sid, const actor_addr& hdl);
/// Handles `stream_msg::drop` messages.
/// @param hdl Handle to the sender.
/// @pre `hdl != nullptr`
virtual error drop(const stream_id& sid, const actor_addr& hdl);
/// Handles `stream_msg::drop` messages. The default implementation calls
/// `abort(reason)` and returns `sec::unhandled_stream_error`.
/// @param hdl Handle to the sender.
/// @param reason Reported error from the source.
/// @pre `hdl != nullptr`
/// @pre `err != none`
virtual error forced_close(const stream_id& sid, const actor_addr& hdl,
error reason);
/// Handles `stream_msg::drop` messages. The default implementation calls
/// `abort(reason)` and returns `sec::unhandled_stream_error`.
/// @param hdl Handle to the sender.
/// @param reason Reported error from the sink.
/// @pre `hdl != nullptr`
/// @pre `err != none`
virtual error forced_drop(const stream_id& sid, const actor_addr& hdl,
error reason);
/// Adds a new sink to the stream.
virtual bool add_sink(const stream_id& sid, strong_actor_ptr origin,
strong_actor_ptr sink_ptr,
mailbox_element::forwarding_stack stages,
message_id handshake_mid, message handshake_data,
stream_priority prio, bool redeployable);
/// Adds the source `hdl` to a stream. Convenience function for calling
/// `in().add_path(sid, hdl).second`.
virtual bool add_source(const stream_id& sid, strong_actor_ptr source_ptr,
strong_actor_ptr original_stage, stream_priority prio,
bool redeployable, response_promise result_cb);
/// Pushes new data to downstream actors by sending batches. The amount of
/// pushed data is limited by the available credit.
virtual void push();
/// Aborts a stream after any stream message handler returned a non-default
/// constructed error `reason` or the parent actor terminates with a
/// non-default error.
/// @param reason Previous error or non-default exit reason of the parent.
virtual void abort(error reason);
/// Closes the stream when the parent terminates with default exit reason or
/// the stream reached its end.
virtual void close();
// -- implementation hooks for all children classes --------------------------
/// Returns the stream edge for incoming data.
virtual stream_gatherer& in() = 0;
/// Returns the stream edge for outgoing data.
virtual stream_scatterer& out() = 0;
/// Returns whether the stream has reached the end and can be discarded
/// safely.
virtual bool done() const = 0;
protected:
// -- implementation hooks for sinks -----------------------------------------
/// Called when the gatherer closes to produce the final stream result for
/// all listeners. The default implementation returns an empty message.
/// @param reason `none` on orderly shutdowns, otherwise the abort reason.
virtual message make_final_result();
/// Called to handle incoming data. The default implementation logs an error
/// (sinks are expected to override this member function).
virtual error process_batch(message& msg);
/// Called when `in().closed()` changes to `true`. The default
/// implementation does nothing.
virtual void input_closed(error reason);
// -- implementation hooks for sources ---------------------------------------
/// Returns a type-erased `stream<T>` as handshake token for downstream
/// actors. Returns an empty message for sinks.
virtual message make_output_token(const stream_id&) const;
/// Called whenever new credit becomes available. The default implementation
/// logs an error (sources are expected to override this hook).
virtual void downstream_demand(outbound_path* ptr, long demand);
/// Called when `out().closed()` changes to `true`. The default
/// implementation does nothing.
virtual void output_closed(error reason);
/// Pointer to the parent actor.
local_actor* self_;
/// Keeps track of pending handshakes.
};
/// A reference counting pointer to a `stream_manager`.
/// @relates stream_manager
using stream_manager_ptr = intrusive_ptr<stream_manager>;
} // namespace caf
#endif // CAF_STREAM_MANAGER_HPP
...@@ -45,9 +45,7 @@ struct stream_msg : tag::boxing_type { ...@@ -45,9 +45,7 @@ struct stream_msg : tag::boxing_type {
/// Identifies content types that only flow downstream. /// Identifies content types that only flow downstream.
flows_downstream, flows_downstream,
/// Identifies content types that only flow upstream. /// Identifies content types that only flow upstream.
flows_upstream, flows_upstream
/// Identifies content types that propagate errors in both directions.
flows_both_ways
}; };
/// Initiates a stream handshake. /// Initiates a stream handshake.
...@@ -80,7 +78,9 @@ struct stream_msg : tag::boxing_type { ...@@ -80,7 +78,9 @@ struct stream_msg : tag::boxing_type {
/// originally receiving the `open` message. No effect when set to /// originally receiving the `open` message. No effect when set to
/// `nullptr`. This mechanism enables pipeline definitions consisting of /// `nullptr`. This mechanism enables pipeline definitions consisting of
/// proxy actors that are replaced with actual actors on demand. /// proxy actors that are replaced with actual actors on demand.
strong_actor_ptr rebind_from; actor_addr rebind_from;
/// Points to sender_, but with a strong reference.
strong_actor_ptr rebind_to;
/// Grants credit to the source. /// Grants credit to the source.
int32_t initial_demand; int32_t initial_demand;
/// Tells the upstream whether rebindings can occur on this path. /// Tells the upstream whether rebindings can occur on this path.
...@@ -114,7 +114,7 @@ struct stream_msg : tag::boxing_type { ...@@ -114,7 +114,7 @@ struct stream_msg : tag::boxing_type {
int64_t acknowledged_id; int64_t acknowledged_id;
}; };
/// Closes a stream after receiving an ACK for the last batch. /// Orderly shuts down a stream after receiving an ACK for the last batch.
struct close { struct close {
/// Allows visitors to dispatch on this tag. /// Allows visitors to dispatch on this tag.
static constexpr flow_label label = flows_downstream; static constexpr flow_label label = flows_downstream;
...@@ -122,59 +122,60 @@ struct stream_msg : tag::boxing_type { ...@@ -122,59 +122,60 @@ struct stream_msg : tag::boxing_type {
using outer_type = stream_msg; using outer_type = stream_msg;
}; };
/// Propagates fatal errors. /// Informs a source that a sink orderly drops out of a stream.
struct abort { struct drop {
/// Allows visitors to dispatch on this tag. /// Allows visitors to dispatch on this tag.
static constexpr flow_label label = flows_both_ways; static constexpr flow_label label = flows_upstream;
/// Allows the testing DSL to unbox this type automagically. /// Allows the testing DSL to unbox this type automagically.
using outer_type = stream_msg; using outer_type = stream_msg;
/// Reason for shutting down the stream.
error reason;
}; };
/// Send by the runtime if a downstream path failed. The receiving actor /// Propagates a fatal error from sources to sinks.
/// awaits a `resume` message afterwards if the downstream path was struct forced_close {
/// redeployable. Otherwise, this results in a fatal error.
struct downstream_failed {
/// Allows visitors to dispatch on this tag. /// Allows visitors to dispatch on this tag.
static constexpr flow_label label = flows_upstream; static constexpr flow_label label = flows_downstream;
/// Allows the testing DSL to unbox this type automagically. /// Allows the testing DSL to unbox this type automagically.
using outer_type = stream_msg; using outer_type = stream_msg;
/// Exit reason of the failing downstream path. /// Reason for shutting down the stream.
error reason; error reason;
}; };
/// Send by the runtime if an upstream path failed. The receiving actor /// Propagates a fatal error from sinks to sources.
/// awaits a `resume` message afterwards if the upstream path was struct forced_drop {
/// redeployable. Otherwise, this results in a fatal error.
struct upstream_failed {
/// Allows visitors to dispatch on this tag. /// Allows visitors to dispatch on this tag.
static constexpr flow_label label = flows_downstream; static constexpr flow_label label = flows_upstream;
/// Allows the testing DSL to unbox this type automagically. /// Allows the testing DSL to unbox this type automagically.
using outer_type = stream_msg; using outer_type = stream_msg;
/// Exit reason of the failing upstream path. /// Reason for shutting down the stream.
error reason; error reason;
}; };
/// Lists all possible options for the payload. /// Lists all possible options for the payload.
using content_alternatives = detail::type_list<open, ack_open, batch, using content_alternatives =
ack_batch, close, abort, detail::type_list<open, ack_open, batch, ack_batch, close, drop,
downstream_failed, forced_close, forced_drop>;
upstream_failed>;
/// Stores one of `content_alternatives`. /// Stores one of `content_alternatives`.
using content_type = variant<open, ack_open, batch, ack_batch, close, using content_type = variant<open, ack_open, batch, ack_batch, close, drop,
abort, downstream_failed, upstream_failed>; forced_close, forced_drop>;
/// ID of the affected stream. /// ID of the affected stream.
stream_id sid; stream_id sid;
/// 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. /// Palyoad of the message.
content_type content; content_type content;
template <class T> template <class T>
stream_msg(stream_id id, T&& x) stream_msg(const stream_id& id, actor_addr addr, T&& x)
: sid(std::move(id)), : sid(std::move(id)),
sender(std::move(addr)),
content(std::forward<T>(x)) { content(std::forward<T>(x)) {
// nop // nop
} }
...@@ -206,8 +207,8 @@ typename std::enable_if< ...@@ -206,8 +207,8 @@ typename std::enable_if<
>::value, >::value,
stream_msg stream_msg
>::type >::type
make(const stream_id& sid, Ts&&... xs) { make(const stream_id& sid, actor_addr addr, Ts&&... xs) {
return {sid, T{std::forward<Ts>(xs)...}}; return {sid, std::move(addr), T{std::forward<Ts>(xs)...}};
} }
template <class Inspector> template <class Inspector>
...@@ -234,25 +235,27 @@ typename Inspector::result_type inspect(Inspector& f, ...@@ -234,25 +235,27 @@ typename Inspector::result_type inspect(Inspector& f,
} }
template <class Inspector> template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, stream_msg::close&) { typename Inspector::result_type inspect(Inspector& f,
stream_msg::close&) {
return f(meta::type_name("close")); return f(meta::type_name("close"));
} }
template <class Inspector> template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, stream_msg::abort& x) { typename Inspector::result_type inspect(Inspector& f,
return f(meta::type_name("abort"), x.reason); stream_msg::drop&) {
return f(meta::type_name("drop"));
} }
template <class Inspector> template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, typename Inspector::result_type inspect(Inspector& f,
stream_msg::downstream_failed& x) { stream_msg::forced_close& x) {
return f(meta::type_name("downstream_failed"), x.reason); return f(meta::type_name("forced_close"), x.reason);
} }
template <class Inspector> template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, typename Inspector::result_type inspect(Inspector& f,
stream_msg::upstream_failed& x) { stream_msg::forced_drop& x) {
return f(meta::type_name("upstream_failed"), x.reason); return f(meta::type_name("forced_drop"), x.reason);
} }
template <class Inspector> template <class Inspector>
......
...@@ -26,40 +26,60 @@ ...@@ -26,40 +26,60 @@
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/stream_msg.hpp" #include "caf/stream_msg.hpp"
#include "caf/intrusive_ptr.hpp" #include "caf/intrusive_ptr.hpp"
#include "caf/stream_handler.hpp" #include "caf/stream_manager.hpp"
#include "caf/scheduled_actor.hpp"
namespace caf { namespace caf {
class stream_msg_visitor { class stream_msg_visitor {
public: public:
using map_type = std::unordered_map<stream_id, intrusive_ptr<stream_handler>>; using map_type = std::unordered_map<stream_id, intrusive_ptr<stream_manager>>;
using iterator = typename map_type::iterator; using iterator = typename map_type::iterator;
using result_type = std::pair<error, iterator>; using result_type = bool;
stream_msg_visitor(scheduled_actor* self, stream_id& sid, stream_msg_visitor(scheduled_actor* self, const stream_msg& msg,
iterator i, iterator last, behavior* bhvr); behavior* bhvr);
result_type operator()(stream_msg::open& x); result_type operator()(stream_msg::open& x);
result_type operator()(stream_msg::ack_open&); result_type operator()(stream_msg::ack_open& x);
result_type operator()(stream_msg::batch&); result_type operator()(stream_msg::batch& x);
result_type operator()(stream_msg::ack_batch&); result_type operator()(stream_msg::ack_batch& x);
result_type operator()(stream_msg::close&); result_type operator()(stream_msg::close& x);
result_type operator()(stream_msg::abort&); result_type operator()(stream_msg::drop& x);
result_type operator()(stream_msg::downstream_failed&); result_type operator()(stream_msg::forced_close& x);
result_type operator()(stream_msg::upstream_failed&); result_type operator()(stream_msg::forced_drop& x);
private: private:
// Invokes `f` on the stream manager. On error calls `abort` on the manager
// and removes it from the streams map.
template <class F>
bool invoke(F f) {
auto e = self_->streams().end();
auto i = self_->streams().find(sid_);
if (i == e)
return false;
auto err = f(i->second);
if (err != none) {
i->second->abort(std::move(err));
self_->streams().erase(i);
} else if (i->second->done()) {
CAF_LOG_DEBUG("manager reported done, remove from streams");
i->second->close();
self_->streams().erase(i);
}
return true;
}
scheduled_actor* self_; scheduled_actor* self_;
stream_id& sid_; const stream_id& sid_;
iterator i_; const actor_addr& sender_;
iterator e_;
behavior* bhvr_; behavior* bhvr_;
}; };
......
...@@ -21,8 +21,10 @@ ...@@ -21,8 +21,10 @@
#define CAF_STREAM_RESULT_HPP #define CAF_STREAM_RESULT_HPP
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/none.hpp"
#include "caf/stream_id.hpp" #include "caf/stream_id.hpp"
#include "caf/stream_handler.hpp" #include "caf/stream_manager.hpp"
#include "caf/meta/type_name.hpp" #include "caf/meta/type_name.hpp"
namespace caf { namespace caf {
...@@ -37,13 +39,17 @@ public: ...@@ -37,13 +39,17 @@ public:
stream_result& operator=(stream_result&&) = default; stream_result& operator=(stream_result&&) = default;
stream_result& operator=(const stream_result&) = default; stream_result& operator=(const stream_result&) = default;
stream_result(none_t) : stream_result() {
// nop
}
stream_result(stream_id sid) : id_(std::move(sid)) { stream_result(stream_id sid) : id_(std::move(sid)) {
// nop // nop
} }
/// Convenience constructor for returning the result of `self->new_stream_result` /// Convenience constructor for returning the result of `self->new_stream_result`
/// and similar functions. /// and similar functions.
stream_result(stream_id sid, stream_handler_ptr sptr) stream_result(stream_id sid, stream_manager_ptr sptr)
: id_(std::move(sid)), : id_(std::move(sid)),
ptr_(std::move(sptr)) { ptr_(std::move(sptr)) {
// nop // nop
...@@ -51,7 +57,7 @@ public: ...@@ -51,7 +57,7 @@ public:
/// Convenience constructor for returning the result of `self->new_stream_result` /// Convenience constructor for returning the result of `self->new_stream_result`
/// and similar functions. /// and similar functions.
stream_result(stream_result other, stream_handler_ptr sptr) stream_result(stream_result other, stream_manager_ptr sptr)
: id_(std::move(other.id_)), : id_(std::move(other.id_)),
ptr_(std::move(sptr)) { ptr_(std::move(sptr)) {
// nop // nop
...@@ -63,7 +69,7 @@ public: ...@@ -63,7 +69,7 @@ public:
} }
/// Returns the handler assigned to this stream_result on this actor. /// Returns the handler assigned to this stream_result on this actor.
inline const stream_handler_ptr& ptr() const { inline const stream_manager_ptr& ptr() const {
return ptr_; return ptr_;
} }
...@@ -74,7 +80,7 @@ public: ...@@ -74,7 +80,7 @@ public:
private: private:
stream_id id_; stream_id id_;
stream_handler_ptr ptr_; stream_manager_ptr ptr_;
}; };
} // namespace caf } // namespace caf
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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_SCATTERER_HPP
#define CAF_STREAM_SCATTERER_HPP
#include <cstddef>
#include "caf/fwd.hpp"
#include "caf/duration.hpp"
#include "caf/optional.hpp"
#include "caf/mailbox_element.hpp"
namespace caf {
/// Type-erased policy for dispatching data to sinks.
class stream_scatterer {
public:
// -- member types -----------------------------------------------------------
using path_type = outbound_path;
using path_ptr = path_type*;
// -- constructors, destructors, and assignment operators --------------------
stream_scatterer() = default;
virtual ~stream_scatterer();
// -- pure virtual memeber functions -----------------------------------------
/// Adds a path to the edge.
/// @returns The added path on success, `nullptr` otherwise.
virtual path_ptr add_path(const stream_id& sid, strong_actor_ptr origin,
strong_actor_ptr sink_ptr,
mailbox_element::forwarding_stack stages,
message_id handshake_mid, message handshake_data,
stream_priority prio, bool redeployable) = 0;
/// Adds a path to a sink and initiates the handshake.
virtual path_ptr confirm_path(const stream_id& sid, const actor_addr& from,
strong_actor_ptr to, long initial_demand,
bool redeployable) = 0;
/// Removes a path from the scatterer.
virtual bool remove_path(const stream_id& sid, const actor_addr& x,
error reason, bool silent) = 0;
/// Returns `true` if there is no data pending and no unacknowledged batch on
/// any path.
virtual bool paths_clean() const = 0;
/// Removes all paths gracefully.
virtual void close() = 0;
/// Removes all paths with an error message.
virtual void abort(error reason) = 0;
/// Returns the number of paths managed on this edge.
virtual long num_paths() const = 0;
/// Returns `true` if no downstream exists, `false` otherwise.
virtual bool closed() const = 0;
/// Returns whether this edge remains open after the last path is removed.
virtual bool continuous() const = 0;
/// Sets whether this edge remains open after the last path is removed.
virtual void continuous(bool value) = 0;
/// Sends batches to sinks.
virtual void emit_batches() = 0;
/// Returns the stored state for `x` if `x` is a known path and associated to
/// `sid`, otherwise `nullptr`.
virtual path_ptr find(const stream_id& sid, const actor_addr& x) = 0;
/// Returns the stored state for `x` if `x` is a known path and associated to
/// `sid`, otherwise `nullptr`.
virtual path_ptr path_at(size_t index) = 0;
/// Returns the currently available credit, depending on the policy in use.
/// For example, a broadcast policy would return the minimum of all available
/// downstream credits.
virtual long credit() const = 0;
/// Returns the size of the output buffer.
virtual long buffered() const = 0;
/// Minimum amount of messages required to emit a batch. A value of 0
/// disables batch delays.
virtual long min_batch_size() const = 0;
/// Maximum amount of messages to put into a single batch. Causes the actor
/// to split a buffer into more batches if necessary.
virtual long max_batch_size() const = 0;
/// Minimum amount of messages we wish to store at the actor in order to emit
/// new batches immediately when receiving new downstream demand.
virtual long min_buffer_size() const = 0;
/// Forces to actor to emit a batch even if the minimum batch size was not
/// reached.
virtual duration max_batch_delay() const = 0;
/// Minimum amount of messages required to emit a batch. A value of 0
/// disables batch delays.
virtual void min_batch_size(long x) = 0;
/// Maximum amount of messages to put into a single batch. Causes the actor
/// to split a buffer into more batches if necessary.
virtual void max_batch_size(long x) = 0;
/// Minimum amount of messages we wish to store at the actor in order to emit
/// new batches immediately when receiving new downstream demand.
virtual void min_buffer_size(long x) = 0;
/// Forces to actor to emit a batch even if the minimum batch size was not
/// reached.
virtual void max_batch_delay(duration x) = 0;
// -- convenience functions --------------------------------------------------
/// Removes a path from the scatterer.
bool remove_path(const stream_id& sid, const strong_actor_ptr& x,
error reason, bool silent);
/// Convenience function for calling `find(x, actor_cast<actor_addr>(x))`.
path_ptr find(const stream_id& sid, const strong_actor_ptr& x);
};
} // namespace caf
#endif // CAF_STREAM_SCATTERER_HPP
...@@ -17,68 +17,88 @@ ...@@ -17,68 +17,88 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_MIXIN_HAS_DOWNSTREAMS_HPP #ifndef CAF_STREAM_SCATTERER_IMPL_HPP
#define CAF_MIXIN_HAS_DOWNSTREAMS_HPP #define CAF_STREAM_SCATTERER_IMPL_HPP
#include <cstddef> #include <cstddef>
#include "caf/sec.hpp" #include "caf/fwd.hpp"
#include "caf/logger.hpp" #include "caf/duration.hpp"
#include "caf/actor_control_block.hpp" #include "caf/stream_edge_impl.hpp"
#include "caf/stream_scatterer.hpp"
namespace caf { namespace caf {
namespace mixin {
/// Mixin for streams with any number of downstreams. /// Type-erased policy for dispatching data to sinks.
template <class Base, class Subtype> class stream_scatterer_impl : public stream_edge_impl<stream_scatterer> {
class has_downstreams : public Base {
public: public:
error add_downstream(strong_actor_ptr& ptr) override { // -- member types -----------------------------------------------------------
CAF_LOG_TRACE(CAF_ARG(ptr));
CAF_ASSERT(ptr != nullptr);
if (out().add_path(ptr))
return none;
return sec::downstream_already_exists;
}
error confirm_downstream(const strong_actor_ptr& rebind_from,
strong_actor_ptr& ptr, long initial_demand,
bool redeployable) override {
CAF_LOG_TRACE(CAF_ARG(ptr) << CAF_ARG(initial_demand)
<< CAF_ARG(redeployable));
CAF_ASSERT(ptr != nullptr);
if (out().confirm_path(rebind_from, ptr, redeployable)) {
auto path = out().find(ptr);
if (!path) {
CAF_LOG_ERROR("Unable to find path after confirming it");
return sec::invalid_downstream;
}
downstream_demand(path, initial_demand);
return none;
}
return sec::invalid_downstream;
}
error push() override {
CAF_LOG_TRACE("");
out().emit_batches();
return none;
}
protected: using super = stream_edge_impl<stream_scatterer>;
virtual void downstream_demand(downstream_path* ptr, long demand) = 0;
// -- constructors, destructors, and assignment operators --------------------
stream_scatterer_impl(local_actor* selfptr);
~stream_scatterer_impl() override;
// -- convenience functions for children classes -----------------------------
/// Removes all paths gracefully.
void close() override;
/// Removes all paths with an error message.
void abort(error reason) override;
/// Returns the total number (sum) of all credit on all paths.
long total_credit() const;
/// Returns the minimum number of credit on all paths.
long min_credit() const;
/// Returns the maximum number of credit on all paths.
long max_credit() const;
/// Folds `paths()` by extracting the `open_credit` from each path.
long fold_credit(long x0, long (*f)(long, long)) const;
// -- overridden functions ---------------------------------------------------
private: path_ptr add_path(const stream_id& sid, strong_actor_ptr origin,
Subtype* dptr() { strong_actor_ptr sink_ptr,
return static_cast<Subtype*>(this); mailbox_element::forwarding_stack stages,
} message_id handshake_mid, message handshake_data,
stream_priority prio, bool redeployable) override;
downstream_policy& out() { path_ptr confirm_path(const stream_id& sid, const actor_addr& from,
return *this->dp(); strong_actor_ptr to, long initial_demand,
} bool redeployable) override;
bool paths_clean() const override;
long min_batch_size() const override;
long max_batch_size() const override;
long min_buffer_size() const override;
duration max_batch_delay() const override;
void min_batch_size(long x) override;
void max_batch_size(long x) override;
void min_buffer_size(long x) override;
void max_batch_delay(duration x) override;
protected:
long min_batch_size_;
long max_batch_size_;
long min_buffer_size_;
duration max_batch_delay_;
}; };
} // namespace mixin
} // namespace caf } // namespace caf
#endif // CAF_MIXIN_HAS_DOWNSTREAMS_HPP #endif // CAF_STREAM_SCATTERER_IMPL_HPP
...@@ -24,27 +24,24 @@ ...@@ -24,27 +24,24 @@
#include "caf/extend.hpp" #include "caf/extend.hpp"
#include "caf/message_id.hpp" #include "caf/message_id.hpp"
#include "caf/stream_handler.hpp" #include "caf/stream_manager.hpp"
#include "caf/upstream_policy.hpp" #include "caf/upstream_policy.hpp"
#include "caf/actor_control_block.hpp" #include "caf/actor_control_block.hpp"
#include "caf/mixin/has_upstreams.hpp"
namespace caf { namespace caf {
/// Represents a terminal stream stage. /// Represents a terminal stream stage.
class stream_sink : public extend<stream_handler, stream_sink>:: class stream_sink : public stream_manager {
with<mixin::has_upstreams> {
public: public:
stream_sink(upstream_policy* in_ptr, strong_actor_ptr&& orig_sender, stream_sink(upstream_policy* in_ptr, strong_actor_ptr&& orig_sender,
std::vector<strong_actor_ptr>&& trailing_stages, message_id mid); std::vector<strong_actor_ptr>&& trailing_stages, message_id mid);
bool done() const override; bool done() const override;
error upstream_batch(strong_actor_ptr& src, int64_t xs_id, error batch(const actor_addr& hdl, long xs_size, message& xs,
long xs_size, message& xs) override; int64_t xs_id) override;
void abort(strong_actor_ptr& cause, const error& reason) override; void abort(error reason) override;
void last_upstream_closed(); void last_upstream_closed();
......
...@@ -20,14 +20,20 @@ ...@@ -20,14 +20,20 @@
#ifndef CAF_STREAM_SINK_IMPL_HPP #ifndef CAF_STREAM_SINK_IMPL_HPP
#define CAF_STREAM_SINK_IMPL_HPP #define CAF_STREAM_SINK_IMPL_HPP
#include "caf/stream_sink.hpp" #include "caf/sec.hpp"
#include "caf/logger.hpp"
#include "caf/message_id.hpp"
#include "caf/stream_manager.hpp"
#include "caf/stream_sink_trait.hpp" #include "caf/stream_sink_trait.hpp"
#include "caf/terminal_stream_scatterer.hpp"
namespace caf { namespace caf {
template <class Fun, class Finalize, class UpstreamPolicy> template <class Fun, class Finalize, class UpstreamPolicy>
class stream_sink_impl : public stream_sink { class stream_sink_impl : public stream_manager {
public: public:
using super = stream_manager;
using trait = stream_sink_trait_t<Fun, Finalize>; using trait = stream_sink_trait_t<Fun, Finalize>;
using state_type = typename trait::state; using state_type = typename trait::state;
...@@ -36,25 +42,32 @@ public: ...@@ -36,25 +42,32 @@ public:
using output_type = typename trait::output; using output_type = typename trait::output;
stream_sink_impl(local_actor* self, stream_sink_impl(local_actor* self, Fun fun, Finalize fin)
strong_actor_ptr&& orig_sender, : fun_(std::move(fun)),
std::vector<strong_actor_ptr>&& svec, message_id mid,
Fun fun, Finalize fin)
: stream_sink(&in_, std::move(orig_sender), std::move(svec), mid),
fun_(std::move(fun)),
fin_(std::move(fin)), fin_(std::move(fin)),
in_(self) { in_(self) {
// nop // nop
} }
expected<long> add_upstream(strong_actor_ptr& ptr, const stream_id& sid, state_type& state() {
stream_priority prio) override { return state_;
if (ptr)
return in().add_path(ptr, sid, prio, min_buffer_size());
return sec::invalid_argument;
} }
error consume(message& msg) override { UpstreamPolicy& in() override {
return in_;
}
terminal_stream_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 (msg.match_elements<vec_type>()) {
auto& xs = msg.get_as<vec_type>(0); auto& xs = msg.get_as<vec_type>(0);
...@@ -62,26 +75,20 @@ public: ...@@ -62,26 +75,20 @@ public:
fun_(state_, x); fun_(state_, x);
return none; return none;
} }
CAF_LOG_ERROR("received unexpected batch type");
return sec::unexpected_message; return sec::unexpected_message;
} }
message finalize() override { message make_final_result() override {
return trait::make_result(state_, fin_); return trait::make_result(state_, fin_);
} }
optional<upstream_policy&> up() override {
return in_;
}
state_type& state() {
return state_;
}
private: private:
state_type state_; state_type state_;
Fun fun_; Fun fun_;
Finalize fin_; Finalize fin_;
UpstreamPolicy in_; UpstreamPolicy in_;
terminal_stream_scatterer out_;
}; };
} // namespace caf } // namespace caf
......
...@@ -20,14 +20,17 @@ ...@@ -20,14 +20,17 @@
#ifndef CAF_STREAM_SOURCE_IMPL_HPP #ifndef CAF_STREAM_SOURCE_IMPL_HPP
#define CAF_STREAM_SOURCE_IMPL_HPP #define CAF_STREAM_SOURCE_IMPL_HPP
#include "caf/logger.hpp"
#include "caf/downstream.hpp" #include "caf/downstream.hpp"
#include "caf/stream_source.hpp" #include "caf/outbound_path.hpp"
#include "caf/stream_manager.hpp"
#include "caf/stream_source_trait.hpp" #include "caf/stream_source_trait.hpp"
#include "caf/invalid_stream_gatherer.hpp"
namespace caf { namespace caf {
template <class Fun, class Predicate, class DownstreamPolicy> template <class Fun, class Predicate, class DownstreamPolicy>
class stream_source_impl : public stream_source { class stream_source_impl : public stream_manager {
public: public:
using trait = stream_source_trait_t<Fun>; using trait = stream_source_trait_t<Fun>;
...@@ -35,30 +38,28 @@ public: ...@@ -35,30 +38,28 @@ public:
using output_type = typename trait::output; using output_type = typename trait::output;
stream_source_impl(local_actor* self, const stream_id& sid, stream_source_impl(local_actor* self, Fun fun, Predicate pred)
Fun fun, Predicate pred) : fun_(std::move(fun)),
: stream_source(&out_),
fun_(std::move(fun)),
pred_(std::move(pred)), pred_(std::move(pred)),
out_(self, sid) { out_(self) {
// nop // nop
} }
void generate(size_t num) override { void generate(size_t num) {
CAF_LOG_TRACE(CAF_ARG(num)); CAF_LOG_TRACE(CAF_ARG(num));
downstream<typename DownstreamPolicy::value_type> ds{out_.buf()}; downstream<typename DownstreamPolicy::value_type> ds{out_.buf()};
fun_(state_, ds, num); fun_(state_, ds, num);
} }
long buf_size() const override { bool done() const override {
return out_.buf_size(); return at_end() && out_.paths_clean();
} }
bool at_end() const override { invalid_stream_gatherer& in() override {
return pred_(state_); return in_;
} }
optional<downstream_policy&> dp() override { DownstreamPolicy& out() override {
return out_; return out_;
} }
...@@ -66,11 +67,44 @@ public: ...@@ -66,11 +67,44 @@ public:
return state_; return state_;
} }
protected:
bool at_end() const {
return pred_(state_);
}
void downstream_demand(outbound_path* path, long) override {
CAF_LOG_TRACE(CAF_ARG(path));
if (!at_end()) {
// produce new elements
auto capacity = out_.credit();
// TODO: fix issue where a source does not emit the last pieces if
// min_batch_size cannot be reached
while (capacity >= out_.min_batch_size()) {
auto current_size = out_.buffered();
auto size_hint = std::min(capacity, out_.max_batch_size());
if (size_hint > current_size)
generate(static_cast<size_t>(size_hint - current_size));
push();
// Set available credit for next iteration.
auto new_capacity = at_end() ? 0l : out_.credit();
if (new_capacity != capacity)
capacity = new_capacity;
}
} 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_; state_type state_;
Fun fun_; Fun fun_;
Predicate pred_; Predicate pred_;
DownstreamPolicy out_; DownstreamPolicy out_;
invalid_stream_gatherer in_;
}; };
} // namespace caf } // namespace caf
......
...@@ -20,18 +20,18 @@ ...@@ -20,18 +20,18 @@
#ifndef CAF_STREAM_STAGE_IMPL_HPP #ifndef CAF_STREAM_STAGE_IMPL_HPP
#define CAF_STREAM_STAGE_IMPL_HPP #define CAF_STREAM_STAGE_IMPL_HPP
#include "caf/sec.hpp"
#include "caf/logger.hpp"
#include "caf/downstream.hpp" #include "caf/downstream.hpp"
#include "caf/stream_stage.hpp" #include "caf/outbound_path.hpp"
#include "caf/stream_manager.hpp"
#include "caf/stream_stage_trait.hpp" #include "caf/stream_stage_trait.hpp"
#include "caf/policy/greedy.hpp"
#include "caf/policy/broadcast.hpp"
namespace caf { namespace caf {
template <class Fun, class Cleanup, template <class Fun, class Cleanup,
class UpstreamPolicy, class DownstreamPolicy> class UpstreamPolicy, class DownstreamPolicy>
class stream_stage_impl : public stream_stage { class stream_stage_impl : public stream_manager {
public: public:
using trait = stream_stage_trait_t<Fun>; using trait = stream_stage_trait_t<Fun>;
...@@ -41,26 +41,43 @@ public: ...@@ -41,26 +41,43 @@ public:
using output_type = typename trait::output; using output_type = typename trait::output;
stream_stage_impl(local_actor* self, stream_stage_impl(local_actor* self, const stream_id&,
const stream_id& sid,
Fun fun, Cleanup cleanup) Fun fun, Cleanup cleanup)
: stream_stage(&in_, &out_), : fun_(std::move(fun)),
fun_(std::move(fun)),
cleanup_(std::move(cleanup)), cleanup_(std::move(cleanup)),
in_(self), in_(self),
out_(self, sid) { out_(self) {
// nop // nop
} }
expected<long> add_upstream(strong_actor_ptr& ptr, const stream_id& sid, state_type& state() {
stream_priority prio) override { return state_;
CAF_LOG_TRACE(CAF_ARG(ptr) << CAF_ARG(sid) << CAF_ARG(prio)); }
if (ptr)
return in().add_path(ptr, sid, prio, out_.credit()); UpstreamPolicy& in() override {
return sec::invalid_argument; return in_;
}
DownstreamPolicy& out() override {
return out_;
}
bool done() const override {
return in_.closed() && out_.closed();
}
protected:
void input_closed(error reason) override {
if (reason == none) {
if (out_.buffered() == 0)
out_.close();
} else {
out_.abort(std::move(reason));
}
} }
error process_batch(message& msg) override { error process_batch(message& msg) override {
CAF_LOG_TRACE(CAF_ARG(msg));
using vec_type = std::vector<output_type>; using vec_type = std::vector<output_type>;
if (msg.match_elements<vec_type>()) { if (msg.match_elements<vec_type>()) {
auto& xs = msg.get_as<vec_type>(0); auto& xs = msg.get_as<vec_type>(0);
...@@ -69,6 +86,7 @@ public: ...@@ -69,6 +86,7 @@ public:
fun_(state_, ds, x); fun_(state_, ds, x);
return none; return none;
} }
CAF_LOG_ERROR("received unexpected batch type");
return sec::unexpected_message; return sec::unexpected_message;
} }
...@@ -76,24 +94,20 @@ public: ...@@ -76,24 +94,20 @@ public:
return make_message(stream<output_type>{x}); return make_message(stream<output_type>{x});
} }
optional<downstream_policy&> dp() override { void downstream_demand(outbound_path* path, long) override {
return out_; CAF_LOG_TRACE(CAF_ARG(path));
} auto hdl = path->hdl;
if(out_.buffered() > 0)
optional<upstream_policy&> up() override { push();
return in_; else if (in_.closed()) {
} // don't pass path->hdl: path can become invalid
auto sid = path->sid;
state_type& state() { out_.remove_path(sid, hdl, none, false);
return state_; }
} auto current_size = out_.buffered();
auto desired_size = out_.credit();
UpstreamPolicy& in() { if (current_size < desired_size)
return in_; in_.assign_credit(desired_size - current_size);
}
DownstreamPolicy& out() {
return out_;
} }
private: private:
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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_TERMINAL_STREAM_SCATTERER_HPP
#define CAF_TERMINAL_STREAM_SCATTERER_HPP
#include "caf/stream_scatterer.hpp"
namespace caf {
/// Special-purpose scatterer for sinks that terminate a stream. A terminal
/// stream scatterer generates credit without downstream actors.
class terminal_stream_scatterer : public stream_scatterer {
public:
terminal_stream_scatterer() = default;
~terminal_stream_scatterer() override;
path_ptr add_path(const stream_id& sid, strong_actor_ptr origin,
strong_actor_ptr sink_ptr,
mailbox_element::forwarding_stack stages,
message_id handshake_mid, message handshake_data,
stream_priority prio, bool redeployable) override;
path_ptr confirm_path(const stream_id& sid, const actor_addr& from,
strong_actor_ptr to, long initial_demand,
bool redeployable) override;
bool remove_path(const stream_id& sid, const actor_addr& x,
error reason, bool silent) override;
bool paths_clean() const override;
void close() override;
void abort(error reason) override;
long num_paths() const override;
bool closed() const override;
bool continuous() const override;
void continuous(bool value) override;
path_type* path_at(size_t index) override;
void emit_batches() override;
path_type* find(const stream_id& sid, const actor_addr& x) override;
long credit() const override;
long buffered() const override;
long min_batch_size() const override;
long max_batch_size() const override;
long min_buffer_size() const override;
duration max_batch_delay() const override;
void min_batch_size(long x) override;
void max_batch_size(long x) override;
void min_buffer_size(long x) override;
void max_batch_delay(duration x) override;
};
} // namespace caf
#endif // CAF_TERMINAL_STREAM_SCATTERER_HPP
...@@ -17,8 +17,8 @@ ...@@ -17,8 +17,8 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_FILTERING_DOWNSTREAM_HPP #ifndef CAF_TOPIC_SCATTERER_HPP
#define CAF_FILTERING_DOWNSTREAM_HPP #define CAF_TOPIC_SCATTERER_HPP
#include <map> #include <map>
#include <tuple> #include <tuple>
...@@ -26,111 +26,79 @@ ...@@ -26,111 +26,79 @@
#include <vector> #include <vector>
#include <functional> #include <functional>
#include "caf/downstream_policy.hpp" #include "caf/buffered_scatterer.hpp"
#include "caf/mixin/buffered_policy.hpp"
#include "caf/policy/broadcast.hpp"
#include "caf/meta/type_name.hpp" #include "caf/meta/type_name.hpp"
namespace caf { namespace caf {
/// A filtering downstream allows stages to fork into multiple lanes, where /// A topic scatterer allows stream nodes to fork into multiple lanes, where
/// each lane carries only a subset of the data. For example, the lane /// each lane carries only a subset of the data. For example, the lane
/// mechanism allows you filter key/value pairs before forwarding them to a set /// mechanism allows you filter key/value pairs before forwarding them to a set
/// of workers in order to handle only a subset of the overall data on each /// of workers.
/// lane. template <class T, class Filter, class KeyCompare, long KeyIndex>
template <class T, class Key, class KeyCompare = std::equal_to<Key>, class topic_scatterer : public buffered_scatterer<T> {
long KeyIndex = 0,
class Base = policy::broadcast<T>>
class filtering_downstream : public Base {
public: public:
/// Base type. /// Base type.
using super = Base; using super = buffered_scatterer<T>;
struct lane { struct lane {
typename super::buffer_type buf; typename super::buffer_type buf;
typename super::path_ptr_list paths; typename super::path_ptr_vec paths;
template <class Inspector> template <class Inspector>
friend typename Inspector::result_type inspect(Inspector& f, lane& x) { friend typename Inspector::result_type inspect(Inspector& f, lane& x) {
return f(meta::type_name("lane"), x.buf, x.paths); return f(meta::type_name("lane"), x.buf, x.paths);
} }
}; };
/// Identifies a lane inside the downstream. Filters are kept in sorted order /// Identifies a lane inside the downstream.
/// and require `Key` to provide `operator<`. using filter_type = Filter;
using filter = std::vector<Key>;
using lanes_map = std::map<filter, lane>; using lanes_map = std::map<filter_type, lane>;
filtering_downstream(local_actor* selfptr, const stream_id& sid) topic_scatterer(local_actor* selfptr) : super(selfptr) {
: super(selfptr, sid) {
// nop // nop
} }
void emit_broadcast() override { using super::remove_path;
fan_out();
for (auto& kvp : 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) {
x->open_credit -= csize;
this->emit_batch(*x, static_cast<size_t>(csize), wrapped_chunk);
}
}
}
void emit_anycast() override { bool remove_path(const stream_id& sid, const actor_addr& x,
fan_out(); error reason, bool silent) override {
for (auto& kvp : lanes_) { auto i = this->iter_find(this->paths_, sid, x);
auto& l = kvp.second; if (i != this->paths_.end()) {
super::sort_by_credit(l.paths); erase_from_lanes(x);
for (auto& x : l.paths) { return super::remove_path(i, std::move(reason), silent);
auto chunk = super::get_chunk(l.buf, x->open_credit);
auto csize = static_cast<long>(chunk.size());
if (csize == 0)
break;
x->open_credit -= csize;
this->emit_batch(*x, static_cast<size_t>(csize),
std::move(make_message(std::move(chunk))));
}
} }
return false;
} }
bool remove_path(strong_actor_ptr& ptr) override { void add_lane(filter_type f) {
erase_from_lanes(ptr);
return Base::remove_path(ptr);
}
void add_lane(filter f) {
std::sort(f); std::sort(f);
lanes_.emplace(std::move(f), typename super::buffer_type{}); lanes_.emplace(std::move(f), typename super::buffer_type{});
} }
/// Sets the filter for `x` to `f` and inserts `x` into the appropriate lane. /// Sets the filter for `x` to `f` and inserts `x` into the appropriate lane.
/// @pre `x` is not registered on *any* lane /// @pre `x` is not registered on *any* lane
void set_filter(const strong_actor_ptr& x, filter f) { template <class Handle>
void set_filter(const stream_id& sid, const Handle& x, filter_type f) {
std::sort(f.begin(), f.end()); std::sort(f.begin(), f.end());
lanes_[std::move(f)].paths.push_back(super::find(x)); lanes_[std::move(f)].paths.push_back(super::find(sid, x));
} }
void update_filter(const strong_actor_ptr& x, filter f) { template <class Handle>
void update_filter(const stream_id& sid, const Handle& x, filter_type f) {
std::sort(f.begin(), f.end()); std::sort(f.begin(), f.end());
erase_from_lanes(x); erase_from_lanes(x);
lanes_[std::move(f)].paths.push_back(super::find(x)); lanes_[std::move(f)].paths.push_back(super::find(sid, x));
} }
const lanes_map& lanes() const { const lanes_map& lanes() const {
return lanes_; return lanes_;
} }
private: protected:
void erase_from_lanes(const strong_actor_ptr& x) { template <class Handle>
void erase_from_lanes(const Handle& x) {
for (auto i = lanes_.begin(); i != lanes_.end(); ++i) for (auto i = lanes_.begin(); i != lanes_.end(); ++i)
if (erase_from_lane(i->second, x)) { if (erase_from_lane(i->second, x)) {
if (i->second.paths.empty()) if (i->second.paths.empty())
...@@ -139,8 +107,9 @@ private: ...@@ -139,8 +107,9 @@ private:
} }
} }
bool erase_from_lane(lane& l, const strong_actor_ptr& x) { template <class Handle>
auto predicate = [&](const downstream_path* y) { bool erase_from_lane(lane& l, const Handle& x) {
auto predicate = [&](const outbound_path* y) {
return x == y->hdl; return x == y->hdl;
}; };
auto e = l.paths.end(); auto e = l.paths.end();
...@@ -162,7 +131,7 @@ private: ...@@ -162,7 +131,7 @@ private:
} }
/// Returns `true` if `x` is selected by `f`, `false` otherwise. /// Returns `true` if `x` is selected by `f`, `false` otherwise.
bool selected(const filter& f, const T& x) { bool selected(const filter_type& f, const T& x) {
using std::get; using std::get;
for (auto& key : f) for (auto& key : f)
if (cmp_(key, get<KeyIndex>(x))) if (cmp_(key, get<KeyIndex>(x)))
...@@ -176,4 +145,4 @@ private: ...@@ -176,4 +145,4 @@ private:
} // namespace caf } // namespace caf
#endif // CAF_FILTERING_DOWNSTREAM_HPP #endif // CAF_TOPIC_SCATTERER_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. *
******************************************************************************/
#include "caf/downstream_policy.hpp"
#include <utility>
#include "caf/send.hpp"
#include "caf/atom.hpp"
#include "caf/stream_aborter.hpp"
#include "caf/downstream_path.hpp"
#include "caf/downstream_policy.hpp"
namespace caf {
downstream_policy::downstream_policy(local_actor* selfptr, const stream_id& id)
: self_(selfptr),
sid_(id),
// TODO: make configurable
min_batch_size_(1),
max_batch_size_(5),
min_buffer_size_(5),
max_batch_delay_(infinite) {
// nop
}
downstream_policy::~downstream_policy() {
// nop
}
long downstream_policy::total_credit() const {
return total_credit(paths_);
}
long downstream_policy::max_credit() const {
return max_credit(paths_);
}
long downstream_policy::min_credit() const {
return min_credit(paths_);
}
bool downstream_policy::add_path(strong_actor_ptr ptr) {
CAF_LOG_TRACE(CAF_ARG(ptr));
auto predicate = [&](const path_uptr& x) { return x->hdl == ptr; };
if (std::none_of(paths_.begin(), paths_.end(), predicate)) {
CAF_LOG_DEBUG("added new downstream path" << CAF_ARG(ptr));
stream_aborter::add(ptr, self_->address(), sid_);
paths_.emplace_back(new downstream_path(std::move(ptr), false));
return true;
}
return false;
}
bool downstream_policy::confirm_path(const strong_actor_ptr& rebind_from,
strong_actor_ptr& ptr,
bool redeployable) {
CAF_LOG_TRACE(CAF_ARG(rebind_from) << CAF_ARG(ptr) << CAF_ARG(redeployable));
auto predicate = [&](const path_uptr& x) { return x->hdl == rebind_from; };
auto e = paths_.end();
auto i = std::find_if(paths_.begin(), e, predicate);
if (i != e) {
(*i)->redeployable = redeployable;
if (rebind_from != ptr)
(*i)->hdl = ptr;
return true;
}
CAF_LOG_INFO("confirming path failed" << CAF_ARG(rebind_from)
<< CAF_ARG(ptr));
return false;
}
bool downstream_policy::remove_path(strong_actor_ptr& ptr) {
auto predicate = [&](const path_uptr& x) { return x->hdl == ptr; };
auto e = paths_.end();
auto i = std::find_if(paths_.begin(), e, predicate);
if (i != e) {
CAF_ASSERT((*i)->hdl != nullptr);
if (i != paths_.end() - 1)
std::swap(*i, paths_.back());
auto x = std::move(paths_.back());
paths_.pop_back();
unsafe_send_as(self_, x->hdl, make<stream_msg::close>(sid_));
stream_aborter::del(x->hdl, self_->address(), sid_);
return true;
}
return false;
}
void downstream_policy::close() {
for (auto& x : paths_)
unsafe_send_as(self_, x->hdl, make<stream_msg::close>(sid_));
paths_.clear();
}
void downstream_policy::abort(strong_actor_ptr& cause, const error& reason) {
CAF_LOG_TRACE(CAF_ARG(cause) << CAF_ARG(reason));
for (auto& x : paths_) {
if (x->hdl != cause)
unsafe_send_as(self_, x->hdl, make<stream_msg::abort>(sid_, reason));
stream_aborter::del(x->hdl, self_->address(), sid_);
}
}
downstream_path* downstream_policy::find(const strong_actor_ptr& ptr) const {
return find(paths_, ptr);
}
void downstream_policy::sort_paths_by_credit() {
sort_by_credit(paths_);
}
void downstream_policy::emit_batch(downstream_path& dest, size_t xs_size,
message xs) {
CAF_LOG_TRACE(CAF_ARG(dest) << CAF_ARG(xs_size) << CAF_ARG(xs));
auto scs = static_cast<int32_t>(xs_size);
auto batch_id = dest.next_batch_id++;
stream_msg::batch batch{scs, std::move(xs), batch_id};
if (dest.redeployable)
dest.unacknowledged_batches.emplace_back(batch_id, batch);
unsafe_send_as(self_, dest.hdl, stream_msg{sid_, std::move(batch)});
}
} // namespace caf
...@@ -17,80 +17,72 @@ ...@@ -17,80 +17,72 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#include "caf/stream_stage.hpp" #include "caf/inbound_path.hpp"
#include "caf/sec.hpp" #include "caf/send.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/upstream_path.hpp" #include "caf/no_stages.hpp"
#include "caf/downstream_path.hpp" #include "caf/local_actor.hpp"
#include "caf/upstream_policy.hpp"
#include "caf/downstream_policy.hpp"
namespace caf { namespace caf {
stream_stage::stream_stage(upstream_policy* in_ptr, downstream_policy* out_ptr) inbound_path::inbound_path(local_actor* selfptr, const stream_id& id,
: in_ptr_(in_ptr), strong_actor_ptr ptr)
out_ptr_(out_ptr) { : self(selfptr),
sid(std::move(id)),
hdl(std::move(ptr)),
prio(stream_priority::normal),
last_acked_batch_id(0),
last_batch_id(0),
assigned_credit(0),
redeployable(false) {
// nop // nop
} }
bool stream_stage::done() const {
return in_ptr_->closed() && out_ptr_->closed();
}
error stream_stage::upstream_batch(strong_actor_ptr& hdl, int64_t xs_id, inbound_path::~inbound_path() {
long xs_size, message& xs) { if (hdl) {
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(xs_size) << CAF_ARG(xs)); if (shutdown_reason == none)
auto path = in().find(hdl); unsafe_send_as(self, hdl, make<stream_msg::drop>(sid, self->address()));
if (path) { else
if (xs_size > path->assigned_credit) unsafe_send_as(self, hdl,
return sec::invalid_stream_state; make<stream_msg::forced_drop>(sid, self->address(),
path->last_batch_id = xs_id; shutdown_reason));
path->assigned_credit -= xs_size;
auto err = process_batch(xs);
if (err == none) {
push();
auto current_size = out().buf_size();
auto desired_size = out().credit();
if (current_size < desired_size)
in().assign_credit(desired_size - current_size);
}
return err;
} }
return sec::invalid_upstream;
} }
error stream_stage::downstream_ack(strong_actor_ptr& hdl, int64_t, void inbound_path::handle_batch(long batch_size, int64_t batch_id) {
long demand) { assigned_credit -= batch_size;
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(demand)); last_batch_id = batch_id;
auto path = out_ptr_->find(hdl);
if (path) {
downstream_demand(path, demand);
return none;
}
return sec::invalid_downstream;
} }
void stream_stage::abort(strong_actor_ptr& cause, const error& reason) { void inbound_path::emit_ack_open(actor_addr rebind_from,
in_ptr_->abort(cause, reason); long initial_demand, bool redeployable) {
out_ptr_->abort(cause, reason); CAF_LOG_TRACE(CAF_ARG(rebind_from) << CAF_ARG(initial_demand)
<< CAF_ARG(redeployable));
assigned_credit = initial_demand;
unsafe_send_as(self, hdl,
make<stream_msg::ack_open>(
sid, self->address(), std::move(rebind_from), self->ctrl(),
static_cast<int32_t>(initial_demand), redeployable));
} }
void stream_stage::last_upstream_closed() { void inbound_path::emit_ack_batch(long new_demand) {
if (out().buf_size() == 0) CAF_LOG_TRACE(CAF_ARG(new_demand));
out().close(); last_acked_batch_id = last_batch_id;
assigned_credit += new_demand;
unsafe_send_as(self, hdl,
make<stream_msg::ack_batch>(sid, self->address(),
static_cast<int32_t>(new_demand),
last_batch_id));
} }
void stream_stage::downstream_demand(downstream_path* path, long demand) { void inbound_path::emit_irregular_shutdown(local_actor* self,
auto hdl = path->hdl; const stream_id& sid,
path->open_credit += demand; const strong_actor_ptr& hdl,
if(out().buf_size() > 0) error reason) {
push(); CAF_LOG_TRACE(CAF_ARG(sid) << CAF_ARG(hdl) << CAF_ARG(reason));
else if (in().closed()) unsafe_send_as(self, hdl, make<stream_msg::forced_drop>(sid, self->address(),
out().remove_path(hdl); // don't pass path->hdl: path can become invalid std::move(reason)));
auto current_size = out().buf_size();
auto desired_size = out().credit();
if (current_size < desired_size)
in().assign_credit(desired_size - current_size);
} }
} // namespace caf } // namespace caf
...@@ -17,43 +17,93 @@ ...@@ -17,43 +17,93 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#include "caf/policy/pull5.hpp" #include "caf/invalid_stream_gatherer.hpp"
#include <numeric>
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/upstream_path.hpp"
namespace caf { namespace caf {
namespace policy {
invalid_stream_gatherer::~invalid_stream_gatherer() {
pull5::~pull5() { // nop
// nop }
}
stream_gatherer::path_ptr
void pull5::fill_assignment_vec(long downstream_credit) { invalid_stream_gatherer::add_path(const stream_id&, strong_actor_ptr,
CAF_LOG_TRACE(CAF_ARG(downstream_credit)); strong_actor_ptr, stream_priority, long, bool,
// Zero-out assignment vector if no credit is available at downstream paths. response_promise) {
if (downstream_credit <= 0) { CAF_LOG_ERROR("invalid_stream_gatherer::add_path called");
for (auto& x : assignment_vec_) return nullptr;
x.second = 0; }
return;
} bool invalid_stream_gatherer::remove_path(const stream_id&, const actor_addr&,
// Assign credit to upstream paths until no more credit is available. We must error, bool) {
// make sure to write to each element in the vector. CAF_LOG_ERROR("invalid_stream_gatherer::remove_path called");
auto available = downstream_credit; return false;
for (auto& p : assignment_vec_) { }
auto& x = p.first->assigned_credit; // current value
auto y = std::min(5l, x + available); void invalid_stream_gatherer::close(message) {
auto delta = y - x; // nop
if (delta >= min_credit_assignment()) { }
p.second = delta;
available -= delta; void invalid_stream_gatherer::abort(error) {
} else { // nop
p.second = 0; }
}
} long invalid_stream_gatherer::num_paths() const {
} return 0;
}
} // namespace policy
bool invalid_stream_gatherer::closed() const {
return true;
}
bool invalid_stream_gatherer::continuous() const {
return false;
}
void invalid_stream_gatherer::continuous(bool) {
// nop
}
stream_gatherer::path_type* invalid_stream_gatherer::path_at(size_t) {
return nullptr;
}
stream_gatherer::path_type* invalid_stream_gatherer::find(const stream_id&,
const actor_addr&) {
return nullptr;
}
long invalid_stream_gatherer::high_watermark() const {
return 0;
}
long invalid_stream_gatherer::min_credit_assignment() const {
return 0;
}
long invalid_stream_gatherer::max_credit() const {
return 0;
}
void invalid_stream_gatherer::high_watermark(long) {
// nop
}
void invalid_stream_gatherer::min_credit_assignment(long) {
// nop
}
void invalid_stream_gatherer::max_credit(long) {
// nop
}
void invalid_stream_gatherer::assign_credit(long) {
// nop
}
long invalid_stream_gatherer::initial_credit(long, path_type*) {
return 0;
}
} // namespace caf } // namespace caf
...@@ -17,71 +17,113 @@ ...@@ -17,71 +17,113 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#include "caf/stream_handler.hpp" #include "caf/invalid_stream_scatterer.hpp"
#include "caf/sec.hpp"
#include "caf/error.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/message.hpp"
#include "caf/expected.hpp"
namespace caf { namespace caf {
stream_handler::~stream_handler() { invalid_stream_scatterer::~invalid_stream_scatterer() {
// nop // nop
} }
error stream_handler::add_downstream(strong_actor_ptr&) { stream_scatterer::path_ptr
CAF_LOG_ERROR("Cannot add downstream to a stream marked as no-downstreams"); invalid_stream_scatterer::add_path(const stream_id&, strong_actor_ptr,
return sec::cannot_add_downstream; strong_actor_ptr,
mailbox_element::forwarding_stack,
message_id, message, stream_priority, bool) {
CAF_LOG_ERROR("invalid_stream_scatterer::add_path called");
return nullptr;
} }
error stream_handler::confirm_downstream(const strong_actor_ptr&, stream_scatterer::path_ptr
strong_actor_ptr&, long, bool) { invalid_stream_scatterer::confirm_path(const stream_id&, const actor_addr&,
CAF_LOG_ERROR("Cannot add downstream to a stream marked as no-downstreams"); strong_actor_ptr, long, bool) {
return sec::cannot_add_downstream; CAF_LOG_ERROR("invalid_stream_scatterer::confirm_path called");
return nullptr;
} }
error stream_handler::downstream_ack(strong_actor_ptr&, int64_t, long) { bool invalid_stream_scatterer::remove_path(const stream_id&, const actor_addr&,
CAF_LOG_ERROR("Received downstream messages in " error, bool) {
"a stream marked as no-downstreams"); CAF_LOG_ERROR("invalid_stream_scatterer::remove_path called");
return sec::invalid_downstream; return false;
} }
error stream_handler::push() { void invalid_stream_scatterer::close() {
CAF_LOG_ERROR("Cannot push to a stream marked as no-downstreams"); // nop
return sec::invalid_downstream; }
void invalid_stream_scatterer::abort(error) {
// nop
}
long invalid_stream_scatterer::num_paths() const {
return 0;
}
bool invalid_stream_scatterer::closed() const {
return true;
}
bool invalid_stream_scatterer::continuous() const {
return false;
}
void invalid_stream_scatterer::continuous(bool) {
// nop
}
stream_scatterer::path_type* invalid_stream_scatterer::path_at(size_t) {
return nullptr;
}
void invalid_stream_scatterer::emit_batches() {
// nop
}
stream_scatterer::path_type* invalid_stream_scatterer::find(const stream_id&,
const actor_addr&) {
return nullptr;
}
long invalid_stream_scatterer::credit() const {
return 0;
}
long invalid_stream_scatterer::buffered() const {
return 0;
} }
expected<long> stream_handler::add_upstream(strong_actor_ptr&, const stream_id&, long invalid_stream_scatterer::min_batch_size() const {
stream_priority) { return 0;
CAF_LOG_ERROR("Cannot add upstream to a stream marked as no-upstreams");
return sec::cannot_add_upstream;
} }
error stream_handler::upstream_batch(strong_actor_ptr&, int64_t, long, long invalid_stream_scatterer::max_batch_size() const {
message&) { return 0;
CAF_LOG_ERROR("Received upstream messages in "
"a stream marked as no-upstreams");
return sec::invalid_upstream;
} }
error stream_handler::close_upstream(strong_actor_ptr&) { long invalid_stream_scatterer::min_buffer_size() const {
CAF_LOG_ERROR("Received upstream messages in " return 0;
"a stream marked as no-upstreams");
return sec::invalid_upstream;
} }
optional<downstream_policy&> stream_handler::dp() { duration invalid_stream_scatterer::max_batch_delay() const {
return none; return infinite;
} }
optional<upstream_policy&> stream_handler::up() { void invalid_stream_scatterer::min_batch_size(long) {
return none; // nop
}
void invalid_stream_scatterer::max_batch_size(long) {
// nop
} }
message stream_handler::make_output_token(const stream_id&) const { void invalid_stream_scatterer::min_buffer_size(long) {
return make_message(); // nop
}
void invalid_stream_scatterer::max_batch_delay(duration) {
// nop
} }
} // namespace caf } // namespace caf
...@@ -17,56 +17,80 @@ ...@@ -17,56 +17,80 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#include "caf/stream_sink.hpp" #include "caf/outbound_path.hpp"
#include "caf/send.hpp" #include "caf/send.hpp"
#include "caf/upstream_path.hpp" #include "caf/logger.hpp"
#include "caf/upstream_policy.hpp" #include "caf/no_stages.hpp"
#include "caf/local_actor.hpp"
namespace caf { namespace caf {
stream_sink::stream_sink(upstream_policy* in_ptr, outbound_path::outbound_path(local_actor* selfptr, const stream_id& id,
strong_actor_ptr&& orig_sender, strong_actor_ptr ptr)
std::vector<strong_actor_ptr>&& trailing_stages, : self(selfptr),
message_id mid) sid(id),
: in_ptr_(in_ptr), hdl(std::move(ptr)),
original_sender_(std::move(orig_sender)), next_batch_id(0),
next_stages_(std::move(trailing_stages)), open_credit(0),
original_msg_id_(mid), redeployable(false),
min_buffer_size_(5) { // TODO: make configurable next_ack_id(0) {
// nop // nop
} }
bool stream_sink::done() const { outbound_path::~outbound_path() {
// we are done streaming if no upstream paths remain CAF_LOG_TRACE(CAF_ARG(shutdown_reason));
return in_ptr_->closed(); if (hdl) {
if (shutdown_reason == none)
unsafe_send_as(self, hdl, make<stream_msg::close>(sid, self->address()));
else
unsafe_send_as(
self, hdl,
make<stream_msg::forced_close>(sid, self->address(), shutdown_reason));
}
if (shutdown_reason != none)
unsafe_response(self, std::move(cd.hdl), no_stages, cd.mid,
std::move(shutdown_reason));
} }
error stream_sink::upstream_batch(strong_actor_ptr& hdl, int64_t xs_id, void outbound_path::handle_ack_open(long initial_credit) {
long xs_size, message& xs) { open_credit = initial_credit;
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(xs_size) << CAF_ARG(xs)); cd.hdl = nullptr;
auto path = in().find(hdl); }
if (path) {
if (xs_size > path->assigned_credit) void outbound_path::emit_open(strong_actor_ptr origin,
return sec::invalid_stream_state; mailbox_element::forwarding_stack stages,
path->last_batch_id = xs_id; message_id handshake_mid, message handshake_data,
path->assigned_credit -= xs_size; stream_priority prio, bool redeployable) {
auto err = consume(xs); CAF_LOG_TRACE(CAF_ARG(origin) << CAF_ARG(stages) << CAF_ARG(handshake_mid)
if (err == none) << CAF_ARG(handshake_data) << CAF_ARG(prio)
in().assign_credit(min_buffer_size()); << CAF_ARG(redeployable));
return err; cd = client_data{origin, handshake_mid};
} hdl->enqueue(
return sec::invalid_upstream; make_mailbox_element(std::move(origin), handshake_mid, std::move(stages),
make_message(make<stream_msg::open>(
sid, self->address(), std::move(handshake_data),
self->ctrl(), hdl, prio, redeployable))),
self->context());
} }
void stream_sink::abort(strong_actor_ptr& cause, const error& reason) { void outbound_path::emit_batch(long xs_size, message xs) {
unsafe_send_as(in_ptr_->self(), original_sender_, reason); CAF_LOG_TRACE(CAF_ARG(xs_size) << CAF_ARG(xs));
in_ptr_->abort(cause, reason); open_credit -= xs_size;
auto bid = next_batch_id++;
stream_msg::batch batch{static_cast<int32_t>(xs_size), std::move(xs), bid};
if (redeployable)
unacknowledged_batches.emplace_back(bid, batch);
unsafe_send_as(self, hdl, stream_msg{sid, self->address(), std::move(batch)});
} }
void stream_sink::last_upstream_closed() { void outbound_path::emit_irregular_shutdown(local_actor* self,
unsafe_response(in().self(), std::move(original_sender_), const stream_id& sid,
std::move(next_stages_), original_msg_id_, finalize()); const strong_actor_ptr& hdl,
error reason) {
CAF_LOG_TRACE(CAF_ARG(reason));
unsafe_send_as(self, hdl, make<stream_msg::forced_close>(sid, self->address(),
std::move(reason)));
} }
} // namespace caf } // namespace caf
...@@ -17,21 +17,28 @@ ...@@ -17,21 +17,28 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#include <utility> #include "caf/detail/pull5_gatherer.hpp"
#include "caf/upstream_path.hpp"
namespace caf { namespace caf {
namespace detail {
upstream_path::upstream_path(strong_actor_ptr ptr, stream_id id, pull5_gatherer::pull5_gatherer(local_actor* selfptr) : super(selfptr) {
stream_priority p)
: hdl(std::move(ptr)),
sid(std::move(id)),
prio(p),
last_acked_batch_id(0),
last_batch_id(0),
assigned_credit(0) {
// nop // nop
} }
void pull5_gatherer::assign_credit(long available) {
CAF_LOG_TRACE(CAF_ARG(available));
for (auto& kvp : assignment_vec_) {
auto x = std::min(available, 5l - kvp.first->assigned_credit);
available -= x;
kvp.second = x;
}
emit_credits();
}
long pull5_gatherer::initial_credit(long, inbound_path*) {
return 5;
}
} // namespace detail
} // namespace caf } // namespace caf
...@@ -17,21 +17,34 @@ ...@@ -17,21 +17,34 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#include "caf/policy/greedy.hpp" #include "caf/random_gatherer.hpp"
#include <numeric>
#include "caf/logger.hpp"
#include "caf/upstream_path.hpp"
namespace caf { namespace caf {
namespace policy {
greedy::~greedy() { random_gatherer::random_gatherer(local_actor* selfptr) : super(selfptr) {
// nop
}
random_gatherer::~random_gatherer() {
// nop // nop
} }
void greedy::fill_assignment_vec(long downstream_credit) { void random_gatherer::assign_credit(long available) {
CAF_LOG_TRACE(CAF_ARG(available));
for (auto& kvp : assignment_vec_) {
auto x = std::min(available, max_credit() - kvp.first->assigned_credit);
available -= x;
kvp.second = x;
}
emit_credits();
}
long random_gatherer::initial_credit(long available, path_type*) {
return std::min(available, max_credit());
}
/*
void random_gatherer::fill_assignment_vec(long downstream_credit) {
CAF_LOG_TRACE(CAF_ARG(downstream_credit)); CAF_LOG_TRACE(CAF_ARG(downstream_credit));
// Zero-out assignment vector if no credit is available at downstream paths. // Zero-out assignment vector if no credit is available at downstream paths.
if (downstream_credit <= 0) { if (downstream_credit <= 0) {
...@@ -54,6 +67,6 @@ void greedy::fill_assignment_vec(long downstream_credit) { ...@@ -54,6 +67,6 @@ void greedy::fill_assignment_vec(long downstream_credit) {
} }
} }
} }
*/
} // namespace policy
} // namespace caf } // namespace caf
...@@ -27,8 +27,11 @@ ...@@ -27,8 +27,11 @@
namespace caf { namespace caf {
response_promise::response_promise() response_promise::response_promise() : self_(nullptr) {
: self_(nullptr) { // nop
}
response_promise::response_promise(none_t) : response_promise() {
// nop // nop
} }
......
...@@ -193,6 +193,12 @@ bool scheduled_actor::cleanup(error&& fail_state, execution_unit* host) { ...@@ -193,6 +193,12 @@ bool scheduled_actor::cleanup(error&& fail_state, execution_unit* host) {
// Clear all state. // Clear all state.
awaited_responses_.clear(); awaited_responses_.clear();
multiplexed_responses_.clear(); multiplexed_responses_.clear();
if (fail_state != none)
for (auto& kvp : streams_)
kvp.second->abort(fail_state);
else
for (auto& kvp : streams_)
kvp.second->close();
streams_.clear(); streams_.clear();
// Dispatch to parent's `cleanup` function. // Dispatch to parent's `cleanup` function.
return local_actor::cleanup(std::move(fail_state), host); return local_actor::cleanup(std::move(fail_state), host);
...@@ -647,27 +653,46 @@ bool scheduled_actor::handle_stream_msg(mailbox_element& x, ...@@ -647,27 +653,46 @@ bool scheduled_actor::handle_stream_msg(mailbox_element& x,
CAF_LOG_TRACE(CAF_ARG(x)); CAF_LOG_TRACE(CAF_ARG(x));
CAF_ASSERT(x.content().match_elements<stream_msg>()); CAF_ASSERT(x.content().match_elements<stream_msg>());
auto& sm = x.content().get_mutable_as<stream_msg>(0); auto& sm = x.content().get_mutable_as<stream_msg>(0);
auto e = streams_.end(); stream_msg_visitor f{this, sm, active_behavior};
stream_msg_visitor f{this, sm.sid, streams_.find(sm.sid), e, active_behavior}; auto result = visit(f, sm.content);
auto res = visit(f, sm.content); if (streams_.empty() && !has_behavior())
auto success = (res.first == none); quit(exit_reason::normal);
auto i = res.second; return result;
if (!success) { }
if (i != e) {
i->second->abort(current_sender(), std::move(res.first)); bool scheduled_actor::add_source(const stream_manager_ptr& mgr, const stream_id& sid,
streams_.erase(i); strong_actor_ptr source_ptr, strong_actor_ptr original_stage,
} stream_priority prio, bool redeployable,
} else if (i != e) { response_promise result_cb) {
if (i->second->done()) { CAF_LOG_TRACE(CAF_ARG(mgr) << CAF_ARG(sid) << CAF_ARG(source_ptr)
streams_.erase(i); << CAF_ARG(original_stage) << CAF_ARG(prio)
CAF_LOG_DEBUG("Stream is done and could be safely erased" << CAF_ARG(redeployable) << CAF_ARG(result_cb));
<< CAF_ARG(sm.sid) << ", remaining streams =" CAF_ASSERT(mgr != nullptr);
<< deep_to_string(streams_).c_str()); if (!source_ptr || !sid.valid())
if (streams_.empty() && !has_behavior()) return false;
quit(exit_reason::normal); return mgr->add_source(sid, std::move(source_ptr),
} std::move(original_stage), prio, redeployable,
} std::move(result_cb));
return success; }
bool scheduled_actor::add_source(const stream_manager_ptr& mgr,
const stream_id& sid,
response_promise result_cb) {
CAF_LOG_TRACE(CAF_ARG(mgr) << CAF_ARG(sid));
CAF_ASSERT(mgr != nullptr);
CAF_ASSERT(current_mailbox_element() != nullptr);
if (!current_mailbox_element()->content().match_elements<stream_msg>())
return false;
auto& sm = current_mailbox_element()->content().get_mutable_as<stream_msg>(0);
if (!holds_alternative<stream_msg::open>(sm.content))
return false;
auto& opn = get<stream_msg::open>(sm.content);
auto source_ptr = std::move(opn.prev_stage);
if (!source_ptr || !sid.valid())
return false;
return mgr->add_source(sid, std::move(source_ptr),
std::move(opn.original_stage), opn.priority, opn.redeployable,
std::move(result_cb));
} }
} // namespace caf } // namespace caf
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. *
******************************************************************************/
#include "caf/stream_gatherer.hpp"
#include "caf/actor_addr.hpp"
#include "caf/actor_cast.hpp"
#include "caf/inbound_path.hpp"
#include "caf/actor_control_block.hpp"
namespace caf {
stream_gatherer::~stream_gatherer() {
// nop
}
bool stream_gatherer::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_gatherer::path_type* stream_gatherer::find(const stream_id& sid,
const strong_actor_ptr& x) {
return find(sid, actor_cast<actor_addr>(x));
}
} // namespace caf
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
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