Commit 6f9cf134 authored by Dominik Charousset's avatar Dominik Charousset

Implement upstream policies

parent ac7db0b2
...@@ -83,8 +83,14 @@ public: ...@@ -83,8 +83,14 @@ public:
/// Returns the size of the output buffer. /// Returns the size of the output buffer.
virtual size_t buf_size() const = 0; virtual size_t buf_size() const = 0;
bool add_path(strong_actor_ptr ptr, bool redeployable); /// 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); bool remove_path(strong_actor_ptr& ptr);
/// Removes all paths. /// Removes all paths.
...@@ -95,6 +101,8 @@ public: ...@@ -95,6 +101,8 @@ public:
optional<path&> find(const strong_actor_ptr& ptr) const; optional<path&> find(const strong_actor_ptr& ptr) const;
size_t available_credit() const;
inline local_actor* self() const { inline local_actor* self() const {
return self_; return self_;
} }
......
...@@ -57,12 +57,13 @@ public: ...@@ -57,12 +57,13 @@ public:
void abort(strong_actor_ptr& cause, const error& reason); void abort(strong_actor_ptr& cause, const error& reason);
error pull(strong_actor_ptr& hdl, size_t n); /// Assigns credit to upstream actors according to the configured policy.
void assign_credit(size_t buf_size, size_t downstream_credit);
error pull(size_t n);
/// Adds a new upstream actor and returns the initial credit.
expected<size_t> add_path(strong_actor_ptr hdl, const stream_id& sid, expected<size_t> add_path(strong_actor_ptr hdl, const stream_id& sid,
stream_priority prio); stream_priority prio, size_t buf_size,
size_t downstream_credit);
bool remove_path(const strong_actor_ptr& hdl); bool remove_path(const strong_actor_ptr& hdl);
...@@ -75,10 +76,20 @@ public: ...@@ -75,10 +76,20 @@ public:
return paths_.empty(); return paths_.empty();
} }
optional<path&> find(const strong_actor_ptr& x) const;
protected: protected:
/// Pointer to the parent actor.
local_actor* self_; local_actor* self_;
/// List of all known paths.
path_list paths_; path_list paths_;
/// Our policy for assigning credit.
policy_ptr policy_; policy_ptr policy_;
/// An assignment vector that's re-used whenever calling the policy.
upstream_policy::assignment_vec policy_vec_;
}; };
} // namespace caf } // namespace caf
......
...@@ -28,6 +28,8 @@ ...@@ -28,6 +28,8 @@
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/stream_msg.hpp" #include "caf/stream_msg.hpp"
#include "caf/meta/type_name.hpp"
namespace caf { namespace caf {
/// Denotes a downstream actor in a stream topology. All downstream actors use /// Denotes a downstream actor in a stream topology. All downstream actors use
...@@ -35,7 +37,7 @@ namespace caf { ...@@ -35,7 +37,7 @@ namespace caf {
class downstream_path { class downstream_path {
public: public:
/// Handle to the downstream actor. /// Handle to the downstream actor.
strong_actor_ptr ptr; strong_actor_ptr hdl;
/// Next expected batch ID. /// Next expected batch ID.
int64_t next_batch_id; int64_t next_batch_id;
...@@ -54,6 +56,12 @@ public: ...@@ -54,6 +56,12 @@ public:
downstream_path(strong_actor_ptr p, bool redeploy); downstream_path(strong_actor_ptr p, bool redeploy);
}; };
template <class Inspector>
typename Inspector::return_type inspect(Inspector& f, downstream_path& x) {
return f(meta::type_name("upstream_path"), x.next_batch_id, x.open_credit,
x.redeployable, x.unacknowledged_batches);
}
} // namespace caf } // namespace caf
#endif // CAF_DOWNSTREAM_PATH_HPP #endif // CAF_DOWNSTREAM_PATH_HPP
...@@ -31,15 +31,11 @@ class downstream_policy { ...@@ -31,15 +31,11 @@ class downstream_policy {
public: public:
virtual ~downstream_policy(); virtual ~downstream_policy();
/// Returns the topic name for `x`. The default implementation returns an
/// empty atom value, indicating a stream without topics.
virtual atom_value categorize(type_erased_value& x) const;
/// Queries the optimal amount of data for the next `push` operation to `x`. /// Queries the optimal amount of data for the next `push` operation to `x`.
virtual size_t desired_buffer_size(const abstract_downstream& x) = 0; virtual size_t available_credit(const abstract_downstream& x) = 0;
/// Pushes data to the downstream paths of `x`, optionally passing the last /// Pushes data to the downstream paths of `x`, optionally passing the last
/// result of `desired_buffer_size` for `x` as second argument. /// result of `available_credit` for `x` as second argument.
virtual void push(abstract_downstream& x, size_t* hint = nullptr) = 0; virtual void push(abstract_downstream& x, size_t* hint = nullptr) = 0;
// TODO: callbacks für new and closed downstream pahts // TODO: callbacks für new and closed downstream pahts
......
...@@ -23,6 +23,7 @@ ...@@ -23,6 +23,7 @@
#include <cstddef> #include <cstddef>
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/logger.hpp"
#include "caf/actor_control_block.hpp" #include "caf/actor_control_block.hpp"
namespace caf { namespace caf {
...@@ -32,17 +33,27 @@ namespace mixin { ...@@ -32,17 +33,27 @@ namespace mixin {
template <class Base, class Subtype> template <class Base, class Subtype>
class has_downstreams : public Base { class has_downstreams : public Base {
public: public:
error add_downstream(strong_actor_ptr& ptr, size_t init, error add_downstream(strong_actor_ptr& ptr) final {
bool is_redeployable) final { CAF_LOG_TRACE(CAF_ARG(ptr));
if (!ptr) CAF_ASSERT(ptr != nullptr);
return sec::invalid_argument; if (out().add_path(ptr))
if (out().add_path(ptr, is_redeployable)) {
dptr()->downstream_demand(ptr, init);
return none; return none;
}
return sec::downstream_already_exists; return sec::downstream_already_exists;
} }
error confirm_downstream(const strong_actor_ptr& rebind_from,
strong_actor_ptr& ptr, size_t initial_demand,
bool redeployable) final {
CAF_LOG_TRACE(CAF_ARG(ptr) << CAF_ARG(initial_demand)
<< CAF_ARG(redeployable));
CAF_ASSERT(ptr != nullptr);
if (out().confirm_path(rebind_from, ptr, redeployable)) {
dptr()->downstream_demand(ptr, initial_demand);
return none;
}
return sec::invalid_downstream;
}
error push(size_t* hint = nullptr) final { error push(size_t* hint = nullptr) final {
if (out().buf_size() > 0) if (out().buf_size() > 0)
out().policy().push(out(), hint); out().policy().push(out(), hint);
......
...@@ -33,13 +33,6 @@ namespace mixin { ...@@ -33,13 +33,6 @@ namespace mixin {
template <class Base, class Subtype> template <class Base, class Subtype>
class has_upstreams : public Base { class has_upstreams : public Base {
public: public:
expected<size_t> add_upstream(strong_actor_ptr& ptr, const stream_id& sid,
stream_priority prio) final {
if (!ptr)
return sec::invalid_argument;
return in().add_path(ptr, sid, prio);
}
error close_upstream(strong_actor_ptr& ptr) final { error close_upstream(strong_actor_ptr& ptr) final {
if (in().remove_path(ptr)) { if (in().remove_path(ptr)) {
if (in().closed()) if (in().closed())
...@@ -49,10 +42,6 @@ public: ...@@ -49,10 +42,6 @@ public:
return sec::invalid_upstream; return sec::invalid_upstream;
} }
error pull(size_t amount) final {
return in().pull(amount);
}
private: private:
Subtype* dptr() { Subtype* dptr() {
return static_cast<Subtype*>(this); return static_cast<Subtype*>(this);
......
...@@ -28,7 +28,7 @@ namespace policy { ...@@ -28,7 +28,7 @@ namespace policy {
class anycast final : public downstream_policy { class anycast final : public downstream_policy {
public: public:
void push(abstract_downstream& out, size_t* hint) override; void push(abstract_downstream& out, size_t* hint) override;
size_t desired_buffer_size(const abstract_downstream& out) override; size_t available_credit(const abstract_downstream& out) override;
}; };
} // namespace policy } // namespace policy
......
...@@ -28,7 +28,7 @@ namespace policy { ...@@ -28,7 +28,7 @@ namespace policy {
class broadcast final : public downstream_policy { class broadcast final : public downstream_policy {
public: public:
void push(abstract_downstream& out, size_t* hint) override; void push(abstract_downstream& out, size_t* hint) override;
size_t desired_buffer_size(const abstract_downstream& out) override; size_t available_credit(const abstract_downstream& out) override;
}; };
} // namespace policy } // namespace policy
......
...@@ -28,9 +28,16 @@ namespace policy { ...@@ -28,9 +28,16 @@ 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 final : public upstream_policy {
public: public:
size_t initial_credit(abstract_upstream& x, upstream_path& y) override; greedy();
void reclaim(abstract_upstream& x, upstream_path& y) override; void assign_credit(assignment_vec& xs, size_t buf_size,
size_t downstream_credit) override;
size_t low_watermark;
size_t high_watermark;
size_t min_buffer_size;
}; };
} // namespace policy } // namespace policy
......
...@@ -346,7 +346,8 @@ public: ...@@ -346,7 +346,8 @@ public:
make_mailbox_element( make_mailbox_element(
ctrl(), res_id, {}, ctrl(), res_id, {},
make<stream_msg::open>(sid, make_message_from_tuple(std::move(ys)), make<stream_msg::open>(sid, make_message_from_tuple(std::move(ys)),
ctrl(), stream_priority::normal, false)), ctrl(), actor_cast<strong_actor_ptr>(dest),
stream_priority::normal, false)),
context()); context());
// install response handler // install response handler
this->add_multiplexed_response_handler( this->add_multiplexed_response_handler(
...@@ -359,6 +360,10 @@ public: ...@@ -359,6 +360,10 @@ public:
auto ptr = make_counted<impl>(this, sid, std::move(dpolicy), 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());
// Add downstream with 0 credit.
// TODO: add multiple downstreams when receiving a composed actor handle
auto next = actor_cast<strong_actor_ptr>(dest);
ptr->add_downstream(next);
streams_.emplace(sid, ptr); streams_.emplace(sid, ptr);
return {std::move(sid), std::move(ptr)}; return {std::move(sid), std::move(ptr)};
} }
...@@ -395,7 +400,8 @@ public: ...@@ -395,7 +400,8 @@ public:
>::value, >::value,
"Expected signature `bool (const State&)` for " "Expected signature `bool (const State&)` for "
"closed_predicate function"); "closed_predicate function");
if (current_mailbox_element()->stages.empty()) { auto& stages = current_mailbox_element()->stages;
if (stages.empty()) {
CAF_LOG_ERROR("cannot create a stream data source without downstream"); CAF_LOG_ERROR("cannot create a stream data source without downstream");
auto rp = make_response_promise(); auto rp = make_response_promise();
rp.deliver(sec::no_downstream_stages_defined); rp.deliver(sec::no_downstream_stages_defined);
...@@ -403,6 +409,8 @@ public: ...@@ -403,6 +409,8 @@ public:
} }
stream_id sid{ctrl(), stream_id sid{ctrl(),
new_request_id(message_priority::normal).integer_value()}; new_request_id(message_priority::normal).integer_value()};
auto next = stages.back();
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>;
if (dpolicy == nullptr) if (dpolicy == nullptr)
...@@ -411,6 +419,9 @@ public: ...@@ -411,6 +419,9 @@ public:
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);
// Add downstream with 0 credit.
// TODO: add multiple downstreams when receiving a composed actor handle
ptr->add_downstream(next);
return {std::move(sid), std::move(ptr)}; return {std::move(sid), std::move(ptr)};
} }
...@@ -435,11 +446,14 @@ public: ...@@ -435,11 +446,14 @@ public:
typename detail::get_callable_trait<Init>::fun_sig typename detail::get_callable_trait<Init>::fun_sig
>::value, >::value,
"Expected signature `void (State&)` for init function"); "Expected signature `void (State&)` for init function");
if (current_mailbox_element()->stages.empty()) { auto& stages = current_mailbox_element()->stages;
if (stages.empty()) {
CAF_LOG_ERROR("cannot create a stream data source without downstream"); CAF_LOG_ERROR("cannot create a stream data source without downstream");
return stream_id{nullptr, 0}; return stream_id{nullptr, 0};
} }
auto sid = in.id(); auto sid = in.id();
auto next = stages.back();
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) if (upolicy == nullptr)
...@@ -451,6 +465,9 @@ public: ...@@ -451,6 +465,9 @@ public:
std::move(cleanup)); std::move(cleanup));
init(ptr->state()); init(ptr->state());
streams_.emplace(sid, ptr); streams_.emplace(sid, ptr);
// Add downstream with 0 credit.
// TODO: add multiple downstreams when receiving a composed actor handle
ptr->add_downstream(next);
return {std::move(sid), std::move(ptr)}; return {std::move(sid), std::move(ptr)};
} }
...@@ -462,19 +479,6 @@ public: ...@@ -462,19 +479,6 @@ public:
std::move(fun), std::move(cleanup)); std::move(fun), std::move(cleanup));
} }
/// Adds `hdl` to an already existing stream as a new downstream path. Note
/// that the new downstream path will *not* receive previous (historic) data.
template <class T, class Handle>
expected<void> add_downstream_path(stream<T>& in, Handle hdl) {
auto i = streams_.find(in.id());
if (i != streams_.end()) {
auto ptr = actor_cast<strong_actor_ptr>(std::move(hdl));
i->second->add_downstream(ptr, 5, false);
return unit;
}
return sec::cannot_add_downstream;
}
/// 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>
stream_result<typename stream_sink_trait_t<Fun, Finalize>::output> stream_result<typename stream_sink_trait_t<Fun, Finalize>::output>
...@@ -641,7 +645,7 @@ protected: ...@@ -641,7 +645,7 @@ protected:
make_mailbox_element( make_mailbox_element(
mptr->sender, mptr->mid, std::move(stages), mptr->sender, mptr->mid, std::move(stages),
make<stream_msg::open>(sid, make_message_from_tuple(std::move(ys)), make<stream_msg::open>(sid, make_message_from_tuple(std::move(ys)),
ctrl(), stream_priority::normal, false)), ctrl(), next, stream_priority::normal, false)),
context()); context());
mptr->mid.mark_as_answered(); mptr->mid.mark_as_answered();
} }
......
...@@ -36,14 +36,21 @@ public: ...@@ -36,14 +36,21 @@ public:
// -- handler for downstream events ------------------------------------------ // -- handler for downstream events ------------------------------------------
/// Add a new downstream actor to the stream. /// Add a new downstream actor to the stream with in-flight
/// `stream_msg::open` message.
/// @param hdl Handle to the new downstream actor.
/// @pre `hdl != nullptr`
virtual error add_downstream(strong_actor_ptr& hdl);
/// Confirms a downstream actor after receiving its `stream_msg::ack_open`.
/// @param hdl Handle to the new downstream actor. /// @param hdl Handle to the new downstream actor.
/// @param initial_demand Credit received with `ack_open`. /// @param initial_demand Credit received with `ack_open`.
/// @param redeployable Denotes whether the runtime can redeploy /// @param redeployable Denotes whether the runtime can redeploy
/// the downstream actor on failure. /// the downstream actor on failure.
/// @pre `hdl != nullptr` /// @pre `hdl != nullptr`
virtual error add_downstream(strong_actor_ptr& hdl, size_t initial_demand, virtual error confirm_downstream(const strong_actor_ptr& rebind_from,
bool redeployable); strong_actor_ptr& hdl, size_t initial_demand,
bool redeployable);
/// Handles ACK message from a downstream actor. /// Handles ACK message from a downstream actor.
/// @pre `hdl != nullptr` /// @pre `hdl != nullptr`
......
...@@ -59,8 +59,10 @@ struct stream_msg : tag::boxing_type { ...@@ -59,8 +59,10 @@ struct stream_msg : tag::boxing_type {
/// Contains a type-erased stream<T> object as first argument followed by /// Contains a type-erased stream<T> object as first argument followed by
/// any number of user-defined additional handshake data. /// any number of user-defined additional handshake data.
message msg; message msg;
/// A pointer to the previous stage in the pipeline. /// Identifies the previous stage in the pipeline.
strong_actor_ptr prev_stage; strong_actor_ptr prev_stage;
/// Identifies the original receiver of this message.
strong_actor_ptr original_stage;
/// Configures the priority for stream elements. /// Configures the priority for stream elements.
stream_priority priority; stream_priority priority;
/// Tells the downstream whether rebindings can occur on this path. /// Tells the downstream whether rebindings can occur on this path.
...@@ -74,6 +76,11 @@ struct stream_msg : tag::boxing_type { ...@@ -74,6 +76,11 @@ struct stream_msg : tag::boxing_type {
static constexpr flow_label label = flows_upstream; static constexpr flow_label label = flows_upstream;
/// Allows the testing DSL to unbox this type automagically. /// Allows the testing DSL to unbox this type automagically.
using outer_type = stream_msg; using outer_type = stream_msg;
/// Allows actors to participate in a stream instead of the actor
/// originally receiving the `open` message. No effect when set to
/// `nullptr`. This mechanism enables pipeline definitions consisting of
/// proxy actors that are replaced with actual actors on demand.
strong_actor_ptr rebind_from;
/// Grants credit to the source. /// Grants credit to the source.
int32_t initial_demand; int32_t initial_demand;
/// Tells the upstream whether rebindings can occur on this path. /// Tells the upstream whether rebindings can occur on this path.
...@@ -205,13 +212,14 @@ make(const stream_id& sid, Ts&&... xs) { ...@@ -205,13 +212,14 @@ make(const stream_id& sid, Ts&&... xs) {
template <class Inspector> template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, stream_msg::open& x) { typename Inspector::result_type inspect(Inspector& f, stream_msg::open& x) {
return f(meta::type_name("open"), x.msg, x.prev_stage, x.priority, return f(meta::type_name("open"), x.msg, x.prev_stage, x.original_stage,
x.redeployable); x.priority, x.redeployable);
} }
template <class Inspector> template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, stream_msg::ack_open& x) { typename Inspector::result_type inspect(Inspector& f, stream_msg::ack_open& x) {
return f(meta::type_name("ack_open"), x.initial_demand, x.redeployable); return f(meta::type_name("ack_open"), x.rebind_from, x.initial_demand,
x.redeployable);
} }
template <class Inspector> template <class Inspector>
......
...@@ -48,6 +48,13 @@ public: ...@@ -48,6 +48,13 @@ public:
// nop // nop
} }
expected<size_t> add_upstream(strong_actor_ptr& ptr, const stream_id& sid,
stream_priority prio) final {
if (ptr)
return in().add_path(ptr, sid, prio, 0, 0);
return sec::invalid_argument;
}
error consume(message& msg) final { error consume(message& msg) final {
using vec_type = std::vector<input_type>; using vec_type = std::vector<input_type>;
if (msg.match_elements<vec_type>()) { if (msg.match_elements<vec_type>()) {
......
...@@ -51,6 +51,14 @@ public: ...@@ -51,6 +51,14 @@ public:
// nop // nop
} }
expected<size_t> add_upstream(strong_actor_ptr& ptr, const stream_id& sid,
stream_priority prio) final {
if (ptr)
return in().add_path(ptr, sid, prio, out_.buf_size(),
out_.available_credit());
return sec::invalid_argument;
}
error process_batch(message& msg) final { error process_batch(message& msg) final {
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>()) {
......
...@@ -27,6 +27,8 @@ ...@@ -27,6 +27,8 @@
#include "caf/stream_priority.hpp" #include "caf/stream_priority.hpp"
#include "caf/actor_control_block.hpp" #include "caf/actor_control_block.hpp"
#include "caf/meta/type_name.hpp"
namespace caf { namespace caf {
/// Denotes an upstream actor in a stream topology. Each upstream actor can /// Denotes an upstream actor in a stream topology. Each upstream actor can
...@@ -54,6 +56,12 @@ public: ...@@ -54,6 +56,12 @@ public:
upstream_path(strong_actor_ptr ptr, stream_id id, stream_priority p); upstream_path(strong_actor_ptr ptr, stream_id id, stream_priority p);
}; };
template <class Inspector>
typename Inspector::return_type inspect(Inspector& f, upstream_path& x) {
return f(meta::type_name("upstream_path"), x.hdl, x.sid, x.prio,
x.last_acked_batch_id, x.last_batch_id, x.assigned_credit);
}
} // namespace caf } // namespace caf
#endif // CAF_UPSTREAM_PATH_HPP #endif // CAF_UPSTREAM_PATH_HPP
...@@ -20,7 +20,9 @@ ...@@ -20,7 +20,9 @@
#ifndef CAF_UPSTREAM_POLICY_HPP #ifndef CAF_UPSTREAM_POLICY_HPP
#define CAF_UPSTREAM_POLICY_HPP #define CAF_UPSTREAM_POLICY_HPP
#include <vector>
#include <cstdint> #include <cstdint>
#include <utility>
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
...@@ -30,11 +32,22 @@ class upstream_policy { ...@@ -30,11 +32,22 @@ class upstream_policy {
public: public:
virtual ~upstream_policy(); virtual ~upstream_policy();
/// Queries the initial credit for the new path `y` in `x`. /// Describes an assignment of credit to an upstream actor.
virtual size_t initial_credit(abstract_upstream& x, upstream_path& y) = 0; using assignment_pair = std::pair<upstream_path*, size_t>;
/// Reclaim credit of closed upstream `y`. /// Describes an assignment of credit to all upstream actors.
virtual void reclaim(abstract_upstream& x, upstream_path& y) = 0; using assignment_vec = std::vector<assignment_pair>;
/// Assigns credit to upstream actors.
/// @param xs Stores assignment decisions. Note that the second element of
/// each pair is uninitialized and must be set to 0 for all paths
/// that do not receive credit.
/// @param buf_size Denotes how many stream elements are currently buffered.
/// @param downstream_credit Denotes how many items we could send downstream.
/// Usually 0, unless the downstream policy avoids
/// small batches.
virtual void assign_credit(assignment_vec& xs, size_t buf_size,
size_t downstream_credit) = 0;
}; };
} // namespace caf } // namespace caf
......
...@@ -57,26 +57,44 @@ size_t abstract_downstream::min_credit() const { ...@@ -57,26 +57,44 @@ size_t abstract_downstream::min_credit() const {
return fold_paths(std::numeric_limits<size_t>::max(), f); return fold_paths(std::numeric_limits<size_t>::max(), f);
} }
bool abstract_downstream::add_path(strong_actor_ptr ptr, bool redeployable) { bool abstract_downstream::add_path(strong_actor_ptr ptr) {
auto predicate = [&](const path_uptr& x) { return x->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)) { if (std::none_of(paths_.begin(), paths_.end(), predicate)) {
paths_.emplace_back(new path(std::move(ptr), redeployable)); 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; return true;
} }
return false; return false;
} }
bool abstract_downstream::remove_path(strong_actor_ptr& ptr) { bool abstract_downstream::remove_path(strong_actor_ptr& ptr) {
auto predicate = [&](const path_uptr& x) { return x->ptr == ptr; }; auto predicate = [&](const path_uptr& x) { return x->hdl == ptr; };
auto e = paths_.end(); auto e = paths_.end();
auto i = std::find_if(paths_.begin(), e, predicate); auto i = std::find_if(paths_.begin(), e, predicate);
if (i != e) { if (i != e) {
CAF_ASSERT((*i)->ptr != nullptr); CAF_ASSERT((*i)->hdl != nullptr);
if (i != paths_.end() - 1) if (i != paths_.end() - 1)
std::swap(*i, paths_.back()); std::swap(*i, paths_.back());
auto x = std::move(paths_.back()); auto x = std::move(paths_.back());
paths_.pop_back(); paths_.pop_back();
unsafe_send_as(self_, x->ptr, make<stream_msg::close>(sid_)); unsafe_send_as(self_, x->hdl, make<stream_msg::close>(sid_));
//policy_->reclaim(this, x); //policy_->reclaim(this, x);
return true; return true;
} }
...@@ -85,21 +103,21 @@ bool abstract_downstream::remove_path(strong_actor_ptr& ptr) { ...@@ -85,21 +103,21 @@ bool abstract_downstream::remove_path(strong_actor_ptr& ptr) {
void abstract_downstream::close() { void abstract_downstream::close() {
for (auto& x : paths_) for (auto& x : paths_)
unsafe_send_as(self_, x->ptr, make<stream_msg::close>(sid_)); unsafe_send_as(self_, x->hdl, make<stream_msg::close>(sid_));
paths_.clear(); paths_.clear();
} }
void abstract_downstream::abort(strong_actor_ptr& cause, const error& reason) { void abstract_downstream::abort(strong_actor_ptr& cause, const error& reason) {
for (auto& x : paths_) for (auto& x : paths_)
if (x->ptr != cause) if (x->hdl != cause)
unsafe_send_as(self_, x->ptr, unsafe_send_as(self_, x->hdl,
make<stream_msg::abort>(this->sid_, reason)); make<stream_msg::abort>(this->sid_, reason));
} }
auto abstract_downstream::find(const strong_actor_ptr& ptr) const auto abstract_downstream::find(const strong_actor_ptr& ptr) const
-> optional<path&> { -> optional<path&> {
auto predicate = [&](const path_uptr& y) { auto predicate = [&](const path_uptr& y) {
return y->ptr == ptr; return y->hdl == ptr;
}; };
auto e = paths_.end(); auto e = paths_.end();
auto i = std::find_if(paths_.begin(), e, predicate); auto i = std::find_if(paths_.begin(), e, predicate);
...@@ -108,6 +126,10 @@ auto abstract_downstream::find(const strong_actor_ptr& ptr) const ...@@ -108,6 +126,10 @@ auto abstract_downstream::find(const strong_actor_ptr& ptr) const
return none; return none;
} }
size_t abstract_downstream::available_credit() const {
return policy().available_credit(*this);
}
void abstract_downstream::send_batch(downstream_path& dest, size_t chunk_size, void abstract_downstream::send_batch(downstream_path& dest, size_t chunk_size,
message chunk) { message chunk) {
auto scs = static_cast<int32_t>(chunk_size); auto scs = static_cast<int32_t>(chunk_size);
...@@ -115,7 +137,7 @@ void abstract_downstream::send_batch(downstream_path& dest, size_t chunk_size, ...@@ -115,7 +137,7 @@ void abstract_downstream::send_batch(downstream_path& dest, size_t chunk_size,
stream_msg::batch batch{scs, std::move(chunk), batch_id}; stream_msg::batch batch{scs, std::move(chunk), batch_id};
if (dest.redeployable) if (dest.redeployable)
dest.unacknowledged_batches.emplace_back(batch_id, batch); dest.unacknowledged_batches.emplace_back(batch_id, batch);
unsafe_send_as(self_, dest.ptr, stream_msg{sid_, std::move(batch)}); unsafe_send_as(self_, dest.hdl, stream_msg{sid_, std::move(batch)});
} }
void abstract_downstream::sort_by_credit() { void abstract_downstream::sort_by_credit() {
......
...@@ -41,53 +41,44 @@ void abstract_upstream::abort(strong_actor_ptr& cause, const error& reason) { ...@@ -41,53 +41,44 @@ void abstract_upstream::abort(strong_actor_ptr& cause, const error& reason) {
unsafe_send_as(self_, x->hdl, make<stream_msg::abort>(x->sid, reason)); unsafe_send_as(self_, x->hdl, make<stream_msg::abort>(x->sid, reason));
} }
error abstract_upstream::pull(strong_actor_ptr& hdl, size_t n) { void abstract_upstream::assign_credit(size_t buf_size,
CAF_ASSERT(hdl != nullptr); size_t downstream_credit) {
auto has_hdl = [&](const path_uptr& x) { policy_->assign_credit(policy_vec_, buf_size, downstream_credit);
CAF_ASSERT(x != nullptr); for (auto& x : policy_vec_) {
return x->hdl == hdl; auto n = x.second;
}; if (n > 0) {
auto e = paths_.end(); auto ptr = x.first;
auto i = std::find_if(paths_.begin(), e, has_hdl); ptr->assigned_credit += n;
if (i != e) { unsafe_send_as(self_, ptr->hdl, make<stream_msg::ack_batch>(
auto& x = *(*i); ptr->sid, static_cast<int32_t>(n),
CAF_ASSERT(x.hdl == hdl); ptr->last_batch_id++));
x.assigned_credit += n; }
unsafe_send_as(self_, x.hdl,
make<stream_msg::ack_batch>(x.sid, static_cast<int32_t>(n),
x.last_batch_id++));
return none;
}
return sec::invalid_upstream;
}
error abstract_upstream::pull(size_t n) {
// TODO: upstream policy
if (!paths_.empty()) {
pull(paths_.front()->hdl, n);
return none;
} }
return sec::invalid_upstream;
} }
expected<size_t> abstract_upstream::add_path(strong_actor_ptr hdl, expected<size_t> abstract_upstream::add_path(strong_actor_ptr hdl,
const stream_id& sid, const stream_id& sid,
stream_priority prio) { stream_priority prio,
size_t buf_size,
size_t downstream_credit) {
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(sid) << CAF_ARG(prio));
CAF_ASSERT(hdl != nullptr); CAF_ASSERT(hdl != nullptr);
auto has_hdl = [&](const path_uptr& x) { if (!find(hdl)) {
CAF_ASSERT(x != nullptr); CAF_LOG_DEBUG("add new upstraem path" << CAF_ARG(hdl));
return x->hdl == hdl;
};
auto e = paths_.end();
auto i = std::find_if(paths_.begin(), e, has_hdl);
if (i == e) {
paths_.emplace_back(new path(std::move(hdl), sid, prio)); paths_.emplace_back(new path(std::move(hdl), sid, prio));
return policy_->initial_credit(*this, *paths_.back()); 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, buf_size, downstream_credit);
paths_.back()->assigned_credit += tmp.back().second;
return tmp.back().second;
} }
return sec::upstream_already_exists; return sec::upstream_already_exists;
} }
bool abstract_upstream::remove_path(const strong_actor_ptr& hdl) { bool abstract_upstream::remove_path(const strong_actor_ptr& hdl) {
// Find element in our paths list.
auto has_hdl = [&](const path_uptr& x) { auto has_hdl = [&](const path_uptr& x) {
CAF_ASSERT(x != nullptr); CAF_ASSERT(x != nullptr);
return x->hdl == hdl; return x->hdl == hdl;
...@@ -95,6 +86,17 @@ bool abstract_upstream::remove_path(const strong_actor_ptr& hdl) { ...@@ -95,6 +86,17 @@ bool abstract_upstream::remove_path(const strong_actor_ptr& hdl) {
auto e = paths_.end(); auto e = paths_.end();
auto i = std::find_if(paths_.begin(), e, has_hdl); auto i = std::find_if(paths_.begin(), e, has_hdl);
if (i != e) { 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) if (i != e - 1)
std::swap(*i, paths_.back()); std::swap(*i, paths_.back());
paths_.pop_back(); paths_.pop_back();
...@@ -103,4 +105,16 @@ bool abstract_upstream::remove_path(const strong_actor_ptr& hdl) { ...@@ -103,4 +105,16 @@ bool abstract_upstream::remove_path(const strong_actor_ptr& hdl) {
return false; 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 } // namespace caf
...@@ -28,7 +28,7 @@ void anycast::push(abstract_downstream& out, size_t* hint) { ...@@ -28,7 +28,7 @@ void anycast::push(abstract_downstream& out, size_t* hint) {
out.anycast(hint); out.anycast(hint);
} }
size_t anycast::desired_buffer_size(const abstract_downstream& out) { size_t anycast::available_credit(const abstract_downstream& out) {
return out.total_credit(); return out.total_credit();
} }
......
...@@ -28,7 +28,7 @@ void broadcast::push(abstract_downstream& out, size_t* hint) { ...@@ -28,7 +28,7 @@ void broadcast::push(abstract_downstream& out, size_t* hint) {
out.broadcast(hint); out.broadcast(hint);
} }
size_t broadcast::desired_buffer_size(const abstract_downstream& out) { size_t broadcast::available_credit(const abstract_downstream& out) {
return out.min_credit(); return out.min_credit();
} }
......
...@@ -22,7 +22,7 @@ ...@@ -22,7 +22,7 @@
namespace caf { namespace caf {
downstream_path::downstream_path(strong_actor_ptr p, bool redeploy) downstream_path::downstream_path(strong_actor_ptr p, bool redeploy)
: ptr(std::move(p)), : hdl(std::move(p)),
next_batch_id(0), next_batch_id(0),
open_credit(0), open_credit(0),
redeployable(redeploy) { redeployable(redeploy) {
......
...@@ -27,8 +27,4 @@ downstream_policy::~downstream_policy() { ...@@ -27,8 +27,4 @@ downstream_policy::~downstream_policy() {
// nop // nop
} }
atom_value downstream_policy::categorize(type_erased_value&) const {
return atom("");
}
} // namespace caf } // namespace caf
...@@ -19,15 +19,48 @@ ...@@ -19,15 +19,48 @@
#include "caf/policy/greedy.hpp" #include "caf/policy/greedy.hpp"
#include <numeric>
#include "caf/logger.hpp"
#include "caf/upstream_path.hpp"
namespace caf { namespace caf {
namespace policy { namespace policy {
size_t greedy::initial_credit(abstract_upstream&, upstream_path&) { greedy::greedy() : low_watermark(0), high_watermark(5), min_buffer_size(5) {
return 5; // nop
} }
void greedy::reclaim(abstract_upstream&, upstream_path&) { void greedy::assign_credit(assignment_vec& xs, size_t buf_size,
// nop size_t downstream_credit) {
CAF_LOG_TRACE(CAF_ARG(xs) << CAF_ARG(buf_size) << CAF_ARG(downstream_credit));
/// Calculate how much credit we can hand out and how much credit we have
/// already assigned. Buffered elements are counted as assigned credit,
/// because we "release" credit only after pushing elements downstram.
size_t max_available = downstream_credit + min_buffer_size;
size_t assigned =
buf_size + std::accumulate(xs.begin(), xs.end(), size_t{0},
[](size_t x, const assignment_pair& y) {
return x + y.first->assigned_credit;
});
if (assigned >= max_available) {
// zero out assignment vector
for (auto& x : xs)
x.second = 0;
return;
}
// Assign credit to upstream paths until no more credit is available. We must
// make sure to write to each element in the vector.
auto available = max_available - assigned;
for (auto& p : xs) {
auto& x = p.first->assigned_credit;
if (x < high_watermark) {
p.second = std::min(high_watermark - x, available);
available -= p.second;
} else {
p.second = 0;
}
}
} }
} // namespace policy } // namespace policy
......
...@@ -67,7 +67,8 @@ void incoming_stream_multiplexer::operator()(stream_msg::open& x) { ...@@ -67,7 +67,8 @@ void incoming_stream_multiplexer::operator()(stream_msg::open& x) {
auto ptr = make_mailbox_element( auto ptr = make_mailbox_element(
cme->sender, cme->mid, std::move(cme->stages), cme->sender, cme->mid, std::move(cme->stages),
make<stream_msg::open>(current_stream_msg_->sid, std::move(x.msg), make<stream_msg::open>(current_stream_msg_->sid, std::move(x.msg),
self_->ctrl(), x.priority, x.redeployable)); self_->ctrl(), successor, x.priority,
x.redeployable));
successor->enqueue(std::move(ptr), self_->context()); successor->enqueue(std::move(ptr), self_->context());
// Send out demand upstream. // Send out demand upstream.
manage_credit(); manage_credit();
......
...@@ -77,9 +77,9 @@ void outgoing_stream_multiplexer::operator()(stream_msg::open& x) { ...@@ -77,9 +77,9 @@ void outgoing_stream_multiplexer::operator()(stream_msg::open& x) {
auto ptr = make_mailbox_element( auto ptr = make_mailbox_element(
cme->sender, message_id::make(), {}, forward_atom::value, cme->sender, cme->sender, message_id::make(), {}, forward_atom::value, cme->sender,
std::move(cme->stages), path->hdl, cme->mid, std::move(cme->stages), path->hdl, cme->mid,
make_message(make<stream_msg::open>(current_stream_msg_->sid, make_message(make<stream_msg::open>(
std::move(x.msg), self_->ctrl(), current_stream_msg_->sid, std::move(x.msg), self_->ctrl(),
x.priority, x.redeployable))); x.original_stage, x.priority, x.redeployable)));
basp()->enqueue(std::move(ptr), self_->context()); basp()->enqueue(std::move(ptr), self_->context());
} }
......
...@@ -31,7 +31,13 @@ stream_handler::~stream_handler() { ...@@ -31,7 +31,13 @@ stream_handler::~stream_handler() {
// nop // nop
} }
error stream_handler::add_downstream(strong_actor_ptr&, size_t, bool) { error stream_handler::add_downstream(strong_actor_ptr&) {
CAF_LOG_ERROR("Cannot add downstream to a stream marked as no-downstreams");
return sec::cannot_add_downstream;
}
error stream_handler::confirm_downstream(const strong_actor_ptr&,
strong_actor_ptr&, size_t, bool) {
CAF_LOG_ERROR("Cannot add downstream to a stream marked as no-downstreams"); CAF_LOG_ERROR("Cannot add downstream to a stream marked as no-downstreams");
return sec::cannot_add_downstream; return sec::cannot_add_downstream;
} }
......
...@@ -72,12 +72,11 @@ auto stream_msg_visitor::operator()(stream_msg::open& x) -> result_type { ...@@ -72,12 +72,11 @@ auto stream_msg_visitor::operator()(stream_msg::open& x) -> result_type {
// store upstream actor // store upstream actor
auto initial_credit = handler->add_upstream(x.prev_stage, sid_, x.priority); auto initial_credit = handler->add_upstream(x.prev_stage, sid_, x.priority);
if (initial_credit) { if (initial_credit) {
// check whether we are a stage in a longer pipeline and send more // send ACK to predecessor
// stream_open messages if required
auto ic = static_cast<int32_t>(*initial_credit); auto ic = static_cast<int32_t>(*initial_credit);
unsafe_send_as(self_, predecessor, unsafe_send_as(self_, predecessor,
make_message(make<stream_msg::ack_open>(std::move(sid_), ic, make_message(make<stream_msg::ack_open>(
false))); std::move(sid_), std::move(x.original_stage), ic, false)));
return {none, i_}; return {none, i_};
} }
self_->streams().erase(i_); self_->streams().erase(i_);
...@@ -104,11 +103,12 @@ auto stream_msg_visitor::operator()(stream_msg::abort& x) -> result_type { ...@@ -104,11 +103,12 @@ auto stream_msg_visitor::operator()(stream_msg::abort& x) -> result_type {
auto stream_msg_visitor::operator()(stream_msg::ack_open& x) -> result_type { auto stream_msg_visitor::operator()(stream_msg::ack_open& x) -> result_type {
CAF_LOG_TRACE(CAF_ARG(x)); CAF_LOG_TRACE(CAF_ARG(x));
if (i_ != e_) if (i_ != e_) {
return {i_->second->add_downstream(self_->current_sender(), auto d = static_cast<size_t>(x.initial_demand);
static_cast<size_t>(x.initial_demand), return {i_->second->confirm_downstream(x.rebind_from,
false), self_->current_sender(), d, false),
i_}; i_};
}
CAF_LOG_DEBUG("received stream_msg::ok for unknown stream"); CAF_LOG_DEBUG("received stream_msg::ok for unknown stream");
return {sec::unexpected_message, e_}; return {sec::unexpected_message, e_};
} }
......
...@@ -41,12 +41,20 @@ bool stream_sink::done() const { ...@@ -41,12 +41,20 @@ bool stream_sink::done() const {
} }
error stream_sink::upstream_batch(strong_actor_ptr& src, size_t xs_size, error stream_sink::upstream_batch(strong_actor_ptr& hdl, size_t xs_size,
message& xs) { message& xs) {
auto err = in_ptr_->pull(src, xs_size); CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(xs_size) << CAF_ARG(xs));
if (!err) auto path = in().find(hdl);
consume(xs); if (path) {
return err; if (xs_size > path->assigned_credit)
return sec::invalid_stream_state;
path->assigned_credit -= xs_size;
auto err = consume(xs);
if (err == none)
in().assign_credit(0, 0);
return err;
}
return sec::invalid_upstream;
} }
void stream_sink::abort(strong_actor_ptr& cause, const error& reason) { void stream_sink::abort(strong_actor_ptr& cause, const error& reason) {
......
...@@ -48,7 +48,7 @@ error stream_source::downstream_demand(strong_actor_ptr& hdl, size_t value) { ...@@ -48,7 +48,7 @@ error stream_source::downstream_demand(strong_actor_ptr& hdl, size_t 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_ptr_->policy().desired_buffer_size(*out_ptr_); auto size_hint = out().available_credit();
if (current_size < size_hint) if (current_size < size_hint)
generate(size_hint - current_size); generate(size_hint - current_size);
return push(&size_hint); return push(&size_hint);
......
...@@ -40,27 +40,32 @@ bool stream_stage::done() const { ...@@ -40,27 +40,32 @@ bool stream_stage::done() const {
} }
error stream_stage::upstream_batch(strong_actor_ptr& hdl, size_t xs_size, error stream_stage::upstream_batch(strong_actor_ptr& hdl, size_t xs_size,
message& xs) { message& xs) {
auto err = in_ptr_->pull(hdl, xs_size); CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(xs_size) << CAF_ARG(xs));
if (!err) { auto path = in().find(hdl);
process_batch(xs); if (path) {
push(); if (xs_size > path->assigned_credit)
return sec::invalid_stream_state;
path->assigned_credit -= xs_size;
auto err = process_batch(xs);
if (err == none) {
push();
in().assign_credit(out().buf_size(), out().available_credit());
}
return err;
} }
return err; return sec::invalid_upstream;
} }
error stream_stage::downstream_demand(strong_actor_ptr& hdl, size_t value) { error stream_stage::downstream_demand(strong_actor_ptr& hdl, size_t value) {
auto path = out_ptr_->find(hdl); auto path = out_ptr_->find(hdl);
if (path) { if (path) {
path->open_credit += value; path->open_credit += value;
if(out_ptr_->buf_size() > 0) { if(out().buf_size() > 0)
return push(); push();
} else if (in().closed() && !out().remove_path(hdl))
if (in_ptr_->closed()) { return sec::invalid_downstream;
if (!out_ptr_->remove_path(hdl)) { in().assign_credit(out().buf_size(), out().available_credit());
return sec::invalid_downstream;
}
}
return none; return none;
} }
return sec::invalid_downstream; return sec::invalid_downstream;
......
...@@ -253,7 +253,7 @@ CAF_TEST(stream_interception) { ...@@ -253,7 +253,7 @@ CAF_TEST(stream_interception) {
// streamer [via stream_serv1 / BASP] --(stream_msg::open)--> stream_serv2 // streamer [via stream_serv1 / BASP] --(stream_msg::open)--> stream_serv2
expect_network_traffic(streamer, stream_serv2); expect_network_traffic(streamer, stream_serv2);
expect((stream_msg::open), expect((stream_msg::open),
from(_).to(stream_serv2).with(_, stream_serv1, _, _, false)); from(_).to(stream_serv2).with(_, stream_serv1, _, _, _, false));
// stream_serv2 [via BASP] --('sys', 'ok', 5)--> stream_serv1 // stream_serv2 [via BASP] --('sys', 'ok', 5)--> stream_serv1
expect_network_traffic(stream_serv2, stream_serv1); expect_network_traffic(stream_serv2, stream_serv1);
expect((atom_value, atom_value, int32_t), expect((atom_value, atom_value, int32_t),
...@@ -261,14 +261,14 @@ CAF_TEST(stream_interception) { ...@@ -261,14 +261,14 @@ CAF_TEST(stream_interception) {
.with(sys_atom::value, ok_atom::value, 5)); .with(sys_atom::value, ok_atom::value, 5));
// stream_serv2 --(stream_msg::open)--> sum_up // stream_serv2 --(stream_msg::open)--> sum_up
expect((stream_msg::open), expect((stream_msg::open),
from(_).to(sum_up).with(_, stream_serv2, _, _, false)); from(_).to(sum_up).with(_, stream_serv2, _, _, _, false));
// sum_up --(stream_msg::ack_open)--> stream_serv2 // sum_up --(stream_msg::ack_open)--> stream_serv2
expect((stream_msg::ack_open), expect((stream_msg::ack_open),
from(sum_up).to(stream_serv2).with(5, _, false)); from(sum_up).to(stream_serv2).with(_, 5, _, false));
// stream_serv2 [via BASP] --(stream_msg::ack_open)--> stream_serv1 // stream_serv2 [via BASP] --(stream_msg::ack_open)--> stream_serv1
expect_network_traffic(stream_serv2, stream_serv1); expect_network_traffic(stream_serv2, stream_serv1);
expect((stream_msg::ack_open), expect((stream_msg::ack_open),
from(stream_serv2).to(stream_serv1).with(5, _, false)); from(stream_serv2).to(stream_serv1).with(_, 5, _, false));
// stream_serv1 --('sys', 'ok', 5)--> stream_serv2 // stream_serv1 --('sys', 'ok', 5)--> stream_serv2
expect_network_traffic(stream_serv1, stream_serv2); expect_network_traffic(stream_serv1, stream_serv2);
expect((atom_value, atom_value, int32_t), expect((atom_value, atom_value, int32_t),
...@@ -276,7 +276,7 @@ CAF_TEST(stream_interception) { ...@@ -276,7 +276,7 @@ CAF_TEST(stream_interception) {
.with(sys_atom::value, ok_atom::value, 5)); .with(sys_atom::value, ok_atom::value, 5));
// stream_serv1 --(stream_msg::ack_open)--> streamer // stream_serv1 --(stream_msg::ack_open)--> streamer
expect((stream_msg::ack_open), expect((stream_msg::ack_open),
from(stream_serv1).to(streamer).with(5, _, false)); from(stream_serv1).to(streamer).with(_, 5, _, false));
// streamer --(stream_msg::batch)--> stream_serv1 // streamer --(stream_msg::batch)--> stream_serv1
expect((stream_msg::batch), from(streamer).to(stream_serv1).with(5, _, 0)); expect((stream_msg::batch), from(streamer).to(stream_serv1).with(5, _, 0));
// stream_serv1 [via BASP] --(stream_msg::batch)--> stream_serv2 // stream_serv1 [via BASP] --(stream_msg::batch)--> stream_serv2
......
This diff is collapsed.
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* 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. *
******************************************************************************/
// this test is essentially a subset of streaming.cpp and tests whether the
// 3-stage pipeline compiles and runs when using a type-safe version of each
// stage
#include <deque>
#include <string>
#include <vector>
#define CAF_SUITE typed_stream_stages
#include "caf/test/dsl.hpp"
using std::string;
using namespace caf;
using file_reader_actor = typed_actor<replies_to<string>
::with<stream<int>, string>>;
using filter_actor = typed_actor<replies_to<stream<int>, string>
::with<stream<int>, string>>;
using sum_up_actor = typed_actor<replies_to<stream<int>, string>::with<int>>;
file_reader_actor::behavior_type file_reader(file_reader_actor::pointer self) {
using buf = std::deque<int>;
return {
[=](std::string& fname) {
CAF_CHECK_EQUAL(fname, "test.txt");
return self->add_source(
// forward file name in handshake to next stage
std::forward_as_tuple(std::move(fname)),
// initialize state
[=](buf& xs) {
xs = buf{1, 2, 3, 4, 5, 6, 7, 8, 9};
},
// get next element
[=](buf& xs, downstream<int>& out, size_t num) {
auto n = std::min(num, xs.size());
for (size_t i = 0; i < n; ++i)
out.push(xs[i]);
xs.erase(xs.begin(), xs.begin() + static_cast<ptrdiff_t>(n));
},
// check whether we reached the end
[=](const buf& xs) {
return xs.empty();
}
);
}
};
}
filter_actor::behavior_type filter(filter_actor::pointer self) {
return {
[=](stream<int>& in, std::string& fname) {
CAF_CHECK_EQUAL(fname, "test.txt");
return self->add_stage(
// input stream
in,
// forward file name in handshake to next stage
std::forward_as_tuple(std::move(fname)),
// initialize state
[=](unit_t&) {
// nop
},
// processing step
[=](unit_t&, downstream<int>& out, int x) {
if ((x & 0x01) != 0)
out.push(x);
},
// cleanup
[=](unit_t&) {
// nop
}
);
}
};
}
sum_up_actor::behavior_type sum_up(sum_up_actor::pointer self) {
return {
[=](stream<int>& in, std::string& fname) {
CAF_CHECK_EQUAL(fname, "test.txt");
return self->add_sink(
// input stream
in,
// initialize state
[](int& x) {
x = 0;
},
// processing step
[](int& x, int y) {
x += y;
},
// cleanup and produce result message
[](int& x) -> int {
return x;
}
);
}
};
}
using pipeline_actor = typed_actor<replies_to<string>::with<int>>;
CAF_TEST(depth3_pipeline) {
actor_system_config cfg;
actor_system sys{cfg};
scoped_actor self{sys};
auto source = sys.spawn(file_reader);
auto stage = sys.spawn(filter);
auto sink = sys.spawn(sum_up);
auto pipeline = sink * stage * source;
static_assert(std::is_same<decltype(pipeline), pipeline_actor>::value,
"pipeline composition returned wrong type");
self->request(pipeline, infinite, "test.txt").receive(
[&](int x) {
CAF_CHECK_EQUAL(x, 25);
},
[&](error& err) {
CAF_FAIL("error: " << sys.render(err));
}
);
}
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