Commit 32321ab7 authored by Dominik Charousset's avatar Dominik Charousset

Eaglery create outbound paths and refine API

* create paths eagerly instead of delaying path creation until
  receiving ack_open
* create outbound paths in "pending mode", i.e., do not emit any batches
  on an outbound path with receiver slot 0
* remove unused state from `stream<T>`
* clean up implemenation for add_outbound_path and add_inbound_path
parent 994385fb
...@@ -41,6 +41,9 @@ public: ...@@ -41,6 +41,9 @@ public:
/// Type of `paths_`. /// Type of `paths_`.
using typename super::map_type; using typename super::map_type;
/// Unique pointer to an outbound path.
using typename super::unique_path_ptr;
/// Enables or disables output per path. /// Enables or disables output per path.
using filter_type = Filter; using filter_type = Filter;
...@@ -127,18 +130,24 @@ public: ...@@ -127,18 +130,24 @@ public:
// -- overridden functions --------------------------------------------------- // -- overridden functions ---------------------------------------------------
typename super::path_ptr add_path(stream_slots slots, bool insert_path(unique_path_ptr ptr) override {
strong_actor_ptr target) override { CAF_LOG_TRACE(CAF_ARG(ptr));
CAF_LOG_TRACE(CAF_ARG(slots) << CAF_ARG(target)); // Make sure state_map_ and paths_ are always equally sorted, otherwise
// Make sure we have state for the slot. // we'll run into UB when calling `zip_foreach`.
state_map_.emplace(slots.sender, path_state{}); CAF_ASSERT(state_map_.size() == this->paths_.size());
// Append to `paths_`. auto slot = ptr->slots.sender;
auto index = this->paths_.size(); // Append to the regular path map.
auto result = super::add_path(slots, target); if (!super::insert_path(std::move(ptr)))
if (result == nullptr) return false;
return nullptr; // Append to the state map.
if (!state_map_.emplace(slot, path_state{}).second) {
super::remove_path(slot, none, true);
return false;
}
return true;
// Make sure state_map_ and paths_ are always equally sorted, otherwise // Make sure state_map_ and paths_ are always equally sorted, otherwise
// we'll run into UB when calling `zip_foreach`. // we'll run into UB when calling `zip_foreach`.
/*
CAF_ASSERT(index == this->paths_.size() - 1); CAF_ASSERT(index == this->paths_.size() - 1);
CAF_ASSERT(result->slots == slots); CAF_ASSERT(result->slots == slots);
CAF_ASSERT(this->paths_.container().back().first == slots.sender); CAF_ASSERT(this->paths_.container().back().first == slots.sender);
...@@ -152,6 +161,7 @@ public: ...@@ -152,6 +161,7 @@ public:
swap(ys[index], *i); swap(ys[index], *i);
} }
return result; return result;
*/
} }
void emit_batches() override { void emit_batches() override {
......
...@@ -88,10 +88,6 @@ public: ...@@ -88,10 +88,6 @@ public:
return buf_.size(); return buf_.size();
} }
message make_handshake_token(stream_slot slot) const override {
return make_message(stream<T>{slot});
}
buffer_type& buf() { buffer_type& buf() {
return buf_; return buf_;
} }
......
...@@ -56,12 +56,6 @@ public: ...@@ -56,12 +56,6 @@ public:
delegate(x); delegate(x);
} }
void operator()(stream_slot in, stream_slot out,
stream_manager_ptr& mgr) override {
// TODO: error handling
self_->build_pipeline(in, out, std::move(mgr));
}
private: private:
void deliver(response_promise& rp, error& x) { void deliver(response_promise& rp, error& x) {
CAF_LOG_DEBUG("report error back to requesting actor"); CAF_LOG_DEBUG("report error back to requesting actor");
......
...@@ -61,11 +61,6 @@ public: ...@@ -61,11 +61,6 @@ public:
/// Called if the message handler returned any "ordinary" value. /// Called if the message handler returned any "ordinary" value.
virtual void operator()(message&) = 0; virtual void operator()(message&) = 0;
/// Called if the message handler returned an `output_stream<...>` or a
/// `stream_result<...>`.
virtual void operator()(stream_slot in, stream_slot out,
stream_manager_ptr& mgr) = 0;
/// Called if the message handler returns "nothing", for example a /// Called if the message handler returns "nothing", for example a
/// default-constructed `optional<T>`. /// default-constructed `optional<T>`.
virtual void operator()(const none_t&) = 0; virtual void operator()(const none_t&) = 0;
...@@ -146,16 +141,14 @@ public: ...@@ -146,16 +141,14 @@ public:
/// Calls `(*this)(x.in(), x.out(), x.ptr())`. /// Calls `(*this)(x.in(), x.out(), x.ptr())`.
template <class Out, class Tuple, class P> template <class Out, class Tuple, class P>
void operator()(output_stream<Out, Tuple, P>& x) { void operator()(output_stream<Out, Tuple, P>&) {
stream_manager_ptr ptr{std::move(x.ptr())}; (*this)();
(*this)(x.in(), x.out(), ptr);
} }
/// Calls `(*this)(x.in(), 0, x.ptr())`. /// Calls `(*this)(x.in(), 0, x.ptr())`.
template <class T, class P> template <class T>
void operator()(stream_result<T, P>& x) { void operator()(stream_result<T>&) {
stream_manager_ptr ptr{std::move(x.ptr())}; (*this)();
(*this)(x.in(), 0, ptr);
} }
// -- visit API: return true if T was visited, false if T was skipped -------- // -- visit API: return true if T was visited, false if T was skipped --------
......
...@@ -28,16 +28,16 @@ namespace caf { ...@@ -28,16 +28,16 @@ namespace caf {
namespace detail { namespace detail {
/// Identifies an unbound sequence of messages. /// Identifies an unbound sequence of messages.
template <class Input, class Result, class Process, class Finalize> template <class Input, class Process, class Finalize>
class stream_sink_driver_impl final : public stream_sink_driver<Input, Result> { class stream_sink_driver_impl final : public stream_sink_driver<Input> {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
using super = stream_sink_driver<Input, Result>; using super = stream_sink_driver<Input>;
using typename super::input_type; using typename super::input_type;
using trait = stream_sink_trait_t<Process, Finalize>; using trait = stream_sink_trait_t<Process>;
using state_type = typename trait::state; using state_type = typename trait::state;
...@@ -48,12 +48,12 @@ public: ...@@ -48,12 +48,12 @@ public:
init(state_); init(state_);
} }
void process(std::vector<input_type>&& xs) override { void process(std::vector<input_type>& xs) override {
return trait::process::invoke(process_, state_, std::move(xs)); return trait::process::invoke(process_, state_, xs);
} }
message make_final_result() override { void finalize(const error& err) override {
return trait::finalize::invoke(finalize_, state_); finalize_(state_, err);
} }
private: private:
......
...@@ -64,18 +64,13 @@ public: ...@@ -64,18 +64,13 @@ public:
CAF_LOG_TRACE(CAF_ARG(x)); CAF_LOG_TRACE(CAF_ARG(x));
using vec_type = std::vector<input_type>; using vec_type = std::vector<input_type>;
if (x.xs.match_elements<vec_type>()) { if (x.xs.match_elements<vec_type>()) {
auto& xs = x.xs.get_mutable_as<vec_type>(0); driver_.process(x.xs.get_mutable_as<vec_type>(0));
driver_.process(std::move(xs));
return; return;
} }
CAF_LOG_ERROR("received unexpected batch type (dropped)"); CAF_LOG_ERROR("received unexpected batch type (dropped)");
} }
protected: protected:
message make_final_result() override {
return driver_.make_final_result();
}
void finalize(const error& reason) override { void finalize(const error& reason) override {
driver_.finalize(reason); driver_.finalize(reason);
} }
......
...@@ -27,48 +27,28 @@ ...@@ -27,48 +27,28 @@
namespace caf { namespace caf {
namespace detail { namespace detail {
template <class Scatterer, class Pull, class Done, class Finalize,
class HandshakeData>
class stream_source_driver_impl;
/// Identifies an unbound sequence of messages. /// Identifies an unbound sequence of messages.
template <class Scatterer, class Pull, class Done, class Finalize, class... Ts> template <class Scatterer, class Pull, class Done, class Finalize>
class stream_source_driver_impl<Scatterer, Pull, Done, Finalize, class stream_source_driver_impl final : public stream_source_driver<Scatterer> {
std::tuple<Ts...>>
final : public stream_source_driver<Scatterer, Ts...> {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
using super = stream_source_driver<Scatterer, Ts...>; using super = stream_source_driver<Scatterer>;
using output_type = typename super::output_type; using output_type = typename super::output_type;
using stream_type = stream<output_type>;
using output_stream_type = typename super::output_stream_type;
using tuple_type = std::tuple<Ts...>;
using handshake_tuple_type = typename super::handshake_tuple_type;
using trait = stream_source_trait_t<Pull>; using trait = stream_source_trait_t<Pull>;
using state_type = typename trait::state; using state_type = typename trait::state;
template <class Init, class Tuple> template <class Init>
stream_source_driver_impl(Init init, Pull f, Done pred, Finalize fin, stream_source_driver_impl(Init init, Pull f, Done pred, Finalize fin)
Tuple&& hs)
: pull_(std::move(f)), : pull_(std::move(f)),
done_(std::move(pred)), done_(std::move(pred)),
finalize_(std::move(fin)), finalize_(std::move(fin)) {
hs_(std::forward<Tuple>(hs)) {
init(state_); init(state_);
} }
handshake_tuple_type make_handshake(stream_slot slot) const override {
return std::tuple_cat(std::make_tuple(stream_type{slot}), hs_);
}
void pull(downstream<output_type>& out, size_t num) override { void pull(downstream<output_type>& out, size_t num) override {
return pull_(state_, out, num); return pull_(state_, out, num);
} }
...@@ -86,7 +66,6 @@ private: ...@@ -86,7 +66,6 @@ private:
Pull pull_; Pull pull_;
Done done_; Done done_;
Finalize finalize_; Finalize finalize_;
tuple_type hs_;
}; };
} // namespace detail } // namespace detail
......
...@@ -72,10 +72,6 @@ public: ...@@ -72,10 +72,6 @@ public:
return hint != this->out_.capacity(); return hint != this->out_.capacity();
} }
message make_handshake(stream_slot slot) const override {
return make_message_from_tuple(driver_.make_handshake(slot));
}
protected: protected:
void finalize(const error& reason) override { void finalize(const error& reason) override {
driver_.finalize(reason); driver_.finalize(reason);
......
...@@ -27,22 +27,16 @@ ...@@ -27,22 +27,16 @@
namespace caf { namespace caf {
namespace detail { namespace detail {
template <class Input, class Scatterer, class Process, class Finalize,
class HandshakeData>
class stream_stage_driver_impl;
/// Default implementation for a `stream_stage_driver` that hardwires `message` /// Default implementation for a `stream_stage_driver` that hardwires `message`
/// as result type and implements `process` and `finalize` using user-provided /// as result type and implements `process` and `finalize` using user-provided
/// function objects (usually lambdas). /// function objects (usually lambdas).
template <class Input, class Scatterer, class Process, class Finalize, template <class Input, class Scatterer, class Process, class Finalize>
class... Ts> class stream_stage_driver_impl final
class stream_stage_driver_impl<Input, Scatterer, Process, Finalize, : public stream_stage_driver<Input, Scatterer> {
std::tuple<Ts...>>
final : public stream_stage_driver<Input, message, Scatterer, Ts...> {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
using super = stream_stage_driver<Input, message, Scatterer, Ts...>; using super = stream_stage_driver<Input, Scatterer>;
using typename super::input_type; using typename super::input_type;
...@@ -50,47 +44,30 @@ public: ...@@ -50,47 +44,30 @@ public:
using typename super::stream_type; using typename super::stream_type;
using typename super::handshake_tuple_type;
using trait = stream_stage_trait_t<Process>; using trait = stream_stage_trait_t<Process>;
using state_type = typename trait::state; using state_type = typename trait::state;
template <class Init, class Tuple> template <class Init>
stream_stage_driver_impl(Init init, Process f, Finalize fin, Tuple&& hs) stream_stage_driver_impl(Init init, Process f, Finalize fin)
: process_(std::move(f)), : process_(std::move(f)),
fin_(std::move(fin)), fin_(std::move(fin)) {
hs_(std::forward<Tuple>(hs)) {
init(state_); init(state_);
} }
handshake_tuple_type make_handshake(stream_slot slot) const override { void process(downstream<output_type>& out,
return std::tuple_cat(std::make_tuple(stream_type{slot}), hs_); std::vector<input_type>& batch) override {
} trait::process::invoke(process_, state_, out, batch);
void process(std::vector<input_type>&& batch,
downstream<output_type>& out) override {
trait::process::invoke(process_, state_, std::move(batch), out);
}
void add_result(message& x) override {
// The default driver assumes to receive only a single result.
result_ = std::move(x);
}
message make_final_result() override {
return std::move(result_);
} }
void finalize(const error&) override { void finalize(const error& err) override {
return fin_(state_); return fin_(state_, err);
} }
private: private:
state_type state_; state_type state_;
Process process_; Process process_;
Finalize fin_; Finalize fin_;
std::tuple<Ts...> hs_;
message result_; message result_;
}; };
......
...@@ -67,18 +67,13 @@ public: ...@@ -67,18 +67,13 @@ public:
CAF_LOG_TRACE(CAF_ARG(x)); CAF_LOG_TRACE(CAF_ARG(x));
using vec_type = std::vector<input_type>; using vec_type = std::vector<input_type>;
if (x.xs.match_elements<vec_type>()) { if (x.xs.match_elements<vec_type>()) {
auto& xs = x.xs.get_mutable_as<vec_type>(0);
downstream<output_type> ds{this->out_.buf()}; downstream<output_type> ds{this->out_.buf()};
driver_.process(std::move(xs), ds); driver_.process(ds, x.xs.get_mutable_as<vec_type>(0));
return; return;
} }
CAF_LOG_ERROR("received unexpected batch type (dropped)"); CAF_LOG_ERROR("received unexpected batch type (dropped)");
} }
message make_handshake(stream_slot slot) const override {
return make_message_from_tuple(driver_.make_handshake(slot));
}
bool congested() const noexcept override { bool congested() const noexcept override {
return this->out_.capacity() == 0; return this->out_.capacity() == 0;
} }
......
...@@ -107,15 +107,25 @@ public: ...@@ -107,15 +107,25 @@ public:
/// A tuple holding all nested scatterers. /// A tuple holding all nested scatterers.
using nested_scatterers = std::tuple<T, Ts...>; using nested_scatterers = std::tuple<T, Ts...>;
/// Pointer to an outbound path.
using typename super::path_ptr;
/// Unique pointer to an outbound path.
using typename super::unique_path_ptr;
/// State held for each slot. /// State held for each slot.
struct non_owning_ptr { struct non_owning_ptr {
outbound_path* ptr; path_ptr ptr;
stream_scatterer* owner; stream_scatterer* owner;
}; };
/// Maps slots to path and nested scatterer. /// Maps slots to path and nested scatterer.
using map_type = detail::unordered_flat_map<stream_slot, non_owning_ptr>; using map_type = detail::unordered_flat_map<stream_slot, non_owning_ptr>;
/// Maps slots to paths that haven't a scatterer assigned yet.
using unassigned_map_type = detail::unordered_flat_map<stream_slot,
unique_path_ptr>;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
fused_scatterer(scheduled_actor* self) fused_scatterer(scheduled_actor* self)
...@@ -135,9 +145,34 @@ public: ...@@ -135,9 +145,34 @@ public:
// return std::get<U>(substreams_); // return std::get<U>(substreams_);
} }
/// Requires a previous call to `add_path` for given slot.
template <class U> template <class U>
void assign(stream_slot slot) { void assign(stream_slot slot) {
paths_.emplace(slot, non_owning_ptr{nullptr, &get<U>()}); // Fetch pointer from the unassigned paths.
auto i = unassigned_paths_.find(slot);
if (i == unassigned_paths_.end()) {
CAF_LOG_ERROR("cannot assign nested scatterer to unknown slot");
return;
}
// Error or not, remove entry from unassigned_paths_ before leaving.
auto cleanup = detail::make_scope_guard([&] {
unassigned_paths_.erase(i);
});
// Transfer ownership to nested scatterer.
auto ptr = i->second.get();
CAF_ASSERT(ptr != nullptr);
auto owner = &get<U>();
if (!owner->insert_path(std::move(i->second))) {
CAF_LOG_ERROR("slot exists as unassigned and assigned");
return;
}
// Store owner and path in our map.
auto er = paths_.emplace(slot, non_owning_ptr{ptr, owner});
if (!er.second) {
CAF_LOG_ERROR("slot already mapped");
owner->remove_path(slot, sec::invalid_stream_state, false);
return;
}
} }
// -- overridden functions --------------------------------------------------- // -- overridden functions ---------------------------------------------------
...@@ -146,32 +181,9 @@ public: ...@@ -146,32 +181,9 @@ public:
return paths_.size(); return paths_.size();
} }
/// Requires a previous call to `assign<T>(slot.sender)`.
path_ptr add_path(stream_slots slots, strong_actor_ptr target) override {
CAF_LOG_TRACE(CAF_ARG(slots) << CAF_ARG(target));
auto e = paths_.end();
auto i = paths_.find(slots.sender);
if (i == e) {
CAF_LOG_ERROR("no scatterer assigned:" << CAF_ARG(slots.sender));
return nullptr;
}
if (i->second.ptr != nullptr) {
CAF_LOG_ERROR("multiple calls to add_path:" << CAF_ARG(slots.sender));
return nullptr;
}
auto result = i->second.owner->add_path(slots, target);
if (result == nullptr) {
CAF_LOG_ERROR("nested scatterer unable to add path:"
<< CAF_ARG(slots.sender));
paths_.erase(i);
return nullptr;
}
i->second.ptr = result;
return result;
}
bool remove_path(stream_slot slot, error reason, bool remove_path(stream_slot slot, error reason,
bool silent) noexcept override { bool silent) noexcept override {
CAF_LOG_TRACE(CAF_ARG(slot) << CAF_ARG(reason) << CAF_ARG(silent));
auto i = paths_.find(slot); auto i = paths_.find(slot);
if (i == paths_.end()) if (i == paths_.end())
return false; return false;
...@@ -229,21 +241,21 @@ public: ...@@ -229,21 +241,21 @@ public:
return result; return result;
} }
message make_handshake_token(stream_slot slot) const override {
auto i = paths_.find(slot);
if (i != paths_.end())
return i->second.owner->make_handshake_token(slot);
CAF_LOG_ERROR("no scatterer available:" << CAF_ARG(slot));
return make_message(stream<message>{slot});
}
void clear_paths() override { void clear_paths() override {
CAF_LOG_TRACE("");
for (auto ptr : ptrs_) for (auto ptr : ptrs_)
ptr->clear_paths(); ptr->clear_paths();
paths_.clear(); paths_.clear();
} }
protected: protected:
bool insert_path(unique_path_ptr ptr) override {
CAF_LOG_TRACE(CAF_ARG(ptr));
CAF_ASSERT(ptr != nullptr);
auto slot = ptr->slots.sender;
return unassigned_paths_.emplace(slot, std::move(ptr)).second;
}
void for_each_path_impl(path_visitor& f) override { void for_each_path_impl(path_visitor& f) override {
for (auto& kvp : paths_) for (auto& kvp : paths_)
f(*kvp.second.ptr); f(*kvp.second.ptr);
...@@ -268,6 +280,7 @@ private: ...@@ -268,6 +280,7 @@ private:
nested_scatterers nested_; nested_scatterers nested_;
stream_scatterer* ptrs_[sizeof...(Ts) + 1]; stream_scatterer* ptrs_[sizeof...(Ts) + 1];
map_type paths_; map_type paths_;
unassigned_map_type unassigned_paths_;
}; };
} // namespace caf } // namespace caf
......
...@@ -34,6 +34,7 @@ template <class> class intrusive_ptr; ...@@ -34,6 +34,7 @@ template <class> class intrusive_ptr;
template <class> class optional; template <class> class optional;
template <class> class param; template <class> class param;
template <class> class stream; template <class> class stream;
template <class> class stream_sink;
template <class> class trivial_match_case; template <class> class trivial_match_case;
template <class> class weak_intrusive_ptr; template <class> class weak_intrusive_ptr;
...@@ -41,7 +42,7 @@ template <class> struct timeout_definition; ...@@ -41,7 +42,7 @@ template <class> struct timeout_definition;
// -- 2 param templates -------------------------------------------------------- // -- 2 param templates --------------------------------------------------------
template <class, class> class make_sink_result; template <class, class> class stream_source;
// -- 3 param templates -------------------------------------------------------- // -- 3 param templates --------------------------------------------------------
...@@ -50,21 +51,19 @@ template <class, class, int> class actor_cast_access; ...@@ -50,21 +51,19 @@ template <class, class, int> class actor_cast_access;
template <class, class, class> class broadcast_scatterer; template <class, class, class> class broadcast_scatterer;
template <class, class, class> class broadcast_topic_scatterer; template <class, class, class> class broadcast_topic_scatterer;
template <class, class, class> class random_topic_scatterer; template <class, class, class> class random_topic_scatterer;
template <class, class, class> class stream_stage;
// -- variadic templates ------------------------------------------------------- // -- variadic templates -------------------------------------------------------
template <class...> class result;
template <class...> class delegated; template <class...> class delegated;
template <class...> class result;
template <class...> class typed_actor; template <class...> class typed_actor;
template <class...> class typed_actor_pointer; template <class...> class typed_actor_pointer;
template <class...> class typed_response_promise;
template <class...> class typed_event_based_actor; template <class...> class typed_event_based_actor;
template <class...> class typed_response_promise;
// -- variadic templates with fixed arguments ---------------------------------- // -- variadic templates with fixed arguments ----------------------------------
template <class, class, class, class, class...> class stream_stage;
template <class, class, class...> class stream_source;
// -- classes ------------------------------------------------------------------ // -- classes ------------------------------------------------------------------
class actor; class actor;
...@@ -100,6 +99,7 @@ class proxy_registry; ...@@ -100,6 +99,7 @@ class proxy_registry;
class stream_manager; class stream_manager;
class actor_companion; class actor_companion;
class mailbox_element; class mailbox_element;
class message_builder;
class message_handler; class message_handler;
class scheduled_actor; class scheduled_actor;
class stream_scatterer; class stream_scatterer;
...@@ -231,7 +231,7 @@ using mailbox_element_ptr = std::unique_ptr<mailbox_element, detail::disposer>; ...@@ -231,7 +231,7 @@ using mailbox_element_ptr = std::unique_ptr<mailbox_element, detail::disposer>;
template <class, class = std::tuple<>, class = stream_manager_ptr> template <class, class = std::tuple<>, class = stream_manager_ptr>
class output_stream; class output_stream;
template <class, class = stream_manager_ptr> template <class = stream_manager_ptr>
class stream_result; class stream_result;
} // namespace caf } // namespace caf
......
...@@ -32,8 +32,6 @@ public: ...@@ -32,8 +32,6 @@ public:
size_t num_paths() const noexcept override; size_t num_paths() const noexcept override;
path_ptr add_path(stream_slots slots, strong_actor_ptr target) override;
bool remove_path(stream_slot slots, error reason, bool remove_path(stream_slot slots, error reason,
bool silent) noexcept override; bool silent) noexcept override;
...@@ -47,9 +45,9 @@ public: ...@@ -47,9 +45,9 @@ public:
size_t buffered() const noexcept override; size_t buffered() const noexcept override;
message make_handshake_token(stream_slot slot) const override;
protected: protected:
bool insert_path(unique_path_ptr) override;
void for_each_path_impl(path_visitor& f) override; void for_each_path_impl(path_visitor& f) override;
bool check_paths_impl(path_algorithm algo, bool check_paths_impl(path_algorithm algo,
......
...@@ -276,6 +276,12 @@ public: ...@@ -276,6 +276,12 @@ public:
return current_element_; return current_element_;
} }
/// Returns a pointer to the currently processed mailbox element.
/// @private
inline void current_mailbox_element(mailbox_element* ptr) {
current_element_ = ptr;
}
/// Adds a unidirectional `monitor` to `whom`. /// Adds a unidirectional `monitor` to `whom`.
/// @note Each call to `monitor` creates a new, independent monitor. /// @note Each call to `monitor` creates a new, independent monitor.
template <class Handle> template <class Handle>
......
...@@ -23,9 +23,7 @@ ...@@ -23,9 +23,7 @@
#include "caf/stream_slot.hpp" #include "caf/stream_slot.hpp"
#include "caf/stream_source.hpp" #include "caf/stream_source.hpp"
#include "caf/detail/type_traits.hpp" #include "caf/detail/implicit_conversions.hpp"
#include "caf/meta/type_name.hpp"
namespace caf { namespace caf {
...@@ -37,7 +35,7 @@ struct make_source_result { ...@@ -37,7 +35,7 @@ struct make_source_result {
using value_type = typename Scatterer::value_type; using value_type = typename Scatterer::value_type;
/// Fully typed stream manager as returned by `make_source`. /// Fully typed stream manager as returned by `make_source`.
using source_type = stream_source<value_type, Scatterer, Ts...>; using source_type = stream_source<value_type, Scatterer>;
/// Pointer to a fully typed stream manager. /// Pointer to a fully typed stream manager.
using source_ptr_type = intrusive_ptr<source_type>; using source_ptr_type = intrusive_ptr<source_type>;
...@@ -50,7 +48,8 @@ struct make_source_result { ...@@ -50,7 +48,8 @@ struct make_source_result {
/// the types of the handshake arguments. /// the types of the handshake arguments.
template <class Scatterer, class... Ts> template <class Scatterer, class... Ts>
using make_source_result_t = using make_source_result_t =
typename make_source_result<Scatterer, detail::decay_t<Ts>...>::type; typename make_source_result<Scatterer,
detail::strip_and_convert_t<Ts>...>::type;
} // namespace caf } // namespace caf
......
...@@ -24,20 +24,20 @@ ...@@ -24,20 +24,20 @@
#include "caf/stream_slot.hpp" #include "caf/stream_slot.hpp"
#include "caf/stream_stage.hpp" #include "caf/stream_stage.hpp"
#include "caf/detail/type_traits.hpp" #include "caf/detail/implicit_conversions.hpp"
namespace caf { namespace caf {
/// Helper trait for deducing an `output_stream` from the arguments to /// Helper trait for deducing an `output_stream` from the arguments to
/// `scheduled_actor::make_stage`. /// `scheduled_actor::make_stage`.
template <class In, class Result, class Scatterer, class... Ts> template <class In, class Scatterer, class... Ts>
class make_stage_result { class make_stage_result {
public: public:
/// Type of a single element. /// Type of a single element.
using value_type = typename Scatterer::value_type; using value_type = typename Scatterer::value_type;
/// Fully typed stream manager as returned by `make_stage`. /// Fully typed stream manager as returned by `make_stage`.
using stage_type = stream_stage<In, Result, value_type, Scatterer, Ts...>; using stage_type = stream_stage<In, value_type, Scatterer>;
/// Pointer to a fully typed stream manager. /// Pointer to a fully typed stream manager.
using stage_ptr_type = intrusive_ptr<stage_type>; using stage_ptr_type = intrusive_ptr<stage_type>;
...@@ -50,8 +50,8 @@ public: ...@@ -50,8 +50,8 @@ public:
/// additional handshake types. Hardwires `message` as result type. /// additional handshake types. Hardwires `message` as result type.
template <class In, class Scatterer, class... Ts> template <class In, class Scatterer, class... Ts>
using make_stage_result_t = using make_stage_result_t =
typename make_stage_result<In, message, Scatterer, typename make_stage_result<In, Scatterer,
detail::decay_t<Ts>...>::type; detail::strip_and_convert_t<Ts>...>::type;
} // namespace caf } // namespace caf
......
...@@ -29,11 +29,12 @@ ...@@ -29,11 +29,12 @@
#include "caf/none.hpp" #include "caf/none.hpp"
#include "caf/intrusive_ptr.hpp" #include "caf/intrusive_ptr.hpp"
#include "caf/message.hpp"
#include "caf/duration.hpp"
#include "caf/behavior.hpp" #include "caf/behavior.hpp"
#include "caf/ref_counted.hpp" #include "caf/duration.hpp"
#include "caf/match_case.hpp"
#include "caf/may_have_timeout.hpp" #include "caf/may_have_timeout.hpp"
#include "caf/message.hpp"
#include "caf/ref_counted.hpp"
#include "caf/timeout_definition.hpp" #include "caf/timeout_definition.hpp"
#include "caf/detail/behavior_impl.hpp" #include "caf/detail/behavior_impl.hpp"
......
...@@ -27,6 +27,7 @@ ...@@ -27,6 +27,7 @@
#include "caf/actor_control_block.hpp" #include "caf/actor_control_block.hpp"
#include "caf/downstream_msg.hpp" #include "caf/downstream_msg.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/logger.hpp"
#include "caf/stream_aborter.hpp" #include "caf/stream_aborter.hpp"
#include "caf/stream_slot.hpp" #include "caf/stream_slot.hpp"
#include "caf/system_messages.hpp" #include "caf/system_messages.hpp"
...@@ -64,8 +65,8 @@ public: ...@@ -64,8 +65,8 @@ public:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
/// Constructs a path for given handle and stream ID. /// Constructs a pending path for given slot and handle.
outbound_path(stream_slots id, strong_actor_ptr ptr); outbound_path(stream_slot sender_slot, strong_actor_ptr receiver_hdl);
~outbound_path(); ~outbound_path();
...@@ -83,10 +84,14 @@ public: ...@@ -83,10 +84,14 @@ public:
/// `xs_size` and increments `next_batch_id` by 1. /// `xs_size` and increments `next_batch_id` by 1.
void emit_batch(local_actor* self, long xs_size, message xs); void emit_batch(local_actor* self, long xs_size, message xs);
/// Calls `emit_batch` for each /// Calls `emit_batch` for each chunk in the cache, whereas each chunk is of
/// size `desired_batch_size`. Does nothing for pending paths.
template <class T> template <class T>
void emit_batches(local_actor* self, std::vector<T>& cache, void emit_batches(local_actor* self, std::vector<T>& cache,
bool force_underfull) { bool force_underfull) {
CAF_LOG_TRACE(CAF_ARG(cache) << CAF_ARG(force_underfull));
if (pending())
return;
CAF_ASSERT(desired_batch_size > 0); CAF_ASSERT(desired_batch_size > 0);
if (cache.size() == desired_batch_size) { if (cache.size() == desired_batch_size) {
emit_batch(self, desired_batch_size, make_message(std::move(cache))); emit_batch(self, desired_batch_size, make_message(std::move(cache)));
...@@ -120,6 +125,14 @@ public: ...@@ -120,6 +125,14 @@ public:
const strong_actor_ptr& hdl, const strong_actor_ptr& hdl,
error reason); error reason);
// -- properties -------------------------------------------------------------
/// Returns whether this path is pending, i.e., didn't receive an `ack_open`
/// yet.
inline bool pending() const noexcept {
return slots.receiver == invalid_stream_slot;
}
// -- member variables ------------------------------------------------------- // -- member variables -------------------------------------------------------
/// Slot IDs for sender (self) and receiver (hdl). /// Slot IDs for sender (self) and receiver (hdl).
......
...@@ -99,6 +99,15 @@ public: ...@@ -99,6 +99,15 @@ public:
return ptr_; return ptr_;
} }
// -- conversions ------------------------------------------------------------
/// Rebinds the `ptr()` property.
template <class Manager>
output_stream<value_type, tuple_type, intrusive_ptr<Manager>>
rebind(Manager* ptr) const noexcept {
return {in_, out_, ptr};
}
private: private:
// -- member variables ------------------------------------------------------- // -- member variables -------------------------------------------------------
......
...@@ -48,8 +48,6 @@ ...@@ -48,8 +48,6 @@
#include "caf/stream.hpp" #include "caf/stream.hpp"
#include "caf/stream_manager.hpp" #include "caf/stream_manager.hpp"
#include "caf/stream_result.hpp" #include "caf/stream_result.hpp"
#include "caf/stream_result_handler_trait.hpp"
#include "caf/stream_result_trait.hpp"
#include "caf/stream_source_trait.hpp" #include "caf/stream_source_trait.hpp"
#include "caf/to_string.hpp" #include "caf/to_string.hpp"
...@@ -313,10 +311,20 @@ public: ...@@ -313,10 +311,20 @@ public:
// -- properties ------------------------------------------------------------- // -- properties -------------------------------------------------------------
/// Returns the queue for storing incoming messages. /// Returns the queue for storing incoming messages.
inline mailbox_type& mailbox() { inline mailbox_type& mailbox() noexcept {
return mailbox_; return mailbox_;
} }
/// Returns map for all active streams.
inline stream_manager_map& stream_managers() noexcept {
return stream_managers_;
}
/// Returns map for all pending streams.
inline stream_manager_map& pending_stream_managers() noexcept {
return pending_stream_managers_;
}
// -- event handlers --------------------------------------------------------- // -- event handlers ---------------------------------------------------------
/// Sets a custom handler for unexpected messages. /// Sets a custom handler for unexpected messages.
...@@ -420,146 +428,122 @@ public: ...@@ -420,146 +428,122 @@ public:
/// @param res_handler Function object for receiving the stream result. /// @param res_handler Function object for receiving the stream result.
/// @param scatterer_type Configures the policy for downstream communication. /// @param scatterer_type Configures the policy for downstream communication.
/// @returns A stream object with a pointer to the generated `stream_manager`. /// @returns A stream object with a pointer to the generated `stream_manager`.
template <class Driver, class... Ts> template <class Driver, class... Ts, class Init, class Pull, class Done,
typename Driver::output_stream_type make_source(Ts&&... xs) { class Finalize = unit_t>
make_source_result_t<typename Driver::scatterer_type, Ts...>
make_source(std::tuple<Ts...> xs, Init init, Pull pull, Done done,
Finalize finalize = {}) {
using detail::make_stream_source; using detail::make_stream_source;
auto mgr = make_stream_source<Driver>(this, std::forward<Ts>(xs)...); auto mgr = make_stream_source<Driver>(this, std::move(init),
return mgr->add_outbound_path(); std::move(pull), std::move(done),
std::move(finalize));
return mgr->add_outbound_path(std::move(xs));
} }
template <class... Ts, class Init, class Pull, class Done, class Finalize, template <class... Ts, class Init, class Pull, class Done,
class Finalize = unit_t,
class Scatterer = class Scatterer =
broadcast_scatterer<typename stream_source_trait_t<Pull>::output>> broadcast_scatterer<typename stream_source_trait_t<Pull>::output>>
make_source_result_t<Scatterer, Ts...> make_source_result_t<Scatterer, Ts...>
make_source(std::tuple<Ts...> xs, Init init, Pull pull, Done done, make_source(std::tuple<Ts...> xs, Init init, Pull pull, Done done,
Finalize finalize, policy::arg<Scatterer> = {}) { Finalize finalize = {}, policy::arg<Scatterer> = {}) {
using tuple_type = std::tuple<detail::decay_t<Ts>...>;
using driver = detail::stream_source_driver_impl<Scatterer, Pull, Done, using driver = detail::stream_source_driver_impl<Scatterer, Pull, Done,
Finalize, tuple_type>; Finalize>;
return make_source<driver>(std::move(init), std::move(pull), return make_source<driver>(std::move(xs), std::move(init), std::move(pull),
std::move(done), std::move(finalize), std::move(done), std::move(finalize));
std::move(xs));
} }
template <class Init, class Pull, class Done, template <class Init, class Pull, class Done, class Finalize = unit_t,
class Scatterer = class Scatterer =
broadcast_scatterer<typename stream_source_trait_t<Pull>::output>, broadcast_scatterer<typename stream_source_trait_t<Pull>::output>,
class Trait = stream_source_trait_t<Pull>> class Trait = stream_source_trait_t<Pull>>
make_source_result_t<Scatterer> make_source_result_t<Scatterer>
make_source(Init init, Pull pull, Done done, make_source(Init init, Pull pull, Done done, Finalize finalize = {},
policy::arg<Scatterer> scatterer_type = {}) { policy::arg<Scatterer> scatterer_type = {}) {
auto finalize = [](typename Trait::state, const error&) {};
return make_source(std::make_tuple(), init, pull, done, finalize, return make_source(std::make_tuple(), init, pull, done, finalize,
scatterer_type); scatterer_type);
} }
template <class Init, class Pull, class Done, class Finalize,
class Scatterer =
broadcast_scatterer<typename stream_source_trait_t<Pull>::output>,
class Trait = stream_source_trait_t<Pull>>
make_source_result_t<Scatterer>
make_source(Init init, Pull pull, Done done, Finalize finalize,
policy::arg<Scatterer> scatterer_type = {}) {
return make_source(std::make_tuple(), init, pull, done, finalize,
scatterer_type);
}
template <class... Ts, class Init, class Pull, class Done,
class Scatterer =
broadcast_scatterer<typename stream_source_trait_t<Pull>::output>,
class Trait = stream_source_trait_t<Pull>>
make_source_result_t<Scatterer, Ts...>
make_source(std::tuple<Ts...> xs, Init init, Pull pull, Done done,
policy::arg<Scatterer> scatterer_type = {}) {
auto finalize = [](typename Trait::state, const error&) {};
return make_source(std::move(xs), init, pull, done, finalize,
scatterer_type);
}
/// Creates a new stream source and adds `dest` as first outbound path to it. /// Creates a new stream source and adds `dest` as first outbound path to it.
template <class ActorHandle, class... Ts, class Init, class Pull, class Done, template <class ActorHandle, class... Ts, class Init, class Pull, class Done,
class HandleResult, class Finalize = unit_t,
class Scatterer = class Scatterer =
broadcast_scatterer<typename stream_source_trait_t<Pull>::output>, broadcast_scatterer<typename stream_source_trait_t<Pull>::output>,
class Trait = stream_source_trait_t<Pull>> class Trait = stream_source_trait_t<Pull>>
detail::enable_if_t< detail::enable_if_t<detail::is_actor_handle<ActorHandle>::value,
detail::is_actor_handle<ActorHandle>::value, typename make_source_result_t<Scatterer>::pointer_type>
typename make_source_result_t<Scatterer, Ts...>::pointer_type>
make_source(const ActorHandle& dest, std::tuple<Ts...> xs, Init init, make_source(const ActorHandle& dest, std::tuple<Ts...> xs, Init init,
Pull pull, Done done, HandleResult handle_res, Pull pull, Done done, Finalize fin = {},
policy::arg<Scatterer> scatterer_arg = {}) { policy::arg<Scatterer> = {}) {
// TODO: type-check whether `dest` is a valid next stage // TODO: type checking of dest
// Expect `handle_result` to have signature `void (expected<T>)`. using output_type = typename Trait::output;
using handle_res_trait = stream_result_handler_trait_t<HandleResult>; using driver = detail::stream_source_driver_impl<Scatterer, Pull, Done,
static_assert(handle_res_trait::valid, Finalize>;
"expected a result handler with signature " auto mgr = detail::make_stream_source<driver>(this, std::move(init),
"'void (expected<T>)'"); std::move(pull),
using handle_res_result = typename handle_res_trait::result; std::move(done),
auto result = make_source(std::move(xs), std::move(init), std::move(pull), std::move(fin));
std::move(done), scatterer_arg); auto tk = std::make_tuple(stream<output_type>{});
auto mid = new_request_id(message_priority::normal); auto handshake = make_message_from_tuple(std::tuple_cat(tk, std::move(xs)));
result.ptr()->add_unsafe_outbound_path( auto slot = mgr->add_unsafe_outbound_path_impl(
actor_cast<strong_actor_ptr>(dest), result.out(), actor_cast<strong_actor_ptr>(dest), std::move(handshake));
actor_cast<strong_actor_ptr>(this), no_stages, mid); if (slot == invalid_stream_slot) {
behavior tmp{ CAF_LOG_WARNING("unable to assign a slot in make_source");
[=](handle_res_result& x) mutable { handle_res(std::move(x)); }, }
[=](error& err) mutable { handle_res(std::move(err)); } return mgr;
};
add_multiplexed_response_handler(mid.response_id(), std::move(tmp));
return result.ptr();
} }
/// Creates a new stream source and adds `dest` as first outbound path to it. /// Creates a new stream source and adds `dest` as first outbound path to it.
template <class ActorHandle, class Init, class Pull, class Done, template <class ActorHandle, class Init, class Pull, class Done,
class HandleResult, class Finalize = unit_t,
class Scatterer = class Scatterer =
broadcast_scatterer<typename stream_source_trait_t<Pull>::output>, broadcast_scatterer<typename stream_source_trait_t<Pull>::output>,
class Trait = stream_source_trait_t<Pull>> class Trait = stream_source_trait_t<Pull>>
detail::enable_if_t<detail::is_actor_handle<ActorHandle>::value, detail::enable_if_t<detail::is_actor_handle<ActorHandle>::value,
typename make_source_result_t<Scatterer>::pointer_type> typename make_source_result_t<Scatterer>::pointer_type>
make_source(const ActorHandle& dest, Init init, Pull pull, Done done, make_source(const ActorHandle& dest, Init init, Pull pull, Done done,
HandleResult handle_res, Finalize fin = {},
policy::arg<Scatterer> scatterer_arg = {}) { policy::arg<Scatterer> scatterer_arg = {}) {
return make_source(dest, std::make_tuple(), std::move(init), return make_source(dest, std::make_tuple(), std::move(init),
std::move(pull), std::move(done), std::move(handle_res), std::move(pull), std::move(done), std::move(fin),
scatterer_arg); scatterer_arg);
} }
template <class Driver, class... Ts> template <class Driver, class... Ts>
stream_result<typename Driver::result_type, typename Driver::sink_ptr_type> stream_result<typename Driver::sink_ptr_type>
make_sink(const stream<typename Driver::input_type>& src, Ts&&... xs) { make_sink(const stream<typename Driver::input_type>& src, Ts&&... xs) {
auto mgr = detail::make_stream_sink<Driver>(this, std::forward<Ts>(xs)...); auto mgr = detail::make_stream_sink<Driver>(this, std::forward<Ts>(xs)...);
return mgr->add_inbound_path(src); return mgr->add_inbound_path(src);
} }
template <class Input, class Init, class Fun, class Finalize, template <class Input, class Init, class Fun, class Finalize = unit_t,
class Trait = stream_sink_trait_t<Fun, Finalize>> class Trait = stream_sink_trait_t<Fun>>
stream_result<typename Trait::result, typename Trait::pointer> stream_result<typename Trait::pointer>
make_sink(const stream<Input>& in, Init init, Fun fun, Finalize fin) { make_sink(const stream<Input>& in, Init init, Fun fun, Finalize fin = {}) {
using driver = detail::stream_sink_driver_impl<typename Trait::input, using driver = detail::stream_sink_driver_impl<typename Trait::input,
typename Trait::result,
Fun, Finalize>; Fun, Finalize>;
return make_sink<driver>(in, std::move(init), std::move(fun), return make_sink<driver>(in, std::move(init), std::move(fun),
std::move(fin)); std::move(fin));
} }
template <class Driver, class In, class... Ts> template <class Driver, class In, class... Ts, class... Us>
typename Driver::output_stream_type make_stage(const stream<In>& src, make_stage_result_t<In, typename Driver::scatterer_type, Ts...>
Ts&&... xs) { make_stage(const stream<In>& src, std::tuple<Ts...> xs, Us&&... ys) {
using detail::make_stream_stage; using detail::make_stream_stage;
auto mgr = make_stream_stage<Driver>(this, std::forward<Ts>(xs)...); auto mgr = make_stream_stage<Driver>(this, std::forward<Us>(ys)...);
auto in = mgr->add_inbound_path(src).in(); auto in = mgr->add_inbound_path(src).in();
auto out = mgr->add_outbound_path().out(); auto out = mgr->add_outbound_path(std::move(xs)).out();
return {in, out, std::move(mgr)}; return {in, out, std::move(mgr)};
} }
template <class In, class... Ts, class Init, class Fun, class Cleanup, template <class In, class... Ts, class Init, class Fun,
class Finalize = unit_t,
class Scatterer = class Scatterer =
broadcast_scatterer<typename stream_stage_trait_t<Fun>::output>, broadcast_scatterer<typename stream_stage_trait_t<Fun>::output>,
class Trait = stream_stage_trait_t<Fun>> class Trait = stream_stage_trait_t<Fun>>
make_stage_result_t<In, Scatterer, Ts...> make_stage_result_t<In, Scatterer, Ts...>
make_stage(const stream<In>& in, std::tuple<Ts...> xs, Init init, Fun fun, make_stage(const stream<In>& in, std::tuple<Ts...> xs, Init init, Fun fun,
Cleanup cleanup, policy::arg<Scatterer> scatterer_type = {}) { Finalize fin = {}, policy::arg<Scatterer> scatterer_type = {}) {
CAF_IGNORE_UNUSED(scatterer_type); CAF_IGNORE_UNUSED(scatterer_type);
CAF_ASSERT(current_mailbox_element() != nullptr); CAF_ASSERT(current_mailbox_element() != nullptr);
CAF_ASSERT( CAF_ASSERT(
...@@ -577,12 +561,21 @@ public: ...@@ -577,12 +561,21 @@ public:
>::value, >::value,
"Expected signature `void (State&, downstream<Out>&, In)` " "Expected signature `void (State&, downstream<Out>&, In)` "
"for consume function"); "for consume function");
using driver = using driver = detail::stream_stage_driver_impl<typename Trait::input,
detail::stream_stage_driver_impl<typename Trait::input, Scatterer, Fun, Scatterer, Fun, Finalize>;
Cleanup, return make_stage<driver>(in, std::move(xs), std::move(init),
std::tuple<detail::decay_t<Ts>...>>; std::move(fun), std::move(fin));
return make_stage<driver>(in, std::move(init), std::move(fun), }
std::move(cleanup), std::move(xs));
template <class In, class Init, class Fun, class Finalize = unit_t,
class Scatterer =
broadcast_scatterer<typename stream_stage_trait_t<Fun>::output>,
class Trait = stream_stage_trait_t<Fun>>
make_stage_result_t<In, Scatterer>
make_stage(const stream<In>& in, Init init, Fun fun, Finalize fin = {},
policy::arg<Scatterer> scatterer_type = {}) {
return make_stage(in, std::make_tuple(), std::move(init), std::move(fun),
std::move(fin), scatterer_type);
} }
/// Returns a stream manager (implementing a continuous stage) without in- or /// Returns a stream manager (implementing a continuous stage) without in- or
...@@ -595,14 +588,12 @@ public: ...@@ -595,14 +588,12 @@ public:
return std::move(ptr); return std::move(ptr);
} }
template <class... Ts, class Init, class Fun, class Cleanup, template <class Init, class Fun, class Cleanup,
class Scatterer = class Scatterer =
broadcast_scatterer<typename stream_stage_trait_t<Fun>::output>, broadcast_scatterer<typename stream_stage_trait_t<Fun>::output>,
class Trait = stream_stage_trait_t<Fun>> class Trait = stream_stage_trait_t<Fun>>
stream_stage_ptr<typename Trait::input, message, typename Trait::output, stream_stage_ptr<typename Trait::input, typename Trait::output, Scatterer>
Scatterer, detail::decay_t<Ts>...> make_continuous_stage(Init init, Fun fun, Cleanup cleanup,
make_continuous_stage(std::tuple<Ts...> xs, Init init, Fun fun,
Cleanup cleanup,
policy::arg<Scatterer> scatterer_type = {}) { policy::arg<Scatterer> scatterer_type = {}) {
CAF_IGNORE_UNUSED(scatterer_type); CAF_IGNORE_UNUSED(scatterer_type);
using input_type = typename Trait::input; using input_type = typename Trait::input;
...@@ -619,33 +610,14 @@ public: ...@@ -619,33 +610,14 @@ public:
>::value, >::value,
"Expected signature `void (State&, downstream<Out>&, In)` " "Expected signature `void (State&, downstream<Out>&, In)` "
"for consume function"); "for consume function");
using driver = using driver = detail::stream_stage_driver_impl<typename Trait::input,
detail::stream_stage_driver_impl<typename Trait::input, Scatterer, Fun, Scatterer, Fun, Cleanup>;
Cleanup,
std::tuple<detail::decay_t<Ts>...>>;
return make_continuous_stage<driver>(std::move(init), std::move(fun), return make_continuous_stage<driver>(std::move(init), std::move(fun),
std::move(cleanup), std::move(xs)); std::move(cleanup));
}
template <class Init, class Fun, class Cleanup,
class Scatterer =
broadcast_scatterer<typename stream_stage_trait_t<Fun>::output>,
class Trait = stream_stage_trait_t<Fun>>
stream_stage_ptr<typename stream_stage_trait_t<Fun>::input, message,
typename stream_stage_trait_t<Fun>::output, Scatterer>
make_continuous_stage(Init init, Fun fun, Cleanup cleanup,
policy::arg<Scatterer> scatterer_type = {}) {
return make_continuous_stage(std::make_tuple(), std::move(init),
std::move(fun), std::move(cleanup),
scatterer_type);
} }
/// @cond PRIVATE /// @cond PRIVATE
/// Builds the pipeline after receiving an `open_stream_msg`. Sends a stream
/// handshake to the next actor if `mgr->out().terminal() == false`.
sec build_pipeline(stream_slot in, stream_slot out, stream_manager_ptr mgr);
// -- timeout management ----------------------------------------------------- // -- timeout management -----------------------------------------------------
/// Requests a new timeout and returns its ID. /// Requests a new timeout and returns its ID.
...@@ -758,7 +730,7 @@ public: ...@@ -758,7 +730,7 @@ public:
virtual void erase_inbound_paths_later(const stream_manager* mgr, virtual void erase_inbound_paths_later(const stream_manager* mgr,
error reason); error reason);
// -- handling of stream message --------------------------------------------- // -- handling of stream messages --------------------------------------------
void handle_upstream_msg(stream_slots slots, actor_addr& sender, void handle_upstream_msg(stream_slots slots, actor_addr& sender,
upstream_msg::ack_open& x); upstream_msg::ack_open& x);
...@@ -857,6 +829,9 @@ public: ...@@ -857,6 +829,9 @@ public:
/// Removes the stream manager mapped to `id` in `O(log n)`. /// Removes the stream manager mapped to `id` in `O(log n)`.
void erase_stream_manager(stream_slot id); void erase_stream_manager(stream_slot id);
/// Removes the stream manager mapped to `id` in `O(log n)`.
void erase_pending_stream_manager(stream_slot id);
/// Removes all entries for `mgr` in `O(n)`. /// Removes all entries for `mgr` in `O(n)`.
void erase_stream_manager(const stream_manager_ptr& mgr); void erase_stream_manager(const stream_manager_ptr& mgr);
......
...@@ -28,49 +28,9 @@ ...@@ -28,49 +28,9 @@
namespace caf { namespace caf {
/// Identifies an unbound sequence of elements. /// Empty marker type for streaming handshakes.
template <class T> template <class T>
class stream { class stream {};
public:
// -- member types -----------------------------------------------------------
/// Type of a single element.
using value_type = T;
// -- constructors and destructors -------------------------------------------
stream(stream&&) = default;
stream(const stream&) = default;
stream& operator=(stream&&) = default;
stream& operator=(const stream&) = default;
explicit stream(stream_slot id = 0) : slot_(id) {
// nop
}
explicit stream(invalid_stream_t) : slot_(0) {
// nop
}
// -- properties -------------------------------------------------------------
/// Returns the actor-specific stream slot ID.
inline stream_slot slot() const {
return slot_;
}
// -- serialization support --------------------------------------------------
template <class Inspector>
friend typename Inspector::result_type inspect(Inspector& f, stream& x) {
return f(meta::type_name("stream"), x.slot_);
}
private:
// -- member variables -------------------------------------------------------
stream_slot slot_;
};
} // namespace caf } // namespace caf
......
...@@ -23,10 +23,17 @@ ...@@ -23,10 +23,17 @@
#include <cstdint> #include <cstdint>
#include <cstddef> #include <cstddef>
#include "caf/actor.hpp"
#include "caf/actor_cast.hpp"
#include "caf/downstream_msg.hpp" #include "caf/downstream_msg.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/mailbox_element.hpp" #include "caf/mailbox_element.hpp"
#include "caf/make_message.hpp"
#include "caf/message_builder.hpp"
#include "caf/output_stream.hpp"
#include "caf/ref_counted.hpp" #include "caf/ref_counted.hpp"
#include "caf/stream_result.hpp"
#include "caf/stream_scatterer.hpp"
#include "caf/stream_slot.hpp" #include "caf/stream_slot.hpp"
#include "caf/upstream_msg.hpp" #include "caf/upstream_msg.hpp"
...@@ -45,25 +52,6 @@ public: ...@@ -45,25 +52,6 @@ public:
~stream_manager() override; ~stream_manager() override;
/// Handles `stream_msg::open` messages by creating a new slot for incoming
/// traffic.
/// @param slot Slot ID used by the sender, i.e., the slot ID for upstream
/// messages back to the sender.
/// @param hdl Handle to the sender.
/// @param original_stage Handle to the initial receiver of the handshake.
/// @param priority Affects credit assignment and maximum bandwidth.
/// @param result_cb Callback for the listener of the final stream result.
/// Ignored when returning `nullptr`, because the previous
/// stage is responsible for it until this manager
/// acknowledges the handshake.
/// @returns An error if the stream manager rejects the handshake.
/// @pre `hdl != nullptr`
/*
virtual error open(stream_slot slot, strong_actor_ptr hdl,
strong_actor_ptr original_stage, stream_priority priority,
response_promise result_cb);
*/
virtual void handle(inbound_path* from, downstream_msg::batch& x); virtual void handle(inbound_path* from, downstream_msg::batch& x);
virtual void handle(inbound_path* from, downstream_msg::close& x); virtual void handle(inbound_path* from, downstream_msg::close& x);
...@@ -98,14 +86,9 @@ public: ...@@ -98,14 +86,9 @@ public:
/// Sends a handshake to `dest`. /// Sends a handshake to `dest`.
/// @pre `dest != nullptr` /// @pre `dest != nullptr`
virtual void send_handshake(strong_actor_ptr dest, stream_slot slot, virtual void send_handshake(strong_actor_ptr next, stream_slot slot,
strong_actor_ptr stream_origin, mailbox_element::forwarding_stack stages,
mailbox_element::forwarding_stack fwd_stack, message handshake);
message_id handshake_mid);
/// Sends a handshake to `dest`.
/// @pre `dest != nullptr`
void send_handshake(strong_actor_ptr dest, stream_slot slot);
// -- implementation hooks for sources --------------------------------------- // -- implementation hooks for sources ---------------------------------------
...@@ -141,15 +124,6 @@ public: ...@@ -141,15 +124,6 @@ public:
/// Removes an input path /// Removes an input path
virtual void remove_input_path(stream_slot slot, error reason, bool silent); virtual void remove_input_path(stream_slot slot, error reason, bool silent);
// -- mutators ---------------------------------------------------------------
/// Adds a response promise to a sink for delivering the final result.
/// @pre `out().terminal() == true`
void add_promise(response_promise x);
/// Calls `x.deliver()`
void deliver_promises(message x);
// -- properties ------------------------------------------------------------- // -- properties -------------------------------------------------------------
/// Returns whether this stream remains open even if no in- or outbound paths /// Returns whether this stream remains open even if no in- or outbound paths
...@@ -175,25 +149,82 @@ public: ...@@ -175,25 +149,82 @@ public:
return self_; return self_;
} }
/// Creates an outbound path to the current sender without any type checking.
/// @pre `out().terminal() == false`
/// @private
template <class Out>
output_stream_t<Out> add_unsafe_outbound_path() {
auto handshake = make_message(stream<Out>{});
return {0, add_unsafe_outbound_path_impl(std::move(handshake)), this};
}
/// Creates an outbound path to the current sender without any type checking. /// Creates an outbound path to the current sender without any type checking.
/// @pre `out().terminal() == false` /// @pre `out().terminal() == false`
/// @private /// @private
template <class Out, class... Ts> template <class Out, class... Ts>
output_stream<Out, Ts...> add_unsafe_outbound_path(); output_stream_t<Out, detail::strip_and_convert_t<Ts>...>
add_unsafe_outbound_path(std::tuple<Ts...> xs) {
auto tk = std::make_tuple(stream<Out>{});
auto handshake = make_message_from_tuple(std::tuple_cat(tk, std::move(xs)));
return {0, add_unsafe_outbound_path_impl(std::move(handshake)), this};
}
/// Creates an outbound path to `next` without any type checking. /// Creates an outbound path to `next`, only checking whether the interface
/// of `next` allows handshakes of type `Out`.
/// @pre `next != nullptr` /// @pre `next != nullptr`
/// @pre `self()->pending_stream_managers_[slot] == this` /// @pre `self()->pending_stream_managers_[slot] == this`
/// @pre `out().terminal() == false` /// @pre `out().terminal() == false`
/// @private /// @private
void add_unsafe_outbound_path(strong_actor_ptr next, stream_slot slot, template <class Out, class Handle>
strong_actor_ptr origin, output_stream_t<Out> add_unsafe_outbound_path(Handle next) {
mailbox_element::forwarding_stack stages, // TODO: type checking
message_id mid); auto handshake = make_message(stream<Out>{});
auto hdl = actor_cast<strong_actor_ptr>(std::move(next));
auto slot = add_unsafe_outbound_path_impl(std::move(hdl),
std::move(handshake));
return {0, slot, this};
}
/// Creates an outbound path to `next`, only checking whether the interface
/// of `next` allows handshakes of type `Out` with arguments `Ts...`.
/// @pre `next != nullptr`
/// @pre `self()->pending_stream_managers_[slot] == this`
/// @pre `out().terminal() == false`
/// @private
template <class Out, class Handle, class... Ts>
output_stream_t<Out, detail::strip_and_convert_t<Ts>...>
add_unsafe_outbound_path(const Handle& next, std::tuple<Ts...> xs) {
// TODO: type checking
auto tk = std::make_tuple(stream<Out>{});
auto handshake = make_message_from_tuple(std::tuple_cat(tk, std::move(xs)));
auto hdl = actor_cast<strong_actor_ptr>(std::move(next));
auto slot = add_unsafe_outbound_path_impl(std::move(hdl),
std::move(handshake));
return {0, slot, this};
}
/// Creates an inbound path to the current sender without any type checking. /// Creates an inbound path to the current sender without any type checking.
template <class Result, class In> /// @pre `current_sender() != nullptr`
stream_result<Result> add_unsafe_inbound_path(const stream<In>& in); /// @pre `out().terminal() == false`
/// @private
template <class In>
stream_result<> add_unsafe_inbound_path(const stream<In>&) {
return {add_unsafe_inbound_path_impl(), this};
}
/// Adds `next` as a new outbound path.
/// @private
stream_slot
add_unsafe_outbound_path_impl(strong_actor_ptr next, message handshake,
mailbox_element::forwarding_stack stages = {});
/// Adds `self_->current_sender()` as outbound path.
/// @private
stream_slot add_unsafe_outbound_path_impl(message handshake);
/// Adds
/// @pre Current message is an `open_stream_msg`.
stream_slot add_unsafe_inbound_path_impl();
protected: protected:
// -- modifiers for self ----------------------------------------------------- // -- modifiers for self -----------------------------------------------------
...@@ -222,10 +253,6 @@ protected: ...@@ -222,10 +253,6 @@ protected:
// -- implementation hooks for sources --------------------------------------- // -- implementation hooks for sources ---------------------------------------
/// Returns a type-erased `stream<T>` as handshake token for downstream
/// actors. Returns an empty message for sinks.
virtual message make_handshake(stream_slot slot) const;
/// Called whenever new credit becomes available. The default implementation /// Called whenever new credit becomes available. The default implementation
/// logs an error (sources are expected to override this hook). /// logs an error (sources are expected to override this hook).
virtual void downstream_demand(outbound_path* ptr, long demand); virtual void downstream_demand(outbound_path* ptr, long demand);
...@@ -234,6 +261,8 @@ protected: ...@@ -234,6 +261,8 @@ protected:
/// implementation does nothing. /// implementation does nothing.
virtual void output_closed(error reason); virtual void output_closed(error reason);
// -- member variables -------------------------------------------------------
/// Points to the parent actor. /// Points to the parent actor.
scheduled_actor* self_; scheduled_actor* self_;
...@@ -246,13 +275,6 @@ protected: ...@@ -246,13 +275,6 @@ protected:
/// Configures the importance of outgoing traffic. /// Configures the importance of outgoing traffic.
stream_priority priority_; stream_priority priority_;
/// Stores response promises for dellivering the final result.
std::vector<response_promise> promises_;
/// Stores promises while a handshake is active. The sink at the associated
/// key becomes responsible for the promise after receiving `ack_open`.
std::map<stream_slot, response_promise> in_flight_promises_;
/// Stores whether this stream shall remain open even if no in- or outbound /// Stores whether this stream shall remain open even if no in- or outbound
/// paths exist. /// paths exist.
bool continuous_; bool continuous_;
...@@ -264,23 +286,4 @@ using stream_manager_ptr = intrusive_ptr<stream_manager>; ...@@ -264,23 +286,4 @@ using stream_manager_ptr = intrusive_ptr<stream_manager>;
} // namespace caf } // namespace caf
#include "caf/output_stream.hpp"
#include "caf/stream_result.hpp"
namespace caf {
template <class Out, class... Ts>
output_stream<Out, Ts...>
stream_manager::add_unsafe_outbound_path() {
return {0, this->assign_next_pending_slot(), this};
}
template <class Result, class In>
stream_result<Result>
stream_manager::add_unsafe_inbound_path(const stream<In>&) {
return {this->assign_next_slot(), this};
}
} // namespace caf
#endif // CAF_STREAM_MANAGER_HPP #endif // CAF_STREAM_MANAGER_HPP
...@@ -21,15 +21,12 @@ ...@@ -21,15 +21,12 @@
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/none.hpp" #include "caf/none.hpp"
#include "caf/stream_manager.hpp"
#include "caf/stream_slot.hpp" #include "caf/stream_slot.hpp"
#include "caf/meta/type_name.hpp"
namespace caf { namespace caf {
/// Terminates a stream by reducing it to a single value. /// Terminates a stream by reducing it to a single value.
template <class T, class Pointer /* = stream_manager_ptr */> template <class Pointer /* = stream_manager_ptr */>
class stream_result { class stream_result {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
...@@ -54,14 +51,14 @@ public: ...@@ -54,14 +51,14 @@ public:
} }
template <class CompatiblePointer> template <class CompatiblePointer>
stream_result(stream_result<T, CompatiblePointer> other) stream_result(stream_result<CompatiblePointer> other)
: in_(other.in()), : in_(other.in()),
ptr_(std::move(other.ptr())) { ptr_(std::move(other.ptr())) {
// nop // nop
} }
template <class CompatiblePointer> template <class CompatiblePointer>
stream_result& operator=(stream_result<T, CompatiblePointer> other) { stream_result& operator=(stream_result<CompatiblePointer> other) {
in_ = other.in(); in_ = other.in();
ptr_ = std::move(other.ptr()); ptr_ = std::move(other.ptr());
return *this; return *this;
......
...@@ -115,12 +115,12 @@ public: ...@@ -115,12 +115,12 @@ public:
/// Returns the current number of paths. /// Returns the current number of paths.
virtual size_t num_paths() const noexcept = 0; virtual size_t num_paths() const noexcept = 0;
/// Adds a path to `target` to the scatterer. /// Adds a pending path to `target` to the scatterer.
/// @returns The added path on success, `nullptr` otherwise. /// @returns The added path on success, `nullptr` otherwise.
virtual path_ptr add_path(stream_slots slots, strong_actor_ptr target) = 0; path_ptr add_path(stream_slot slot, strong_actor_ptr target);
/// Removes a path from the scatterer and returns it. /// Removes a path from the scatterer and returns it.
virtual bool remove_path(stream_slot slots, error reason, virtual bool remove_path(stream_slot slot, error reason,
bool silent) noexcept = 0; bool silent) noexcept = 0;
/// Returns the path associated to `slots` or `nullptr`. /// Returns the path associated to `slots` or `nullptr`.
...@@ -164,16 +164,15 @@ public: ...@@ -164,16 +164,15 @@ public:
/// Returns the size of the output buffer. /// Returns the size of the output buffer.
virtual size_t buffered() const noexcept = 0; virtual size_t buffered() const noexcept = 0;
/// Returns `make_message(stream<T>{slot})`, where `T` is the value type of
/// this scatterer.
virtual message make_handshake_token(stream_slot slot) const = 0;
/// Silently removes all paths. /// Silently removes all paths.
virtual void clear_paths() = 0; virtual void clear_paths() = 0;
protected: protected:
// -- customization points --------------------------------------------------- // -- customization points ---------------------------------------------------
/// Inserts `ptr` to the implementation-specific container.
virtual bool insert_path(unique_path_ptr ptr) = 0;
/// Applies `f` to each path. /// Applies `f` to each path.
virtual void for_each_path_impl(path_visitor& f) = 0; virtual void for_each_path_impl(path_visitor& f) = 0;
......
...@@ -59,8 +59,6 @@ public: ...@@ -59,8 +59,6 @@ public:
size_t num_paths() const noexcept override; size_t num_paths() const noexcept override;
path_ptr add_path(stream_slots slots, strong_actor_ptr target) override;
bool remove_path(stream_slot slots, error reason, bool remove_path(stream_slot slots, error reason,
bool silent) noexcept override; bool silent) noexcept override;
...@@ -69,6 +67,8 @@ public: ...@@ -69,6 +67,8 @@ public:
void clear_paths() override; void clear_paths() override;
protected: protected:
bool insert_path(unique_path_ptr ptr) override;
void for_each_path_impl(path_visitor& f) override; void for_each_path_impl(path_visitor& f) override;
bool check_paths_impl(path_algorithm algo, bool check_paths_impl(path_algorithm algo,
......
...@@ -29,15 +29,13 @@ ...@@ -29,15 +29,13 @@
namespace caf { namespace caf {
template <class In, class Result> template <class In>
class stream_sink : public virtual stream_manager { class stream_sink : public virtual stream_manager {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
using input_type = In; using input_type = In;
using result_type = Result;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
stream_sink(scheduled_actor* self) : stream_manager(self) { stream_sink(scheduled_actor* self) : stream_manager(self) {
...@@ -47,14 +45,14 @@ public: ...@@ -47,14 +45,14 @@ public:
// -- properties ------------------------------------------------------------- // -- properties -------------------------------------------------------------
/// Creates a new input path to the current sender. /// Creates a new input path to the current sender.
stream_result<result_type, intrusive_ptr<stream_sink>> stream_result<intrusive_ptr<stream_sink>>
add_inbound_path(const stream<input_type>&) { add_inbound_path(const stream<input_type>&) {
return {this->assign_next_slot(), this}; return {add_unsafe_inbound_path_impl(), this};
} }
}; };
template <class In, class Result> template <class In>
using stream_sink_ptr = intrusive_ptr<stream_sink<In, Result>>; using stream_sink_ptr = intrusive_ptr<stream_sink<In>>;
} // namespace caf } // namespace caf
......
...@@ -31,17 +31,15 @@ ...@@ -31,17 +31,15 @@
namespace caf { namespace caf {
/// Identifies an unbound sequence of messages. /// Identifies an unbound sequence of messages.
template <class Input, class Result> template <class Input>
class stream_sink_driver { class stream_sink_driver {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
using input_type = Input; using input_type = Input;
using result_type = Result;
/// Implemented `stream_sink` interface. /// Implemented `stream_sink` interface.
using sink_type = stream_sink<input_type, result_type>; using sink_type = stream_sink<input_type>;
/// Smart pointer to the interface type. /// Smart pointer to the interface type.
using sink_ptr_type = intrusive_ptr<sink_type>; using sink_ptr_type = intrusive_ptr<sink_type>;
...@@ -54,11 +52,7 @@ public: ...@@ -54,11 +52,7 @@ public:
// -- virtual functions ------------------------------------------------------ // -- virtual functions ------------------------------------------------------
/// Produces a result message after closing the last inbound path. /// Called after closing the last inbound path.
virtual message make_final_result() {
return make_message();
}
virtual void finalize(const error&) { virtual void finalize(const error&) {
// nop // nop
} }
...@@ -66,7 +60,7 @@ public: ...@@ -66,7 +60,7 @@ public:
// -- pure virtual functions ------------------------------------------------- // -- pure virtual functions -------------------------------------------------
/// Processes a single batch. /// Processes a single batch.
virtual void process(std::vector<input_type>&& batch) = 0; virtual void process(std::vector<input_type>& batch) = 0;
}; };
} // namespace caf } // namespace caf
......
...@@ -33,7 +33,7 @@ namespace detail { ...@@ -33,7 +33,7 @@ namespace detail {
struct stream_sink_trait_invoke_one { struct stream_sink_trait_invoke_one {
template <class F, class State, class In> template <class F, class State, class In>
static void invoke(F& f, State& st, std::vector<In>&& xs) { static void invoke(F& f, State& st, std::vector<In>& xs) {
for (auto& x : xs) for (auto& x : xs)
f(st, std::move(x)); f(st, std::move(x));
} }
...@@ -41,25 +41,8 @@ struct stream_sink_trait_invoke_one { ...@@ -41,25 +41,8 @@ struct stream_sink_trait_invoke_one {
struct stream_sink_trait_invoke_all { struct stream_sink_trait_invoke_all {
template <class F, class State, class In> template <class F, class State, class In>
static void invoke(F& f, State& st, std::vector<In>&& xs) { static void invoke(F& f, State& st, std::vector<In>& xs) {
f(st, std::move(xs)); f(st, xs);
}
};
// -- result helper to support void and non-void functions ---------------------
struct stream_sink_trait_default_finalize {
template <class F, class State>
static message invoke(F& f, State& st) {
return make_message(f(st));
}
};
struct stream_sink_trait_void_finalize {
template <class F, class State>
static message invoke(F& f, State& st) {
f(st);
return make_message();
} }
}; };
...@@ -68,7 +51,7 @@ struct stream_sink_trait_void_finalize { ...@@ -68,7 +51,7 @@ struct stream_sink_trait_void_finalize {
// -- trait implementation ----------------------------------------------------- // -- trait implementation -----------------------------------------------------
/// Base type for all sink traits. /// Base type for all sink traits.
template <class State, class In, class Result> template <class State, class In>
struct stream_sink_trait_base { struct stream_sink_trait_base {
/// Defines the state element for the function objects. /// Defines the state element for the function objects.
using state = State; using state = State;
...@@ -76,57 +59,26 @@ struct stream_sink_trait_base { ...@@ -76,57 +59,26 @@ struct stream_sink_trait_base {
/// Defines the type of a single stream element. /// Defines the type of a single stream element.
using input = In; using input = In;
/// Defines the result type of the sink.
using result = Result;
/// Defines a pointer to a sink. /// Defines a pointer to a sink.
using pointer = stream_sink_ptr<input, result>; using pointer = stream_sink_ptr<input>;
}; };
/// Defines required type aliases for stream sinks. /// Defines required type aliases for stream sinks.
template <class Fun, class Fin> template <class Fun>
struct stream_sink_trait; struct stream_sink_trait;
/// Specializes the trait for element-wise processing with result. /// Specializes the trait for element-wise processing.
template <class State, class In, class Out>
struct stream_sink_trait<void(State&, In), Out(State&)>
: stream_sink_trait_base<State, In, Out> {
/// Defines a helper for dispatching to the finalizing function object.
using finalize = detail::stream_sink_trait_default_finalize;
/// Defines a helper for dispatching to the processing function object.
using process = detail::stream_sink_trait_invoke_one;
};
/// Specializes the trait for element-wise processing without result.
template <class State, class In> template <class State, class In>
struct stream_sink_trait<void(State&, In), void(State&)> struct stream_sink_trait<void(State&, In)>
: stream_sink_trait_base<State, In, void> { : stream_sink_trait_base<State, In> {
/// Defines a helper for dispatching to the finalizing function object.
using finalize = detail::stream_sink_trait_void_finalize;
/// Defines a helper for dispatching to the processing function object. /// Defines a helper for dispatching to the processing function object.
using process = detail::stream_sink_trait_invoke_one; using process = detail::stream_sink_trait_invoke_one;
}; };
/// Specializes the trait for batch-wise processing with result. /// Specializes the trait for batch-wise processing.
template <class State, class In, class Out>
struct stream_sink_trait<void(State&, std::vector<In>&&), Out(State&)>
: stream_sink_trait_base<State, In, Out> {
/// Defines a helper for dispatching to the finalizing function object.
using finalize = detail::stream_sink_trait_default_finalize;
/// Defines a helper for dispatching to the processing function object.
using process = detail::stream_sink_trait_invoke_all;
};
/// Specializes the trait for batch-wise processing without result.
template <class State, class In> template <class State, class In>
struct stream_sink_trait<void(State&, std::vector<In>&&), void(State&)> struct stream_sink_trait<void(State&, std::vector<In>&)>
: stream_sink_trait_base<State, In, void> { : stream_sink_trait_base<State, In> {
/// Defines a helper for dispatching to the finalizing function object.
using finalize = detail::stream_sink_trait_void_finalize;
/// Defines a helper for dispatching to the processing function object. /// Defines a helper for dispatching to the processing function object.
using process = detail::stream_sink_trait_invoke_all; using process = detail::stream_sink_trait_invoke_all;
}; };
...@@ -134,10 +86,9 @@ struct stream_sink_trait<void(State&, std::vector<In>&&), void(State&)> ...@@ -134,10 +86,9 @@ struct stream_sink_trait<void(State&, std::vector<In>&&), void(State&)>
// -- convenience alias -------------------------------------------------------- // -- convenience alias --------------------------------------------------------
/// Derives a sink trait from the signatures of Fun and Fin. /// Derives a sink trait from the signatures of Fun and Fin.
template <class Fun, class Fin> template <class Fun>
using stream_sink_trait_t = using stream_sink_trait_t =
stream_sink_trait<typename detail::get_callable_trait<Fun>::fun_sig, stream_sink_trait<typename detail::get_callable_trait<Fun>::fun_sig>;
typename detail::get_callable_trait<Fin>::fun_sig>;
} // namespace caf } // namespace caf
......
...@@ -23,13 +23,14 @@ ...@@ -23,13 +23,14 @@
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/intrusive_ptr.hpp" #include "caf/intrusive_ptr.hpp"
#include "caf/logger.hpp"
#include "caf/stream_manager.hpp" #include "caf/stream_manager.hpp"
#include "caf/detail/type_traits.hpp" #include "caf/detail/type_traits.hpp"
namespace caf { namespace caf {
template <class Out, class Scatterer, class... Ts> template <class Out, class Scatterer>
class stream_source : public virtual stream_manager { class stream_source : public virtual stream_manager {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
...@@ -47,23 +48,49 @@ public: ...@@ -47,23 +48,49 @@ public:
} }
/// Creates a new output path to the current sender. /// Creates a new output path to the current sender.
output_stream<Out, std::tuple<Ts...>, intrusive_ptr<stream_source>> output_stream<Out, std::tuple<>, intrusive_ptr<stream_source>>
add_outbound_path() { add_outbound_path() {
return {0, this->assign_next_pending_slot(), this}; CAF_LOG_TRACE("");
return this->add_unsafe_outbound_path<Out>().rebind(this);
}
/// Creates a new output path to the current sender with custom handshake.
template <class... Ts>
output_stream<Out, std::tuple<detail::strip_and_convert_t<Ts>...>,
intrusive_ptr<stream_source>>
add_outbound_path(std::tuple<Ts...> xs) {
CAF_LOG_TRACE(CAF_ARG(xs));
return this->add_unsafe_outbound_path<Out>(std::move(xs)).rebind(this);
}
/// Creates a new output path to the current sender.
template <class Handle>
output_stream<Out, std::tuple<>, intrusive_ptr<stream_source>>
add_outbound_path(const Handle& next) {
CAF_LOG_TRACE(CAF_ARG(next));
return this->add_unsafe_outbound_path<Out>(next).rebind(this);
}
/// Creates a new output path to the current sender with custom handshake.
template <class Handle, class... Ts>
output_stream<Out, std::tuple<detail::strip_and_convert_t<Ts>...>,
intrusive_ptr<stream_source>>
add_outbound_path(const Handle& next, std::tuple<Ts...> xs) {
CAF_LOG_TRACE(CAF_ARG(next) << CAF_ARG(xs));
return this->add_unsafe_outbound_path<Out>(next, std::move(xs))
.rebind(this);
} }
protected: protected:
Scatterer out_; Scatterer out_;
}; };
template <class Out, class Scatterer, class... HandshakeData> template <class Out, class Scatterer>
using stream_source_ptr = using stream_source_ptr = intrusive_ptr<stream_source<Out, Scatterer>>;
intrusive_ptr<stream_source<Out, Scatterer, HandshakeData...>>;
template <class Scatterer, class... Ts> template <class Scatterer>
using stream_source_ptr_t = using stream_source_ptr_t =
stream_source_ptr<typename Scatterer::value_type, Scatterer, stream_source_ptr<typename Scatterer::value_type, Scatterer>;
detail::decay_t<Ts>...>;
} // namespace caf } // namespace caf
......
...@@ -27,7 +27,7 @@ ...@@ -27,7 +27,7 @@
namespace caf { namespace caf {
/// Identifies an unbound sequence of messages. /// Identifies an unbound sequence of messages.
template <class Scatterer, class... Ts> template <class Scatterer>
class stream_source_driver { class stream_source_driver {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
...@@ -42,18 +42,14 @@ public: ...@@ -42,18 +42,14 @@ public:
using stream_type = stream<output_type>; using stream_type = stream<output_type>;
/// Tuple for creating the `open_stream_msg` handshake. /// Tuple for creating the `open_stream_msg` handshake.
using handshake_tuple_type = std::tuple<stream_type, Ts...>; using handshake_tuple_type = std::tuple<stream_type>;
/// Implemented `stream_source` interface. /// Implemented `stream_source` interface.
using source_type = stream_source<output_type, scatterer_type, Ts...>; using source_type = stream_source<output_type, scatterer_type>;
/// Smart pointer to the interface type. /// Smart pointer to the interface type.
using source_ptr_type = intrusive_ptr<source_type>; using source_ptr_type = intrusive_ptr<source_type>;
/// Type of the output stream including handshake argument types.
using output_stream_type = output_stream<output_type, std::tuple<Ts...>,
source_ptr_type>;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
virtual ~stream_source_driver() { virtual ~stream_source_driver() {
...@@ -69,9 +65,6 @@ public: ...@@ -69,9 +65,6 @@ public:
// -- pure virtual functions ------------------------------------------------- // -- pure virtual functions -------------------------------------------------
/// Generates handshake data for the next actor in the pipeline.
virtual handshake_tuple_type make_handshake(stream_slot slot) const = 0;
/// Generates more stream elements. /// Generates more stream elements.
virtual void pull(downstream<output_type>& out, size_t num) = 0; virtual void pull(downstream<output_type>& out, size_t num) = 0;
......
...@@ -28,16 +28,15 @@ ...@@ -28,16 +28,15 @@
namespace caf { namespace caf {
template <class In, class Result, class Out, class Scatterer, template <class In, class Out, class Scatterer>
class... HandshakeData> class stream_stage : public stream_source<Out, Scatterer>,
class stream_stage : public stream_source<Out, Scatterer, HandshakeData...>, public stream_sink<In> {
public stream_sink<In, Result> {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
using left_super = stream_source<Out, Scatterer, HandshakeData...>; using left_super = stream_source<Out, Scatterer>;
using right_super = stream_sink<In, Result>; using right_super = stream_sink<In>;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
...@@ -49,10 +48,9 @@ public: ...@@ -49,10 +48,9 @@ public:
} }
}; };
template <class In, class Result, class Out, class Scatterer, template <class In, class Out, class Scatterer>
class... HandshakeData>
using stream_stage_ptr = using stream_stage_ptr =
intrusive_ptr<stream_stage<In, Result, Out, Scatterer, HandshakeData...>>; intrusive_ptr<stream_stage<In, Out, Scatterer>>;
} // namespace caf } // namespace caf
......
...@@ -29,7 +29,7 @@ ...@@ -29,7 +29,7 @@
namespace caf { namespace caf {
/// Encapsulates user-provided functionality for generating a stream stage. /// Encapsulates user-provided functionality for generating a stream stage.
template <class Input, class Result, class Scatterer, class... Ts> template <class Input, class Scatterer>
class stream_stage_driver { class stream_stage_driver {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
...@@ -37,10 +37,6 @@ public: ...@@ -37,10 +37,6 @@ public:
/// Element type of the input stream. /// Element type of the input stream.
using input_type = Input; using input_type = Input;
/// Result type shipped to the client after processing all elements of the
/// stream.
using result_type = Result;
/// Policy for distributing data to outbound paths. /// Policy for distributing data to outbound paths.
using scatterer_type = Scatterer; using scatterer_type = Scatterer;
...@@ -50,21 +46,12 @@ public: ...@@ -50,21 +46,12 @@ public:
/// Type of the output stream. /// Type of the output stream.
using stream_type = stream<output_type>; using stream_type = stream<output_type>;
/// Tuple for creating the `open_stream_msg` handshake.
using handshake_tuple_type = std::tuple<stream_type, Ts...>;
/// Implemented `stream_stage` interface. /// Implemented `stream_stage` interface.
using stage_type = stream_stage<input_type, result_type, using stage_type = stream_stage<input_type, output_type, Scatterer>;
output_type, Scatterer, Ts...>;
/// Smart pointer to the interface type. /// Smart pointer to the interface type.
using stage_ptr_type = intrusive_ptr<stage_type>; using stage_ptr_type = intrusive_ptr<stage_type>;
/// Type of the output stream including handshake argument types.
using output_stream_type = output_stream<output_type, std::tuple<Ts...>,
stage_ptr_type>;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
virtual ~stream_stage_driver() { virtual ~stream_stage_driver() {
...@@ -73,22 +60,14 @@ public: ...@@ -73,22 +60,14 @@ public:
// -- virtual functions ------------------------------------------------------ // -- virtual functions ------------------------------------------------------
/// Generates handshake data for the next actor in the pipeline.
virtual handshake_tuple_type make_handshake(stream_slot slot) const = 0;
/// Processes a single batch. /// Processes a single batch.
virtual void process(std::vector<input_type>&& batch, virtual void process(downstream<output_type>& out,
downstream<output_type>& out) = 0; std::vector<input_type>& batch) = 0;
/// Handles the result of an outbound path.
virtual void add_result(message&) = 0;
/// Produces a result message after receiving the result messages of all
/// outbound paths and closing all paths.
virtual message make_final_result() = 0;
/// Cleans up any state. /// Cleans up any state.
virtual void finalize(const error&) = 0; virtual void finalize(const error&) {
// nop
}
}; };
} // namespace caf } // namespace caf
......
...@@ -32,19 +32,19 @@ namespace detail { ...@@ -32,19 +32,19 @@ namespace detail {
// -- invoke helper to support element-wise and batch-wise processing ---------- // -- invoke helper to support element-wise and batch-wise processing ----------
struct stream_stage_trait_invoke_one { struct stream_stage_trait_invoke_one {
template <class F, class State, class In, class Out> template <class F, class State, class Out, class In>
static void invoke(F& f, State& st, std::vector<In>&& xs, static void invoke(F& f, State& st, downstream<Out>& out,
downstream<Out>& out) { std::vector<In>& xs) {
for (auto& x : xs) for (auto& x : xs)
f(st, out, std::move(x)); f(st, out, std::move(x));
} }
}; };
struct stream_stage_trait_invoke_all { struct stream_stage_trait_invoke_all {
template <class F, class State, class In, class Out> template <class F, class State, class Out, class In>
static void invoke(F& f, State& st, std::vector<In>&& xs, static void invoke(F& f, State& st, downstream<Out>& out,
downstream<Out>& out) { std::vector<In>& xs) {
f(st, out, std::move(xs)); f(st, out, xs);
} }
}; };
...@@ -66,7 +66,7 @@ struct stream_stage_trait<void (State&, downstream<Out>&, In)> { ...@@ -66,7 +66,7 @@ struct stream_stage_trait<void (State&, downstream<Out>&, In)> {
}; };
template <class State, class In, class Out> template <class State, class In, class Out>
struct stream_stage_trait<void (State&, downstream<Out>&, std::vector<In>&&)> { struct stream_stage_trait<void (State&, downstream<Out>&, std::vector<In>&)> {
using state = State; using state = State;
using input = In; using input = In;
using output = Out; using output = Out;
......
...@@ -73,10 +73,6 @@ public: ...@@ -73,10 +73,6 @@ public:
void operator()(const none_t&) override { void operator()(const none_t&) override {
(*this)(); (*this)();
} }
void operator()(stream_slot, stream_slot, stream_manager_ptr&) override {
(*this)();
}
}; };
} // namespace <anonymous> } // namespace <anonymous>
......
...@@ -19,8 +19,10 @@ ...@@ -19,8 +19,10 @@
#include "caf/invalid_stream_scatterer.hpp" #include "caf/invalid_stream_scatterer.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/message_builder.hpp"
#include "caf/outbound_path.hpp" #include "caf/outbound_path.hpp"
#include "caf/stream.hpp" #include "caf/stream.hpp"
#include "caf/stream_slot.hpp"
namespace caf { namespace caf {
...@@ -37,11 +39,6 @@ size_t invalid_stream_scatterer::num_paths() const noexcept { ...@@ -37,11 +39,6 @@ size_t invalid_stream_scatterer::num_paths() const noexcept {
return 0; return 0;
} }
auto invalid_stream_scatterer::add_path(stream_slots, strong_actor_ptr)
-> path_ptr {
return nullptr;
}
bool invalid_stream_scatterer::remove_path(stream_slot, error, bool) noexcept { bool invalid_stream_scatterer::remove_path(stream_slot, error, bool) noexcept {
return false; return false;
} }
...@@ -66,8 +63,8 @@ size_t invalid_stream_scatterer::buffered() const noexcept { ...@@ -66,8 +63,8 @@ size_t invalid_stream_scatterer::buffered() const noexcept {
return 0u; return 0u;
} }
message invalid_stream_scatterer::make_handshake_token(stream_slot slot) const { bool invalid_stream_scatterer::insert_path(unique_path_ptr) {
return make_message(stream<message>{slot}); return false;
} }
void invalid_stream_scatterer::for_each_path_impl(path_visitor&) { void invalid_stream_scatterer::for_each_path_impl(path_visitor&) {
......
...@@ -25,9 +25,10 @@ ...@@ -25,9 +25,10 @@
namespace caf { namespace caf {
outbound_path::outbound_path(stream_slots id, strong_actor_ptr ptr) outbound_path::outbound_path(stream_slot sender_slot,
: slots(id), strong_actor_ptr receiver_hdl)
hdl(std::move(ptr)), : slots(sender_slot, invalid_stream_slot),
hdl(std::move(receiver_hdl)),
next_batch_id(1), next_batch_id(1),
open_credit(0), open_credit(0),
desired_batch_size(0), desired_batch_size(0),
......
...@@ -400,44 +400,6 @@ void scheduled_actor::quit(error x) { ...@@ -400,44 +400,6 @@ void scheduled_actor::quit(error x) {
setf(is_terminated_flag); setf(is_terminated_flag);
} }
// -- stream management --------------------------------------------------------
sec scheduled_actor::build_pipeline(stream_slot in, stream_slot out,
stream_manager_ptr mgr) {
CAF_LOG_TRACE(CAF_ARG(in) << CAF_ARG(out) << CAF_ARG(mgr));
auto fail = [&](sec code) {
auto rp = make_response_promise();
rp.deliver(code);
return code;
};
// Check arguments.
if (mgr == nullptr || (in | out) == 0) {
CAF_LOG_ERROR("build_pipeline called with invalid arguments");
return sec::invalid_stream_state;
}
// Get handle to the next stage in the pipeline.
auto next = take_current_next_stage();
if (in != 0) {
CAF_ASSERT(stream_managers_[in] == mgr);
// Sinks must always terminate the stream and store a response promise to
// ship the final result.
if (mgr->out().terminal()) {
CAF_ASSERT(out == 0);
if (next != nullptr)
return fail(sec::cannot_add_downstream);
mgr->add_promise(make_response_promise());
}
}
if (out != 0 && mgr->out().path(out) == nullptr) {
if (next == nullptr)
return fail(sec::no_downstream_stages_defined);
mgr->add_unsafe_outbound_path(std::move(next), out, current_sender(),
take_current_forwarding_stack(),
current_message_id());
}
return sec::none;
}
// -- timeout management ------------------------------------------------------- // -- timeout management -------------------------------------------------------
uint64_t scheduled_actor::set_receive_timeout(actor_clock::time_point x) { uint64_t scheduled_actor::set_receive_timeout(actor_clock::time_point x) {
...@@ -970,7 +932,13 @@ void scheduled_actor::erase_stream_manager(stream_slot id) { ...@@ -970,7 +932,13 @@ void scheduled_actor::erase_stream_manager(stream_slot id) {
stream_ticks_.stop(); stream_ticks_.stop();
} }
void scheduled_actor::erase_pending_stream_manager(stream_slot id) {
CAF_LOG_TRACE(CAF_ARG(id));
pending_stream_managers_.erase(id);
}
void scheduled_actor::erase_stream_manager(const stream_manager_ptr& mgr) { void scheduled_actor::erase_stream_manager(const stream_manager_ptr& mgr) {
{ // Lifetime scope of first iterator pair.
auto i = stream_managers_.begin(); auto i = stream_managers_.begin();
auto e = stream_managers_.end(); auto e = stream_managers_.end();
while (i != e) while (i != e)
...@@ -978,6 +946,16 @@ void scheduled_actor::erase_stream_manager(const stream_manager_ptr& mgr) { ...@@ -978,6 +946,16 @@ void scheduled_actor::erase_stream_manager(const stream_manager_ptr& mgr) {
i = stream_managers_.erase(i); i = stream_managers_.erase(i);
else else
++i; ++i;
}
{ // Lifetime scope of second iterator pair.
auto i = pending_stream_managers_.begin();
auto e = pending_stream_managers_.end();
while (i != e)
if (i->second == mgr)
i = pending_stream_managers_.erase(i);
else
++i;
}
if (stream_managers_.empty()) if (stream_managers_.empty())
stream_ticks_.stop(); stream_ticks_.stop();
} }
...@@ -987,9 +965,6 @@ scheduled_actor::handle_open_stream_msg(mailbox_element& x) { ...@@ -987,9 +965,6 @@ scheduled_actor::handle_open_stream_msg(mailbox_element& x) {
CAF_LOG_TRACE(CAF_ARG(x)); CAF_LOG_TRACE(CAF_ARG(x));
// Fetches a stream manger from a behavior. // Fetches a stream manger from a behavior.
struct visitor : detail::invoke_result_visitor { struct visitor : detail::invoke_result_visitor {
stream_manager_ptr ptr;
stream_slot in_slot;
stream_slot out_slot;
void operator()() override { void operator()() override {
// nop // nop
} }
...@@ -1002,13 +977,6 @@ scheduled_actor::handle_open_stream_msg(mailbox_element& x) { ...@@ -1002,13 +977,6 @@ scheduled_actor::handle_open_stream_msg(mailbox_element& x) {
// nop // nop
} }
void operator()(stream_slot in, stream_slot out,
stream_manager_ptr& x) override {
ptr = std::move(x);
in_slot = in;
out_slot = out;
}
void operator()(const none_t&) override { void operator()(const none_t&) override {
// nop // nop
} }
...@@ -1048,6 +1016,7 @@ scheduled_actor::handle_open_stream_msg(mailbox_element& x) { ...@@ -1048,6 +1016,7 @@ scheduled_actor::handle_open_stream_msg(mailbox_element& x) {
CAF_LOG_DEBUG("no match in behavior, fall back to default handler"); CAF_LOG_DEBUG("no match in behavior, fall back to default handler");
return fallback(); return fallback();
case match_case::result::match: { case match_case::result::match: {
/*
if (f.ptr == nullptr) { if (f.ptr == nullptr) {
CAF_LOG_WARNING("actor did not return a stream manager after " CAF_LOG_WARNING("actor did not return a stream manager after "
"handling open_stream_msg"); "handling open_stream_msg");
...@@ -1060,6 +1029,7 @@ scheduled_actor::handle_open_stream_msg(mailbox_element& x) { ...@@ -1060,6 +1029,7 @@ scheduled_actor::handle_open_stream_msg(mailbox_element& x) {
path->emit_ack_open(this, actor_cast<actor_addr>(osm.original_stage)); path->emit_ack_open(this, actor_cast<actor_addr>(osm.original_stage));
// Propagate handshake down the pipeline. // Propagate handshake down the pipeline.
build_pipeline(f.in_slot, f.out_slot, std::move(f.ptr)); build_pipeline(f.in_slot, f.out_slot, std::move(f.ptr));
*/
return im_success; return im_success;
} }
default: default:
......
...@@ -60,14 +60,25 @@ void stream_manager::handle(inbound_path*, downstream_msg::forced_close& x) { ...@@ -60,14 +60,25 @@ void stream_manager::handle(inbound_path*, downstream_msg::forced_close& x) {
bool stream_manager::handle(stream_slots slots, upstream_msg::ack_open& x) { bool stream_manager::handle(stream_slots slots, upstream_msg::ack_open& x) {
CAF_LOG_TRACE(CAF_ARG(slots) << CAF_ARG(x)); CAF_LOG_TRACE(CAF_ARG(slots) << CAF_ARG(x));
auto path = out().add_path(slots.invert(), x.rebind_to); auto ptr = out().path(slots.receiver);
if (path == nullptr) if (ptr == nullptr)
return false; return false;
path->open_credit = x.initial_demand; if (!ptr->pending()) {
path->desired_batch_size = x.desired_batch_size; CAF_LOG_ERROR("received repeated ack_open");
return false;
}
if (ptr->hdl != x.rebind_from) {
CAF_LOG_ERROR("received ack_open with invalid rebind_from");
return false;
}
if (x.rebind_from != x.rebind_to) {
ptr->hdl = x.rebind_to;
}
ptr->slots.receiver = slots.sender;
ptr->open_credit = x.initial_demand;
ptr->desired_batch_size = x.desired_batch_size;
--pending_handshakes_; --pending_handshakes_;
push(); push();
in_flight_promises_.erase(slots.sender);
return true; return true;
} }
...@@ -98,19 +109,10 @@ void stream_manager::stop() { ...@@ -98,19 +109,10 @@ void stream_manager::stop() {
error tmp; error tmp;
finalize(tmp); finalize(tmp);
self_->erase_inbound_paths_later(this); self_->erase_inbound_paths_later(this);
if (!promises_.empty())
deliver_promises(make_final_result());
} }
void stream_manager::abort(error reason) { void stream_manager::abort(error reason) {
CAF_LOG_TRACE(CAF_ARG(reason)); CAF_LOG_TRACE(CAF_ARG(reason));
if (!promises_.empty() || !in_flight_promises_.empty()) {
auto msg = make_message(reason);
deliver_promises(msg);
for (auto& kvp : in_flight_promises_)
kvp.second.deliver(msg);
in_flight_promises_.clear();
}
out().abort(reason); out().abort(reason);
finalize(reason); finalize(reason);
self_->erase_inbound_paths_later(this, std::move(reason)); self_->erase_inbound_paths_later(this, std::move(reason));
...@@ -127,27 +129,19 @@ bool stream_manager::congested() const noexcept { ...@@ -127,27 +129,19 @@ bool stream_manager::congested() const noexcept {
return false; return false;
} }
void stream_manager::send_handshake(strong_actor_ptr dest, stream_slot slot, void stream_manager::send_handshake(strong_actor_ptr next, stream_slot slot,
strong_actor_ptr client, mailbox_element::forwarding_stack stages,
mailbox_element::forwarding_stack fwd_stack, message handshake) {
message_id mid) { CAF_ASSERT(next != nullptr);
CAF_ASSERT(dest != nullptr); CAF_ASSERT(slot != invalid_stream_slot);
++pending_handshakes_; ++pending_handshakes_;
in_flight_promises_.emplace( next->enqueue(
slot, response_promise{self()->ctrl(), client, fwd_stack, mid}); make_mailbox_element(self_->ctrl(), make_message_id(), std::move(stages),
dest->enqueue( open_stream_msg{slot, std::move(handshake),
make_mailbox_element(std::move(client), mid, std::move(fwd_stack), self_->ctrl(), next, priority_}),
open_stream_msg{slot, make_handshake(slot),
self_->ctrl(), dest, priority_}),
self_->context()); self_->context());
} }
void stream_manager::send_handshake(strong_actor_ptr dest, stream_slot slot) {
mailbox_element::forwarding_stack fwd_stack;
send_handshake(std::move(dest), slot, nullptr, std::move(fwd_stack),
make_message_id());
}
bool stream_manager::generate_messages() { bool stream_manager::generate_messages() {
return false; return false;
} }
...@@ -184,28 +178,67 @@ void stream_manager::remove_input_path(stream_slot slot, error reason, ...@@ -184,28 +178,67 @@ void stream_manager::remove_input_path(stream_slot slot, error reason,
self_->erase_inbound_path_later(slot, std::move(reason)); self_->erase_inbound_path_later(slot, std::move(reason));
} }
void stream_manager::add_promise(response_promise x) { stream_slot stream_manager::add_unsafe_outbound_path_impl(
CAF_LOG_TRACE(CAF_ARG(x)); strong_actor_ptr next, message handshake,
CAF_ASSERT(out().terminal()); mailbox_element::forwarding_stack stages) {
promises_.emplace_back(std::move(x)); CAF_LOG_TRACE(CAF_ARG(next) << CAF_ARG(handshake) << CAF_ARG(stages));
CAF_ASSERT(out().terminal() == false);
if (next == nullptr) {
CAF_LOG_WARNING("add_outbound_path called with next == nullptr");
auto rp = self_->make_response_promise();
rp.deliver(sec::no_downstream_stages_defined);
return invalid_stream_slot;
}
auto slot = self_->assign_next_pending_slot_to(this);
auto path = out().add_path(slot, next);
CAF_IGNORE_UNUSED(path);
CAF_ASSERT(path != nullptr);
// Build pipeline by forwarding handshake along the path.
send_handshake(std::move(next), slot, std::move(stages),
std::move(handshake));
generate_messages();
return slot;
} }
void stream_manager::deliver_promises(message x) { stream_slot stream_manager::add_unsafe_outbound_path_impl(message handshake) {
CAF_LOG_TRACE(CAF_ARG(x)); CAF_LOG_TRACE(CAF_ARG(handshake));
for (auto& p : promises_) // Call take_current_next_stage explicitly before calling
p.deliver(x); // take_current_forwarding_stack to avoid UB due to undefined evaluation
promises_.clear(); // order.
auto next = self_->take_current_next_stage();
auto stages = self_->take_current_forwarding_stack();
// Sources simply forward the original request.
return add_unsafe_outbound_path_impl(std::move(next), std::move(handshake),
std::move(stages));
} }
void stream_manager::add_unsafe_outbound_path( stream_slot stream_manager::add_unsafe_inbound_path_impl() {
strong_actor_ptr next, stream_slot slot, strong_actor_ptr origin, CAF_LOG_TRACE("");
mailbox_element::forwarding_stack stages, message_id mid) { auto x = self_->current_mailbox_element();
CAF_ASSERT(next != nullptr); if (x == nullptr || !x->content().match_elements<open_stream_msg>()) {
CAF_ASSERT(out().terminal() == false); CAF_LOG_ERROR("add_unsafe_inbound_path called, but current message "
// Build pipeline by forwarding handshake along the path. "is not an open_stream_msg");
send_handshake(std::move(next), slot, std::move(origin), return invalid_stream_slot;
std::move(stages), mid); }
generate_messages(); auto& osm = x->content().get_mutable_as<open_stream_msg>(0);
auto fail = [&](sec code) {
stream_slots path_id{osm.slot, 0};
inbound_path::emit_irregular_shutdown(self_, path_id,
std::move(osm.prev_stage), code);
return invalid_stream_slot;
};
if (out().terminal() && self_->current_next_stage() != nullptr) {
// Sinks must always terminate the stream.
CAF_LOG_WARNING("add_unsafe_inbound_path called in a sink, but the "
"handshake has further stages");
return fail(sec::cannot_add_downstream);
}
auto slot = assign_next_slot();
stream_slots path_id{osm.slot, slot};
auto ptr = self_->make_inbound_path(this, path_id, std::move(osm.prev_stage));
CAF_ASSERT(ptr != nullptr);
ptr->emit_ack_open(self_, actor_cast<actor_addr>(osm.original_stage));
return slot;
} }
stream_slot stream_manager::assign_next_slot() { stream_slot stream_manager::assign_next_slot() {
...@@ -233,11 +266,6 @@ void stream_manager::output_closed(error) { ...@@ -233,11 +266,6 @@ void stream_manager::output_closed(error) {
// nop // nop
} }
message stream_manager::make_handshake(stream_slot) const {
CAF_LOG_ERROR("stream_manager::make_handshake called");
return none;
}
void stream_manager::downstream_demand(outbound_path*, long) { void stream_manager::downstream_demand(outbound_path*, long) {
CAF_LOG_ERROR("stream_manager::downstream_demand called"); CAF_LOG_ERROR("stream_manager::downstream_demand called");
} }
......
...@@ -96,8 +96,17 @@ bool stream_scatterer::terminal() const noexcept { ...@@ -96,8 +96,17 @@ bool stream_scatterer::terminal() const noexcept {
return false; return false;
} }
stream_scatterer::path_ptr stream_scatterer::add_path(stream_slot slot,
strong_actor_ptr target) {
CAF_LOG_TRACE(CAF_ARG(slot) << CAF_ARG(target));
unique_path_ptr ptr{new outbound_path(slot, std::move(target))};
auto result = ptr.get();
return insert_path(std::move(ptr)) ? result : nullptr;
}
void stream_scatterer::about_to_erase(outbound_path* ptr, bool silent, void stream_scatterer::about_to_erase(outbound_path* ptr, bool silent,
error* reason) { error* reason) {
CAF_LOG_TRACE(CAF_ARG(ptr) << CAF_ARG(silent) << CAF_ARG(reason));
if (!silent) { if (!silent) {
if (reason == nullptr) if (reason == nullptr)
ptr->emit_regular_shutdown(self_); ptr->emit_regular_shutdown(self_);
......
...@@ -34,24 +34,13 @@ stream_scatterer_impl::~stream_scatterer_impl() { ...@@ -34,24 +34,13 @@ stream_scatterer_impl::~stream_scatterer_impl() {
// nop // nop
} }
outbound_path* stream_scatterer_impl::add_path(stream_slots slots,
strong_actor_ptr target) {
CAF_LOG_TRACE(CAF_ARG(slots) << CAF_ARG(target));
auto res = paths_.emplace(slots.sender, nullptr);
if (res.second) {
auto ptr = new outbound_path(slots, std::move(target));
res.first->second.reset(ptr);
return ptr;
}
return nullptr;
}
size_t stream_scatterer_impl::num_paths() const noexcept { size_t stream_scatterer_impl::num_paths() const noexcept {
return paths_.size(); return paths_.size();
} }
bool stream_scatterer_impl::remove_path(stream_slot slot, error reason, bool stream_scatterer_impl::remove_path(stream_slot slot, error reason,
bool silent) noexcept { bool silent) noexcept {
CAF_LOG_TRACE(CAF_ARG(slot) << CAF_ARG(reason) << CAF_ARG(silent));
auto i = paths_.find(slot); auto i = paths_.find(slot);
if (i != paths_.end()) { if (i != paths_.end()) {
about_to_erase(i->second.get(), silent, &reason); about_to_erase(i->second.get(), silent, &reason);
...@@ -70,6 +59,14 @@ void stream_scatterer_impl::clear_paths() { ...@@ -70,6 +59,14 @@ void stream_scatterer_impl::clear_paths() {
paths_.clear(); paths_.clear();
} }
bool stream_scatterer_impl::insert_path(unique_path_ptr ptr) {
CAF_LOG_TRACE(CAF_ARG(ptr));
CAF_ASSERT(ptr != nullptr);
auto slot = ptr->slots.sender;
CAF_ASSERT(slot != invalid_stream_slot);
return paths_.emplace(slot, std::move(ptr)).second;
}
void stream_scatterer_impl::for_each_path_impl(path_visitor& f) { void stream_scatterer_impl::for_each_path_impl(path_visitor& f) {
for (auto& kvp : paths_) for (auto& kvp : paths_)
f(*kvp.second); f(*kvp.second);
......
...@@ -86,9 +86,10 @@ public: ...@@ -86,9 +86,10 @@ public:
} }
void add_path_to(entity& x, int desired_batch_size) { void add_path_to(entity& x, int desired_batch_size) {
auto ptr = bs.add_path({next_slot++, 0}, x.ctrl()); auto ptr = bs.add_path(next_slot++, x.ctrl());
CAF_REQUIRE(ptr != nullptr); CAF_REQUIRE(ptr != nullptr);
ptr->desired_batch_size = desired_batch_size; ptr->desired_batch_size = desired_batch_size;
ptr->slots.receiver = x.next_slot++;
paths.emplace_back(ptr); paths.emplace_back(ptr);
} }
......
...@@ -78,21 +78,21 @@ TESTEE(sum_up) { ...@@ -78,21 +78,21 @@ TESTEE(sum_up) {
return { return {
[=](stream<int>& in, const string& fname) { [=](stream<int>& in, const string& fname) {
CAF_CHECK_EQUAL(fname, "numbers.txt"); CAF_CHECK_EQUAL(fname, "numbers.txt");
using int_ptr = int*;
return self->make_sink( return self->make_sink(
// input stream // input stream
in, in,
// initialize state // initialize state
[](unit_t&) { [=](int_ptr& x) {
// nop x = &self->state.x;
}, },
// processing step // processing step
[=](unit_t&, int y) { [](int_ptr& x , int y) {
self->state.x += y; *x += y;
}, },
// cleanup and produce result message // cleanup
[=](unit_t&) -> int { [=](int_ptr&, const error&) {
CAF_MESSAGE(self->name() << " is done"); CAF_MESSAGE(self->name() << " is done");
return self->state.x;
} }
); );
}, },
...@@ -104,13 +104,11 @@ TESTEE(sum_up) { ...@@ -104,13 +104,11 @@ TESTEE(sum_up) {
} }
TESTEE_STATE(stream_multiplexer) { TESTEE_STATE(stream_multiplexer) {
stream_stage_ptr<int, message, int, broadcast_scatterer<int>, string> stage; stream_stage_ptr<int, int, broadcast_scatterer<int>> stage;
}; };
TESTEE(stream_multiplexer) { TESTEE(stream_multiplexer) {
self->state.stage = self->make_continuous_stage( self->state.stage = self->make_continuous_stage(
// handshake data
std::make_tuple(std::string{"numbers.txt"}),
// initialize state // initialize state
[](unit_t&) { [](unit_t&) {
// nop // nop
...@@ -120,14 +118,15 @@ TESTEE(stream_multiplexer) { ...@@ -120,14 +118,15 @@ TESTEE(stream_multiplexer) {
out.push(x); out.push(x);
}, },
// cleanup // cleanup
[=](unit_t&) { [=](unit_t&, const error&) {
CAF_MESSAGE(self->name() << " is done"); CAF_MESSAGE(self->name() << " is done");
} }
); );
return { return {
[=](join_atom) { [=](join_atom) {
CAF_MESSAGE("received 'join' request"); CAF_MESSAGE("received 'join' request");
return self->state.stage->add_outbound_path(); return self->state.stage->add_outbound_path(
std::make_tuple("numbers.txt"));
}, },
[=](const stream<int>& in, std::string& fname) { [=](const stream<int>& in, std::string& fname) {
CAF_CHECK_EQUAL(fname, "numbers.txt"); CAF_CHECK_EQUAL(fname, "numbers.txt");
......
...@@ -105,21 +105,22 @@ TESTEE_STATE(sum_up) { ...@@ -105,21 +105,22 @@ TESTEE_STATE(sum_up) {
}; };
TESTEE(sum_up) { TESTEE(sum_up) {
using intptr = int*;
return { return {
[=](stream<int>& in) { [=](stream<int>& in) {
return self->make_sink( return self->make_sink(
// input stream // input stream
in, in,
// initialize state // initialize state
[](unit_t&) { [=](intptr& x) {
// nop x = &self->state.x;
}, },
// processing step // processing step
[=](unit_t&, int y) { [](intptr& x, int y) {
self->state.x += y; *x += y;
}, },
// cleanup and produce result message // cleanup and produce result message
[=](unit_t&) { [=](intptr&, const error&) {
CAF_MESSAGE(self->name() << " is done"); CAF_MESSAGE(self->name() << " is done");
} }
); );
...@@ -150,7 +151,7 @@ TESTEE(collect) { ...@@ -150,7 +151,7 @@ TESTEE(collect) {
self->state.strings.emplace_back(std::move(y)); self->state.strings.emplace_back(std::move(y));
}, },
// cleanup and produce result message // cleanup and produce result message
[=](unit_t&) { [=](unit_t&, const error&) {
CAF_MESSAGE(self->name() << " is done"); CAF_MESSAGE(self->name() << " is done");
} }
); );
...@@ -170,6 +171,8 @@ using scatterer = fused_scatterer<int_scatterer, string_scatterer>; ...@@ -170,6 +171,8 @@ using scatterer = fused_scatterer<int_scatterer, string_scatterer>;
class fused_stage : public stream_manager { class fused_stage : public stream_manager {
public: public:
using super = stream_manager;
fused_stage(scheduled_actor* self) : stream_manager(self), out_(self) { fused_stage(scheduled_actor* self) : stream_manager(self), out_(self) {
continuous(true); continuous(true);
} }
...@@ -184,12 +187,14 @@ public: ...@@ -184,12 +187,14 @@ public:
using int_vec = std::vector<int>; using int_vec = std::vector<int>;
using string_vec = std::vector<string>; using string_vec = std::vector<string>;
if (batch.xs.match_elements<int_vec>()) { if (batch.xs.match_elements<int_vec>()) {
CAF_MESSAGE("handle an integer batch");
auto& xs = batch.xs.get_mutable_as<int_vec>(0); auto& xs = batch.xs.get_mutable_as<int_vec>(0);
auto& buf = out_.get<int_scatterer>().buf(); auto& buf = out_.get<int_scatterer>().buf();
buf.insert(buf.end(), xs.begin(), xs.end()); buf.insert(buf.end(), xs.begin(), xs.end());
return; return;
} }
if (batch.xs.match_elements<string_vec>()) { if (batch.xs.match_elements<string_vec>()) {
CAF_MESSAGE("handle a string batch");
auto& xs = batch.xs.get_mutable_as<string_vec>(0); auto& xs = batch.xs.get_mutable_as<string_vec>(0);
auto& buf = out_.get<string_scatterer>().buf(); auto& buf = out_.get<string_scatterer>().buf();
buf.insert(buf.end(), xs.begin(), xs.end()); buf.insert(buf.end(), xs.begin(), xs.end());
...@@ -198,10 +203,6 @@ public: ...@@ -198,10 +203,6 @@ public:
CAF_LOG_ERROR("received unexpected batch type (dropped)"); CAF_LOG_ERROR("received unexpected batch type (dropped)");
} }
message make_handshake(stream_slot slot) const override {
return out_.make_handshake_token(slot);
}
bool congested() const noexcept override { bool congested() const noexcept override {
return out_.capacity() == 0; return out_.capacity() == 0;
} }
...@@ -212,7 +213,6 @@ public: ...@@ -212,7 +213,6 @@ public:
private: private:
scatterer out_; scatterer out_;
std::map<stream_slot, stream_scatterer*> scatterers_;
}; };
TESTEE_STATE(stream_multiplexer) { TESTEE_STATE(stream_multiplexer) {
...@@ -231,18 +231,18 @@ TESTEE(stream_multiplexer) { ...@@ -231,18 +231,18 @@ TESTEE(stream_multiplexer) {
}, },
[=](join_atom, strings_atom) { [=](join_atom, strings_atom) {
auto& stg = self->state.stage; auto& stg = self->state.stage;
CAF_MESSAGE("received 'join' request for integers"); CAF_MESSAGE("received 'join' request for strings");
auto result = stg->add_unsafe_outbound_path<string>(); auto result = stg->add_unsafe_outbound_path<string>();
stg->out().assign<string_scatterer>(result.out()); stg->out().assign<string_scatterer>(result.out());
return result; return result;
}, },
[=](const stream<int>& in) { [=](const stream<int>& in) {
CAF_MESSAGE("received handshake for integers"); CAF_MESSAGE("received handshake for integers");
return self->state.stage->add_unsafe_inbound_path<void>(in); return self->state.stage->add_unsafe_inbound_path(in);
}, },
[=](const stream<string>& in) { [=](const stream<string>& in) {
CAF_MESSAGE("received handshake for strings"); CAF_MESSAGE("received handshake for strings");
return self->state.stage->add_unsafe_inbound_path<void>(in); return self->state.stage->add_unsafe_inbound_path(in);
} }
}; };
} }
......
...@@ -181,7 +181,6 @@ public: ...@@ -181,7 +181,6 @@ public:
: super(cfg), : super(cfg),
mbox(unit, unit, unit, unit, unit), mbox(unit, unit, unit, unit, unit),
name_(cstr_name), name_(cstr_name),
next_slot_(static_cast<stream_slot>(id())),
global_time_(global_time), global_time_(global_time),
tick_emitter_(global_time == nullptr ? clock_type::now() tick_emitter_(global_time == nullptr ? clock_type::now()
: *global_time) { : *global_time) {
...@@ -236,9 +235,6 @@ public: ...@@ -236,9 +235,6 @@ public:
void start_streaming(entity& ref, int num_messages) { void start_streaming(entity& ref, int num_messages) {
CAF_REQUIRE_NOT_EQUAL(num_messages, 0); CAF_REQUIRE_NOT_EQUAL(num_messages, 0);
auto slot = next_slot_++;
CAF_MESSAGE(name_ << " starts streaming to " << ref.name()
<< " on slot " << slot);
using scatterer = broadcast_scatterer<int>; using scatterer = broadcast_scatterer<int>;
struct driver final : public stream_source_driver<scatterer> { struct driver final : public stream_source_driver<scatterer> {
public: public:
...@@ -246,10 +242,6 @@ public: ...@@ -246,10 +242,6 @@ public:
// nop // nop
} }
handshake_tuple_type make_handshake(stream_slot slot) const override {
return std::make_tuple(stream_type{slot});
}
void pull(downstream<int>& out, size_t hint) override { void pull(downstream<int>& out, size_t hint) override {
auto y = std::min(sentinel_, x_ + static_cast<int>(hint)); auto y = std::min(sentinel_, x_ + static_cast<int>(hint));
while (x_ < y) while (x_ < y)
...@@ -263,40 +255,25 @@ public: ...@@ -263,40 +255,25 @@ public:
int x_; int x_;
int sentinel_; int sentinel_;
}; };
auto ptr = detail::make_stream_source<driver>(this, num_messages); auto mgr = detail::make_stream_source<driver>(this, num_messages);
ptr->send_handshake(ref.ctrl(), slot); auto res = mgr->add_outbound_path(ref.ctrl());
ptr->generate_messages(); CAF_MESSAGE(name_ << " starts streaming to " << ref.name()
pending_managers_.emplace(slot, std::move(ptr)); << " on slot " << res.out());
} }
void forward_to(entity& ref) { void forward_to(entity& ref) {
auto slot = next_slot_++;
CAF_MESSAGE(name_ << " starts forwarding to " << ref.name()
<< " on slot " << slot);
using scatterer = broadcast_scatterer<int>; using scatterer = broadcast_scatterer<int>;
struct driver final : public stream_stage_driver<int, none_t, scatterer> { struct driver final : public stream_stage_driver<int, scatterer> {
public: public:
driver(vector<int>* log) : log_(log) { driver(vector<int>* log) : log_(log) {
// nop // nop
} }
handshake_tuple_type make_handshake(stream_slot slot) const override { void process(downstream<int>& out, vector<int>& batch) override {
return std::make_tuple(stream_type{slot});
}
void process(vector<int>&& batch, downstream<int>& out) override {
log_->insert(log_->end(), batch.begin(), batch.end()); log_->insert(log_->end(), batch.begin(), batch.end());
out.append(batch.begin(), batch.end()); out.append(batch.begin(), batch.end());
} }
void add_result(message&) override {
// nop
}
message make_final_result() override {
return none;
}
void finalize(const error&) override { void finalize(const error&) override {
// nop // nop
} }
...@@ -305,27 +282,25 @@ public: ...@@ -305,27 +282,25 @@ public:
vector<int>* log_; vector<int>* log_;
}; };
forwarder = detail::make_stream_stage<driver>(this, &data); forwarder = detail::make_stream_stage<driver>(this, &data);
forwarder->send_handshake(ref.ctrl(), slot); auto res = forwarder->add_outbound_path(ref.ctrl());
pending_managers_.emplace(slot, forwarder); CAF_MESSAGE(name_ << " starts forwarding to " << ref.name()
<< " on slot " << res.out());
} }
void operator()(open_stream_msg& hs) { void operator()(open_stream_msg& hs) {
TRACE(name_, stream_handshake_msg, TRACE(name_, stream_handshake_msg,
CAF_ARG2("sender", name_of(hs.prev_stage))); CAF_ARG2("sender", name_of(hs.prev_stage)));
auto slot = next_slot_++;
//stream_slots id{slot, hs.sender_slot};
stream_slots id{hs.slot, slot};
// Create required state if no forwarder exists yet, otherwise `forward_to` // Create required state if no forwarder exists yet, otherwise `forward_to`
// was called and we run as a stage. // was called and we run as a stage.
auto mgr = forwarder; stream_sink_ptr<int> mgr = forwarder;
if (mgr == nullptr) { if (mgr == nullptr) {
struct driver final : public stream_sink_driver<int, void> { struct driver final : public stream_sink_driver<int> {
public: public:
driver(std::vector<int>* log) : log_(log) { driver(std::vector<int>* log) : log_(log) {
// nop // nop
} }
void process(std::vector<int>&& xs) override { void process(std::vector<int>& xs) override {
log_->insert(log_->end(), xs.begin(), xs.end()); log_->insert(log_->end(), xs.begin(), xs.end());
} }
private: private:
...@@ -333,11 +308,9 @@ public: ...@@ -333,11 +308,9 @@ public:
}; };
mgr = detail::make_stream_sink<driver>(this, &data); mgr = detail::make_stream_sink<driver>(this, &data);
} }
managers_.emplace(slot, mgr); CAF_REQUIRE(hs.msg.match_elements<stream<int>>());
// Create a new queue in the mailbox for incoming traffic. auto& in = hs.msg.get_as<stream<int>>(0);
auto path = make_inbound_path(mgr, id, std::move(hs.prev_stage)); mgr->add_inbound_path(in);
CAF_REQUIRE_NOT_EQUAL(path, nullptr);
path->emit_ack_open(this, actor_cast<actor_addr>(hs.original_stage));
} }
void operator()(stream_slots slots, actor_addr& sender, void operator()(stream_slots slots, actor_addr& sender,
...@@ -345,33 +318,14 @@ public: ...@@ -345,33 +318,14 @@ public:
TRACE(name_, ack_open, CAF_ARG(slots), TRACE(name_, ack_open, CAF_ARG(slots),
CAF_ARG2("sender", name_of(x.rebind_to)), CAF_ARG(x)); CAF_ARG2("sender", name_of(x.rebind_to)), CAF_ARG(x));
CAF_REQUIRE_EQUAL(sender, x.rebind_to); CAF_REQUIRE_EQUAL(sender, x.rebind_to);
// Get the manager for that stream, move it from `pending_managers_` to scheduled_actor::handle_upstream_msg(slots, sender, x);
// `managers_`, and handle `x`.
auto i = pending_managers_.find(slots.receiver);
CAF_REQUIRE_NOT_EQUAL(i, pending_managers_.end());
auto res = managers_.emplace(slots.receiver, std::move(i->second));
pending_managers_.erase(i);
CAF_REQUIRE(res.second);
res.first->second->handle(slots, x);
res.first->second->out().force_emit_batches();
} }
void operator()(stream_slots slots, actor_addr& sender, void operator()(stream_slots slots, actor_addr& sender,
upstream_msg::ack_batch& x) { upstream_msg::ack_batch& x) {
TRACE(name_, ack_batch, CAF_ARG(slots), TRACE(name_, ack_batch, CAF_ARG(slots),
CAF_ARG2("sender", name_of(sender)), CAF_ARG(x)); CAF_ARG2("sender", name_of(sender)), CAF_ARG(x));
// Get the manager for that stream. scheduled_actor::handle_upstream_msg(slots, sender, x);
auto i = managers_.find(slots.receiver);
CAF_REQUIRE_NOT_EQUAL(i, managers_.end());
i->second->handle(slots, x);
i->second->out().force_emit_batches();
if (i->second->done()) {
CAF_MESSAGE(name_ << " is done sending batches");
// Purge the manager from the map.
auto mgr = i->second;
erase_stream_manager(mgr);
mgr->stop();
}
} }
void advance_time() { void advance_time() {
...@@ -380,7 +334,7 @@ public: ...@@ -380,7 +334,7 @@ public:
auto f = [&](tick_type x) { auto f = [&](tick_type x) {
if (x % ticks_per_force_batches_interval == 0) { if (x % ticks_per_force_batches_interval == 0) {
// Force batches on all output paths. // Force batches on all output paths.
for (auto& kvp : managers_) for (auto& kvp : stream_managers())
kvp.second->out().force_emit_batches(); kvp.second->out().force_emit_batches();
} }
if (x % ticks_per_credit_interval == 0) { if (x % ticks_per_credit_interval == 0) {
...@@ -422,20 +376,6 @@ public: ...@@ -422,20 +376,6 @@ public:
CAF_FAIL("unexpected function call"); CAF_FAIL("unexpected function call");
} }
void erase_stream_manager(stream_slot id) {
managers_.erase(id);
}
void erase_stream_manager(const stream_manager_ptr& mgr) {
auto i = managers_.begin();
auto e = managers_.end();
while (i != e)
if (i->second == mgr)
i = managers_.erase(i);
else
++i;
}
time_point now() { time_point now() {
return global_time_ == nullptr ? clock_type::now() : *global_time_; return global_time_ == nullptr ? clock_type::now() : *global_time_;
} }
...@@ -444,11 +384,8 @@ public: ...@@ -444,11 +384,8 @@ public:
mboxqueue mbox; mboxqueue mbox;
const char* name_; const char* name_;
stream_slot next_slot_ = 1;
vector<int> data; // Keeps track of all received data from all batches. vector<int> data; // Keeps track of all received data from all batches.
stream_manager_ptr forwarder; stream_stage_ptr<int, int, broadcast_scatterer<int>> forwarder;
std::map<stream_slot, stream_manager_ptr> managers_;
std::map<stream_slot, stream_manager_ptr> pending_managers_;
tick_type ticks_per_force_batches_interval; tick_type ticks_per_force_batches_interval;
tick_type ticks_per_credit_interval; tick_type ticks_per_credit_interval;
...@@ -474,7 +411,9 @@ struct msg_visitor { ...@@ -474,7 +411,9 @@ struct msg_visitor {
result_type operator()(is_default_async, default_queue&, mailbox_element& x) { result_type operator()(is_default_async, default_queue&, mailbox_element& x) {
CAF_REQUIRE_EQUAL(x.content().type_token(), CAF_REQUIRE_EQUAL(x.content().type_token(),
make_type_token<open_stream_msg>()); make_type_token<open_stream_msg>());
self->current_mailbox_element(&x);
(*self)(x.content().get_mutable_as<open_stream_msg>(0)); (*self)(x.content().get_mutable_as<open_stream_msg>(0));
self->current_mailbox_element(nullptr);
return intrusive::task_result::resume; return intrusive::task_result::resume;
} }
...@@ -485,6 +424,7 @@ struct msg_visitor { ...@@ -485,6 +424,7 @@ struct msg_visitor {
result_type operator()(is_umsg, umsg_queue&, mailbox_element& x) { result_type operator()(is_umsg, umsg_queue&, mailbox_element& x) {
CAF_REQUIRE(x.content().type_token() == make_type_token<upstream_msg>()); CAF_REQUIRE(x.content().type_token() == make_type_token<upstream_msg>());
self->current_mailbox_element(&x);
auto& um = x.content().get_mutable_as<upstream_msg>(0); auto& um = x.content().get_mutable_as<upstream_msg>(0);
auto f = detail::make_overload( auto f = detail::make_overload(
[&](upstream_msg::ack_open& y) { [&](upstream_msg::ack_open& y) {
...@@ -501,6 +441,7 @@ struct msg_visitor { ...@@ -501,6 +441,7 @@ struct msg_visitor {
} }
); );
visit(f, um.content); visit(f, um.content);
self->current_mailbox_element(nullptr);
return intrusive::task_result::resume; return intrusive::task_result::resume;
} }
...@@ -508,6 +449,7 @@ struct msg_visitor { ...@@ -508,6 +449,7 @@ struct msg_visitor {
policy::downstream_messages::nested_queue_type& q, policy::downstream_messages::nested_queue_type& q,
mailbox_element& x) { mailbox_element& x) {
CAF_REQUIRE(x.content().type_token() == make_type_token<downstream_msg>()); CAF_REQUIRE(x.content().type_token() == make_type_token<downstream_msg>());
self->current_mailbox_element(&x);
auto inptr = q.policy().handler.get(); auto inptr = q.policy().handler.get();
if (inptr == nullptr) if (inptr == nullptr)
return intrusive::task_result::stop; return intrusive::task_result::stop;
...@@ -525,13 +467,13 @@ struct msg_visitor { ...@@ -525,13 +467,13 @@ struct msg_visitor {
[&](downstream_msg::close& y) { [&](downstream_msg::close& y) {
TRACE(self->name(), close, CAF_ARG(dm.slots)); TRACE(self->name(), close, CAF_ARG(dm.slots));
auto slots = dm.slots; auto slots = dm.slots;
auto i = self->managers_.find(slots.receiver); auto i = self->stream_managers().find(slots.receiver);
CAF_REQUIRE_NOT_EQUAL(i, self->managers_.end()); CAF_REQUIRE_NOT_EQUAL(i, self->stream_managers().end());
i->second->handle(inptr, y); i->second->handle(inptr, y);
q.policy().handler.reset(); q.policy().handler.reset();
qs.erase_later(slots.receiver); qs.erase_later(slots.receiver);
if (!i->second->done()) { if (!i->second->done()) {
self->managers_.erase(i); self->stream_managers().erase(i);
} else { } else {
// Close the manager and remove it on all registered slots. // Close the manager and remove it on all registered slots.
auto mgr = i->second; auto mgr = i->second;
...@@ -544,7 +486,9 @@ struct msg_visitor { ...@@ -544,7 +486,9 @@ struct msg_visitor {
CAF_FAIL("did not expect downstream_msg::forced_close"); CAF_FAIL("did not expect downstream_msg::forced_close");
return intrusive::task_result::stop; return intrusive::task_result::stop;
}); });
return visit(f, dm.content); auto result = visit(f, dm.content);
self->current_mailbox_element(nullptr);
return result;
} }
// -- member variables ------------------------------------------------------- // -- member variables -------------------------------------------------------
...@@ -610,8 +554,8 @@ struct fixture { ...@@ -610,8 +554,8 @@ struct fixture {
entity* xs[] = {&alice, &bob, &carl}; entity* xs[] = {&alice, &bob, &carl};
for (auto x : xs) { for (auto x : xs) {
CAF_CHECK(get<2>(x->mbox.queues()).queues().empty()); CAF_CHECK(get<2>(x->mbox.queues()).queues().empty());
CAF_CHECK(x->pending_managers_.empty()); CAF_CHECK(x->pending_stream_managers().empty());
CAF_CHECK(x->managers_.empty()); CAF_CHECK(x->stream_managers().empty());
} }
} }
...@@ -627,7 +571,7 @@ struct fixture { ...@@ -627,7 +571,7 @@ struct fixture {
template <class... Ts> template <class... Ts>
void next_cycle(Ts&... xs) { void next_cycle(Ts&... xs) {
entity* es[] = {&xs...}; entity* es[] = {&xs...};
//CAF_MESSAGE("advance clock by " << tc.credit_interval.count() << "ns"); CAF_MESSAGE("advance clock by " << tc.credit_interval.count() << "ns");
sched.clock().current_time += tc.credit_interval; sched.clock().current_time += tc.credit_interval;
for (auto e : es) for (auto e : es)
e->advance_time(); e->advance_time();
...@@ -653,7 +597,7 @@ struct fixture { ...@@ -653,7 +597,7 @@ struct fixture {
bool done_streaming() { bool done_streaming() {
entity* es[] = {&alice, &bob, &carl}; entity* es[] = {&alice, &bob, &carl};
return std::all_of(std::begin(es), std::end(es), return std::all_of(std::begin(es), std::end(es),
[](entity* e) { return e->managers_.empty(); }); [](entity* e) { return e->stream_managers().empty(); });
} }
}; };
...@@ -672,7 +616,9 @@ CAF_TEST_FIXTURE_SCOPE(native_streaming_classes_tests, fixture) ...@@ -672,7 +616,9 @@ CAF_TEST_FIXTURE_SCOPE(native_streaming_classes_tests, fixture)
CAF_TEST(depth_2_pipeline_30_items) { CAF_TEST(depth_2_pipeline_30_items) {
alice.start_streaming(bob, 30); alice.start_streaming(bob, 30);
loop(alice, bob); loop(alice, bob);
next_cycle(alice, bob); // a single credit round is enough next_cycle(alice, bob); // emit first ack_batch
loop(alice, bob);
next_cycle(alice, bob); // to emit final ack_batch
loop(alice, bob); loop(alice, bob);
CAF_CHECK_EQUAL(bob.data, make_iota(0, 30)); CAF_CHECK_EQUAL(bob.data, make_iota(0, 30));
} }
...@@ -688,10 +634,12 @@ CAF_TEST(depth_3_pipeline_30_items) { ...@@ -688,10 +634,12 @@ CAF_TEST(depth_3_pipeline_30_items) {
bob.forward_to(carl); bob.forward_to(carl);
alice.start_streaming(bob, 30); alice.start_streaming(bob, 30);
loop(alice, bob, carl); loop(alice, bob, carl);
next_cycle(alice, bob, carl); next_cycle(alice, bob, carl); // emit first ack_batch
loop(alice, bob, carl); loop(alice, bob, carl);
next_cycle(alice, bob, carl); next_cycle(alice, bob, carl);
loop(alice, bob, carl); loop(alice, bob, carl);
next_cycle(alice, bob, carl); // emit final ack_batch
loop(alice, bob, carl);
CAF_CHECK_EQUAL(bob.data, make_iota(0, 30)); CAF_CHECK_EQUAL(bob.data, make_iota(0, 30));
CAF_CHECK_EQUAL(carl.data, make_iota(0, 30)); CAF_CHECK_EQUAL(carl.data, make_iota(0, 30));
} }
......
...@@ -56,96 +56,100 @@ void push_from_buf(buf& xs, downstream<int>& out, size_t num) { ...@@ -56,96 +56,100 @@ void push_from_buf(buf& xs, downstream<int>& out, size_t num) {
std::function<bool(const buf&)> is_done(scheduled_actor* self) { std::function<bool(const buf&)> is_done(scheduled_actor* self) {
return [=](const buf& xs) { return [=](const buf& xs) {
if (xs.empty()) { if (xs.empty()) {
CAF_MESSAGE(self->name() << " is done sending"); CAF_MESSAGE(self->name() << " exhausted its buffer");
return true; return true;
} }
return false; return false;
}; };
} }
template <class T>
std::function<void (T&, const error&)> fin(scheduled_actor* self) {
return [=](T&, const error& err) {
if (err == none) {
CAF_MESSAGE(self->name() << " is done");
} else {
CAF_MESSAGE(self->name() << " aborted with error");
}
};
}
VARARGS_TESTEE(file_reader, size_t buf_size) { VARARGS_TESTEE(file_reader, size_t buf_size) {
return { return {
[=](string& fname) -> output_stream_t<int, string> { [=](string& fname) -> output_stream_t<int> {
CAF_CHECK_EQUAL(fname, "numbers.txt"); CAF_CHECK_EQUAL(fname, "numbers.txt");
CAF_CHECK_EQUAL(self->mailbox().empty(), true); CAF_CHECK_EQUAL(self->mailbox().empty(), true);
return self->make_source( return self->make_source(
// forward file name in handshake to next stage
std::forward_as_tuple(std::move(fname)),
init(buf_size), init(buf_size),
push_from_buf, push_from_buf,
is_done(self) is_done(self),
fin<buf>(self)
); );
}, },
[=](string& fname, actor dest) { [=](string& fname, actor next) {
CAF_CHECK_EQUAL(fname, "numbers.txt"); CAF_CHECK_EQUAL(fname, "numbers.txt");
CAF_CHECK_EQUAL(self->mailbox().empty(), true); CAF_CHECK_EQUAL(self->mailbox().empty(), true);
auto rp = self->make_response_promise();
self->make_source( self->make_source(
// next stage in the stream next,
dest,
// forward file name in handshake to next stage
std::forward_as_tuple(std::move(fname)),
init(buf_size), init(buf_size),
push_from_buf, push_from_buf,
is_done(self), is_done(self),
[=](expected<int> x) mutable { fin<buf>(self)
CAF_MESSAGE(self->name() << " received the result");
rp.deliver(std::move(x));
}
); );
return rp;
} }
}; };
} }
TESTEE_STATE(sum_up) {
int x = 0;
};
TESTEE(sum_up) { TESTEE(sum_up) {
using intptr = int*;
return { return {
[=](stream<int>& in, const string& fname) { [=](stream<int>& in) {
CAF_CHECK_EQUAL(fname, "numbers.txt");
return self->make_sink( return self->make_sink(
// input stream // input stream
in, in,
// initialize state // initialize state
[](int& x) { [=](intptr& x) {
x = 0; x = &self->state.x;
}, },
// processing step // processing step
[](int& x, int y) { [](intptr& x, int y) {
x += y; *x += y;
}, },
// cleanup and produce result message fin<intptr>(self)
[=](int& x) -> int {
CAF_MESSAGE(self->name() << " is done");
return x;
}
); );
} }
}; };
} }
TESTEE_STATE(delayed_sum_up) {
int x = 0;
};
TESTEE(delayed_sum_up) { TESTEE(delayed_sum_up) {
using intptr = int*;
self->set_default_handler(skip); self->set_default_handler(skip);
return { return {
[=](ok_atom) { [=](ok_atom) {
self->become( self->become(
[=](stream<int>& in, const std::string& fname) { [=](stream<int>& in) {
CAF_CHECK_EQUAL(fname, "numbers.txt"); self->set_default_handler(print_and_drop);
return self->make_sink( return self->make_sink(
// input stream // input stream
in, in,
// initialize state // initialize state
[](int& x) { [=](intptr& x) {
x = 0; x = &self->state.x;
}, },
// processing step // processing step
[](int& x, int y) { [](intptr& x, int y) {
x += y; *x += y;
}, },
// cleanup and produce result message // cleanup
[=](int& x) -> int { fin<intptr>(self)
CAF_MESSAGE(self->name() << " is done");
return x;
}
); );
} }
); );
...@@ -156,8 +160,8 @@ TESTEE(delayed_sum_up) { ...@@ -156,8 +160,8 @@ TESTEE(delayed_sum_up) {
TESTEE(broken_sink) { TESTEE(broken_sink) {
CAF_IGNORE_UNUSED(self); CAF_IGNORE_UNUSED(self);
return { return {
[=](stream<int>&, const std::string& fname) { [=](stream<int>&, const actor&) {
CAF_CHECK_EQUAL(fname, "numbers.txt"); // nop
} }
}; };
} }
...@@ -165,13 +169,10 @@ TESTEE(broken_sink) { ...@@ -165,13 +169,10 @@ TESTEE(broken_sink) {
TESTEE(filter) { TESTEE(filter) {
CAF_IGNORE_UNUSED(self); CAF_IGNORE_UNUSED(self);
return { return {
[=](stream<int>& in, std::string& fname) { [=](stream<int>& in) {
CAF_CHECK_EQUAL(fname, "numbers.txt");
return self->make_stage( return self->make_stage(
// input stream // input stream
in, in,
// forward file name in handshake to next stage
std::forward_as_tuple(std::move(fname)),
// initialize state // initialize state
[](unit_t&) { [](unit_t&) {
// nop // nop
...@@ -182,9 +183,7 @@ TESTEE(filter) { ...@@ -182,9 +183,7 @@ TESTEE(filter) {
out.push(x); out.push(x);
}, },
// cleanup // cleanup
[=](unit_t&) { fin<unit_t>(self)
CAF_MESSAGE(self->name() << " is done");
}
); );
} }
}; };
...@@ -193,13 +192,10 @@ TESTEE(filter) { ...@@ -193,13 +192,10 @@ TESTEE(filter) {
TESTEE(doubler) { TESTEE(doubler) {
CAF_IGNORE_UNUSED(self); CAF_IGNORE_UNUSED(self);
return { return {
[=](stream<int>& in, std::string& fname) { [=](stream<int>& in) {
CAF_CHECK_EQUAL(fname, "numbers.txt");
return self->make_stage( return self->make_stage(
// input stream // input stream
in, in,
// forward file name in handshake to next stage
std::forward_as_tuple(std::move(fname)),
// initialize state // initialize state
[](unit_t&) { [](unit_t&) {
// nop // nop
...@@ -209,9 +205,7 @@ TESTEE(doubler) { ...@@ -209,9 +205,7 @@ TESTEE(doubler) {
out.push(x * 2); out.push(x * 2);
}, },
// cleanup // cleanup
[=](unit_t&) { fin<unit_t>(self)
CAF_MESSAGE(self->name() << " is done");
}
); );
} }
}; };
...@@ -228,11 +222,6 @@ struct fixture : test_coordinator_fixture<> { ...@@ -228,11 +222,6 @@ struct fixture : test_coordinator_fixture<> {
} }
}; };
error fail_state(const actor& x) {
auto ptr = actor_cast<abstract_actor*>(x);
return dynamic_cast<monitorable_actor&>(*ptr).fail_state();
}
} // namespace <anonymous> } // namespace <anonymous>
// -- unit tests --------------------------------------------------------------- // -- unit tests ---------------------------------------------------------------
...@@ -246,7 +235,7 @@ CAF_TEST(depth_2_pipeline_50_items) { ...@@ -246,7 +235,7 @@ CAF_TEST(depth_2_pipeline_50_items) {
CAF_MESSAGE("initiate stream handshake"); CAF_MESSAGE("initiate stream handshake");
self->send(snk * src, "numbers.txt"); self->send(snk * src, "numbers.txt");
expect((string), from(self).to(src).with("numbers.txt")); expect((string), from(self).to(src).with("numbers.txt"));
expect((open_stream_msg), from(self).to(snk)); expect((open_stream_msg), from(src).to(snk));
expect((upstream_msg::ack_open), from(snk).to(src)); expect((upstream_msg::ack_open), from(snk).to(src));
CAF_MESSAGE("start data transmission (a single batch)"); CAF_MESSAGE("start data transmission (a single batch)");
expect((downstream_msg::batch), from(src).to(snk)); expect((downstream_msg::batch), from(src).to(snk));
...@@ -257,9 +246,7 @@ CAF_TEST(depth_2_pipeline_50_items) { ...@@ -257,9 +246,7 @@ CAF_TEST(depth_2_pipeline_50_items) {
expect((upstream_msg::ack_batch), from(snk).to(src)); expect((upstream_msg::ack_batch), from(snk).to(src));
CAF_MESSAGE("expect close message from src and then result from snk"); CAF_MESSAGE("expect close message from src and then result from snk");
expect((downstream_msg::close), from(src).to(snk)); expect((downstream_msg::close), from(src).to(snk));
expect((int), from(snk).to(self).with(1275)); CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.x, 1275);
CAF_CHECK_EQUAL(fail_state(snk), exit_reason::normal);
CAF_CHECK_EQUAL(fail_state(src), exit_reason::normal);
} }
CAF_TEST(depth_2_pipeline_setup2_50_items) { CAF_TEST(depth_2_pipeline_setup2_50_items) {
...@@ -280,10 +267,7 @@ CAF_TEST(depth_2_pipeline_setup2_50_items) { ...@@ -280,10 +267,7 @@ CAF_TEST(depth_2_pipeline_setup2_50_items) {
expect((upstream_msg::ack_batch), from(snk).to(src)); expect((upstream_msg::ack_batch), from(snk).to(src));
CAF_MESSAGE("expect close message from src and then result from snk"); CAF_MESSAGE("expect close message from src and then result from snk");
expect((downstream_msg::close), from(src).to(snk)); expect((downstream_msg::close), from(src).to(snk));
expect((int), from(snk).to(src).with(1275)); CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.x, 1275);
expect((int), from(src).to(self).with(1275));
CAF_CHECK_EQUAL(fail_state(snk), exit_reason::normal);
CAF_CHECK_EQUAL(fail_state(src), exit_reason::normal);
} }
CAF_TEST(delayed_depth_2_pipeline_50_items) { CAF_TEST(delayed_depth_2_pipeline_50_items) {
...@@ -293,13 +277,13 @@ CAF_TEST(delayed_depth_2_pipeline_50_items) { ...@@ -293,13 +277,13 @@ CAF_TEST(delayed_depth_2_pipeline_50_items) {
CAF_MESSAGE("initiate stream handshake"); CAF_MESSAGE("initiate stream handshake");
self->send(snk * src, "numbers.txt"); self->send(snk * src, "numbers.txt");
expect((string), from(self).to(src).with("numbers.txt")); expect((string), from(self).to(src).with("numbers.txt"));
expect((open_stream_msg), from(self).to(snk)); expect((open_stream_msg), from(src).to(snk));
disallow((upstream_msg::ack_open), from(snk).to(src)); disallow((upstream_msg::ack_open), from(snk).to(src));
disallow((upstream_msg::forced_drop), from(snk).to(src)); disallow((upstream_msg::forced_drop), from(snk).to(src));
CAF_MESSAGE("send 'ok' to trigger sink to handle open_stream_msg"); CAF_MESSAGE("send 'ok' to trigger sink to handle open_stream_msg");
self->send(snk, ok_atom::value); self->send(snk, ok_atom::value);
expect((ok_atom), from(self).to(snk)); expect((ok_atom), from(self).to(snk));
expect((open_stream_msg), from(self).to(snk)); expect((open_stream_msg), from(src).to(snk));
expect((upstream_msg::ack_open), from(snk).to(src)); expect((upstream_msg::ack_open), from(snk).to(src));
CAF_MESSAGE("start data transmission (a single batch)"); CAF_MESSAGE("start data transmission (a single batch)");
expect((downstream_msg::batch), from(src).to(snk)); expect((downstream_msg::batch), from(src).to(snk));
...@@ -310,9 +294,7 @@ CAF_TEST(delayed_depth_2_pipeline_50_items) { ...@@ -310,9 +294,7 @@ CAF_TEST(delayed_depth_2_pipeline_50_items) {
expect((upstream_msg::ack_batch), from(snk).to(src)); expect((upstream_msg::ack_batch), from(snk).to(src));
CAF_MESSAGE("expect close message from src and then result from snk"); CAF_MESSAGE("expect close message from src and then result from snk");
expect((downstream_msg::close), from(src).to(snk)); expect((downstream_msg::close), from(src).to(snk));
expect((int), from(snk).to(self).with(1275)); CAF_CHECK_EQUAL(deref<delayed_sum_up_actor>(snk).state.x, 1275);
CAF_CHECK_EQUAL(fail_state(snk), exit_reason::normal);
CAF_CHECK_EQUAL(fail_state(src), exit_reason::normal);
} }
CAF_TEST(depth_2_pipeline_500_items) { CAF_TEST(depth_2_pipeline_500_items) {
...@@ -322,7 +304,7 @@ CAF_TEST(depth_2_pipeline_500_items) { ...@@ -322,7 +304,7 @@ CAF_TEST(depth_2_pipeline_500_items) {
CAF_MESSAGE("initiate stream handshake"); CAF_MESSAGE("initiate stream handshake");
self->send(snk * src, "numbers.txt"); self->send(snk * src, "numbers.txt");
expect((string), from(self).to(src).with("numbers.txt")); expect((string), from(self).to(src).with("numbers.txt"));
expect((open_stream_msg), from(self).to(snk)); expect((open_stream_msg), from(src).to(snk));
expect((upstream_msg::ack_open), from(snk).to(src)); expect((upstream_msg::ack_open), from(snk).to(src));
CAF_MESSAGE("start data transmission (loop until src sends 'close')"); CAF_MESSAGE("start data transmission (loop until src sends 'close')");
do { do {
...@@ -340,9 +322,7 @@ CAF_TEST(depth_2_pipeline_500_items) { ...@@ -340,9 +322,7 @@ CAF_TEST(depth_2_pipeline_500_items) {
} while (!received<downstream_msg::close>(snk)); } while (!received<downstream_msg::close>(snk));
CAF_MESSAGE("expect close message from src and then result from snk"); CAF_MESSAGE("expect close message from src and then result from snk");
expect((downstream_msg::close), from(src).to(snk)); expect((downstream_msg::close), from(src).to(snk));
expect((int), from(snk).to(self).with(125250)); CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.x, 125250);
CAF_CHECK_EQUAL(fail_state(snk), exit_reason::normal);
CAF_CHECK_EQUAL(fail_state(src), exit_reason::normal);
} }
CAF_TEST(depth_2_pipeline_error_during_handshake) { CAF_TEST(depth_2_pipeline_error_during_handshake) {
...@@ -352,9 +332,9 @@ CAF_TEST(depth_2_pipeline_error_during_handshake) { ...@@ -352,9 +332,9 @@ CAF_TEST(depth_2_pipeline_error_during_handshake) {
CAF_MESSAGE("initiate stream handshake"); CAF_MESSAGE("initiate stream handshake");
self->send(snk * src, "numbers.txt"); self->send(snk * src, "numbers.txt");
expect((std::string), from(self).to(src).with("numbers.txt")); expect((std::string), from(self).to(src).with("numbers.txt"));
expect((open_stream_msg), from(self).to(snk)); expect((open_stream_msg), from(src).to(snk));
expect((upstream_msg::forced_drop), from(snk).to(src)); expect((upstream_msg::forced_drop), from(snk).to(src));
expect((error), from(snk).to(self).with(sec::stream_init_failed)); // TODO: expect((error), from(snk).to(self).with(sec::stream_init_failed));
} }
CAF_TEST(depth_2_pipeline_error_at_source) { CAF_TEST(depth_2_pipeline_error_at_source) {
...@@ -365,7 +345,7 @@ CAF_TEST(depth_2_pipeline_error_at_source) { ...@@ -365,7 +345,7 @@ CAF_TEST(depth_2_pipeline_error_at_source) {
CAF_MESSAGE("initiate stream handshake"); CAF_MESSAGE("initiate stream handshake");
self->send(snk * src, "numbers.txt"); self->send(snk * src, "numbers.txt");
expect((string), from(self).to(src).with("numbers.txt")); expect((string), from(self).to(src).with("numbers.txt"));
expect((open_stream_msg), from(self).to(snk)); expect((open_stream_msg), from(src).to(snk));
expect((upstream_msg::ack_open), from(snk).to(src)); expect((upstream_msg::ack_open), from(snk).to(src));
CAF_MESSAGE("start data transmission (and abort source)"); CAF_MESSAGE("start data transmission (and abort source)");
self->send_exit(src, exit_reason::kill); self->send_exit(src, exit_reason::kill);
...@@ -373,9 +353,7 @@ CAF_TEST(depth_2_pipeline_error_at_source) { ...@@ -373,9 +353,7 @@ CAF_TEST(depth_2_pipeline_error_at_source) {
expect((exit_msg), from(self).to(src)); expect((exit_msg), from(self).to(src));
CAF_MESSAGE("expect close message from src and then result from snk"); CAF_MESSAGE("expect close message from src and then result from snk");
expect((downstream_msg::forced_close), from(src).to(snk)); expect((downstream_msg::forced_close), from(src).to(snk));
expect((error), from(snk).to(self)); // TODO: expect((error), from(snk).to(self));
CAF_CHECK_EQUAL(fail_state(snk), exit_reason::normal);
CAF_CHECK_EQUAL(fail_state(src), exit_reason::kill);
} }
CAF_TEST(depth_2_pipelin_error_at_sink) { CAF_TEST(depth_2_pipelin_error_at_sink) {
...@@ -386,16 +364,14 @@ CAF_TEST(depth_2_pipelin_error_at_sink) { ...@@ -386,16 +364,14 @@ CAF_TEST(depth_2_pipelin_error_at_sink) {
CAF_MESSAGE("initiate stream handshake"); CAF_MESSAGE("initiate stream handshake");
self->send(snk * src, "numbers.txt"); self->send(snk * src, "numbers.txt");
expect((string), from(self).to(src).with("numbers.txt")); expect((string), from(self).to(src).with("numbers.txt"));
expect((open_stream_msg), from(self).to(snk)); expect((open_stream_msg), from(src).to(snk));
CAF_MESSAGE("start data transmission (and abort sink)"); CAF_MESSAGE("start data transmission (and abort sink)");
self->send_exit(snk, exit_reason::kill); self->send_exit(snk, exit_reason::kill);
expect((upstream_msg::ack_open), from(snk).to(src)); expect((upstream_msg::ack_open), from(snk).to(src));
expect((exit_msg), from(self).to(snk)); expect((exit_msg), from(self).to(snk));
CAF_MESSAGE("expect close and result messages from snk"); CAF_MESSAGE("expect close and result messages from snk");
expect((upstream_msg::forced_drop), from(snk).to(src)); expect((upstream_msg::forced_drop), from(snk).to(src));
expect((error), from(snk).to(self)); // TODO: expect((error), from(snk).to(self));
CAF_CHECK_EQUAL(fail_state(src), exit_reason::normal);
CAF_CHECK_EQUAL(fail_state(snk), exit_reason::kill);
} }
CAF_TEST(depth_3_pipeline_50_items) { CAF_TEST(depth_3_pipeline_50_items) {
...@@ -413,8 +389,8 @@ CAF_TEST(depth_3_pipeline_50_items) { ...@@ -413,8 +389,8 @@ CAF_TEST(depth_3_pipeline_50_items) {
CAF_MESSAGE("initiate stream handshake"); CAF_MESSAGE("initiate stream handshake");
self->send(snk * stg * src, "numbers.txt"); self->send(snk * stg * src, "numbers.txt");
expect((string), from(self).to(src).with("numbers.txt")); expect((string), from(self).to(src).with("numbers.txt"));
expect((open_stream_msg), from(self).to(stg)); expect((open_stream_msg), from(src).to(stg));
expect((open_stream_msg), from(self).to(snk)); expect((open_stream_msg), from(stg).to(snk));
expect((upstream_msg::ack_open), from(snk).to(stg)); expect((upstream_msg::ack_open), from(snk).to(stg));
expect((upstream_msg::ack_open), from(stg).to(src)); expect((upstream_msg::ack_open), from(stg).to(src));
CAF_MESSAGE("start data transmission (a single batch)"); CAF_MESSAGE("start data transmission (a single batch)");
...@@ -430,7 +406,7 @@ CAF_TEST(depth_3_pipeline_50_items) { ...@@ -430,7 +406,7 @@ CAF_TEST(depth_3_pipeline_50_items) {
CAF_MESSAGE("the stage shuts down and the sink produces its final result"); CAF_MESSAGE("the stage shuts down and the sink produces its final result");
expect((upstream_msg::ack_batch), from(snk).to(stg)); expect((upstream_msg::ack_batch), from(snk).to(stg));
expect((downstream_msg::close), from(stg).to(snk)); expect((downstream_msg::close), from(stg).to(snk));
expect((int), from(snk).to(self).with(625)); CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.x, 625);
} }
CAF_TEST(depth_4_pipeline_500_items) { CAF_TEST(depth_4_pipeline_500_items) {
...@@ -443,16 +419,16 @@ CAF_TEST(depth_4_pipeline_500_items) { ...@@ -443,16 +419,16 @@ CAF_TEST(depth_4_pipeline_500_items) {
CAF_MESSAGE("initiate stream handshake"); CAF_MESSAGE("initiate stream handshake");
self->send(snk * stg2 * stg1 * src, "numbers.txt"); self->send(snk * stg2 * stg1 * src, "numbers.txt");
expect((string), from(self).to(src).with("numbers.txt")); expect((string), from(self).to(src).with("numbers.txt"));
expect((open_stream_msg), from(self).to(stg1)); expect((open_stream_msg), from(src).to(stg1));
expect((open_stream_msg), from(self).to(stg2)); expect((open_stream_msg), from(stg1).to(stg2));
expect((open_stream_msg), from(self).to(snk)); expect((open_stream_msg), from(stg2).to(snk));
expect((upstream_msg::ack_open), from(snk).to(stg2)); expect((upstream_msg::ack_open), from(snk).to(stg2));
expect((upstream_msg::ack_open), from(stg2).to(stg1)); expect((upstream_msg::ack_open), from(stg2).to(stg1));
expect((upstream_msg::ack_open), from(stg1).to(src)); expect((upstream_msg::ack_open), from(stg1).to(src));
CAF_MESSAGE("start data transmission"); CAF_MESSAGE("start data transmission");
sched.run_dispatch_loop(cycle); sched.run_dispatch_loop(cycle);
CAF_MESSAGE("check sink result"); CAF_MESSAGE("check sink result");
expect((int), from(snk).to(self).with(125000)); CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.x, 125000);
} }
CAF_TEST_FIXTURE_SCOPE_END() CAF_TEST_FIXTURE_SCOPE_END()
...@@ -100,16 +100,20 @@ TESTEE(log_producer) { ...@@ -100,16 +100,20 @@ TESTEE(log_producer) {
} }
return false; return false;
}, },
unit,
policy::arg<scatterer>::value policy::arg<scatterer>::value
); );
res.ptr()->out().set_filter(res.out(), lvl); auto& out = res.ptr()->out();
static_assert(std::is_same<decltype(out), scatterer&>::value,
"source has wrong scatterer type");
out.set_filter(res.out(), lvl);
return res; return res;
} }
}; };
} }
TESTEE_STATE(log_dispatcher) { TESTEE_STATE(log_dispatcher) {
stream_stage_ptr<value_type, message, value_type, scatterer> stage; stream_stage_ptr<value_type, value_type, scatterer> stage;
}; };
TESTEE(log_dispatcher) { TESTEE(log_dispatcher) {
...@@ -123,7 +127,7 @@ TESTEE(log_dispatcher) { ...@@ -123,7 +127,7 @@ TESTEE(log_dispatcher) {
out.push(std::move(x)); out.push(std::move(x));
}, },
// cleanup // cleanup
[=](unit_t&) { [=](unit_t&, const error&) {
CAF_MESSAGE(self->name() << " is done"); CAF_MESSAGE(self->name() << " is done");
}, },
policy::arg<scatterer>::value policy::arg<scatterer>::value
...@@ -161,7 +165,7 @@ TESTEE(log_consumer) { ...@@ -161,7 +165,7 @@ TESTEE(log_consumer) {
self->state.log.emplace_back(std::move(x)); self->state.log.emplace_back(std::move(x));
}, },
// cleanup and produce result message // cleanup and produce result message
[=](unit_t&) { [=](unit_t&, const error&) {
CAF_MESSAGE(self->name() << " is done"); CAF_MESSAGE(self->name() << " is done");
} }
); );
......
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