Commit 5f3e6cf6 authored by Dominik Charousset's avatar Dominik Charousset

Overhaul streaming policies design

parent 87c8ab7c
...@@ -15,9 +15,7 @@ set (LIBCAF_CORE_SRCS ...@@ -15,9 +15,7 @@ set (LIBCAF_CORE_SRCS
src/abstract_channel.cpp src/abstract_channel.cpp
src/abstract_composable_behavior.cpp src/abstract_composable_behavior.cpp
src/abstract_coordinator.cpp src/abstract_coordinator.cpp
src/abstract_downstream.cpp
src/abstract_group.cpp src/abstract_group.cpp
src/abstract_upstream.cpp
src/actor.cpp src/actor.cpp
src/actor_addr.cpp src/actor_addr.cpp
src/actor_companion.cpp src/actor_companion.cpp
...@@ -29,7 +27,6 @@ set (LIBCAF_CORE_SRCS ...@@ -29,7 +27,6 @@ set (LIBCAF_CORE_SRCS
src/actor_registry.cpp src/actor_registry.cpp
src/actor_system.cpp src/actor_system.cpp
src/actor_system_config.cpp src/actor_system_config.cpp
src/anycast.cpp
src/atom.cpp src/atom.cpp
src/attachable.cpp src/attachable.cpp
src/behavior.cpp src/behavior.cpp
...@@ -37,7 +34,6 @@ set (LIBCAF_CORE_SRCS ...@@ -37,7 +34,6 @@ set (LIBCAF_CORE_SRCS
src/behavior_stack.cpp src/behavior_stack.cpp
src/blocking_actor.cpp src/blocking_actor.cpp
src/blocking_behavior.cpp src/blocking_behavior.cpp
src/broadcast.cpp
src/concatenated_tuple.cpp src/concatenated_tuple.cpp
src/config_option.cpp src/config_option.cpp
src/decorated_tuple.cpp src/decorated_tuple.cpp
...@@ -78,6 +74,7 @@ set (LIBCAF_CORE_SRCS ...@@ -78,6 +74,7 @@ set (LIBCAF_CORE_SRCS
src/parse_ini.cpp src/parse_ini.cpp
src/private_thread.cpp src/private_thread.cpp
src/proxy_registry.cpp src/proxy_registry.cpp
src/pull5.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
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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_ABSTRACT_DOWNSTREAM_HPP
#define CAF_ABSTRACT_DOWNSTREAM_HPP
#include <set>
#include <vector>
#include <memory>
#include <numeric>
#include <unordered_map>
#include "caf/fwd.hpp"
#include "caf/atom.hpp"
#include "caf/optional.hpp"
#include "caf/stream_id.hpp"
namespace caf {
class abstract_downstream {
public:
using path = downstream_path;
using path_cref = const path&;
/// A unique pointer to a downstream path.
using path_uptr = std::unique_ptr<path>;
/// A raw pointer to a downstream path.
using path_ptr = path*;
/// Stores all available paths.
using path_list = std::vector<path_uptr>;
/// List of views to paths.
using path_ptr_list = std::vector<path_ptr>;
using policy_ptr = std::unique_ptr<downstream_policy>;
abstract_downstream(local_actor* selfptr, const stream_id& sid,
policy_ptr ptr);
virtual ~abstract_downstream();
/// Returns the total available credit for all sinks in `xs` in O(n).
template <class PathContainer>
static long total_credit(const PathContainer& xs) {
return fold(xs, 0u,
[=](long x, const typename PathContainer::value_type& y) {
return x + y->open_credit;
});
}
/// Returns the total available credit for all sinks in `paths_` in O(n).
long total_credit() const;
/// Returns the maximum credit of all sinks in `paths_` in O(n).
template <class PathContainer>
static long max_credit(const PathContainer& xs) {
return fold(xs, 0,
[](long x, const typename PathContainer::value_type& y) {
return std::max(x, y->open_credit);
});
}
/// Returns the maximum credit of all sinks in `paths_` in O(n).
long max_credit() const;
/// Returns the minimal credit of all sinks in `xs` in O(n).
template <class PathContainer>
static long min_credit(const PathContainer& xs) {
return fold(xs, std::numeric_limits<long>::max(),
[](long x, const typename PathContainer::value_type& y) {
return std::min(x, y->open_credit);
});
}
/// Returns the minimal net credit of all sinks in `paths_` in O(n).
long min_credit() const;
/// Broadcasts the first `*hint` elements of the buffer on all paths. If
/// `hint == nullptr` then `min_credit()` is used instead.
virtual void broadcast(long* hint = nullptr) = 0;
/// Sends `*hint` elements of the buffer to available paths. If
/// `hint == nullptr` then `total_credit()` is used instead.
virtual void anycast(long* hint = nullptr) = 0;
/// Returns the size of the output buffer.
virtual long buf_size() const = 0;
/// Adds a path with in-flight `stream_msg::open` message.
bool add_path(strong_actor_ptr ptr);
/// Confirms a path and properly initialize its state.
bool confirm_path(const strong_actor_ptr& rebind_from, strong_actor_ptr& ptr,
bool is_redeployable);
/// Removes a downstream path without aborting the stream.
bool remove_path(strong_actor_ptr& ptr);
/// Removes all paths.
void close();
/// Sends an abort message to all paths and closes the stream.
void abort(strong_actor_ptr& cause, const error& reason);
template <class PathContainer>
static path* find(PathContainer& xs, const strong_actor_ptr& ptr) {
auto predicate = [&](const typename PathContainer::value_type& y) {
return y->hdl == ptr;
};
auto e = xs.end();
auto i = std::find_if(xs.begin(), e, predicate);
if (i != e)
return &(*(*i)); // Ugly, but works for both raw and smart pointers.
return nullptr;
}
path* find(const strong_actor_ptr& ptr) const;
long total_net_credit() const;
virtual long num_paths() const;
inline local_actor* self() const {
return self_;
}
inline downstream_policy& policy() const {
return *policy_;
}
inline const stream_id& sid() const {
return sid_;
}
inline bool closed() const {
return paths_.empty();
}
/// Returns how many items should be stored on individual paths in order to
/// minimize latency between received demand and sent batches.
inline long min_buffer_size() const {
return min_buffer_size_;
}
/// Returns all currently available paths on this downstream.
const path_list& paths() const {
return paths_;
}
protected:
void send_batch(downstream_path& dest, long chunk_size, message chunk);
/// Sorts `xs` in descending order by available credit.
template <class PathContainer>
static void sort_by_credit(PathContainer& xs) {
using value_type = typename PathContainer::value_type;
auto cmp = [](const value_type& x, const value_type& y) {
return x->open_credit > y->open_credit;
};
std::sort(xs.begin(), xs.end(), cmp);
}
/// Sorts `paths_` in descending order by available credit.
void sort_by_credit();
template <class T, class PathContainer, class F>
static T fold(PathContainer& xs, T init, F f) {
auto b = xs.begin();
auto e = xs.end();
return b != e ? std::accumulate(b, e, init, f) : 0;
}
local_actor* self_;
stream_id sid_;
path_list paths_;
long min_buffer_size_;
std::unique_ptr<downstream_policy> policy_;
};
} // namespace caf
#endif // CAF_ABSTRACT_DOWNSTREAM_HPP
...@@ -26,93 +26,33 @@ ...@@ -26,93 +26,33 @@
#include "caf/stream_msg.hpp" #include "caf/stream_msg.hpp"
#include "caf/make_message.hpp" #include "caf/make_message.hpp"
#include "caf/downstream_path.hpp" #include "caf/downstream_path.hpp"
#include "caf/abstract_downstream.hpp"
namespace caf { namespace caf {
/// Grants access to an output stream buffer.
template <class T> template <class T>
class downstream : public abstract_downstream { class downstream {
public: public:
/// A chunk of data for sending batches downstream. // -- member types -----------------------------------------------------------
using chunk_type = std::vector<T>;
/// A queue of items for temporary storage before moving them into chunks. /// A queue of items for temporary storage before moving them into chunks.
using queue_type = std::deque<T>; using queue_type = std::deque<T>;
downstream(local_actor* ptr, const stream_id& sid, // -- constructors, destructors, and assignment operators --------------------
typename abstract_downstream::policy_ptr pptr)
: abstract_downstream(ptr, sid, std::move(pptr)) { downstream(queue_type& q) : buf_(q) {
// nop // nop
} }
// -- queue access -----------------------------------------------------------
template <class... Ts> template <class... Ts>
void push(Ts&&... xs) { void push(Ts&&... xs) {
buf_.emplace_back(std::forward<Ts>(xs)...); buf_.emplace_back(std::forward<Ts>(xs)...);
} }
queue_type& buf() {
return buf_;
}
const queue_type& buf() const {
return buf_;
}
void broadcast(long* hint) override {
auto chunk = get_chunk(hint ? *hint : min_credit());
auto csize = static_cast<long>(chunk.size());
if (csize == 0)
return;
auto wrapped_chunk = make_message(std::move(chunk));
for (auto& x : paths_) {
x->open_credit -= csize;
send_batch(*x, csize, wrapped_chunk);
}
}
void anycast(long*) override {
this->sort_by_credit();
for (auto& x : paths_) {
auto chunk = get_chunk(x->open_credit);
auto csize = chunk.size();
if (csize == 0)
return;
x->open_credit -= csize;
send_batch(*x, static_cast<long>(csize),
std::move(make_message(std::move(chunk))));
}
}
long buf_size() const override {
return static_cast<long>(buf_.size());
}
protected: protected:
/// @pre `n <= buf_.size()` queue_type& buf_;
static chunk_type get_chunk(queue_type& buf, long n) {
CAF_ASSERT(n >= 0);
chunk_type xs;
if (n > 0) {
xs.reserve(static_cast<size_t>(n));
if (static_cast<size_t>(n) < buf.size()) {
auto first = buf.begin();
auto last = first + static_cast<ptrdiff_t>(n);
std::move(first, last, std::back_inserter(xs));
buf.erase(first, last);
} else {
std::move(buf.begin(), buf.end(), std::back_inserter(xs));
buf.clear();
}
}
return xs;
}
/// @pre `n <= buf_.size()`
chunk_type get_chunk(long n) {
return get_chunk(buf_, n);
}
queue_type buf_;
}; };
} // namespace caf } // namespace caf
......
...@@ -23,26 +23,230 @@ ...@@ -23,26 +23,230 @@
#include <cstddef> #include <cstddef>
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/duration.hpp"
#include "caf/downstream_path.hpp"
namespace caf { namespace caf {
/// Implements dispatching to downstream paths. /// Type-erased policy for dispatching to downstream paths.
class downstream_policy { class downstream_policy {
public: public:
// -- member types -----------------------------------------------------------
/// A raw pointer to a downstream path.
using path_ptr = downstream_path*;
/// A unique pointer to a downstream path.
using path_uptr = std::unique_ptr<downstream_path>;
/// Stores all available paths.
using path_uptr_list = std::vector<path_uptr>;
/// List of views to paths.
using path_ptr_list = std::vector<path_ptr>;
// -- constructors, destructors, and assignment operators --------------------
downstream_policy(local_actor* selfptr, const stream_id& id);
virtual ~downstream_policy(); virtual ~downstream_policy();
/// Returns an accumulated value of all individual net credits. For example, // -- communication to downstream actors -------------------------------------
/// a broadcast policy would return the minimum value.
virtual long total_net_credit(const abstract_downstream& x) = 0; /// Tries sending batches to downstream actors.
/// @param hint Optionally passes the last result of `total_credit` to
/// avoid recalculation.
virtual void emit_batches() = 0;
// -- feedback to upstream policy --------------------------------------------
/// 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;
// -- static utility functions for path_uptr_list and path_ptr_list access ---
/// Sorts `xs` in descending order by available credit.
template <class PathContainer>
static void sort_by_credit(PathContainer& xs) {
using value_type = typename PathContainer::value_type;
auto cmp = [](const value_type& x, const value_type& y) {
return x->open_credit > y->open_credit;
};
std::sort(xs.begin(), xs.end(), cmp);
}
template <class T, class PathContainer, class F>
static T fold(PathContainer& xs, T init, F f) {
auto b = xs.begin();
auto e = xs.end();
return b != e ? std::accumulate(b, e, init, f) : 0;
}
/// Returns the total available credit for all sinks in `xs` in O(n).
template <class PathContainer>
static long total_credit(const PathContainer& xs) {
return fold(xs, 0u,
[=](long x, const typename PathContainer::value_type& y) {
return x + y->open_credit;
});
}
/// Returns the maximum credit of all sinks in `paths_` in O(n).
template <class PathContainer>
static long max_credit(const PathContainer& xs) {
return fold(xs, 0,
[](long x, const typename PathContainer::value_type& y) {
return std::max(x, y->open_credit);
});
}
/// Returns the minimal credit of all sinks in `xs` in O(n).
template <class PathContainer>
static long min_credit(const PathContainer& xs) {
return fold(xs, std::numeric_limits<long>::max(),
[](long x, const typename PathContainer::value_type& y) {
return std::min(x, y->open_credit);
});
}
template <class PathContainer>
static downstream_path* find(PathContainer& xs, const strong_actor_ptr& ptr) {
auto predicate = [&](const typename PathContainer::value_type& y) {
return y->hdl == ptr;
};
auto e = xs.end();
auto i = std::find_if(xs.begin(), e, predicate);
if (i != e)
return &(*(*i)); // Ugly, but works for both raw and smart pointers.
return nullptr;
}
// -- credit observers -------------------------------------------------------
/// Returns the sum of all credits of all sinks in `paths_` in O(n).
long total_credit() const;
/// Returns the maximum credit of all sinks in `paths_` in O(n).
long max_credit() const;
/// Returns the minimal net credit of all sinks in `paths_` in O(n).
long min_credit() const;
// -- type-erased access to the buffer ---------------------------------------
/// Returns the size of the output buffer.
virtual long buf_size() const = 0;
// -- path management --------------------------------------------------------
/// Adds a path with in-flight `stream_msg::open` message.
bool add_path(strong_actor_ptr ptr);
/// Confirms a path and properly initialize its state.
bool confirm_path(const strong_actor_ptr& rebind_from, strong_actor_ptr& ptr,
bool is_redeployable);
/// Removes a downstream path without aborting the stream.
bool remove_path(strong_actor_ptr& ptr);
/// Returns the state for `ptr.
downstream_path* find(const strong_actor_ptr& ptr) const;
/// Removes all paths.
void close();
/// Sends an abort message to all downstream actors and closes the stream.
void abort(strong_actor_ptr& cause, const error& reason);
/// Returns `true` if no downstream exists, `false` otherwise.
inline bool closed() const {
return paths_.empty();
}
inline long num_paths() const {
return static_cast<long>(paths_.size());
}
inline const path_uptr_list& paths() const {
return paths_;
}
// -- required stream state --------------------------------------------------
inline local_actor* self() const {
return self_;
}
const stream_id& sid() const {
return sid_;
}
// -- configuration parameters -----------------------------------------------
/// Minimum amount of messages required to emit a batch. A value of 0
/// disables batch delays.
inline long min_batch_size() const {
return min_batch_size_;
}
/// Maximum amount of messages to put into a single batch. Causes the actor
/// to split a buffer into more batches if necessary.
inline long max_batch_size() const {
return max_batch_size_;
}
/// Minimum amount of messages we wish to store at the actor in order to emit
/// new batches immediately when receiving new downstream demand.
inline long min_buffer_size() const {
return min_buffer_size_;
}
/// Forces to actor to emit a batch even if the minimum batch size was not
/// reached.
inline duration max_batch_delay() const {
return max_batch_delay_;
}
protected:
// -- virtually dispatched implementation details ----------------------------
/// Broadcasts the first `*hint` elements of the buffer on all paths. If
/// `hint == nullptr` then `min_credit()` is used instead.
virtual void emit_broadcast() = 0;
/// Sends `*hint` elements of the buffer to available paths. If
/// `hint == nullptr` then `total_credit()` is used instead.
virtual void emit_anycast() = 0;
// -- utility functions for derived classes ----------------------------------
/// Sorts `paths_` according to the open downstream credit.
void sort_paths_by_credit();
/// Emits the type-erased batch `xs` to `dest`.
void emit_batch(downstream_path& dest, size_t xs_size, message xs);
// -- required stream state --------------------------------------------------
local_actor* self_;
stream_id sid_;
// -- configuration parameters -----------------------------------------------
long min_batch_size_;
long max_batch_size_;
long min_buffer_size_;
/// Pushes data to the downstream paths of `x`, optionally passing the last duration max_batch_delay_;
/// result of `available_credit` for `x` as second argument.
virtual void push(abstract_downstream& x, long* hint = nullptr) = 0;
// TODO: callbacks für new and closed downstream pahts // -- path management --------------------------------------------------------
/// Reclaim credit of closed downstream `y`. path_uptr_list paths_;
//virtual void reclaim(abstract_downstream& x, downstream_path& y) = 0;
}; };
} // namespace caf } // namespace caf
......
...@@ -26,7 +26,11 @@ ...@@ -26,7 +26,11 @@
#include <vector> #include <vector>
#include <functional> #include <functional>
#include "caf/downstream.hpp" #include "caf/downstream_policy.hpp"
#include "caf/mixin/buffered_policy.hpp"
#include "caf/policy/broadcast.hpp"
#include "caf/meta/type_name.hpp" #include "caf/meta/type_name.hpp"
...@@ -38,18 +42,19 @@ namespace caf { ...@@ -38,18 +42,19 @@ namespace caf {
/// of workers in order to handle only a subset of the overall data on each /// of workers in order to handle only a subset of the overall data on each
/// lane. /// lane.
template <class T, class Key, class KeyCompare = std::equal_to<Key>, template <class T, class Key, class KeyCompare = std::equal_to<Key>,
long KeyIndex = 0> long KeyIndex = 0,
class filtering_downstream : public downstream<T> { class Base = policy::broadcast<T>>
class filtering_downstream : public Base {
public: public:
/// Base type. /// Base type.
using super = downstream<T>; using super = Base;
struct lane { struct lane {
typename super::queue_type queue; typename super::buffer_type buf;
typename super::path_ptr_list paths; typename super::path_ptr_list 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.queue, x.paths); return f(meta::type_name("lane"), x.buf, x.paths);
} }
}; };
...@@ -59,48 +64,46 @@ public: ...@@ -59,48 +64,46 @@ public:
using lanes_map = std::map<filter, lane>; using lanes_map = std::map<filter, lane>;
filtering_downstream(local_actor* ptr, const stream_id& sid, filtering_downstream(local_actor* selfptr, const stream_id& sid)
typename abstract_downstream::policy_ptr pptr) : super(selfptr, sid) {
: super(ptr, sid, std::move(pptr)) {
// nop // nop
} }
void broadcast(long* hint) override { void emit_broadcast() override {
fan_out(); fan_out();
for (auto& kvp : lanes_) { for (auto& kvp : lanes_) {
auto& l = kvp.second; auto& l = kvp.second;
auto chunk = super::get_chunk(l.queue, auto chunk = super::get_chunk(l.buf, super::min_credit(l.paths));
hint ? *hint : super::min_credit(l.paths));
auto csize = static_cast<long>(chunk.size()); auto csize = static_cast<long>(chunk.size());
if (csize == 0) if (csize == 0)
continue; continue;
auto wrapped_chunk = make_message(std::move(chunk)); auto wrapped_chunk = make_message(std::move(chunk));
for (auto& x : l.paths) { for (auto& x : l.paths) {
x->open_credit -= csize; x->open_credit -= csize;
super::send_batch(*x, csize, wrapped_chunk); this->emit_batch(*x, csize, wrapped_chunk);
} }
} }
} }
void anycast(long*) override { void emit_anycast() override {
fan_out(); fan_out();
for (auto& kvp : lanes_) { for (auto& kvp : lanes_) {
auto& l = kvp.second; auto& l = kvp.second;
super::sort_by_credit(l.paths); super::sort_by_credit(l.paths);
for (auto& x : l.paths) { for (auto& x : l.paths) {
auto chunk = super::get_chunk(l.queue, x->open_credit); auto chunk = super::get_chunk(l.buf, x->open_credit);
auto csize = static_cast<long>(chunk.size()); auto csize = static_cast<long>(chunk.size());
if (csize == 0) if (csize == 0)
break; break;
x->open_credit -= csize; x->open_credit -= csize;
super::send_batch(*x, csize, std::move(make_message(std::move(chunk)))); this->emit_batch(*x, csize, std::move(make_message(std::move(chunk))));
} }
} }
} }
void add_lane(filter f) { void add_lane(filter f) {
std::sort(f); std::sort(f);
lanes_.emplace(std::move(f), typename super::queue_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.
...@@ -147,7 +150,7 @@ private: ...@@ -147,7 +150,7 @@ private:
for (auto& kvp : lanes_) for (auto& kvp : lanes_)
for (auto& x : this->buf_) for (auto& x : this->buf_)
if (selected(kvp.first, x)) if (selected(kvp.first, x))
kvp.second.queue.push_back(x); kvp.second.buf.push_back(x);
this->buf_.clear(); this->buf_.clear();
} }
......
...@@ -99,13 +99,11 @@ class mailbox_element; ...@@ -99,13 +99,11 @@ class mailbox_element;
class message_handler; class message_handler;
class scheduled_actor; class scheduled_actor;
class response_promise; class response_promise;
class abstract_upstream;
class downstream_policy; 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;
class stream_msg_visitor; class stream_msg_visitor;
class abstract_downstream;
class actor_control_block; class actor_control_block;
class actor_system_config; class actor_system_config;
class uniform_type_info_map; class uniform_type_info_map;
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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_BUFFERED_POLICYS_HPP
#define CAF_MIXIN_BUFFERED_POLICYS_HPP
#include <deque>
#include <vector>
#include <cstddef>
#include <iterator>
#include "caf/sec.hpp"
#include "caf/logger.hpp"
#include "caf/actor_control_block.hpp"
namespace caf {
namespace mixin {
/// Mixin for streams with any number of downstreams. `Subtype` must provide a
/// member function `buf()` returning a queue with `std::deque`-like interface.
template <class T, class Base, class Subtype = Base>
class buffered_policy : public Base {
public:
using value_type = T;
using buffer_type = std::deque<value_type>;
using chunk_type = std::vector<value_type>;
template <class... Ts>
buffered_policy(Ts&&... xs) : Base(std::forward<Ts>(xs)...) {
// nop
}
template <class T0, class... Ts>
void push(T0&& x, Ts&&... xs) {
buf_.emplace_back(std::forward<T0>(x), std::forward<Ts>(xs)...);
}
/// @pre `n <= buf_.size()`
static chunk_type get_chunk(buffer_type& buf, long n) {
chunk_type xs;
if (n > 0) {
xs.reserve(static_cast<size_t>(n));
if (static_cast<size_t>(n) < buf.size()) {
auto first = buf.begin();
auto last = first + static_cast<ptrdiff_t>(n);
std::move(first, last, std::back_inserter(xs));
buf.erase(first, last);
} else {
std::move(buf.begin(), buf.end(), std::back_inserter(xs));
buf.clear();
}
}
return xs;
}
chunk_type get_chunk(long n) {
return get_chunk(buf_, n);
}
long buf_size() const override {
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, 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() {
return buf_;
}
const buffer_type& buf() const {
return buf_;
}
protected:
buffer_type buf_;
};
} // namespace mixin
} // namespace caf
#endif // CAF_MIXIN_BUFFERED_POLICYS_HPP
...@@ -54,10 +54,9 @@ public: ...@@ -54,10 +54,9 @@ public:
return sec::invalid_downstream; return sec::invalid_downstream;
} }
error push(long* hint = nullptr) override { error push() override {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
if (out().buf_size() > 0) out().emit_batches();
out().policy().push(out(), hint);
return none; return none;
} }
...@@ -66,8 +65,8 @@ private: ...@@ -66,8 +65,8 @@ private:
return static_cast<Subtype*>(this); return static_cast<Subtype*>(this);
} }
abstract_downstream& out() { downstream_policy& out() {
return dptr()->out(); return *this->dp();
} }
}; };
......
...@@ -47,7 +47,7 @@ private: ...@@ -47,7 +47,7 @@ private:
return static_cast<Subtype*>(this); return static_cast<Subtype*>(this);
} }
abstract_upstream& in() { upstream_policy& in() {
return dptr()->in(); return dptr()->in();
} }
}; };
......
...@@ -22,13 +22,23 @@ ...@@ -22,13 +22,23 @@
#include "caf/downstream_policy.hpp" #include "caf/downstream_policy.hpp"
#include "caf/mixin/buffered_policy.hpp"
namespace caf { namespace caf {
namespace policy { namespace policy {
class anycast final : public downstream_policy { template <class T, class Base = mixin::buffered_policy<T, downstream_policy>>
class anycast : public Base {
public: public:
long total_net_credit(const abstract_downstream& out) override; void emit_batches() override {
void push(abstract_downstream& out, long* hint) override; this->emit_anycast();
}
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();
}
}; };
} // namespace policy } // namespace policy
......
...@@ -17,22 +17,27 @@ ...@@ -17,22 +17,27 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_UPSTREAM_HPP #ifndef CAF_POLICY_ARG_HPP
#define CAF_UPSTREAM_HPP #define CAF_POLICY_ARG_HPP
#include "caf/abstract_upstream.hpp" #include "caf/downstream_policy.hpp"
#include "caf/mixin/buffered_policy.hpp"
namespace caf { namespace caf {
namespace policy {
template <class T> /// Provides a wrapper to pass policy arguments to functions.
class upstream : public abstract_upstream { template <class... Ts>
struct arg {
public: public:
upstream(local_actor* selfptr, typename abstract_upstream::policy_ptr ptr) static const arg value;
: abstract_upstream(selfptr, std::move(ptr)) {
// nop
}
}; };
template <class... Ts>
const arg<Ts...> arg<Ts...>::value = arg<Ts...>{};
} // namespace policy
} // namespace caf } // namespace caf
#endif // CAF_UPSTREAM_HPP #endif // CAF_POLICY_ARG_HPP
...@@ -22,16 +22,27 @@ ...@@ -22,16 +22,27 @@
#include "caf/downstream_policy.hpp" #include "caf/downstream_policy.hpp"
#include "caf/mixin/buffered_policy.hpp"
namespace caf { namespace caf {
namespace policy { namespace policy {
class broadcast final : public downstream_policy { template <class T, class Base = mixin::buffered_policy<T, downstream_policy>>
class broadcast : public Base {
public: public:
void push(abstract_downstream& out, long* hint) override; template <class... Ts>
long total_net_credit(const abstract_downstream& x) override; broadcast(Ts&&... xs) : Base(std::forward<Ts>(xs)...) {
// nop
}
void emit_batches() override {
this->emit_broadcast();
}
inline static std::unique_ptr<downstream_policy> make() { long credit() const override {
return std::unique_ptr<downstream_policy>(new broadcast); // We receive messages until we have exhausted all downstream credit and
// have filled our buffer to its minimum size.
return this->min_credit() + this->min_buffer_size();
} }
}; };
......
...@@ -26,20 +26,16 @@ namespace caf { ...@@ -26,20 +26,16 @@ namespace caf {
namespace policy { namespace policy {
/// Sends ACKs as early and often as possible. /// Sends ACKs as early and often as possible.
class greedy final : public upstream_policy { class greedy : public upstream_policy {
public: public:
greedy(); template <class... Ts>
greedy(Ts&&... xs) : upstream_policy(std::forward<Ts>(xs)...) {
void assign_credit(assignment_vec& xs, // nop
long total_downstream_net_credit) override; }
long low_watermark;
long high_watermark; ~greedy() override;
inline static std::unique_ptr<upstream_policy> make() { void fill_assignment_vec(long downstream_credit) override;
return std::unique_ptr<upstream_policy>(new greedy);
}
}; };
} // namespace policy } // namespace policy
......
...@@ -17,24 +17,28 @@ ...@@ -17,24 +17,28 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#include "caf/policy/broadcast.hpp" #ifndef CAF_POLICY_PULL5_HPP
#define CAF_POLICY_PULL5_HPP
#include "caf/abstract_downstream.hpp" #include "caf/upstream_policy.hpp"
namespace caf { namespace caf {
namespace policy { namespace policy {
long broadcast::total_net_credit(const abstract_downstream& out) { /// Sends ACKs as early and often as possible.
// The buffer on `out` is shared on all paths. Our total available number of class pull5 : public upstream_policy {
// net credit is thus calculates as `av_min + mb - bs`, where `av_min` is the public:
// minimum of available credits on all paths, `mb` is the minimum buffer template <class... Ts>
// size, and `bs` is the current buffer size. pull5(Ts&&... xs) : upstream_policy(std::forward<Ts>(xs)...) {
return out.min_credit() + out.min_buffer_size() - out.buf_size(); // nop
} }
void broadcast::push(abstract_downstream& out, long* hint) { ~pull5() override;
out.broadcast(hint);
} void fill_assignment_vec(long downstream_credit) override;
};
} // namespace policy } // namespace policy
} // namespace caf } // namespace caf
#endif // CAF_POLICY_PULL5_HPP
...@@ -17,25 +17,36 @@ ...@@ -17,25 +17,36 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#include "caf/policy/anycast.hpp" #ifndef CAF_POLICY_PUSH5_HPP
#define CAF_POLICY_PUSH5_HPP
#include "caf/abstract_downstream.hpp" #include "caf/downstream_policy.hpp"
#include "caf/mixin/buffered_policy.hpp"
namespace caf { namespace caf {
namespace policy { namespace policy {
long anycast::total_net_credit(const abstract_downstream& out) { template <class T, class Base = mixin::buffered_policy<T, downstream_policy>>
// The total amount of available net credit is calculated as: class push5 : public Base {
// `av + (n * mb) - bs`, where `av` is the sum of all available credit on all public:
// paths, `n` is the number of downstream paths, `mb` is the minimum buffer template <class... Ts>
// size, and `bs` is the current buffer size. push5(Ts&&... xs) : Base(std::forward<Ts>(xs)...) {
return (out.total_credit() + (out.num_paths() * out.min_buffer_size())) this->min_batch_size_ = 1;
- out.buf_size(); this->max_batch_size_ = 5;
} this->min_buffer_size_ = 5;
}
void emit_batches() override {
this->emit_broadcast();
}
void anycast::push(abstract_downstream& out, long* hint) { long credit() const override {
out.anycast(hint); return this->min_credit() + this->min_buffer_size();
} }
};
} // namespace policy } // namespace policy
} // namespace caf } // namespace caf
#endif // CAF_POLICY_PUSH5_HPP
...@@ -44,6 +44,7 @@ ...@@ -44,6 +44,7 @@
#include "caf/to_string.hpp" #include "caf/to_string.hpp"
#include "caf/policy/arg.hpp"
#include "caf/policy/greedy.hpp" #include "caf/policy/greedy.hpp"
#include "caf/policy/anycast.hpp" #include "caf/policy/anycast.hpp"
#include "caf/policy/broadcast.hpp" #include "caf/policy/broadcast.hpp"
...@@ -319,11 +320,13 @@ public: ...@@ -319,11 +320,13 @@ public:
// Starts a new stream. // Starts a new stream.
template <class Handle, class... Ts, class Init, class Getter, template <class Handle, class... Ts, class Init, class Getter,
class ClosedPredicate, class ResHandler> class ClosedPredicate, class ResHandler,
class DownstreamPolicy =
policy::broadcast<typename stream_source_trait_t<Getter>::output>>
annotated_stream<typename stream_source_trait_t<Getter>::output, Ts...> annotated_stream<typename stream_source_trait_t<Getter>::output, Ts...>
new_stream(const Handle& dest, std::tuple<Ts...> xs, Init init, Getter getter, new_stream(const Handle& dest, std::tuple<Ts...> xs, Init init, Getter getter,
ClosedPredicate pred, ResHandler res_handler, ClosedPredicate pred, ResHandler res_handler,
downstream_policy_ptr dpolicy = nullptr) { policy::arg<DownstreamPolicy> = {}) {
using type = typename stream_source_trait_t<Getter>::output; using type = typename stream_source_trait_t<Getter>::output;
using state_type = typename stream_source_trait_t<Getter>::state; using state_type = typename stream_source_trait_t<Getter>::state;
static_assert(std::is_same< static_assert(std::is_same<
...@@ -361,11 +364,9 @@ public: ...@@ -361,11 +364,9 @@ public:
res_id.response_id(), res_id.response_id(),
stream_result_trait_t<ResHandler>::make_result_handler(res_handler)); stream_result_trait_t<ResHandler>::make_result_handler(res_handler));
// install stream handler // install stream handler
using impl = stream_source_impl<Getter, ClosedPredicate>; using impl = stream_source_impl<Getter, ClosedPredicate, DownstreamPolicy>;
if (dpolicy == nullptr) auto ptr = make_counted<impl>(this, sid, std::move(getter),
dpolicy.reset(new policy::anycast); std::move(pred));
auto ptr = make_counted<impl>(this, sid, std::move(dpolicy),
std::move(getter), std::move(pred));
init(ptr->state()); init(ptr->state());
// Add downstream with 0 credit. // Add downstream with 0 credit.
// TODO: add multiple downstreams when receiving a composed actor handle // TODO: add multiple downstreams when receiving a composed actor handle
...@@ -377,20 +378,25 @@ public: ...@@ -377,20 +378,25 @@ public:
// Starts a new stream. // Starts a new stream.
template <class Handle, class Init, class Getter, template <class Handle, class Init, class Getter,
class ClosedPredicate, class ResHandler> class ClosedPredicate, class ResHandler,
class DownstreamPolicy =
policy::broadcast<typename stream_source_trait_t<Getter>::output>>
stream<typename stream_source_trait_t<Getter>::output> stream<typename stream_source_trait_t<Getter>::output>
new_stream(const Handle& dest, Init init, Getter getter, new_stream(const Handle& dest, Init init, Getter getter,
ClosedPredicate pred, ResHandler res_handler) { ClosedPredicate pred, ResHandler res_handler,
policy::arg<DownstreamPolicy> parg = {}) {
return new_stream(dest, std::make_tuple(), std::move(init), return new_stream(dest, std::make_tuple(), std::move(init),
std::move(getter), std::move(pred), std::move(getter), std::move(pred),
std::move(res_handler)); std::move(res_handler), parg);
} }
/// Adds a stream source to this actor. /// Adds a stream source to this actor.
template <class Init, class... Ts, class Getter, class ClosedPredicate> template <class Init, class... Ts, class Getter, class ClosedPredicate,
class DownstreamPolicy =
policy::broadcast<typename stream_source_trait_t<Getter>::output>>
annotated_stream<typename stream_source_trait_t<Getter>::output, Ts...> annotated_stream<typename stream_source_trait_t<Getter>::output, Ts...>
add_source(std::tuple<Ts...> xs, Init init, Getter getter, add_source(std::tuple<Ts...> xs, Init init, Getter getter,
ClosedPredicate pred, downstream_policy_ptr dpolicy = nullptr) { ClosedPredicate pred, policy::arg<DownstreamPolicy> = {}) {
CAF_ASSERT(current_mailbox_element() != nullptr); CAF_ASSERT(current_mailbox_element() != nullptr);
using type = typename stream_source_trait_t<Getter>::output; using type = typename stream_source_trait_t<Getter>::output;
using state_type = typename stream_source_trait_t<Getter>::state; using state_type = typename stream_source_trait_t<Getter>::state;
...@@ -418,10 +424,8 @@ public: ...@@ -418,10 +424,8 @@ public:
auto next = stages.back(); auto next = stages.back();
CAF_ASSERT(next != nullptr); CAF_ASSERT(next != nullptr);
fwd_stream_handshake<type>(sid, xs); fwd_stream_handshake<type>(sid, xs);
using impl = stream_source_impl<Getter, ClosedPredicate>; using impl = stream_source_impl<Getter, ClosedPredicate, DownstreamPolicy>;
if (dpolicy == nullptr) auto ptr = make_counted<impl>(this, sid,
dpolicy.reset(new policy::anycast);
auto ptr = make_counted<impl>(this, sid, std::move(dpolicy),
std::move(getter), std::move(pred)); std::move(getter), std::move(pred));
init(ptr->state()); init(ptr->state());
streams_.emplace(sid, ptr); streams_.emplace(sid, ptr);
...@@ -431,19 +435,25 @@ public: ...@@ -431,19 +435,25 @@ public:
return {std::move(sid), std::move(ptr)}; return {std::move(sid), std::move(ptr)};
} }
template <class Init, class Getter, class ClosedPredicate> template <class Init, class Getter, class ClosedPredicate,
class DownstreamPolicy =
policy::broadcast<typename stream_source_trait_t<Getter>::output>>
stream<typename stream_source_trait_t<Getter>::output> stream<typename stream_source_trait_t<Getter>::output>
add_source(Init init, Getter getter, ClosedPredicate pred) { add_source(Init init, Getter getter, ClosedPredicate pred,
policy::arg<DownstreamPolicy> parg = {}) {
return add_source(std::make_tuple(), std::move(init), return add_source(std::make_tuple(), std::move(init),
std::move(getter), std::move(pred)); std::move(getter), std::move(pred), parg);
} }
/// Adds a stream stage to this actor. /// Adds a stream stage to this actor.
template <class In, class... Ts, class Init, class Fun, class Cleanup> template <class In, class... Ts, class Init, class Fun, class Cleanup,
class UpstreamPolicy = policy::greedy,
class DownstreamPolicy =
policy::broadcast<typename stream_stage_trait_t<Fun>::output>>
annotated_stream<typename stream_stage_trait_t<Fun>::output, Ts...> annotated_stream<typename stream_stage_trait_t<Fun>::output, Ts...>
add_stage(const stream<In>& in, std::tuple<Ts...> xs, Init init, Fun fun, add_stage(const stream<In>& in, std::tuple<Ts...> xs, Init init, Fun fun,
Cleanup cleanup_fun, upstream_policy_ptr upolicy = nullptr, Cleanup cleanup_fun,
downstream_policy_ptr dpolicy = nullptr) { policy::arg<UpstreamPolicy, DownstreamPolicy> = {}) {
CAF_ASSERT(current_mailbox_element() != nullptr); CAF_ASSERT(current_mailbox_element() != nullptr);
using output_type = typename stream_stage_trait_t<Fun>::output; using output_type = typename stream_stage_trait_t<Fun>::output;
using state_type = typename stream_stage_trait_t<Fun>::state; using state_type = typename stream_stage_trait_t<Fun>::state;
...@@ -461,13 +471,9 @@ public: ...@@ -461,13 +471,9 @@ public:
auto next = stages.back(); auto next = stages.back();
CAF_ASSERT(next != nullptr); CAF_ASSERT(next != nullptr);
fwd_stream_handshake<output_type>(sid, xs); fwd_stream_handshake<output_type>(sid, xs);
using impl = stream_stage_impl<Fun, Cleanup>; using impl = stream_stage_impl<Fun, Cleanup,
if (upolicy == nullptr) UpstreamPolicy, DownstreamPolicy>;
upolicy.reset(new policy::greedy); auto ptr = make_counted<impl>(this, sid, std::move(fun),
if (dpolicy == nullptr)
dpolicy.reset(new policy::anycast);
auto ptr = make_counted<impl>(this, sid, std::move(upolicy),
std::move(dpolicy), std::move(fun),
std::move(cleanup_fun)); std::move(cleanup_fun));
init(ptr->state()); init(ptr->state());
streams_.emplace(sid, ptr); streams_.emplace(sid, ptr);
...@@ -486,10 +492,11 @@ public: ...@@ -486,10 +492,11 @@ public:
} }
/// Adds a stream sink to this actor. /// Adds a stream sink to this actor.
template <class In, class Init, class Fun, class Finalize> template <class In, class Init, class Fun,
class Finalize, class UpstreamPolicy = policy::greedy>
stream_result<typename stream_sink_trait_t<Fun, Finalize>::output> stream_result<typename stream_sink_trait_t<Fun, Finalize>::output>
add_sink(const stream<In>& in, Init init, Fun fun, Finalize finalize_fun, add_sink(const stream<In>& in, Init init, Fun fun, Finalize finalize_fun,
upstream_policy_ptr upolicy = nullptr) { policy::arg<UpstreamPolicy> = {}) {
CAF_ASSERT(current_mailbox_element() != nullptr); CAF_ASSERT(current_mailbox_element() != nullptr);
//using output_type = typename stream_sink_trait_t<Fun, Finalize>::output; //using output_type = typename stream_sink_trait_t<Fun, Finalize>::output;
using state_type = typename stream_sink_trait_t<Fun, Finalize>::state; using state_type = typename stream_sink_trait_t<Fun, Finalize>::state;
...@@ -509,11 +516,8 @@ public: ...@@ -509,11 +516,8 @@ public:
CAF_LOG_ERROR("add_sink called outside of a message handler"); CAF_LOG_ERROR("add_sink called outside of a message handler");
return {stream_id{nullptr, 0}, nullptr}; return {stream_id{nullptr, 0}, nullptr};
} }
using impl = stream_sink_impl<Fun, Finalize>; using impl = stream_sink_impl<Fun, Finalize, UpstreamPolicy>;
if (upolicy == nullptr) auto ptr = make_counted<impl>(this, std::move(mptr->sender),
upolicy.reset(new policy::greedy);
auto ptr = make_counted<impl>(this, std::move(upolicy),
std::move(mptr->sender),
std::move(mptr->stages), mptr->mid, std::move(mptr->stages), mptr->mid,
std::move(fun), std::move(finalize_fun)); std::move(fun), std::move(finalize_fun));
init(ptr->state()); init(ptr->state());
......
...@@ -60,7 +60,7 @@ public: ...@@ -60,7 +60,7 @@ public:
/// Push new data to downstream actors by sending batches. The amount of /// Push new data to downstream actors by sending batches. The amount of
/// pushed data is limited by `hint` or the available credit if /// pushed data is limited by `hint` or the available credit if
/// `hint == nullptr`. /// `hint == nullptr`.
virtual error push(long* hint = nullptr); virtual error push();
// -- handler for upstream events -------------------------------------------- // -- handler for upstream events --------------------------------------------
...@@ -85,15 +85,15 @@ public: ...@@ -85,15 +85,15 @@ public:
virtual bool done() const = 0; virtual bool done() const = 0;
/// Returns the downstream if this handler is a sink or stage.
virtual optional<abstract_downstream&> get_downstream();
/// Returns the upstream if this handler is a source or stage.
virtual optional<abstract_upstream&> get_upstream();
/// Returns a type-erased `stream<T>` as handshake token for downstream /// Returns a type-erased `stream<T>` as handshake token for downstream
/// actors. Returns an empty message for sinks. /// actors. Returns an empty message for sinks.
virtual message make_output_token(const stream_id&) const; virtual message make_output_token(const stream_id&) const;
/// Returns the downstream policy if this handler is a sink or stage.
virtual optional<downstream_policy&> dp();
/// Returns the upstream policy if this handler is a source or stage.
virtual optional<upstream_policy&> up();
}; };
/// A reference counting pointer to a `stream_handler`. /// A reference counting pointer to a `stream_handler`.
......
...@@ -36,7 +36,7 @@ namespace caf { ...@@ -36,7 +36,7 @@ namespace caf {
class stream_sink : public extend<stream_handler, stream_sink>:: class stream_sink : public extend<stream_handler, stream_sink>::
with<mixin::has_upstreams> { with<mixin::has_upstreams> {
public: public:
stream_sink(abstract_upstream* 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;
...@@ -48,7 +48,7 @@ public: ...@@ -48,7 +48,7 @@ public:
void last_upstream_closed(); void last_upstream_closed();
inline abstract_upstream& in() { inline upstream_policy& in() {
return *in_ptr_; return *in_ptr_;
} }
...@@ -64,7 +64,7 @@ protected: ...@@ -64,7 +64,7 @@ protected:
virtual message finalize() = 0; virtual message finalize() = 0;
private: private:
abstract_upstream* in_ptr_; upstream_policy* in_ptr_;
strong_actor_ptr original_sender_; strong_actor_ptr original_sender_;
std::vector<strong_actor_ptr> next_stages_; std::vector<strong_actor_ptr> next_stages_;
message_id original_msg_id_; message_id original_msg_id_;
......
...@@ -26,7 +26,7 @@ ...@@ -26,7 +26,7 @@
namespace caf { namespace caf {
template <class Fun, class Finalize> template <class Fun, class Finalize, class UpstreamPolicy>
class stream_sink_impl : public stream_sink { class stream_sink_impl : public stream_sink {
public: public:
using trait = stream_sink_trait_t<Fun, Finalize>; using trait = stream_sink_trait_t<Fun, Finalize>;
...@@ -37,14 +37,14 @@ public: ...@@ -37,14 +37,14 @@ public:
using output_type = typename trait::output; using output_type = typename trait::output;
stream_sink_impl(local_actor* self, std::unique_ptr<upstream_policy> policy, stream_sink_impl(local_actor* self,
strong_actor_ptr&& orig_sender, strong_actor_ptr&& orig_sender,
std::vector<strong_actor_ptr>&& svec, message_id mid, std::vector<strong_actor_ptr>&& svec, message_id mid,
Fun fun, Finalize fin) Fun fun, Finalize fin)
: stream_sink(&in_, std::move(orig_sender), std::move(svec), mid), : stream_sink(&in_, std::move(orig_sender), std::move(svec), mid),
fun_(std::move(fun)), fun_(std::move(fun)),
fin_(std::move(fin)), fin_(std::move(fin)),
in_(self, std::move(policy)) { in_(self) {
// nop // nop
} }
...@@ -70,7 +70,7 @@ public: ...@@ -70,7 +70,7 @@ public:
return trait::make_result(state_, fin_); return trait::make_result(state_, fin_);
} }
optional<abstract_upstream&> get_upstream() override { optional<upstream_policy&> up() override {
return in_; return in_;
} }
...@@ -82,7 +82,7 @@ private: ...@@ -82,7 +82,7 @@ private:
state_type state_; state_type state_;
Fun fun_; Fun fun_;
Finalize fin_; Finalize fin_;
upstream<output_type> in_; UpstreamPolicy in_;
}; };
} // namespace caf } // namespace caf
......
...@@ -33,7 +33,7 @@ namespace caf { ...@@ -33,7 +33,7 @@ namespace caf {
class stream_source : public extend<stream_handler, stream_source>:: class stream_source : public extend<stream_handler, stream_source>::
with<mixin::has_downstreams> { with<mixin::has_downstreams> {
public: public:
stream_source(abstract_downstream* out_ptr); stream_source(downstream_policy* out_ptr);
~stream_source() override; ~stream_source() override;
...@@ -43,7 +43,7 @@ public: ...@@ -43,7 +43,7 @@ public:
void abort(strong_actor_ptr& cause, const error& reason) override; void abort(strong_actor_ptr& cause, const error& reason) override;
inline abstract_downstream& out() { inline downstream_policy& out() {
return *out_ptr_; return *out_ptr_;
} }
...@@ -62,7 +62,7 @@ protected: ...@@ -62,7 +62,7 @@ protected:
virtual bool at_end() const = 0; virtual bool at_end() const = 0;
private: private:
abstract_downstream* out_ptr_; downstream_policy* out_ptr_;
}; };
} // namespace caf } // namespace caf
......
...@@ -26,7 +26,7 @@ ...@@ -26,7 +26,7 @@
namespace caf { namespace caf {
template <class Fun, class Predicate> template <class Fun, class Predicate, class DownstreamPolicy>
class stream_source_impl : public stream_source { class stream_source_impl : public stream_source {
public: public:
using trait = stream_source_trait_t<Fun>; using trait = stream_source_trait_t<Fun>;
...@@ -35,30 +35,30 @@ public: ...@@ -35,30 +35,30 @@ public:
using output_type = typename trait::output; using output_type = typename trait::output;
stream_source_impl(local_actor* self, stream_source_impl(local_actor* self, const stream_id& sid,
const stream_id& sid, Fun fun, Predicate pred)
std::unique_ptr<downstream_policy> policy, Fun fun,
Predicate pred)
: stream_source(&out_), : stream_source(&out_),
fun_(std::move(fun)), fun_(std::move(fun)),
pred_(std::move(pred)), pred_(std::move(pred)),
out_(self, sid, std::move(policy)) { out_(self, sid) {
// nop // nop
} }
void generate(size_t num) override { void generate(size_t num) override {
fun_(state_, out_, num); CAF_LOG_TRACE(CAF_ARG(num));
downstream<typename DownstreamPolicy::value_type> ds{out_.buf()};
fun_(state_, ds, num);
} }
long buf_size() const override { long buf_size() const override {
return static_cast<long>(out_.buf().size()); return out_.buf_size();
} }
bool at_end() const override { bool at_end() const override {
return pred_(state_); return pred_(state_);
} }
optional<abstract_downstream&> get_downstream() override { optional<downstream_policy&> dp() override {
return out_; return out_;
} }
...@@ -70,7 +70,7 @@ private: ...@@ -70,7 +70,7 @@ private:
state_type state_; state_type state_;
Fun fun_; Fun fun_;
Predicate pred_; Predicate pred_;
downstream<output_type> out_; DownstreamPolicy out_;
}; };
} // namespace caf } // namespace caf
......
...@@ -35,7 +35,7 @@ class stream_stage : public extend<stream_handler, stream_stage>:: ...@@ -35,7 +35,7 @@ class stream_stage : public extend<stream_handler, stream_stage>::
public: public:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
stream_stage(abstract_upstream* in_ptr, abstract_downstream* out_ptr); stream_stage(upstream_policy* in_ptr, downstream_policy* out_ptr);
// -- overrides -------------------------------------------------------------- // -- overrides --------------------------------------------------------------
...@@ -49,11 +49,11 @@ public: ...@@ -49,11 +49,11 @@ public:
void last_upstream_closed(); void last_upstream_closed();
inline abstract_upstream& in() { inline upstream_policy& in() {
return *in_ptr_; return *in_ptr_;
} }
inline abstract_downstream& out() { inline downstream_policy& out() {
return *out_ptr_; return *out_ptr_;
} }
...@@ -61,8 +61,8 @@ protected: ...@@ -61,8 +61,8 @@ protected:
virtual error process_batch(message& msg) = 0; virtual error process_batch(message& msg) = 0;
private: private:
abstract_upstream* in_ptr_; upstream_policy* in_ptr_;
abstract_downstream* out_ptr_; downstream_policy* out_ptr_;
}; };
} // namespace caf } // namespace caf
......
...@@ -25,17 +25,13 @@ ...@@ -25,17 +25,13 @@
#include "caf/stream_stage.hpp" #include "caf/stream_stage.hpp"
#include "caf/stream_stage_trait.hpp" #include "caf/stream_stage_trait.hpp"
namespace caf { #include "caf/policy/greedy.hpp"
#include "caf/policy/broadcast.hpp"
struct default_stream_stage_policy {
template <class InputType>
using upstream_type = upstream<InputType>;
template <class OutputType> namespace caf {
using downstream_type = downstream<OutputType>;
};
template <class Fun, class Cleanup, class Policy = default_stream_stage_policy> template <class Fun, class Cleanup,
class UpstreamPolicy, class DownstreamPolicy>
class stream_stage_impl : public stream_stage { class stream_stage_impl : public stream_stage {
public: public:
using trait = stream_stage_trait_t<Fun>; using trait = stream_stage_trait_t<Fun>;
...@@ -46,20 +42,14 @@ public: ...@@ -46,20 +42,14 @@ public:
using output_type = typename trait::output; using output_type = typename trait::output;
using upstream_type = typename Policy::template upstream_type<output_type>;
using downstream_type = typename Policy::template downstream_type<output_type>;
stream_stage_impl(local_actor* self, stream_stage_impl(local_actor* self,
const stream_id& sid, const stream_id& sid,
typename abstract_upstream::policy_ptr in_policy,
typename abstract_downstream::policy_ptr out_policy,
Fun fun, Cleanup cleanup) Fun fun, Cleanup cleanup)
: stream_stage(&in_, &out_), : stream_stage(&in_, &out_),
fun_(std::move(fun)), fun_(std::move(fun)),
cleanup_(std::move(cleanup)), cleanup_(std::move(cleanup)),
in_(self, std::move(in_policy)), in_(self),
out_(self, sid, std::move(out_policy)) { out_(self, sid) {
// nop // nop
} }
...@@ -67,7 +57,7 @@ public: ...@@ -67,7 +57,7 @@ public:
stream_priority prio) override { stream_priority prio) override {
CAF_LOG_TRACE(CAF_ARG(ptr) << CAF_ARG(sid) << CAF_ARG(prio)); CAF_LOG_TRACE(CAF_ARG(ptr) << CAF_ARG(sid) << CAF_ARG(prio));
if (ptr) if (ptr)
return in().add_path(ptr, sid, prio, out_.total_net_credit()); return in().add_path(ptr, sid, prio, out_.credit());
return sec::invalid_argument; return sec::invalid_argument;
} }
...@@ -75,8 +65,9 @@ public: ...@@ -75,8 +65,9 @@ public:
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);
downstream<typename DownstreamPolicy::value_type> ds{out_.buf()};
for (auto& x : xs) for (auto& x : xs)
fun_(state_, out_, x); fun_(state_, ds, x);
return none; return none;
} }
return sec::unexpected_message; return sec::unexpected_message;
...@@ -86,11 +77,11 @@ public: ...@@ -86,11 +77,11 @@ public:
return make_message(stream<output_type>{x}); return make_message(stream<output_type>{x});
} }
optional<abstract_downstream&> get_downstream() override { optional<downstream_policy&> dp() override {
return out_; return out_;
} }
optional<abstract_upstream&> get_upstream() override { optional<upstream_policy&> up() override {
return in_; return in_;
} }
...@@ -98,11 +89,11 @@ public: ...@@ -98,11 +89,11 @@ public:
return state_; return state_;
} }
upstream_type& in() { UpstreamPolicy& in() {
return in_; return in_;
} }
downstream_type& out() { DownstreamPolicy& out() {
return out_; return out_;
} }
...@@ -110,8 +101,8 @@ private: ...@@ -110,8 +101,8 @@ private:
state_type state_; state_type state_;
Fun fun_; Fun fun_;
Cleanup cleanup_; Cleanup cleanup_;
upstream_type in_; UpstreamPolicy in_;
downstream_type out_; DownstreamPolicy out_;
}; };
} // namespace caf } // namespace caf
......
...@@ -30,7 +30,19 @@ namespace caf { ...@@ -30,7 +30,19 @@ namespace caf {
class upstream_policy { class upstream_policy {
public: public:
virtual ~upstream_policy(); // -- member types -----------------------------------------------------------
/// A raw pointer to a downstream path.
using path_ptr = upstream_path*;
/// A unique pointer to a upstream path.
using path_uptr = std::unique_ptr<upstream_path>;
/// Stores all available paths.
using path_uptr_list = std::vector<path_uptr>;
/// List of views to paths.
using path_ptr_list = std::vector<path_ptr>;
/// Describes an assignment of credit to an upstream actor. /// Describes an assignment of credit to an upstream actor.
using assignment_pair = std::pair<upstream_path*, long>; using assignment_pair = std::pair<upstream_path*, long>;
...@@ -38,16 +50,109 @@ public: ...@@ -38,16 +50,109 @@ public:
/// Describes an assignment of credit to all upstream actors. /// Describes an assignment of credit to all upstream actors.
using assignment_vec = std::vector<assignment_pair>; using assignment_vec = std::vector<assignment_pair>;
/// Assigns credit to upstream actors. // -- constructors, destructors, and assignment operators --------------------
/// @param xs Stores assignment decisions. Note that the second element of
/// each pair is uninitialized and must be set to 0 for all paths upstream_policy(local_actor* selfptr);
/// that do not receive credit.
/// @param total_downstream_net_credit Denotes how many items we could send virtual ~upstream_policy();
/// downstream. A negative value indicates
/// that we are already buffering more // -- path management --------------------------------------------------------
/// items than we can send.
virtual void assign_credit(assignment_vec& xs, /// Returns `true` if all upstream paths are closed and this upstream is not
long total_downstream_net_credit) = 0; /// 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
/// exists.
inline bool continuous() const {
return continuous_;
}
/// Sets whether this upstream remains open even if no more upstream path
/// exists.
inline void continuous(bool value) {
continuous_ = value;
}
/// Sends an abort message to all upstream actors and closes the stream.
void abort(strong_actor_ptr& cause, const error& reason);
/// Assigns credit to upstream actors according to the current capacity of
/// all downstream actors (and a minimum buffer size) combined.
void assign_credit(long downstream_capacity);
/// Adds a new upstream actor and returns the initial credit.
expected<long> add_path(strong_actor_ptr hdl, const stream_id& sid,
stream_priority prio,
long downstream_credit);
bool remove_path(const strong_actor_ptr& hdl);
upstream_path* find(const strong_actor_ptr& x) const;
// -- required state ---------------------------------------------------------
inline local_actor* self() const {
return self_;
}
// -- configuration parameters -----------------------------------------------
/// Returns the point at which an actor stops sending out demand immediately
/// (waiting for the available credit to first drop below the watermark).
long high_watermark() const {
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.
long min_credit_assignment() const {
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.
long max_credit() const {
return max_credit_;
}
/// Sets the maximum credit assigned to a single upstream actors.
void max_credit(long x) {
max_credit_ = x;
}
protected:
/// Assigns new credit to upstream actors by filling `assignment_vec_`.
virtual void fill_assignment_vec(long downstream_credit) = 0;
/// Pointer to the parent actor.
local_actor* self_;
/// List of all known paths.
path_uptr_list paths_;
/// An assignment vector that's re-used whenever calling the policy.
assignment_vec assignment_vec_;
/// Stores whether this stream remains open even if all paths have been
/// closed.
bool continuous_;
long high_watermark_;
long min_credit_assignment_;
long max_credit_;
}; };
} // 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. *
******************************************************************************/
#include <utility>
#include "caf/abstract_downstream.hpp"
#include "caf/send.hpp"
#include "caf/downstream_path.hpp"
#include "caf/downstream_policy.hpp"
namespace caf {
abstract_downstream::abstract_downstream(local_actor* selfptr,
const stream_id& sid,
std::unique_ptr<downstream_policy> ptr)
: self_(selfptr),
sid_(sid),
min_buffer_size_(5), // TODO: make configurable
policy_(std::move(ptr)) {
// nop
}
abstract_downstream::~abstract_downstream() {
// nop
}
long abstract_downstream::total_credit() const {
return total_credit(paths_);
}
long abstract_downstream::max_credit() const {
return max_credit(paths_);
}
long abstract_downstream::min_credit() const {
return min_credit(paths_);
}
bool abstract_downstream::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));
paths_.emplace_back(new path(std::move(ptr), false));
return true;
}
return false;
}
bool abstract_downstream::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 abstract_downstream::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_));
//policy_->reclaim(this, x);
return true;
}
return false;
}
void abstract_downstream::close() {
for (auto& x : paths_)
unsafe_send_as(self_, x->hdl, make<stream_msg::close>(sid_));
paths_.clear();
}
void abstract_downstream::abort(strong_actor_ptr& cause, const error& reason) {
for (auto& x : paths_)
if (x->hdl != cause)
unsafe_send_as(self_, x->hdl,
make<stream_msg::abort>(this->sid_, reason));
}
abstract_downstream::path*
abstract_downstream::find(const strong_actor_ptr& ptr) const {
return find(paths_, ptr);
}
long abstract_downstream::total_net_credit() const {
return policy().total_net_credit(*this);
}
long abstract_downstream::num_paths() const {
return static_cast<long>(paths_.size());
}
void abstract_downstream::send_batch(downstream_path& dest, long chunk_size,
message chunk) {
auto scs = static_cast<int32_t>(chunk_size);
auto batch_id = dest.next_batch_id++;
stream_msg::batch batch{scs, std::move(chunk), 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)});
}
void abstract_downstream::sort_by_credit() {
sort_by_credit(paths_);
}
} // 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. *
******************************************************************************/
#include "caf/abstract_upstream.hpp"
#include "caf/send.hpp"
#include "caf/stream_msg.hpp"
namespace caf {
abstract_upstream::abstract_upstream(local_actor* selfptr,
abstract_upstream::policy_ptr p)
: self_(selfptr),
policy_(std::move(p)),
continuous_(false) {
// nop
}
abstract_upstream::~abstract_upstream() {
// nop
}
void abstract_upstream::abort(strong_actor_ptr& cause, const error& reason) {
for (auto& x : paths_)
if (x->hdl != cause)
unsafe_send_as(self_, x->hdl, make<stream_msg::abort>(x->sid, reason));
}
void abstract_upstream::assign_credit(long total_downstream_net_credit) {
policy_->assign_credit(policy_vec_, total_downstream_net_credit);
for (auto& x : policy_vec_) {
auto n = x.second;
if (n > 0) {
auto ptr = x.first;
ptr->assigned_credit += n;
unsafe_send_as(self_, ptr->hdl, make<stream_msg::ack_batch>(
ptr->sid, static_cast<int32_t>(n),
ptr->last_batch_id++));
}
}
}
expected<long> abstract_upstream::add_path(strong_actor_ptr hdl,
const stream_id& sid,
stream_priority prio,
long total_downstream_net_credit) {
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(sid) << CAF_ARG(prio)
<< CAF_ARG(total_downstream_net_credit));
CAF_ASSERT(hdl != nullptr);
if (!find(hdl)) {
CAF_LOG_DEBUG("add new upstraem path" << CAF_ARG(hdl));
paths_.emplace_back(new path(std::move(hdl), sid, prio));
policy_vec_.emplace_back(paths_.back().get(), 0);
// use a one-shot actor to calculate initial credit
upstream_policy::assignment_vec tmp;
tmp.emplace_back(paths_.back().get(), 0);
policy_->assign_credit(tmp, total_downstream_net_credit);
paths_.back()->assigned_credit += tmp.back().second;
return tmp.back().second;
}
return sec::upstream_already_exists;
}
bool abstract_upstream::remove_path(const strong_actor_ptr& hdl) {
CAF_LOG_TRACE(CAF_ARG(hdl));
// Find element in our paths list.
auto has_hdl = [&](const path_uptr& x) {
CAF_ASSERT(x != nullptr);
return x->hdl == hdl;
};
auto e = paths_.end();
auto i = std::find_if(paths_.begin(), e, has_hdl);
if (i != e) {
// Also find and erase this element from our policy vector.
auto has_ptr = [&](const upstream_policy::assignment_pair& p) {
return p.first == i->get();
};
auto e2 = policy_vec_.end();
auto i2 = std::find_if(policy_vec_.begin(), e2, has_ptr);
if (i2 != e2) {
std::swap(*i2, policy_vec_.back());
policy_vec_.pop_back();
}
// Drop path from list.
if (i != e - 1)
std::swap(*i, paths_.back());
paths_.pop_back();
return true;
}
return false;
}
auto abstract_upstream::find(const strong_actor_ptr& x) const
-> optional<path&> {
CAF_ASSERT(x != nullptr);
auto pred = [&](const path_uptr& y) {
CAF_ASSERT(y != nullptr);
return x == y->hdl;
};
auto e = paths_.end();
auto i = std::find_if(paths_.begin(), e, pred);
return i != e ? i->get() : nullptr;
}
} // namespace caf
...@@ -19,12 +19,118 @@ ...@@ -19,12 +19,118 @@
#include "caf/downstream_policy.hpp" #include "caf/downstream_policy.hpp"
#include <utility>
#include "caf/send.hpp"
#include "caf/atom.hpp" #include "caf/atom.hpp"
#include "caf/downstream_path.hpp"
#include "caf/downstream_policy.hpp"
namespace caf { 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() { downstream_policy::~downstream_policy() {
// nop // 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));
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_));
//policy_->reclaim(this, x);
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) {
for (auto& x : paths_)
if (x->hdl != cause)
unsafe_send_as(self_, x->hdl,
make<stream_msg::abort>(this->sid_, reason));
}
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 } // namespace caf
...@@ -27,27 +27,28 @@ ...@@ -27,27 +27,28 @@
namespace caf { namespace caf {
namespace policy { namespace policy {
greedy::greedy() : low_watermark(0), high_watermark(5) { greedy::~greedy() {
// nop // nop
} }
void greedy::assign_credit(assignment_vec& xs, void greedy::fill_assignment_vec(long downstream_credit) {
long total_downstream_net_credit) { CAF_LOG_TRACE(CAF_ARG(downstream_credit));
CAF_LOG_TRACE(CAF_ARG(xs) << CAF_ARG(total_downstream_net_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 (total_downstream_net_credit <= 0) { if (downstream_credit <= 0) {
for (auto& x : xs) for (auto& x : assignment_vec_)
x.second = 0; x.second = 0;
return; return;
} }
// Assign credit to upstream paths until no more credit is available. We must // Assign credit to upstream paths until no more credit is available. We must
// make sure to write to each element in the vector. // make sure to write to each element in the vector.
auto available = total_downstream_net_credit; auto available = downstream_credit;
for (auto& p : xs) { for (auto& p : assignment_vec_) {
auto& x = p.first->assigned_credit; auto& x = p.first->assigned_credit; // current value
if (x < high_watermark) { auto y = std::min(max_credit(), x + available);
p.second = std::min(high_watermark - x, available); auto delta = y - x;
available -= p.second; if (delta >= min_credit_assignment()) {
p.second = delta;
available -= delta;
} else { } else {
p.second = 0; p.second = 0;
} }
......
...@@ -17,98 +17,43 @@ ...@@ -17,98 +17,43 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_ABSTRACT_UPSTREAM_HPP #include "caf/policy/pull5.hpp"
#define CAF_ABSTRACT_UPSTREAM_HPP
#include <vector> #include <numeric>
#include <memory>
#include <unordered_map>
#include "caf/fwd.hpp" #include "caf/logger.hpp"
#include "caf/upstream_path.hpp" #include "caf/upstream_path.hpp"
#include "caf/upstream_policy.hpp"
namespace caf { namespace caf {
namespace policy {
class abstract_upstream {
public: pull5::~pull5() {
using path = upstream_path; // nop
}
using path_cref = const path&;
void pull5::fill_assignment_vec(long downstream_credit) {
using path_uptr = std::unique_ptr<path>; CAF_LOG_TRACE(CAF_ARG(downstream_credit));
// Zero-out assignment vector if no credit is available at downstream paths.
using path_ptr = path*; if (downstream_credit <= 0) {
for (auto& x : assignment_vec_)
using path_list = std::vector<path_uptr>; x.second = 0;
return;
using path_ptr_list = std::vector<path_ptr>;
/// Stores available paths sorted by priority.
using path_map = std::unordered_map<stream_priority, path_ptr_list>;
using policy_ptr = std::unique_ptr<upstream_policy>;
/// @pre `self != nullptr`
/// @pre `policy != nullptr`
abstract_upstream(local_actor* selfptr, policy_ptr p);
virtual ~abstract_upstream();
void abort(strong_actor_ptr& cause, const error& reason);
/// Assigns credit to upstream actors according to the configured policy.
void assign_credit(long total_downstream_net_credit);
/// Adds a new upstream actor and returns the initial credit.
expected<long> add_path(strong_actor_ptr hdl, const stream_id& sid,
stream_priority prio,
long total_downstream_net_credit);
bool remove_path(const strong_actor_ptr& hdl);
inline local_actor* self() const {
return self_;
} }
// Assign credit to upstream paths until no more credit is available. We must
/// Returns `true` if all upstream paths are closed and this upstream is not // make sure to write to each element in the vector.
/// flagged as `continuous`, `false` otherwise. auto available = downstream_credit;
inline bool closed() const { for (auto& p : assignment_vec_) {
return paths_.empty() && !continuous_; auto& x = p.first->assigned_credit; // current value
auto y = std::min(5l, x + available);
auto delta = y - x;
if (delta >= min_credit_assignment()) {
p.second = delta;
available -= delta;
} else {
p.second = 0;
} }
/// Returns whether this upstream remains open even if no more upstream path
/// exists.
inline bool continuous() const {
return continuous_;
} }
}
/// Sets whether this upstream remains open even if no more upstream path } // namespace policy
/// exists.
inline void continuous(bool value) {
continuous_ = value;
}
optional<path&> find(const strong_actor_ptr& x) const;
protected:
/// Pointer to the parent actor.
local_actor* self_;
/// List of all known paths.
path_list paths_;
/// Our policy for assigning credit.
policy_ptr policy_;
/// An assignment vector that's re-used whenever calling the policy.
upstream_policy::assignment_vec policy_vec_;
/// Stores whether this stream remains open even if all paths have been
/// closed.
bool continuous_;
};
} // namespace caf } // namespace caf
#endif // CAF_ABSTRACT_UPSTREAM_HPP
...@@ -48,7 +48,7 @@ error stream_handler::downstream_demand(strong_actor_ptr&, long) { ...@@ -48,7 +48,7 @@ error stream_handler::downstream_demand(strong_actor_ptr&, long) {
return sec::invalid_downstream; return sec::invalid_downstream;
} }
error stream_handler::push(long*) { error stream_handler::push() {
CAF_LOG_ERROR("Cannot push to a stream marked as no-downstreams"); CAF_LOG_ERROR("Cannot push to a stream marked as no-downstreams");
return sec::invalid_downstream; return sec::invalid_downstream;
} }
...@@ -71,11 +71,11 @@ error stream_handler::close_upstream(strong_actor_ptr&) { ...@@ -71,11 +71,11 @@ error stream_handler::close_upstream(strong_actor_ptr&) {
return sec::invalid_upstream; return sec::invalid_upstream;
} }
optional<abstract_downstream&> stream_handler::get_downstream() { optional<downstream_policy&> stream_handler::dp() {
return none; return none;
} }
optional<abstract_upstream&> stream_handler::get_upstream() { optional<upstream_policy&> stream_handler::up() {
return none; return none;
} }
......
...@@ -20,11 +20,12 @@ ...@@ -20,11 +20,12 @@
#include "caf/stream_sink.hpp" #include "caf/stream_sink.hpp"
#include "caf/send.hpp" #include "caf/send.hpp"
#include "caf/abstract_upstream.hpp" #include "caf/upstream_path.hpp"
#include "caf/upstream_policy.hpp"
namespace caf { namespace caf {
stream_sink::stream_sink(abstract_upstream* in_ptr, stream_sink::stream_sink(upstream_policy* in_ptr,
strong_actor_ptr&& orig_sender, strong_actor_ptr&& orig_sender,
std::vector<strong_actor_ptr>&& trailing_stages, std::vector<strong_actor_ptr>&& trailing_stages,
message_id mid) message_id mid)
......
...@@ -23,11 +23,11 @@ ...@@ -23,11 +23,11 @@
#include "caf/error.hpp" #include "caf/error.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/downstream_path.hpp" #include "caf/downstream_path.hpp"
#include "caf/abstract_downstream.hpp" #include "caf/downstream_policy.hpp"
namespace caf { namespace caf {
stream_source::stream_source(abstract_downstream* out_ptr) : out_ptr_(out_ptr) { stream_source::stream_source(downstream_policy* out_ptr) : out_ptr_(out_ptr) {
// nop // nop
} }
...@@ -48,10 +48,11 @@ error stream_source::downstream_demand(strong_actor_ptr& hdl, long value) { ...@@ -48,10 +48,11 @@ error stream_source::downstream_demand(strong_actor_ptr& hdl, long value) {
if (!at_end()) { if (!at_end()) {
// produce new elements // produce new elements
auto current_size = buf_size(); auto current_size = buf_size();
auto size_hint = out().total_net_credit(); auto size_hint = std::min(out().credit(),
out().max_batch_size());
if (current_size < size_hint) if (current_size < size_hint)
generate(static_cast<size_t>(size_hint - current_size)); generate(static_cast<size_t>(size_hint - current_size));
return push(&size_hint); return push();
} }
// transmit cached elements before closing paths // transmit cached elements before closing paths
if (buf_size() > 0) if (buf_size() > 0)
...@@ -67,9 +68,10 @@ void stream_source::abort(strong_actor_ptr& cause, const error& reason) { ...@@ -67,9 +68,10 @@ void stream_source::abort(strong_actor_ptr& cause, const error& reason) {
} }
void stream_source::generate() { void stream_source::generate() {
CAF_LOG_TRACE("");
if (!at_end()) { if (!at_end()) {
auto current_size = buf_size(); auto current_size = buf_size();
auto size_hint = out().total_net_credit(); auto size_hint = out().credit();
if (current_size < size_hint) if (current_size < size_hint)
generate(static_cast<size_t>(size_hint - current_size)); generate(static_cast<size_t>(size_hint - current_size));
} }
......
...@@ -21,20 +21,18 @@ ...@@ -21,20 +21,18 @@
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/upstream_path.hpp"
#include "caf/downstream_path.hpp" #include "caf/downstream_path.hpp"
#include "caf/abstract_upstream.hpp" #include "caf/upstream_policy.hpp"
#include "caf/downstream_policy.hpp" #include "caf/downstream_policy.hpp"
#include "caf/abstract_downstream.hpp"
namespace caf { namespace caf {
stream_stage::stream_stage(abstract_upstream* in_ptr, stream_stage::stream_stage(upstream_policy* in_ptr, downstream_policy* out_ptr)
abstract_downstream* out_ptr)
: in_ptr_(in_ptr), : in_ptr_(in_ptr),
out_ptr_(out_ptr) { out_ptr_(out_ptr) {
// nop // nop
} }
bool stream_stage::done() const { bool stream_stage::done() const {
return in_ptr_->closed() && out_ptr_->closed(); return in_ptr_->closed() && out_ptr_->closed();
} }
...@@ -50,7 +48,10 @@ error stream_stage::upstream_batch(strong_actor_ptr& hdl, long xs_size, ...@@ -50,7 +48,10 @@ error stream_stage::upstream_batch(strong_actor_ptr& hdl, long xs_size,
auto err = process_batch(xs); auto err = process_batch(xs);
if (err == none) { if (err == none) {
push(); push();
in().assign_credit(out().total_net_credit()); 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 err;
} }
...@@ -66,7 +67,10 @@ error stream_stage::downstream_demand(strong_actor_ptr& hdl, long value) { ...@@ -66,7 +67,10 @@ error stream_stage::downstream_demand(strong_actor_ptr& hdl, long value) {
push(); push();
else if (in().closed() && !out().remove_path(hdl)) else if (in().closed() && !out().remove_path(hdl))
return sec::invalid_downstream; return sec::invalid_downstream;
in().assign_credit(out().total_net_credit()); auto current_size = out().buf_size();
auto desired_size = out().credit();
if (current_size < desired_size)
in().assign_credit(desired_size - current_size);
return none; return none;
} }
return sec::invalid_downstream; return sec::invalid_downstream;
......
...@@ -19,10 +19,117 @@ ...@@ -19,10 +19,117 @@
#include "caf/upstream_policy.hpp" #include "caf/upstream_policy.hpp"
#include "caf/send.hpp"
#include "caf/stream_msg.hpp"
#include "caf/upstream_path.hpp"
namespace caf { namespace caf {
// -- constructors, destructors, and assignment operators ----------------------
upstream_policy::upstream_policy(local_actor* selfptr)
: self_(selfptr),
continuous_(false),
high_watermark_(5),
min_credit_assignment_(1),
max_credit_(5) {
// nop
}
upstream_policy::~upstream_policy() { upstream_policy::~upstream_policy() {
// nop // nop
} }
// -- path management ----------------------------------------------------------
void upstream_policy::abort(strong_actor_ptr& cause, const error& reason) {
for (auto& x : paths_)
if (x->hdl != cause)
unsafe_send_as(self_, x->hdl, make<stream_msg::abort>(x->sid, reason));
}
void upstream_policy::assign_credit(long downstream_capacity) {
CAF_LOG_TRACE(CAF_ARG(downstream_capacity));
auto used_capacity = 0l;
for (auto& x : assignment_vec_)
used_capacity += static_cast<long>(x.first->assigned_credit);
CAF_LOG_DEBUG(CAF_ARG(used_capacity));
if (used_capacity >= downstream_capacity)
return;
fill_assignment_vec(downstream_capacity - used_capacity);
for (auto& x : assignment_vec_) {
auto n = x.second;
if (n > 0) {
auto ptr = x.first;
ptr->assigned_credit += n;
CAF_LOG_DEBUG("ack batch" << ptr->last_batch_id
<< "with" << n << "new capacity");
unsafe_send_as(self_, ptr->hdl,
make<stream_msg::ack_batch>(ptr->sid,
static_cast<int32_t>(n),
ptr->last_batch_id++));
}
}
}
expected<long> upstream_policy::add_path(strong_actor_ptr hdl,
const stream_id& sid,
stream_priority prio,
long downstream_credit) {
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(sid) << CAF_ARG(prio)
<< CAF_ARG(downstream_credit));
CAF_ASSERT(hdl != nullptr);
if (!find(hdl)) {
CAF_LOG_DEBUG("add new upstraem path" << CAF_ARG(hdl));
auto ptr = new upstream_path(std::move(hdl), sid, prio);
paths_.emplace_back(ptr);
assignment_vec_.emplace_back(ptr, 0);
if (downstream_credit > 0)
ptr->assigned_credit = std::min(max_credit_, downstream_credit);
return ptr->assigned_credit;
}
return sec::upstream_already_exists;
}
bool upstream_policy::remove_path(const strong_actor_ptr& hdl) {
CAF_LOG_TRACE(CAF_ARG(hdl));
// Find element in our paths list.
auto has_hdl = [&](const path_uptr& x) {
CAF_ASSERT(x != nullptr);
return x->hdl == hdl;
};
auto e = paths_.end();
auto i = std::find_if(paths_.begin(), e, has_hdl);
if (i != e) {
// Also find and erase this element from our policy vector.
auto has_ptr = [&](const upstream_policy::assignment_pair& p) {
return p.first == i->get();
};
auto e2 = assignment_vec_.end();
auto i2 = std::find_if(assignment_vec_.begin(), e2, has_ptr);
if (i2 != e2) {
std::swap(*i2, assignment_vec_.back());
assignment_vec_.pop_back();
}
// Drop path from list.
if (i != e - 1)
std::swap(*i, paths_.back());
paths_.pop_back();
return true;
}
return false;
}
upstream_path* upstream_policy::find(const strong_actor_ptr& x) const {
CAF_ASSERT(x != nullptr);
auto pred = [&](const path_uptr& y) {
CAF_ASSERT(y != nullptr);
return x == y->hdl;
};
auto e = paths_.end();
auto i = std::find_if(paths_.begin(), e, pred);
return i != e ? i->get() : nullptr;
}
} // namespace caf } // namespace caf
...@@ -32,6 +32,9 @@ ...@@ -32,6 +32,9 @@
#include "caf/stream_source.hpp" #include "caf/stream_source.hpp"
#include "caf/filtering_downstream.hpp" #include "caf/filtering_downstream.hpp"
#include "caf/policy/pull5.hpp"
#include "caf/policy/push5.hpp"
#include "caf/io/middleman.hpp" #include "caf/io/middleman.hpp"
#include "caf/io/basp_broker.hpp" #include "caf/io/basp_broker.hpp"
#include "caf/io/network/test_multiplexer.hpp" #include "caf/io/network/test_multiplexer.hpp"
...@@ -80,13 +83,12 @@ public: ...@@ -80,13 +83,12 @@ public:
struct peer_data { struct peer_data {
filter_type filter; filter_type filter;
downstream<element_type> out; policy::push5<element_type> out;
stream_id incoming_sid; stream_id incoming_sid;
peer_data(filter_type y, local_actor* self, const stream_id& sid, peer_data(filter_type y, local_actor* self, const stream_id& sid)
abstract_downstream::policy_ptr pp)
: filter(std::move(y)), : filter(std::move(y)),
out(self, sid, std::move(pp)) { out(self, sid) {
// nop // nop
} }
...@@ -143,7 +145,7 @@ public: ...@@ -143,7 +145,7 @@ public:
error downstream_demand(strong_actor_ptr& hdl, long new_demand) override; error downstream_demand(strong_actor_ptr& hdl, long new_demand) override;
error push(long* hint) override; error push() override;
expected<long> add_upstream(strong_actor_ptr& hdl, const stream_id& sid, expected<long> add_upstream(strong_actor_ptr& hdl, const stream_id& sid,
stream_priority prio) override; stream_priority prio) override;
...@@ -158,14 +160,18 @@ public: ...@@ -158,14 +160,18 @@ public:
message make_output_token(const stream_id&) const override; message make_output_token(const stream_id&) const override;
long total_downstream_net_credit() const; long downstream_credit() const;
long downstream_buffer_size() const;
void assign_credit();
private: private:
void new_stream(const strong_actor_ptr& hdl, const stream_type& token, void new_stream(const strong_actor_ptr& hdl, const stream_type& token,
message msg); message msg);
core_state* state_; core_state* state_;
upstream<element_type> in_; policy::pull5 in_;
local_downstream local_subscribers_; local_downstream local_subscribers_;
peer_map peers_; peer_map peers_;
}; };
...@@ -222,16 +228,14 @@ const char* core_state::name = "core"; ...@@ -222,16 +228,14 @@ const char* core_state::name = "core";
stream_governor::stream_governor(core_state* state) stream_governor::stream_governor(core_state* state)
: state_(state), : state_(state),
in_(state->self, policy::greedy::make()), in_(state->self),
local_subscribers_(state->self, state->sid, policy::broadcast::make()) { local_subscribers_(state->self, state->sid) {
// nop // nop
} }
stream_governor::peer_data* stream_governor::add_peer(strong_actor_ptr hdl, stream_governor::peer_data* stream_governor::add_peer(strong_actor_ptr hdl,
filter_type filter) { filter_type filter) {
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(filter)); CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(filter));
abstract_downstream::policy_ptr pp{new policy::broadcast}; auto ptr = new peer_data{std::move(filter), state_->self, state_->sid};
auto ptr = new peer_data{std::move(filter), state_->self,
state_->sid, std::move(pp)};
ptr->out.add_path(hdl); ptr->out.add_path(hdl);
auto res = peers_.emplace(std::move(hdl), peer_data_ptr{ptr}); auto res = peers_.emplace(std::move(hdl), peer_data_ptr{ptr});
return res.second ? ptr : nullptr; return res.second ? ptr : nullptr;
...@@ -280,7 +284,9 @@ error stream_governor::downstream_demand(strong_actor_ptr& hdl, long value) { ...@@ -280,7 +284,9 @@ error stream_governor::downstream_demand(strong_actor_ptr& hdl, long value) {
auto path = local_subscribers_.find(hdl); auto path = local_subscribers_.find(hdl);
if (path) { if (path) {
path->open_credit += value; path->open_credit += value;
return push(nullptr); push();
assign_credit();
return none;
} }
auto i = peers_.find(hdl); auto i = peers_.find(hdl);
if (i != peers_.end()) { if (i != peers_.end()) {
...@@ -289,19 +295,21 @@ error stream_governor::downstream_demand(strong_actor_ptr& hdl, long value) { ...@@ -289,19 +295,21 @@ error stream_governor::downstream_demand(strong_actor_ptr& hdl, long value) {
return sec::invalid_stream_state; return sec::invalid_stream_state;
CAF_LOG_DEBUG("grant" << value << "new credit to" << hdl); CAF_LOG_DEBUG("grant" << value << "new credit to" << hdl);
pp->open_credit += value; pp->open_credit += value;
return push(nullptr); push();
assign_credit();
return none;
} }
return sec::invalid_downstream; return sec::invalid_downstream;
} }
error stream_governor::push(long* hint) { error stream_governor::push() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
if (local_subscribers_.buf_size() > 0) if (local_subscribers_.buf_size() > 0)
local_subscribers_.policy().push(local_subscribers_, hint); local_subscribers_.emit_batches();
for (auto& kvp : peers_) { for (auto& kvp : peers_) {
auto& out = kvp.second->out; auto& out = kvp.second->out;
if (out.buf_size() > 0) if (out.buf_size() > 0)
out.policy().push(out, hint); out.emit_batches();
} }
return none; return none;
} }
...@@ -311,7 +319,7 @@ expected<long> stream_governor::add_upstream(strong_actor_ptr& hdl, ...@@ -311,7 +319,7 @@ expected<long> stream_governor::add_upstream(strong_actor_ptr& hdl,
stream_priority prio) { stream_priority prio) {
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(sid) << CAF_ARG(prio)); CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(sid) << CAF_ARG(prio));
if (hdl) if (hdl)
return in_.add_path(hdl, sid, prio, total_downstream_net_credit()); return in_.add_path(hdl, sid, prio, downstream_credit());
return sec::invalid_argument; return sec::invalid_argument;
} }
...@@ -341,17 +349,15 @@ error stream_governor::upstream_batch(strong_actor_ptr& hdl, long xs_size, ...@@ -341,17 +349,15 @@ error stream_governor::upstream_batch(strong_actor_ptr& hdl, long xs_size,
if (selected(kvp.second->filter, x)) if (selected(kvp.second->filter, x))
out.push(x); out.push(x);
if (out.buf_size() > 0) { if (out.buf_size() > 0) {
out.policy().push(out); out.emit_batches();
} }
} }
// Move elements from `xs` to the buffer for local subscribers. // Move elements from `xs` to the buffer for local subscribers.
for (auto& x : vec) for (auto& x : vec)
local_subscribers_.push(std::move(x)); local_subscribers_.push(std::move(x));
local_subscribers_.policy().push(local_subscribers_); local_subscribers_.emit_batches();
// Grant new credit to upstream if possible. // Grant new credit to upstream if possible.
auto available = total_downstream_net_credit(); assign_credit();
if (available > 0)
in_.assign_credit(available);
return none; return none;
} }
...@@ -372,9 +378,12 @@ void stream_governor::abort(strong_actor_ptr& hdl, const error& reason) { ...@@ -372,9 +378,12 @@ void stream_governor::abort(strong_actor_ptr& hdl, const error& reason) {
auto& pd = *i->second; auto& pd = *i->second;
state_->self->streams().erase(pd.incoming_sid); state_->self->streams().erase(pd.incoming_sid);
peers_.erase(i); peers_.erase(i);
} else {
in_.abort(hdl, reason);
} }
} else { } else {
local_subscribers_.abort(hdl, reason); local_subscribers_.abort(hdl, reason);
in_.abort(hdl, reason);
for (auto& kvp : peers_) for (auto& kvp : peers_)
kvp.second->out.abort(hdl, reason); kvp.second->out.abort(hdl, reason);
peers_.clear(); peers_.clear();
...@@ -389,11 +398,37 @@ message stream_governor::make_output_token(const stream_id& x) const { ...@@ -389,11 +398,37 @@ message stream_governor::make_output_token(const stream_id& x) const {
return make_message(stream<element_type>{x}); return make_message(stream<element_type>{x});
} }
long stream_governor::total_downstream_net_credit() const { long stream_governor::downstream_credit() const {
auto net_credit = local_subscribers_.total_net_credit(); auto min_peer_credit = [&] {
return std::accumulate(peers_.begin(), peers_.end(),
std::numeric_limits<long>::max(),
[](long x, const peer_map::value_type& y) {
return std::min(x, y.second->out.min_credit());
});
};
constexpr long min_buffer_size = 5l;
if (local_subscribers_.num_paths() == 0)
return (peers_.empty() ? 0l : min_peer_credit()) + min_buffer_size;
return (peers_.empty()
? local_subscribers_.min_credit()
: std::min(local_subscribers_.min_credit(), min_peer_credit()))
+ min_buffer_size;
}
long stream_governor::downstream_buffer_size() const {
auto result = local_subscribers_.buf_size();
for (auto& kvp : peers_) for (auto& kvp : peers_)
net_credit = std::min(net_credit, kvp.second->out.total_net_credit()); result += std::max(result, kvp.second->out.buf_size());
return net_credit; return result;
}
void stream_governor::assign_credit() {
CAF_LOG_TRACE("");
auto current_size = downstream_buffer_size();
auto desired_size = downstream_credit();
CAF_LOG_DEBUG(CAF_ARG(current_size) << CAF_ARG(desired_size));
if (current_size < desired_size)
in_.assign_credit(desired_size - current_size);
} }
void stream_governor::new_stream(const strong_actor_ptr& hdl, void stream_governor::new_stream(const strong_actor_ptr& hdl,
...@@ -560,7 +595,8 @@ void driver(event_based_actor* self, const actor& sink) { ...@@ -560,7 +595,8 @@ void driver(event_based_actor* self, const actor& sink) {
// Handle result of the stream. // Handle result of the stream.
[](expected<void>) { [](expected<void>) {
// nop // nop
} },
policy::arg<policy::push5<element_type>>::value
); );
} }
...@@ -587,7 +623,8 @@ void consumer(stateful_actor<consumer_state>* self, filter_type ts, ...@@ -587,7 +623,8 @@ void consumer(stateful_actor<consumer_state>* self, filter_type ts,
// Cleanup. // Cleanup.
[](unit_t&) { [](unit_t&) {
// nop // nop
} },
policy::arg<policy::pull5>::value
); );
}, },
[=](get_atom) { [=](get_atom) {
...@@ -642,6 +679,7 @@ CAF_TEST(two_peers) { ...@@ -642,6 +679,7 @@ CAF_TEST(two_peers) {
CAF_REQUIRE(!sched.has_job()); CAF_REQUIRE(!sched.has_job());
// Spin up driver on core1. // Spin up driver on core1.
auto d1 = sys.spawn(driver, core1); auto d1 = sys.spawn(driver, core1);
CAF_MESSAGE("d1: " << to_string(d1));
sched.run_once(); sched.run_once();
expect((stream_msg::open), from(_).to(core1).with(_, d1, _, _, false)); expect((stream_msg::open), from(_).to(core1).with(_, d1, _, _, false));
expect((stream_msg::ack_open), from(core1).to(d1).with(_, 5, _, false)); expect((stream_msg::ack_open), from(core1).to(d1).with(_, 5, _, false));
......
...@@ -31,6 +31,9 @@ ...@@ -31,6 +31,9 @@
#include "caf/filtering_downstream.hpp" #include "caf/filtering_downstream.hpp"
#include "caf/policy/greedy.hpp"
#include "caf/policy/broadcast.hpp"
using std::cout; using std::cout;
using std::endl; using std::endl;
using std::string; using std::string;
...@@ -47,14 +50,6 @@ using filter_type = std::vector<key_type>; ...@@ -47,14 +50,6 @@ using filter_type = std::vector<key_type>;
using element_type = std::pair<key_type, value_type>; using element_type = std::pair<key_type, value_type>;
struct stage_policy {
template <class InputType>
using upstream_type = upstream<InputType>;
template <class OutputType>
using downstream_type = filtering_downstream<OutputType, key_type>;
};
struct process_t { struct process_t {
void operator()(unit_t&, downstream<element_type>& out, element_type x) { void operator()(unit_t&, downstream<element_type>& out, element_type x) {
out.push(std::move(x)); out.push(std::move(x));
...@@ -72,7 +67,9 @@ struct cleanup_t { ...@@ -72,7 +67,9 @@ struct cleanup_t {
constexpr cleanup_t cleanup_fun = cleanup_t{}; constexpr cleanup_t cleanup_fun = cleanup_t{};
struct stream_splitter_state { struct stream_splitter_state {
using stage_impl = stream_stage_impl<process_t, cleanup_t, stage_policy>; using stage_impl = stream_stage_impl<process_t, cleanup_t, policy::greedy,
filtering_downstream<element_type,
key_type>>;
intrusive_ptr<stage_impl> stage; intrusive_ptr<stage_impl> stage;
static const char* name; static const char* name;
}; };
...@@ -83,12 +80,11 @@ behavior stream_splitter(stateful_actor<stream_splitter_state>* self) { ...@@ -83,12 +80,11 @@ behavior stream_splitter(stateful_actor<stream_splitter_state>* self) {
stream_id id{self->ctrl(), stream_id id{self->ctrl(),
self->new_request_id(message_priority::normal).integer_value()}; self->new_request_id(message_priority::normal).integer_value()};
using impl = stream_splitter_state::stage_impl; using impl = stream_splitter_state::stage_impl;
std::unique_ptr<upstream_policy> upolicy{new policy::greedy}; self->state.stage = make_counted<impl>(self, id, process_fun, cleanup_fun);
std::unique_ptr<downstream_policy> dpolicy{new policy::broadcast};
self->state.stage = make_counted<impl>(self, id, std::move(upolicy),
std::move(dpolicy), process_fun,
cleanup_fun);
self->state.stage->in().continuous(true); self->state.stage->in().continuous(true);
// force the splitter to collect credit until reaching 3 in order
// to receive only full batches from upstream (making tests a lot easier)
self->state.stage->in().min_credit_assignment(3);
self->streams().emplace(id, self->state.stage); self->streams().emplace(id, self->state.stage);
return { return {
[=](join_atom, filter_type filter) -> stream<element_type> { [=](join_atom, filter_type filter) -> stream<element_type> {
...@@ -236,14 +232,14 @@ CAF_TEST(fork_setup) { ...@@ -236,14 +232,14 @@ CAF_TEST(fork_setup) {
{"key2", "b"}, {"key2", "b"},
{"key1", "c"}}, {"key1", "c"}},
0)); 0));
expect((stream_msg::batch),
from(splitter).to(d1)
.with(3, batch{{"key1", "a"}, {"key1", "b"}, {"key1", "c"}}, 0));
expect((stream_msg::batch), expect((stream_msg::batch),
from(splitter).to(d2) from(splitter).to(d2)
.with(2, batch{{"key2", "a"}, {"key2", "b"}}, 0)); .with(2, batch{{"key2", "a"}, {"key2", "b"}}, 0));
expect((stream_msg::ack_batch), from(d1).to(splitter).with(3, 0)); expect((stream_msg::batch),
from(splitter).to(d1)
.with(3, batch{{"key1", "a"}, {"key1", "b"}, {"key1", "c"}}, 0));
expect((stream_msg::ack_batch), from(d2).to(splitter).with(2, 0)); expect((stream_msg::ack_batch), from(d2).to(splitter).with(2, 0));
expect((stream_msg::ack_batch), from(d1).to(splitter).with(3, 0));
expect((stream_msg::ack_batch), from(splitter).to(src).with(5, 0)); expect((stream_msg::ack_batch), from(splitter).to(src).with(5, 0));
// Second batch. // Second batch.
expect((stream_msg::batch), expect((stream_msg::batch),
...@@ -294,14 +290,14 @@ CAF_TEST(fork_setup) { ...@@ -294,14 +290,14 @@ CAF_TEST(fork_setup) {
{"key2", "b"}, {"key2", "b"},
{"key1", "c"}}, {"key1", "c"}},
0)); 0));
expect((stream_msg::batch),
from(splitter).to(d1)
.with(3, batch{{"key1", "a"}, {"key1", "b"}, {"key1", "c"}}, 2));
expect((stream_msg::batch), expect((stream_msg::batch),
from(splitter).to(d2) from(splitter).to(d2)
.with(2, batch{{"key2", "a"}, {"key2", "b"}}, 2)); .with(2, batch{{"key2", "a"}, {"key2", "b"}}, 2));
expect((stream_msg::ack_batch), from(d1).to(splitter).with(3, 2)); expect((stream_msg::batch),
from(splitter).to(d1)
.with(3, batch{{"key1", "a"}, {"key1", "b"}, {"key1", "c"}}, 2));
expect((stream_msg::ack_batch), from(d2).to(splitter).with(2, 2)); expect((stream_msg::ack_batch), from(d2).to(splitter).with(2, 2));
expect((stream_msg::ack_batch), from(d1).to(splitter).with(3, 2));
expect((stream_msg::ack_batch), from(splitter).to(src2).with(5, 0)); expect((stream_msg::ack_batch), from(splitter).to(src2).with(5, 0));
// Second batch. // Second batch.
expect((stream_msg::batch), expect((stream_msg::batch),
......
...@@ -26,6 +26,9 @@ ...@@ -26,6 +26,9 @@
#define CAF_SUITE streaming #define CAF_SUITE streaming
#include "caf/test/dsl.hpp" #include "caf/test/dsl.hpp"
#include "caf/policy/pull5.hpp"
#include "caf/policy/push5.hpp"
using std::cout; using std::cout;
using std::endl; using std::endl;
using std::string; using std::string;
...@@ -62,7 +65,8 @@ behavior file_reader(stateful_actor<file_reader_state>* self) { ...@@ -62,7 +65,8 @@ behavior file_reader(stateful_actor<file_reader_state>* self) {
// check whether we reached the end // check whether we reached the end
[=](const buf& xs) { [=](const buf& xs) {
return xs.empty(); return xs.empty();
} },
policy::arg<policy::push5<int>>::value
); );
} }
}; };
...@@ -99,7 +103,8 @@ void streamer(stateful_actor<streamer_state>* self, const actor& dest) { ...@@ -99,7 +103,8 @@ void streamer(stateful_actor<streamer_state>* self, const actor& dest) {
// handle result of the stream // handle result of the stream
[=](expected<int>) { [=](expected<int>) {
// nop // nop
} },
policy::arg<policy::push5<int>>::value
); );
} }
...@@ -130,7 +135,8 @@ behavior filter(stateful_actor<filter_state>* self) { ...@@ -130,7 +135,8 @@ behavior filter(stateful_actor<filter_state>* self) {
// cleanup // cleanup
[=](unit_t&) { [=](unit_t&) {
// nop // nop
} },
policy::arg<policy::pull5, policy::push5<int>>::value
); );
} }
}; };
...@@ -175,7 +181,8 @@ behavior sum_up(stateful_actor<sum_up_state>* self) { ...@@ -175,7 +181,8 @@ behavior sum_up(stateful_actor<sum_up_state>* self) {
// cleanup and produce result message // cleanup and produce result message
[](int& x) -> int { [](int& x) -> int {
return x; return x;
} },
policy::arg<policy::pull5>::value
); );
} }
}; };
...@@ -205,7 +212,8 @@ behavior drop_all(stateful_actor<drop_all_state>* self) { ...@@ -205,7 +212,8 @@ behavior drop_all(stateful_actor<drop_all_state>* self) {
// cleanup and produce void "result" // cleanup and produce void "result"
[](unit_t&) { [](unit_t&) {
CAF_LOG_INFO("drop_all done"); CAF_LOG_INFO("drop_all done");
} },
policy::arg<policy::pull5>::value
); );
} }
}; };
...@@ -244,7 +252,8 @@ void nores_streamer(stateful_actor<nores_streamer_state>* self, ...@@ -244,7 +252,8 @@ void nores_streamer(stateful_actor<nores_streamer_state>* self,
// handle result of the stream // handle result of the stream
[=](expected<void>) { [=](expected<void>) {
// nop // nop
} },
policy::arg<policy::push5<int>>::value
); );
} }
...@@ -264,11 +273,9 @@ behavior stream_multiplexer(stateful_actor<stream_multiplexer_state>* self) { ...@@ -264,11 +273,9 @@ behavior stream_multiplexer(stateful_actor<stream_multiplexer_state>* self) {
}; };
stream_id id{self->ctrl(), stream_id id{self->ctrl(),
self->new_request_id(message_priority::normal).integer_value()}; self->new_request_id(message_priority::normal).integer_value()};
using impl = stream_stage_impl<decltype(process), decltype(cleanup)>; using impl = stream_stage_impl<decltype(process), decltype(cleanup),
std::unique_ptr<upstream_policy> upolicy{new policy::greedy}; policy::pull5, policy::push5<int>>;
std::unique_ptr<downstream_policy> dpolicy{new policy::broadcast}; self->state.stage = make_counted<impl>(self, id, process, cleanup);
self->state.stage = make_counted<impl>(self, id, std::move(upolicy),
std::move(dpolicy), process, cleanup);
self->state.stage->in().continuous(true); self->state.stage->in().continuous(true);
self->streams().emplace(id, self->state.stage); self->streams().emplace(id, self->state.stage);
return { return {
...@@ -290,7 +297,7 @@ behavior stream_multiplexer(stateful_actor<stream_multiplexer_state>* self) { ...@@ -290,7 +297,7 @@ behavior stream_multiplexer(stateful_actor<stream_multiplexer_state>* self) {
// We only need to add a new stream to the map, the runtime system will // We only need to add a new stream to the map, the runtime system will
// take care of adding a new upstream and sending the handshake. // take care of adding a new upstream and sending the handshake.
self->streams().emplace(sid.id(), self->state.stage); self->streams().emplace(sid.id(), self->state.stage);
} },
}; };
} }
......
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