Commit df9971c1 authored by Dominik Charousset's avatar Dominik Charousset

Re-organize filtering and bugfixing

parent 7787c1b8
......@@ -80,6 +80,8 @@ public:
actor& operator=(actor&&) = default;
actor& operator=(const actor&) = default;
actor(std::nullptr_t);
actor(const scoped_actor&);
explicit actor(const unsafe_actor_handle_init_t&) CAF_DEPRECATED;
......@@ -108,12 +110,9 @@ public:
return *this;
}
actor& operator=(const scoped_actor& x);
actor& operator=(std::nullptr_t);
inline actor& operator=(std::nullptr_t) {
ptr_.reset();
return *this;
}
actor& operator=(const scoped_actor& x);
/// Queries whether this actor handle is valid.
inline explicit operator bool() const {
......
......@@ -106,7 +106,7 @@ public:
actor_system_config();
actor_system_config(actor_system_config&&) = default;
actor_system_config(actor_system_config&&);
actor_system_config(const actor_system_config&) = delete;
actor_system_config& operator=(const actor_system_config&) = delete;
......@@ -276,8 +276,8 @@ public:
// -- backward compatibility -------------------------------------------------
std::string& logger_filename CAF_DEPRECATED = logger_file_name;
std::string& logger_filter CAF_DEPRECATED = logger_component_filter;
std::string& logger_filename CAF_DEPRECATED;
std::string& logger_filter CAF_DEPRECATED;
// -- config parameters of the middleman -------------------------------------
......
......@@ -77,6 +77,8 @@
#include "caf/message_builder.hpp"
#include "caf/message_handler.hpp"
#include "caf/response_handle.hpp"
#include "caf/fused_scatterer.hpp"
#include "caf/random_gatherer.hpp"
#include "caf/system_messages.hpp"
#include "caf/abstract_channel.hpp"
#include "caf/may_have_timeout.hpp"
......@@ -87,12 +89,15 @@
#include "caf/event_based_actor.hpp"
#include "caf/primitive_variant.hpp"
#include "caf/timeout_definition.hpp"
#include "caf/broadcast_scatterer.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/composable_behavior.hpp"
#include "caf/typed_actor_pointer.hpp"
#include "caf/scoped_execution_unit.hpp"
#include "caf/typed_response_promise.hpp"
#include "caf/random_topic_scatterer.hpp"
#include "caf/broadcast_topic_scatterer.hpp"
#include "caf/typed_event_based_actor.hpp"
#include "caf/abstract_composable_behavior.hpp"
......
......@@ -40,6 +40,7 @@ public:
}
void emit_batches() override {
CAF_LOG_TRACE("");
auto chunk = this->get_chunk(this->min_credit());
auto csize = static_cast<long>(chunk.size());
CAF_LOG_TRACE(CAF_ARG(chunk));
......
......@@ -17,8 +17,8 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_TOPIC_SCATTERER_HPP
#define CAF_TOPIC_SCATTERER_HPP
#ifndef CAF_BROADCAST_TOPIC_SCATTERER_HPP
#define CAF_BROADCAST_TOPIC_SCATTERER_HPP
#include <map>
#include <tuple>
......@@ -26,19 +26,17 @@
#include <vector>
#include <functional>
#include "caf/buffered_scatterer.hpp"
#include "caf/topic_scatterer.hpp"
namespace caf {
/// A topic scatterer that delivers data in broadcast fashion to all sinks.
template <class T, class Filter,
class KeyCompare = std::equal_to<typename Filter::value_type>,
long KeyIndex = 0>
template <class T, class Filter, class Select>
class broadcast_topic_scatterer
: public topic_scatterer<T, Filter, KeyCompare, KeyIndex> {
: public topic_scatterer<T, Filter, Select> {
public:
/// Base type.
using super = buffered_scatterer<T>;
using super = topic_scatterer<T, Filter, Select>;
broadcast_topic_scatterer(local_actor* selfptr) : super(selfptr) {
// nop
......@@ -51,6 +49,7 @@ public:
}
void emit_batches() override {
CAF_LOG_TRACE("");
this->fan_out();
for (auto& kvp : this->lanes_) {
auto& l = kvp.second;
......@@ -60,8 +59,8 @@ public:
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);
CAF_ASSERT(x->open_credit >= csize);
x->emit_batch(csize, wrapped_chunk);
}
}
}
......@@ -69,4 +68,4 @@ public:
} // namespace caf
#endif // CAF_TOPIC_SCATTERER_HPP
#endif // CAF_BROADCAST_TOPIC_SCATTERER_HPP
......@@ -17,57 +17,66 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_STREAM_SINK_HPP
#define CAF_STREAM_SINK_HPP
#ifndef CAF_DETAIL_APPLY_ARGS_HPP
#define CAF_DETAIL_APPLY_ARGS_HPP
#include <vector>
#include <utility>
#include "caf/extend.hpp"
#include "caf/message_id.hpp"
#include "caf/stream_manager.hpp"
#include "caf/upstream_policy.hpp"
#include "caf/actor_control_block.hpp"
#include "caf/detail/int_list.hpp"
#include "caf/detail/type_list.hpp"
namespace caf {
/// Represents a terminal stream stage.
class stream_sink : public stream_manager {
public:
stream_sink(upstream_policy* in_ptr, strong_actor_ptr&& orig_sender,
std::vector<strong_actor_ptr>&& trailing_stages, message_id mid);
bool done() const override;
error batch(const actor_addr& hdl, long xs_size, message& xs,
int64_t xs_id) override;
void abort(error reason) override;
void last_upstream_closed();
inline upstream_policy& in() {
return *in_ptr_;
}
long min_buffer_size() const {
return min_buffer_size_;
}
protected:
/// Consumes a batch.
virtual error consume(message& xs) = 0;
/// Computes the final result after consuming all batches of the stream.
virtual message finalize() = 0;
private:
upstream_policy* in_ptr_;
strong_actor_ptr original_sender_;
std::vector<strong_actor_ptr> next_stages_;
message_id original_msg_id_;
long min_buffer_size_;
};
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_STREAM_SINK_HPP
#endif // CAF_DETAIL_APPLY_ARGS_HPP
......@@ -122,6 +122,11 @@ public:
(*this)();
}
template <class T>
void operator()(stream<T>&) {
(*this)();
}
// visit API - returns true if T was visited, false if T was skipped
template <class T>
......@@ -138,6 +143,11 @@ public:
return false;
}
template <class T>
bool visit(stream<T>&) {
return true;
}
inline bool visit(optional<skip_t>& x) {
if (x)
return false;
......
......@@ -35,6 +35,10 @@ public:
// nop
}
using super::min_batch_size;
using super::max_batch_size;
using super::min_buffer_size;
long min_batch_size() const override {
return 1;
}
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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_STREAM_DISTRIBUTION_TREE_HPP
#define CAF_DETAIL_STREAM_DISTRIBUTION_TREE_HPP
#include <memory>
#include <unordered_map>
#include "caf/ref_counted.hpp"
#include "caf/stream_manager.hpp"
#include "caf/random_gatherer.hpp"
#include "caf/scheduled_actor.hpp"
#include "caf/broadcast_scatterer.hpp"
namespace caf {
namespace detail {
/// A stream distribution tree consist of peers forming an acyclic graph. The
/// user is responsible for making sure peers do not form a loop. Data is
/// flooded along the tree. Each peer serves any number of subscribers. The
/// policy of the tree enables subscriptions to different chunks of the whole
/// stream (substreams).
///
/// The tree uses two CAF streams between each pair of peers for transmitting
/// data. This automatically adds backpressure to the system, i.e., no peer can
/// overwhelm others.
///
/// Policies need to provide the following member types and functions:
///
/// ~~~{.cpp}
/// struct policy {
/// /// Any number of user-defined arguments.
/// policy(...);
///
/// /// Tuple of available substream scatterers to subscribers. Each
/// /// element of the tuple is a subtype of `topic_scatterer<T>` (or
/// /// provides a similar interface).
/// using substream_scatterers = ...;
///
/// /// A compile-time index type identifying individual substreams.
/// using substream_index_type = ...;
///
/// /// Represent a single topic for filtering stream data.
/// using topic_type = ...;
///
/// /// Groups multiple topics into a single selection filter.
/// using filter_type = ...;
///
/// /// Policy for gathering data from peers.
/// using gatherer_type = ...;
///
/// /// Policy for scattering data to peers.
/// using scatterer_type = ...;
///
/// /// Decides whether a filter applies to a given message.
/// bool selected(const filter_type& sieve, const message& sand) const;
///
/// // + one overload to `selected` for each substream data type.
///
/// /// Accesses individual substream scatterers by index.
/// auto& substream(substream_scatterers&, substream_index_type);
///
/// /// Accesses individual substream scatterers by stream ID.
/// stream_scatterer* substream(substream_scatterers&, const stream_id&);
///
/// /// Returns true if the substreams have no data pending and the policy
/// /// decides to shutdown this peer.
/// bool at_end() const;
///
/// /// Returns true if outgoing data originating on this peer is
/// /// forwarded to all substreams. Otherwise, the substreams receive data
/// /// produced on other peers exclusively.
/// bool substream_local_data() const;
///
/// /// Handles a batch if matches the data types managed by any of the
/// /// substreams. Returns `none` if the batch is a peer message, otherwise
/// /// the error resulting from handling the batch.
/// optional<error> batch(const stream_id& sid, const actor_addr& hdl,
/// long xs_size, message& xs, int64_t xs_id);
///
/// ///
/// void push_to_substreams(message msg);
/// };
/// ~~~
template <class Policy>
class stream_distribution_tree : public stream_manager {
public:
// -- nested types -----------------------------------------------------------
using super = stream_manager;
using gatherer_type = typename Policy::gatherer_type;
using scatterer_type = typename Policy::scatterer_type;
// --- constructors and destructors ------------------------------------------
template <class... Ts>
stream_distribution_tree(scheduled_actor* selfptr, Ts&&... xs)
: self_(selfptr),
in_(selfptr),
out_(selfptr),
policy_(this, std::forward<Ts>(xs)...) {
// nop
}
~stream_distribution_tree() override {
// nop
}
// -- Accessors --------------------------------------------------------------
inline Policy& policy() {
return policy_;
}
inline const Policy& policy() const {
return policy_;
}
void close_remote_input();
// -- Overridden member functions of `stream_manager` ------------------------
/// Terminates the core actor with log output `log_message` if `at_end`
/// returns `true`.
void shutdown_if_at_end(const char* log_message) {
CAF_IGNORE_UNUSED(log_message);
if (policy_.at_end()) {
CAF_LOG_DEBUG(log_message);
self_->quit(caf::exit_reason::user_shutdown);
}
}
// -- overridden member functions of `stream_manager` ------------------------
error ack_open(const stream_id& sid, const actor_addr& rebind_from,
strong_actor_ptr rebind_to, long initial_demand,
bool redeployable) override {
CAF_LOG_TRACE(CAF_ARG(sid) << CAF_ARG(rebind_from) << CAF_ARG(rebind_to)
<< CAF_ARG(initial_demand) << CAF_ARG(redeployable));
auto res = super::ack_open(sid, rebind_from, rebind_to, initial_demand,
redeployable);
if (res == none)
policy_.ack_open_success(sid, rebind_from, rebind_to);
else
policy_.ack_open_failure(sid, rebind_from, rebind_to, res);
return res;
}
error process_batch(message& xs) override {
CAF_LOG_TRACE(CAF_ARG(xs));
policy_.handle_batch(xs);
return none;
}
error batch(const stream_id& sid, const actor_addr& hdl, long xs_size,
message& xs, int64_t xs_id) override {
CAF_LOG_TRACE(CAF_ARG(sid) << CAF_ARG(hdl) << CAF_ARG(xs_size)
<< CAF_ARG(xs) << CAF_ARG(xs_id));
policy_.before_handle_batch(sid, hdl, xs_size, xs, xs_id);
auto res = super::batch(sid, hdl, xs_size, xs, xs_id);
policy_.after_handle_batch(sid, hdl, xs_id);
push();
return res;
}
bool done() const override {
return false;
}
gatherer_type& in() override {
return in_;
}
scatterer_type& out() override {
return out_;
}
void downstream_demand(outbound_path*, long) override {
CAF_LOG_TRACE("");
push();
in_.assign_credit(out_.credit());
}
error close(const stream_id& sid, const actor_addr& hdl) override {
CAF_LOG_TRACE(CAF_ARG(sid) << CAF_ARG(hdl));
if (in_.remove_path(sid, hdl, none, true))
return policy_.path_closed(sid, hdl);
return none;
}
error drop(const stream_id& sid, const actor_addr& hdl) override {
CAF_LOG_TRACE(CAF_ARG(sid) << CAF_ARG(hdl));
if (out_.remove_path(sid, hdl, none, true))
return policy_.path_dropped(sid, hdl);
return none;
}
error forced_close(const stream_id& sid, const actor_addr& hdl,
error reason) override {
CAF_LOG_TRACE(CAF_ARG(sid) << CAF_ARG(hdl) << CAF_ARG(reason));
if (in_.remove_path(sid, hdl, reason, true))
return policy_.path_force_closed(sid, hdl, std::move(reason));
return none;
}
error forced_drop(const stream_id& sid, const actor_addr& hdl,
error reason) override {
if (out_.remove_path(sid, hdl, reason, true))
return policy_.path_force_dropped(sid, hdl, std::move(reason));
return none;
}
scheduled_actor* self() {
return self_;
}
private:
scheduled_actor* self_;
gatherer_type in_;
scatterer_type out_;
Policy policy_;
};
} // namespace detail
} // namespace caf
#endif // CAF_DETAIL_STREAM_DISTRIBUTION_TREE_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_FUSED_SCATTERER
#define CAF_FUSED_SCATTERER
#include <tuple>
#include <cstddef>
#include "caf/logger.hpp"
#include "caf/stream_scatterer.hpp"
namespace caf {
namespace detail {
/// Utility function for repeating `x` for a given template parameter pack.
template <class T, class U>
U pack_repeat(U x) {
return x;
}
template <class Iter>
class ptr_array_initializer {
public:
ptr_array_initializer(Iter first) : i_(first) {
// nop
}
void operator()() {
// end of recursion
}
template <class T, class... Ts>
void operator()(T& x, Ts&... xs) {
*i_ = &x;
++i_;
(*this)(xs...);
}
private:
Iter i_;
};
struct scatterer_selector {
inline stream_scatterer* operator()(const message&) {
return nullptr;
}
template <class T, class... Ts>
stream_scatterer* operator()(const message& msg, T& x, Ts&... xs) {
if (msg.match_element<stream<typename T::value_type>>(0))
return &x;
return (*this)(msg, xs...);
}
};
} // namespace detail
/// A scatterer that delegates to any number of sub-scatterers. Data is only
/// pushed to the main scatterer `T` per default.
template <class T, class... Ts>
class fused_scatterer : public stream_scatterer {
public:
using substreams_tuple = std::tuple<T, Ts...>;
using pointer = stream_scatterer*;
using const_pointer = const pointer;
using iterator = pointer*;
using const_iterator = const pointer*;
fused_scatterer(local_actor* self)
: substreams_(self, detail::pack_repeat<Ts>(self)...) {
detail::ptr_array_initializer<pointer*> f{ptrs_};
auto indices = detail::get_indices(substreams_);
detail::apply_args(f, indices, substreams_);
}
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 {
CAF_LOG_TRACE(CAF_ARG(sid) << CAF_ARG(origin) << CAF_ARG(sink_ptr)
<< CAF_ARG(stages) << CAF_ARG(handshake_mid)
<< CAF_ARG(handshake_data) << CAF_ARG(prio)
<< CAF_ARG(redeployable));
auto ptr = substream_by_handshake_type(handshake_data);
if (!ptr)
return nullptr;
return ptr->add_path(sid, std::move(origin), std::move(sink_ptr),
std::move(stages), handshake_mid,
std::move(handshake_data), prio, redeployable);
}
path_ptr confirm_path(const stream_id& sid, const actor_addr& from,
strong_actor_ptr to, long initial_demand,
bool redeployable) override {
CAF_LOG_TRACE(CAF_ARG(sid) << CAF_ARG(from) << CAF_ARG(to)
<< CAF_ARG(initial_demand) << CAF_ARG(redeployable));
return first_hit([&](pointer ptr) -> outbound_path* {
// Note: we cannot blindly try `confirm_path` on each scatterer, because
// this will trigger forced_close messages.
if (ptr->find(sid, from) == nullptr)
return nullptr;
return ptr->confirm_path(sid, from, to, initial_demand, redeployable);
});
}
bool remove_path(const stream_id& sid, const actor_addr& addr, error reason,
bool silent) override {
CAF_LOG_TRACE(CAF_ARG(sid) << CAF_ARG(addr) << CAF_ARG(reason)
<< CAF_ARG(silent));
return std::any_of(begin(), end(), [&](pointer x) {
return x->remove_path(sid, addr, reason, silent);
});
}
bool paths_clean() const override {
return std::all_of(begin(), end(),
[&](const_pointer x) { return x->paths_clean(); });
}
void close() override {
CAF_LOG_TRACE("");
for (auto ptr : ptrs_)
ptr->close();
}
void abort(error reason) override {
CAF_LOG_TRACE(CAF_ARG(reason));
for (auto ptr : ptrs_)
ptr->abort(reason);
}
long num_paths() const override {
return std::accumulate(begin(), end(), 0l, [](long x, const_pointer y) {
return x + y->num_paths();
});
}
bool closed() const override {
return std::all_of(begin(), end(),
[&](const_pointer x) { return x->closed(); });
}
bool continuous() const override {
return std::any_of(begin(), end(),
[&](const_pointer x) { return x->continuous(); });
}
void continuous(bool value) override {
for (auto ptr : ptrs_)
ptr->continuous(value);
}
void emit_batches() override {
CAF_LOG_TRACE("");
for (auto ptr : ptrs_)
ptr->emit_batches();
}
path_ptr find(const stream_id& sid, const actor_addr& x) override{
return first_hit([&](const_pointer ptr) { return ptr->find(sid, x); });
}
path_ptr path_at(size_t idx) override {
auto i = std::begin(ptrs_);
auto e = std::end(ptrs_);
while (i != e) {
auto np = static_cast<size_t>((*i)->num_paths());
if (idx < np)
return (*i)->path_at(idx);
idx -= np;
}
return nullptr;
}
long credit() const override {
return std::accumulate(
begin(), end(), std::numeric_limits<long>::max(),
[](long x, pointer y) { return std::min(x, y->credit()); });
}
long buffered() const override {
return std::accumulate(begin(), end(), 0l, [](long x, const_pointer y) {
return x + y->buffered();
});
}
long min_batch_size() const override {
return main_stream().min_batch_size();
}
long max_batch_size() const override {
return main_stream().max_batch_size();
}
long min_buffer_size() const override {
return main_stream().min_buffer_size();
}
duration max_batch_delay() const override {
return main_stream().max_batch_delay();
}
void min_batch_size(long x) override {
main_stream().min_batch_size(x);
}
void max_batch_size(long x) override {
main_stream().max_batch_size(x);
}
void min_buffer_size(long x) override {
main_stream().min_buffer_size(x);
}
void max_batch_delay(duration x) override {
main_stream().max_batch_delay(x);
}
template <class... Us>
void push(Us&&... xs) {
main_stream().push(std::forward<Us>(xs)...);
}
iterator begin() {
return std::begin(ptrs_);
}
const_iterator begin() const {
return std::begin(ptrs_);
}
iterator end() {
return std::end(ptrs_);
}
const_iterator end() const {
return std::end(ptrs_);
}
T& main_stream() {
return std::get<0>(substreams_);
}
const T& main_stream() const {
return std::get<0>(substreams_);
}
template <size_t I>
typename std::tuple_element<I, substreams_tuple>::type& substream() {
return std::get<I>(substreams_);
}
template <size_t I>
const typename std::tuple_element<I, substreams_tuple>::type&
substream() const {
return std::get<I>(substreams_);
}
stream_scatterer* substream_by_handshake_type(const message& msg) {
detail::scatterer_selector f;
auto indices = detail::get_indices(substreams_);
return detail::apply_args_prefixed(f, indices, substreams_, msg);
}
private:
template <class F>
path_ptr first_hit(F f) {
for (auto ptr : ptrs_) {
auto result = f(ptr);
if (result != nullptr)
return result;
}
return nullptr;
}
substreams_tuple substreams_;
stream_scatterer* ptrs_[sizeof...(Ts) + 1];
};
} // namespace caf
#endif // CAF_FUSED_SCATTERER
......@@ -43,6 +43,9 @@ template <class> struct timeout_definition;
template <class, class, int> class actor_cast_access;
template <class, class, class> class broadcast_topic_scatterer;
template <class, class, class> class random_topic_scatterer;
// -- variadic templates -------------------------------------------------------
template <class...> class result;
......@@ -54,6 +57,7 @@ template <class...> class typed_event_based_actor;
// -- variadic templates with 1 fixed argument ---------------------------------
template <class, class...> class fused_scatterer;
template <class, class...> class annotated_stream;
// -- classes ------------------------------------------------------------------
......@@ -74,7 +78,6 @@ class serializer;
class actor_proxy;
class local_actor;
class ref_counted;
class stream_sink;
class actor_config;
class actor_system;
class deserializer;
......@@ -83,7 +86,6 @@ class message_view;
class scoped_actor;
class inbound_path;
class outbound_path;
class stream_source;
class abstract_actor;
class abstract_group;
class actor_registry;
......@@ -91,6 +93,7 @@ class blocking_actor;
class execution_unit;
class proxy_registry;
class stream_manager;
class random_gatherer;
class stream_gatherer;
class actor_companion;
class mailbox_element;
......@@ -184,6 +187,7 @@ class manager;
namespace detail {
template <class> class type_erased_value_impl;
template <class> class stream_distribution_tree;
class disposer;
class message_data;
......
......@@ -31,13 +31,11 @@
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>
template <class T, class Filter, class Select>
class random_topic_scatterer
: public topic_scatterer<T, Filter, KeyCompare, KeyIndex> {
: public topic_scatterer<T, Filter, Select> {
public:
using super = topic_scatterer<T, Filter, KeyCompare, KeyIndex>;
using super = topic_scatterer<T, Filter, Select>;
random_topic_scatterer(local_actor* selfptr) : super(selfptr) {
// nop
......@@ -50,6 +48,7 @@ public:
}
void emit_batches() override {
CAF_LOG_TRACE("");
this->fan_out();
for (auto& kvp : this->lanes_) {
auto& l = kvp.second;
......
......@@ -44,6 +44,7 @@
#include "caf/stream_source_impl.hpp"
#include "caf/stream_result_trait.hpp"
#include "caf/broadcast_scatterer.hpp"
#include "caf/terminal_stream_scatterer.hpp"
#include "caf/to_string.hpp"
......@@ -481,8 +482,6 @@ public:
CAF_IGNORE_UNUSED(policies);
CAF_ASSERT(current_mailbox_element() != nullptr);
CAF_ASSERT(current_mailbox_element()->content().match_elements<stream_msg>());
auto& sm = current_mailbox_element()->content().get_as<stream_msg>(0);
CAF_ASSERT(holds_alternative<stream_msg::open>(sm.content));
using output_type = typename stream_stage_trait_t<Fun>::output;
using state_type = typename stream_stage_trait_t<Fun>::state;
static_assert(std::is_same<
......@@ -526,6 +525,36 @@ public:
std::move(cleanup), policies);
}
/// Creates a new stream sink of type T.
/// @pre `current_mailbox_element()` is a `stream_msg::open` handshake
/// @param in The input of the sink.
/// @param f Callback for initializing the object after successful creation.
/// @param xs Parameter pack for creating the instance of T.
/// @returns A stream object with a pointer to the generated `stream_manager`.
template <class T, class In, class SuccessCallback, class... Ts>
stream_result<typename T::output_type>
make_sink_impl(const stream<In>& in, SuccessCallback f, Ts&&... xs) {
CAF_ASSERT(current_mailbox_element() != nullptr);
CAF_ASSERT(current_mailbox_element()->content().match_elements<stream_msg>());
auto& sm = current_mailbox_element()->content().get_as<stream_msg>(0);
CAF_ASSERT(holds_alternative<stream_msg::open>(sm.content));
auto& opn = get<stream_msg::open>(sm.content);
auto sid = in.id();
auto next = take_current_next_stage();
auto ptr = make_counted<T>(this, std::forward<Ts>(xs)...);
auto rp = make_response_promise();
if (!add_source(ptr, sid, std::move(opn.prev_stage),
std::move(opn.original_stage), opn.priority,
opn.redeployable, rp)) {
CAF_LOG_ERROR("cannot create stream stage without source");
rp.deliver(sec::cannot_add_upstream);
return none;
}
f(*ptr);
streams_.emplace(in.id(), ptr);
return {in.id(), std::move(ptr)};
}
/// Creates a new stream sink.
/// @pre `current_mailbox_element()` is a `stream_msg::open` handshake
/// @param in The input of the sink.
......@@ -535,16 +564,12 @@ public:
/// @param gatherer_type Sets the policy for upstream communication.
/// @returns A stream object with a pointer to the generated `stream_manager`.
template <class In, class Init, class Fun, class Finalize,
class Gatherer = random_gatherer>
class Gatherer = random_gatherer,
class Scatterer = terminal_stream_scatterer>
stream_result<typename stream_sink_trait_t<Fun, Finalize>::output>
make_sink(const stream<In>& in, Init init, Fun fun, Finalize finalize,
policy::arg<Gatherer> gatherer_type = {}) {
CAF_IGNORE_UNUSED(gatherer_type);
CAF_ASSERT(current_mailbox_element() != nullptr);
CAF_ASSERT(current_mailbox_element()->content().match_elements<stream_msg>());
auto& sm = current_mailbox_element()->content().get_as<stream_msg>(0);
CAF_ASSERT(holds_alternative<stream_msg::open>(sm.content));
auto& opn = get<stream_msg::open>(sm.content);
policy::arg<Gatherer, Scatterer> policies = {}) {
CAF_IGNORE_UNUSED(policies);
using state_type = typename stream_sink_trait_t<Fun, Finalize>::state;
static_assert(std::is_same<
void (state_type&),
......@@ -557,21 +582,12 @@ public:
>::value,
"Expected signature `void (State&, Input)` "
"for consume function");
auto sid = in.id();
auto next = take_current_next_stage();
using impl = stream_sink_impl<Fun, Finalize, Gatherer>;
auto ptr = make_counted<impl>(this, std::move(fun), std::move(finalize));
auto rp = make_response_promise();
if (!add_source(ptr, sid, std::move(opn.prev_stage),
std::move(opn.original_stage), opn.priority,
opn.redeployable, rp)) {
CAF_LOG_ERROR("cannot create stream stage without source");
rp.deliver(sec::cannot_add_upstream);
return none;
}
init(ptr->state());
streams_.emplace(in.id(), ptr);
return {in.id(), std::move(ptr)};
using impl = stream_sink_impl<Fun, Finalize, Gatherer, Scatterer>;
auto initializer = [&](impl& x) {
init(x.state());
};
return make_sink_impl<impl>(in, initializer, std::move(fun),
std::move(finalize));
}
inline streams_map& streams() {
......
......@@ -152,6 +152,7 @@ public:
using super::remove_path;
bool remove_path(path_uptr_iter i, error reason, bool silent) {
CAF_LOG_TRACE(CAF_ARG(reason) << CAF_ARG(silent));
auto e = paths_.end();
if (i == e) {
CAF_LOG_DEBUG("unable to remove path");
......
......@@ -147,6 +147,13 @@ public:
/// safely.
virtual bool done() const = 0;
// -- implementation hooks for sources ---------------------------------------
/// Tries to generate new messages for the stream. This member function does
/// nothing on stages and sinks, but can trigger a source to produce more
/// messages.
virtual bool generate_messages();
protected:
// -- implementation hooks for sinks -----------------------------------------
......
......@@ -219,8 +219,8 @@ typename Inspector::result_type inspect(Inspector& f, stream_msg::open& x) {
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, stream_msg::ack_open& x) {
return f(meta::type_name("ack_open"), x.rebind_from, x.initial_demand,
x.redeployable);
return f(meta::type_name("ack_open"), x.rebind_from, x.rebind_to,
x.initial_demand, x.redeployable);
}
template <class Inspector>
......@@ -260,7 +260,7 @@ typename Inspector::result_type inspect(Inspector& f,
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, stream_msg& x) {
return f(meta::type_name("stream_msg"), x.sid, x.content);
return f(meta::type_name("stream_msg"), x.sid, x.sender, x.content);
}
} // namespace caf
......
......@@ -42,28 +42,55 @@ public:
~stream_scatterer_impl() override;
// -- convenience functions for children classes -----------------------------
/// Removes all paths gracefully.
void close() override;
// -- static utility functions for operating on paths ------------------------
/// Removes all paths with an error message.
void abort(error reason) override;
/// Returns the total number (sum) of all credit on all paths.
/// Folds `paths()` by extracting the `open_credit` from each path.
template <class PathContainer, class F>
static long fold_credit(const PathContainer& xs, long x0, F f) {
auto g = [f](long x, const typename PathContainer::value_type& y) {
return f(x, y->open_credit);
};
return super::fold(xs, x0, std::move(g));
}
/// Returns the total number (sum) of all credit in `xs`.
template <class PathContainer>
static long total_credit(const PathContainer& xs) {
return fold_credit(xs, 0l, [](long x, long y) { return x + y; });
}
/// Returns the minimum number of credit in `xs`.
template <class PathContainer>
static long min_credit(const PathContainer& xs) {
return !xs.empty()
? fold_credit(xs, std::numeric_limits<long>::max(),
[](long x, long y) { return std::min(x, y); })
: 0;
}
template <class PathContainer>
static long max_credit(const PathContainer& xs) {
return fold_credit(xs, 0l, [](long x, long y) { return std::max(x, y); });
}
// -- convenience functions for children classes -----------------------------
/// Returns the total number (sum) of all credit in `paths()`.
long total_credit() const;
/// Returns the minimum number of credit on all paths.
/// Returns the minimum number of credit in `paths()`.
long min_credit() const;
/// Returns the maximum number of credit on all paths.
/// Returns the maximum number of credit in `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 ---------------------------------------------------
void close() override;
path_ptr add_path(const stream_id& sid, strong_actor_ptr origin,
strong_actor_ptr sink_ptr,
mailbox_element::forwarding_stack stages,
......
......@@ -25,11 +25,10 @@
#include "caf/message_id.hpp"
#include "caf/stream_manager.hpp"
#include "caf/stream_sink_trait.hpp"
#include "caf/terminal_stream_scatterer.hpp"
namespace caf {
template <class Fun, class Finalize, class UpstreamPolicy>
template <class Fun, class Finalize, class Gatherer, class Scatterer>
class stream_sink_impl : public stream_manager {
public:
using super = stream_manager;
......@@ -53,11 +52,11 @@ public:
return state_;
}
UpstreamPolicy& in() override {
Gatherer& in() override {
return in_;
}
terminal_stream_scatterer& out() override {
Scatterer& out() override {
return out_;
}
......@@ -87,8 +86,8 @@ private:
state_type state_;
Fun fun_;
Finalize fin_;
UpstreamPolicy in_;
terminal_stream_scatterer out_;
Gatherer in_;
Scatterer out_;
};
} // namespace caf
......
......@@ -45,12 +45,6 @@ public:
// nop
}
void generate(size_t num) {
CAF_LOG_TRACE(CAF_ARG(num));
downstream<typename DownstreamPolicy::value_type> ds{out_.buf()};
fun_(state_, ds, num);
}
bool done() const override {
return at_end() && out_.paths_clean();
}
......@@ -63,6 +57,16 @@ public:
return out_;
}
bool generate_messages() override {
// produce new elements
auto capacity = out_.credit() - out_.buffered();
if (capacity <= 0)
return false;
downstream<typename DownstreamPolicy::value_type> ds{out_.buf()};
fun_(state_, ds, capacity);
return true;
}
state_type& state() {
return state_;
}
......@@ -75,21 +79,8 @@ protected:
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;
}
generate_messages();
push();
} else if (out_.buffered() > 0) {
push();
} else {
......
......@@ -36,7 +36,7 @@ namespace caf {
/// 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
/// of workers.
template <class T, class Filter, class KeyCompare, long KeyIndex>
template <class T, class Filter, class Select>
class topic_scatterer : public buffered_scatterer<T> {
public:
/// Base type.
......@@ -64,9 +64,11 @@ public:
bool remove_path(const stream_id& sid, const actor_addr& x,
error reason, bool silent) override {
CAF_LOG_TRACE(CAF_ARG(sid) << CAF_ARG(x)
<< CAF_ARG(reason) << CAF_ARG(silent));
auto i = this->iter_find(this->paths_, sid, x);
if (i != this->paths_.end()) {
erase_from_lanes(x);
erase_from_lanes(i->get());
return super::remove_path(i, std::move(reason), silent);
}
return false;
......@@ -81,39 +83,41 @@ public:
/// @pre `x` is not registered on *any* lane
template <class Handle>
void set_filter(const stream_id& sid, const Handle& x, filter_type f) {
std::sort(f.begin(), f.end());
lanes_[std::move(f)].paths.push_back(super::find(sid, x));
}
template <class Handle>
void update_filter(const stream_id& sid, const Handle& x, filter_type f) {
std::sort(f.begin(), f.end());
erase_from_lanes(x);
lanes_[std::move(f)].paths.push_back(super::find(sid, x));
CAF_LOG_TRACE(CAF_ARG(sid) << CAF_ARG(x) << CAF_ARG(f));
auto ptr = super::find(sid, x);
if (!ptr) {
CAF_LOG_WARNING("unable to set filter for unknown path");
return;
}
erase_from_lanes(ptr);
lanes_[std::move(f)].paths.push_back(ptr);
}
const lanes_map& lanes() const {
return lanes_;
}
Select& selector() {
return select_;
}
const Select& selector() const {
return select_;
}
protected:
template <class Handle>
void erase_from_lanes(const Handle& x) {
void erase_from_lanes(typename super::path_ptr ptr) {
for (auto i = lanes_.begin(); i != lanes_.end(); ++i)
if (erase_from_lane(i->second, x)) {
if (erase_from_lane(i->second, ptr)) {
if (i->second.paths.empty())
lanes_.erase(i);
return;
}
}
template <class Handle>
bool erase_from_lane(lane& l, const Handle& x) {
auto predicate = [&](const outbound_path* y) {
return x == y->hdl;
};
bool erase_from_lane(lane& l, typename super::path_ptr ptr) {
auto e = l.paths.end();
auto i = std::find_if(l.paths.begin(), e, predicate);
auto i = std::find(l.paths.begin(), e, ptr);
if (i != e) {
l.paths.erase(i);
return true;
......@@ -132,15 +136,13 @@ protected:
/// Returns `true` if `x` is selected by `f`, `false` otherwise.
bool selected(const filter_type& f, const T& x) {
using std::get;
for (auto& key : f)
if (cmp_(key, get<KeyIndex>(x)))
return true;
if (select_(f, x))
return true;
return false;
}
lanes_map lanes_;
KeyCompare cmp_;
Select select_;
};
} // namespace caf
......
......@@ -36,6 +36,10 @@
namespace caf {
actor::actor(std::nullptr_t) : ptr_(nullptr) {
// nop
}
actor::actor(const scoped_actor& x) : ptr_(actor_cast<strong_actor_ptr>(x)) {
// nop
}
......@@ -52,6 +56,11 @@ actor::actor(actor_control_block* ptr, bool add_ref) : ptr_(ptr, add_ref) {
// nop
}
actor& actor::operator=(std::nullptr_t) {
ptr_.reset();
return *this;
}
actor& actor::operator=(const scoped_actor& x) {
ptr_ = actor_cast<strong_actor_ptr>(x);
return *this;
......
......@@ -99,6 +99,8 @@ actor_system_config::~actor_system_config() {
actor_system_config::actor_system_config()
: cli_helptext_printed(false),
slave_mode(false),
logger_filename(logger_file_name),
logger_filter(logger_component_filter),
slave_mode_fun(nullptr) {
// add `vector<T>` and `stream<T>` for each statically known type
add_message_type_impl<stream<actor>>("stream<@actor>");
......@@ -216,6 +218,71 @@ actor_system_config::actor_system_config()
error_renderers.emplace(atom("exit"), render_exit_reason);
}
actor_system_config::actor_system_config(actor_system_config&& other)
: cli_helptext_printed(other.cli_helptext_printed),
slave_mode(other.slave_mode),
slave_name(std::move(other.slave_name)),
bootstrap_node(std::move(other.bootstrap_node)),
args_remainder(std::move(other.args_remainder)),
scheduler_policy(other.scheduler_policy),
scheduler_max_threads(other.scheduler_max_threads),
scheduler_max_throughput(other.scheduler_max_throughput),
scheduler_enable_profiling(std::move(other.scheduler_enable_profiling)),
scheduler_profiling_ms_resolution(
other.scheduler_profiling_ms_resolution),
scheduler_profiling_output_file(other.scheduler_profiling_output_file),
work_stealing_aggressive_poll_attempts(
other.work_stealing_aggressive_poll_attempts),
work_stealing_aggressive_steal_interval(
other.work_stealing_aggressive_steal_interval),
work_stealing_moderate_poll_attempts(
other.work_stealing_moderate_poll_attempts),
work_stealing_moderate_steal_interval(
other.work_stealing_moderate_steal_interval),
work_stealing_moderate_sleep_duration_us(
other.work_stealing_moderate_sleep_duration_us),
work_stealing_relaxed_steal_interval(
other.work_stealing_relaxed_steal_interval),
work_stealing_relaxed_sleep_duration_us(
other.work_stealing_relaxed_sleep_duration_us),
logger_file_name(std::move(other.logger_file_name)),
logger_file_format(std::move(other.logger_file_format)),
logger_console(other.logger_console),
logger_console_format(std::move(other.logger_console_format)),
logger_component_filter(std::move(other.logger_component_filter)),
logger_verbosity(other.logger_verbosity),
logger_inline_output(other.logger_inline_output),
logger_filename(logger_file_name),
logger_filter(logger_component_filter),
middleman_network_backend(other.middleman_network_backend),
middleman_app_identifier(std::move(other.middleman_app_identifier)),
middleman_enable_automatic_connections(
other.middleman_enable_automatic_connections),
middleman_max_consecutive_reads(other.middleman_max_consecutive_reads),
middleman_heartbeat_interval(other.middleman_heartbeat_interval),
middleman_detach_utility_actors(other.middleman_detach_utility_actors),
middleman_detach_multiplexer(other.middleman_detach_multiplexer),
opencl_device_ids(std::move(other.opencl_device_ids)),
openssl_certificate(std::move(other.openssl_certificate)),
openssl_key(std::move(other.openssl_key)),
openssl_passphrase(std::move(other.openssl_passphrase)),
openssl_capath(std::move(other.openssl_capath)),
openssl_cafile(std::move(other.openssl_cafile)),
value_factories_by_name(std::move(other.value_factories_by_name)),
value_factories_by_rtti(std::move(other.value_factories_by_rtti)),
actor_factories(std::move(other.actor_factories)),
module_factories(std::move(other.module_factories)),
hook_factories(std::move(other.hook_factories)),
group_module_factories(std::move(other.group_module_factories)),
type_names_by_rtti(std::move(other.type_names_by_rtti)),
error_renderers(std::move(other.error_renderers)),
named_actor_configs(std::move(other.named_actor_configs)),
slave_mode_fun(other.slave_mode_fun),
custom_options_(std::move(other.custom_options_)),
options_(std::move(other.options_)) {
// nop
}
std::string
actor_system_config::make_help_text(const std::vector<message::cli_arg>& xs) {
auto is_no_caf_option = [](const message::cli_arg& arg) {
......
......@@ -653,6 +653,10 @@ bool scheduled_actor::handle_stream_msg(mailbox_element& x,
CAF_LOG_TRACE(CAF_ARG(x));
CAF_ASSERT(x.content().match_elements<stream_msg>());
auto& sm = x.content().get_mutable_as<stream_msg>(0);
if (sm.sender == nullptr) {
CAF_LOG_ERROR("received a stream_msg with invalid sender field");
return false;
}
stream_msg_visitor f{this, sm, active_behavior};
auto result = visit(f, sm.content);
if (streams_.empty() && !has_behavior())
......@@ -660,16 +664,24 @@ bool scheduled_actor::handle_stream_msg(mailbox_element& x,
return result;
}
bool scheduled_actor::add_source(const stream_manager_ptr& mgr, const stream_id& sid,
strong_actor_ptr source_ptr, strong_actor_ptr original_stage,
stream_priority prio, bool redeployable,
response_promise result_cb) {
bool scheduled_actor::add_source(const stream_manager_ptr& mgr,
const stream_id& sid,
strong_actor_ptr source_ptr,
strong_actor_ptr original_stage,
stream_priority prio, bool redeployable,
response_promise result_cb) {
CAF_LOG_TRACE(CAF_ARG(mgr) << CAF_ARG(sid) << CAF_ARG(source_ptr)
<< CAF_ARG(original_stage) << CAF_ARG(prio)
<< CAF_ARG(redeployable) << CAF_ARG(result_cb));
CAF_ASSERT(mgr != nullptr);
if (!source_ptr || !sid.valid())
if (!source_ptr) {
CAF_LOG_ERROR("cannot add invalid source");
return false;
}
if (!sid.valid()) {
CAF_LOG_ERROR("cannot add source with invalid stream ID");
return false;
}
return mgr->add_source(sid, std::move(source_ptr),
std::move(original_stage), prio, redeployable,
std::move(result_cb));
......@@ -681,18 +693,22 @@ bool scheduled_actor::add_source(const stream_manager_ptr& mgr,
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>())
if (!current_mailbox_element()->content().match_elements<stream_msg>()) {
CAF_LOG_ERROR("scheduled_actor::add_source called outside "
"a stream_msg handler");
return false;
}
auto& sm = current_mailbox_element()->content().get_mutable_as<stream_msg>(0);
if (!holds_alternative<stream_msg::open>(sm.content))
if (!holds_alternative<stream_msg::open>(sm.content)) {
CAF_LOG_ERROR("scheduled_actor::add_source called outside "
"a stream_msg::open handler");
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));
std::move(opn.original_stage), opn.priority,
opn.redeployable, std::move(result_cb));
}
} // namespace caf
......@@ -189,6 +189,10 @@ void stream_manager::push() {
out().emit_batches();
}
bool stream_manager::generate_messages() {
return false;
}
message stream_manager::make_final_result() {
return none;
}
......
......@@ -98,23 +98,15 @@ void stream_scatterer_impl::abort(error reason) {
}
long stream_scatterer_impl::total_credit() const {
return fold_credit(0l, [](long x, long y) { return x + y; });
return total_credit(paths_);
}
long stream_scatterer_impl::min_credit() const {
return !paths_.empty()
? fold_credit(std::numeric_limits<long>::max(),
[](long x, long y) { return std::min(x, y); })
: 0;
return min_credit(paths_);
}
long stream_scatterer_impl::max_credit() const {
return fold_credit(0l, [](long x, long y) { return std::max(x, y); });
}
long stream_scatterer_impl::fold_credit(long x0, long (*f)(long, long)) const {
return fold(paths_, x0,
[f](long x, const path_uptr& y) { return f(x, y->open_credit); });
return max_credit(paths_);
}
long stream_scatterer_impl::min_batch_size() const {
......
......@@ -29,22 +29,23 @@ terminal_stream_scatterer::~terminal_stream_scatterer() {
stream_scatterer::path_ptr
terminal_stream_scatterer::add_path(const stream_id&, strong_actor_ptr,
strong_actor_ptr,
mailbox_element::forwarding_stack,
message_id, message, stream_priority, bool) {
strong_actor_ptr,
mailbox_element::forwarding_stack,
message_id, message, stream_priority,
bool) {
CAF_LOG_ERROR("terminal_stream_scatterer::add_path called");
return nullptr;
}
stream_scatterer::path_ptr
terminal_stream_scatterer::confirm_path(const stream_id&, const actor_addr&,
strong_actor_ptr, long, bool) {
strong_actor_ptr, long, bool) {
CAF_LOG_ERROR("terminal_stream_scatterer::confirm_path called");
return nullptr;
}
bool terminal_stream_scatterer::remove_path(const stream_id&, const actor_addr&,
error, bool) {
error, bool) {
CAF_LOG_ERROR("terminal_stream_scatterer::remove_path called");
return false;
}
......@@ -85,8 +86,8 @@ void terminal_stream_scatterer::emit_batches() {
// nop
}
stream_scatterer::path_type* terminal_stream_scatterer::find(const stream_id&,
const actor_addr&) {
stream_scatterer::path_type*
terminal_stream_scatterer::find(const stream_id&, const actor_addr&) {
return nullptr;
}
......
......@@ -36,6 +36,7 @@
using std::cout;
using std::endl;
using std::vector;
using std::string;
using namespace caf;
......@@ -66,10 +67,20 @@ struct cleanup_t {
constexpr cleanup_t cleanup_fun = cleanup_t{};
struct selected_t {
template <class T>
bool operator()(const std::vector<string>& filter, const T& x) {
for (auto& prefix : filter)
if (get<0>(x).compare(0, prefix.size(), prefix.c_str()) == 0)
return true;
return false;
}
};
struct stream_splitter_state {
using stage_impl = stream_stage_impl<
process_t, cleanup_t, random_gatherer,
random_topic_scatterer<element_type, std::vector<key_type>>>;
random_topic_scatterer<element_type, std::vector<key_type>, selected_t>>;
intrusive_ptr<stage_impl> stage;
static const char* name;
};
......@@ -137,7 +148,7 @@ behavior storage(stateful_actor<storage_state>* self,
[](unit_t&) {
CAF_LOG_INFO("storage done");
},
policy::arg<detail::pull5_gatherer>::value
policy::arg<detail::pull5_gatherer, terminal_stream_scatterer>::value
);
},
[=](get_atom) {
......@@ -206,7 +217,7 @@ CAF_TEST(fork_setup) {
expect((stream_msg::open),
from(_).to(d1).with(_, splitter, _, _, _, false));
expect((stream_msg::ack_open),
from(d1).to(splitter).with(_, 5, _, false));
from(d1).to(splitter).with(_, _, 5, _, false));
CAF_MESSAGE("spawn second sink");
auto d2 = sys.spawn(storage, splitter, filter_type{"key2"});
sched.run_once();
......@@ -215,7 +226,7 @@ CAF_TEST(fork_setup) {
expect((stream_msg::open),
from(_).to(d2).with(_, splitter, _, _, _, false));
expect((stream_msg::ack_open),
from(d2).to(splitter).with(_, 5, _, false));
from(d2).to(splitter).with(_, _, 5, _, false));
CAF_MESSAGE("spawn source");
auto src = sys.spawn(nores_streamer, splitter);
sched.run_once();
......@@ -223,7 +234,7 @@ CAF_TEST(fork_setup) {
expect((stream_msg::open),
from(_).to(splitter).with(_, src, _, _, _, false));
expect((stream_msg::ack_open),
from(splitter).to(src).with(_, 5, _, false));
from(splitter).to(src).with(_, _, 5, _, false));
// First batch.
expect((stream_msg::batch),
from(src).to(splitter)
......@@ -281,7 +292,7 @@ CAF_TEST(fork_setup) {
expect((stream_msg::open),
from(_).to(splitter).with(_, src2, _, _, _, false));
expect((stream_msg::ack_open),
from(splitter).to(src2).with(_, 5, _, false));
from(splitter).to(src2).with(_, _, 5, _, false));
// First batch.
expect((stream_msg::batch),
from(src2).to(splitter)
......
This diff is collapsed.
......@@ -183,7 +183,7 @@ behavior sum_up(stateful_actor<sum_up_state>* self) {
[](int& x) -> int {
return x;
},
policy::arg<detail::pull5_gatherer>::value
policy::arg<detail::pull5_gatherer, terminal_stream_scatterer>::value
);
}
};
......@@ -214,7 +214,7 @@ behavior drop_all(stateful_actor<drop_all_state>* self) {
[](unit_t&) {
CAF_LOG_INFO("drop_all done");
},
policy::arg<detail::pull5_gatherer>::value
policy::arg<detail::pull5_gatherer, terminal_stream_scatterer>::value
);
}
};
......@@ -380,7 +380,7 @@ CAF_TEST(depth2_pipeline) {
expect((stream_msg::open),
from(self).to(sink).with(_, source, _, _, _, _, false));
// source <----(stream_msg::ack_open)------ sink
expect((stream_msg::ack_open), from(sink).to(source).with(_, 5, _, false));
expect((stream_msg::ack_open), from(sink).to(source).with(_, _, 5, _, false));
// source ----(stream_msg::batch)---> sink
expect((stream_msg::batch),
from(source).to(sink).with(5, std::vector<int>{1, 2, 3, 4, 5}, 0));
......@@ -429,9 +429,10 @@ CAF_TEST(depth3_pipeline_order1) {
CAF_CHECK(!deref(stage).streams().empty());
CAF_CHECK(!deref(sink).streams().empty());
// sink --(stream_msg::ack_open)--> stage
expect((stream_msg::ack_open), from(sink).to(stage).with(_, 5, _, false));
expect((stream_msg::ack_open), from(sink).to(stage).with(_, _, 5, _, false));
// stage --(stream_msg::ack_open)--> source
expect((stream_msg::ack_open), from(stage).to(source).with(_, 5, _, false));
expect((stream_msg::ack_open),
from(stage).to(source).with(_, _, 5, _, false));
// source --(stream_msg::batch)--> stage
expect((stream_msg::batch),
from(source).to(stage).with(5, std::vector<int>{1, 2, 3, 4, 5}, 0));
......@@ -483,7 +484,8 @@ CAF_TEST(depth3_pipeline_order2) {
expect((stream_msg::open),
from(self).to(stage).with(_, source, _, _, _, false));
// stage --(stream_msg::ack_open)--> source
expect((stream_msg::ack_open), from(stage).to(source).with(_, 5, _, false));
expect((stream_msg::ack_open),
from(stage).to(source).with(_, _, 5, _, false));
// source --(stream_msg::batch)--> stage
expect((stream_msg::batch),
from(source).to(stage).with(5, std::vector<int>{1, 2, 3, 4, 5}, 0));
......@@ -516,7 +518,7 @@ CAF_TEST(depth3_pipeline_order2) {
expect((stream_msg::open),
from(self).to(sink).with(_, stage, _, _, _, false));
// sink --(stream_msg::ack_open)--> stage (finally)
expect((stream_msg::ack_open), from(sink).to(stage).with(_, 5, _, false));
expect((stream_msg::ack_open), from(sink).to(stage).with(_, _, 5, _, false));
// stage --(stream_msg::ack_batch)--> source
// The stage has now emptied its buffer and is able to grant more credit.
expect((stream_msg::ack_batch), from(stage).to(source).with(5, 3));
......@@ -570,7 +572,7 @@ CAF_TEST(depth2_pipeline_streamer) {
expect((stream_msg::open),
from(source).to(sink).with(_, source, _, _, _, false));
// source <----(stream_msg::ack_open)------ sink
expect((stream_msg::ack_open), from(sink).to(source).with(_, 5, _, false));
expect((stream_msg::ack_open), from(sink).to(source).with(_, _, 5, _, false));
// source ----(stream_msg::batch)---> sink
expect((stream_msg::batch),
from(source).to(sink).with(5, std::vector<int>{1, 2, 3, 4, 5}, 0));
......@@ -598,7 +600,7 @@ CAF_TEST(stream_without_result) {
expect((stream_msg::open),
from(source).to(sink).with(_, source, _, _, _, false));
// source <----(stream_msg::ack_open)------ sink
expect((stream_msg::ack_open), from(sink).to(source).with(_, 5, _, false));
expect((stream_msg::ack_open), from(sink).to(source).with(_, _, 5, _, false));
// source ----(stream_msg::batch)---> sink
expect((stream_msg::batch),
from(source).to(sink).with(5, std::vector<int>{1, 2, 3, 4, 5}, 0));
......@@ -629,7 +631,7 @@ CAF_TEST(multiplexed_pipeline) {
expect((stream_msg::open),
from(_).to(d1).with(_, multiplexer, _, _, _, false));
expect((stream_msg::ack_open),
from(d1).to(multiplexer).with(_, 5, _, false));
from(d1).to(multiplexer).with(_, _, 5, _, false));
CAF_MESSAGE("spawn second sink");
auto d2 = sys.spawn(joining_drop_all);
sched.run_once();
......@@ -637,7 +639,7 @@ CAF_TEST(multiplexed_pipeline) {
expect((stream_msg::open),
from(_).to(d2).with(_, multiplexer, _, _, _, false));
expect((stream_msg::ack_open),
from(d2).to(multiplexer).with(_, 5, _, false));
from(d2).to(multiplexer).with(_, _, 5, _, false));
CAF_MESSAGE("spawn source");
auto src = sys.spawn(nores_streamer, multiplexer);
sched.run_once();
......@@ -645,7 +647,7 @@ CAF_TEST(multiplexed_pipeline) {
expect((stream_msg::open),
from(_).to(multiplexer).with(_, src, _, _, _, false));
expect((stream_msg::ack_open),
from(multiplexer).to(src).with(_, 5, _, false));
from(multiplexer).to(src).with(_, _, 5, _, false));
// First batch.
expect((stream_msg::batch),
from(src).to(multiplexer).with(5, std::vector<int>{1, 2, 3, 4, 5}, 0));
......@@ -676,7 +678,7 @@ CAF_TEST(multiplexed_pipeline) {
expect((stream_msg::open),
from(_).to(multiplexer).with(_, src2, _, _, _, false));
expect((stream_msg::ack_open),
from(multiplexer).to(src2).with(_, 5, _, false));
from(multiplexer).to(src2).with(_, _, 5, _, false));
// First batch.
expect((stream_msg::batch),
from(src2).to(multiplexer).with(5, std::vector<int>{1, 2, 3, 4, 5}, 0));
......
......@@ -58,7 +58,7 @@ namespace caf {
template <class... Ts>
bool operator==(const message& x, const std::tuple<Ts...>& y) {
return msg_cmp_rec<0>(x, y);
return x.size() == sizeof...(Ts) && msg_cmp_rec<0>(x, y);
}
template <class T>
......
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