Commit 16351b4a authored by Dominik Charousset's avatar Dominik Charousset Committed by Dominik Charousset

Switch to a new internal source/stage/sink design

parent ee3bd69c
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2017 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_STREAM_SINK_DRIVER_IMPL_HPP
#define CAF_STREAM_SINK_DRIVER_IMPL_HPP
#include "caf/none.hpp"
#include "caf/stream_sink_driver.hpp"
#include "caf/stream_sink_trait.hpp"
namespace caf {
namespace detail {
/// Identifies an unbound sequence of messages.
template <class Input, class Process, class Finalize>
class stream_sink_driver_impl final : public stream_sink_driver<Input> {
public:
// -- member types -----------------------------------------------------------
using super = stream_sink_driver<Input>;
using input_type = typename super::input_type;
using trait = stream_sink_trait_t<Process, Finalize>;
using output_type = typename trait::output;
using state_type = typename trait::state;
template <class Init>
stream_sink_driver_impl(Init init, Process f, Finalize fin)
: process_(std::move(f)),
finalize_(std::move(fin)) {
init(state_);
}
void process(std::vector<input_type>&& xs) override {
return trait::process::invoke(process_, state_, std::move(xs));
}
message finalize() override {
return trait::finalize::invoke(finalize_, state_);
}
private:
Process process_;
Finalize finalize_;
state_type state_;
};
} // namespace detail
} // namespace caf
#endif // CAF_STREAM_SINK_DRIVER_IMPL_HPP
...@@ -33,31 +33,23 @@ ...@@ -33,31 +33,23 @@
namespace caf { namespace caf {
namespace detail { namespace detail {
template <class Fun, class Finalize> template <class Driver>
class stream_sink_impl : public stream_manager { class stream_sink_impl : public stream_manager {
public: public:
using super = stream_manager; using super = stream_manager;
using trait = stream_sink_trait_t<Fun, Finalize>; using driver_type = Driver;
using state_type = typename trait::state; using input_type = typename driver_type::input_type;
using input_type = typename trait::input; template <class... Ts>
stream_sink_impl(local_actor* self, Ts&&... xs)
using output_type = typename trait::output;
stream_sink_impl(local_actor* self, Fun fun, Finalize fin)
: stream_manager(self), : stream_manager(self),
fun_(std::move(fun)), driver_(std::forward<Ts>(xs)...),
fin_(std::move(fin)),
out_(self) { out_(self) {
// nop // nop
} }
state_type& state() {
return state_;
}
terminal_stream_scatterer& out() override { terminal_stream_scatterer& out() override {
return out_; return out_;
} }
...@@ -70,9 +62,8 @@ public: ...@@ -70,9 +62,8 @@ 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_as<vec_type>(0); auto& xs = x.xs.get_mutable_as<vec_type>(0);
for (auto& x : xs) driver_.process(std::move(xs));
fun_(state_, std::move(x));
return none; return none;
} }
CAF_LOG_ERROR("received unexpected batch type"); CAF_LOG_ERROR("received unexpected batch type");
...@@ -81,23 +72,18 @@ public: ...@@ -81,23 +72,18 @@ public:
protected: protected:
message make_final_result() override { message make_final_result() override {
return trait::make_result(state_, fin_); return driver_.finalize();
} }
private: private:
state_type state_; driver_type driver_;
Fun fun_;
Finalize fin_;
terminal_stream_scatterer out_; terminal_stream_scatterer out_;
}; };
template <class Init, class Fun, class Finalize> template <class Driver, class... Ts>
stream_manager_ptr make_stream_sink(local_actor* self, Init init, Fun f, stream_manager_ptr make_stream_sink(local_actor* self, Ts&&... xs) {
Finalize fin) { using impl = stream_sink_impl<Driver>;
using impl = stream_sink_impl<Fun, Finalize>; return make_counted<impl>(self, std::forward<Ts>(xs)...);
auto ptr = make_counted<impl>(self, std::move(f), std::move(fin));
init(ptr->state());
return ptr;
} }
} // namespace detail } // namespace detail
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2017 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_STREAM_SOURCE_DRIVER_IMPL_HPP
#define CAF_STREAM_SOURCE_DRIVER_IMPL_HPP
#include "caf/none.hpp"
#include "caf/stream_source_driver.hpp"
#include "caf/stream_source_trait.hpp"
namespace caf {
namespace detail {
template <class Output, class Pull, class Done, class HandshakeData>
class stream_source_driver_impl;
/// Identifies an unbound sequence of messages.
template <class Output, class Pull, class Done, class... Ts>
class stream_source_driver_impl<Output, Pull, Done, std::tuple<Ts...>> final
: public stream_source_driver<Output, Ts...> {
public:
// -- member types -----------------------------------------------------------
using super = stream_source_driver<Output, Ts...>;
using output_type = typename super::output_type;
using stream_type = stream<output_type>;
using annotated_stream_type = typename super::annotated_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 state_type = typename trait::state;
template <class Init, class Tuple>
stream_source_driver_impl(Init init, Pull f, Done pred, Tuple&& hs)
: pull_(std::move(f)),
done_(std::move(pred)),
hs_(std::forward<Tuple>(hs)) {
init(state_);
}
handshake_tuple_type make_handshake() override {
return std::tuple_cat(std::make_tuple(none), hs_);
}
void pull(downstream<output_type>& out, size_t num) override {
return pull_(state_, out, num);
}
bool done() const noexcept override {
return done_(state_);
}
private:
state_type state_;
Pull pull_;
Done done_;
tuple_type hs_;
};
} // namespace detail
} // namespace caf
#endif // CAF_STREAM_SOURCE_DRIVER_IMPL_HPP
...@@ -20,6 +20,7 @@ ...@@ -20,6 +20,7 @@
#define CAF_DETAIL_STREAM_SOURCE_IMPL_HPP #define CAF_DETAIL_STREAM_SOURCE_IMPL_HPP
#include "caf/downstream.hpp" #include "caf/downstream.hpp"
#include "caf/fwd.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/make_counted.hpp" #include "caf/make_counted.hpp"
#include "caf/outbound_path.hpp" #include "caf/outbound_path.hpp"
...@@ -31,70 +32,69 @@ ...@@ -31,70 +32,69 @@
namespace caf { namespace caf {
namespace detail { namespace detail {
template <class Fun, class Predicate, class DownstreamPolicy> template <class Driver, class Scatterer>
class stream_source_impl : public stream_manager { class stream_source_impl : public stream_manager {
public: public:
using trait = stream_source_trait_t<Fun>; // -- static asserts ---------------------------------------------------------
using state_type = typename trait::state; static_assert(std::is_same<typename Driver::output_type,
typename Scatterer::value_type>::value,
"Driver and Scatterer have incompatible types.");
using output_type = typename trait::output; // -- member types -----------------------------------------------------------
stream_source_impl(local_actor* self, Fun fun, Predicate pred) using driver_type = Driver;
using scatterer_type = Scatterer;
using output_type = typename driver_type::output_type;
// -- constructors, destructors, and assignment operators --------------------
template <class... Ts>
stream_source_impl(local_actor* self, Ts&&... xs)
: stream_manager(self), : stream_manager(self),
at_end_(false), at_end_(false),
fun_(std::move(fun)), driver_(std::forward<Ts>(xs)...),
pred_(std::move(pred)),
out_(self) { out_(self) {
// nop // nop
} }
// -- implementation of virtual functions ------------------------------------
bool done() const override { bool done() const override {
return at_end_ && out_.clean(); return at_end_ && out_.clean();
} }
Scatterer& out() override {
DownstreamPolicy& out() override {
return out_; return out_;
} }
bool generate_messages() override { bool generate_messages() override {
if (at_end_) if (at_end_)
return false; return false;
auto hint = out_.capacity(); auto hint = out_.capacity();
if (hint == 0) if (hint == 0)
return false; return false;
downstream<typename DownstreamPolicy::value_type> ds{out_.buf()}; downstream<output_type> ds{out_.buf()};
fun_(state_, ds, hint); driver_.pull(ds, hint);
if (pred_(state_)) if (driver_.done())
at_end_ = true; at_end_ = true;
return hint != out_.capacity(); return hint != out_.capacity();
} }
state_type& state() {
return state_;
}
bool at_end() const noexcept {
return at_end_;;
}
private: private:
bool at_end_; bool at_end_;
state_type state_; Driver driver_;
Fun fun_; Scatterer out_;
Predicate pred_;
DownstreamPolicy out_;
}; };
template <class Init, class Fun, class Predicate, class Scatterer> template <class Driver,
stream_manager_ptr make_stream_source(local_actor* self, Init init, Fun f, class Scatterer = broadcast_scatterer<typename Driver::output_type>,
Predicate p, policy::arg<Scatterer>) { class... Ts>
using impl = stream_source_impl < Fun, Predicate, Scatterer>; stream_manager_ptr make_stream_source(local_actor* self, Ts&&... xs) {
auto ptr = make_counted<impl>(self, std::move(f), std::move(p)); using impl = stream_source_impl<Driver, Scatterer>;
init(ptr->state()); return make_counted<impl>(self, std::forward<Ts>(xs)...);
return ptr;
} }
} // namespace detail } // namespace detail
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2017 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_STREAM_STAGE_DRIVER_IMPL_HPP
#define CAF_STREAM_STAGE_DRIVER_IMPL_HPP
#include "caf/none.hpp"
#include "caf/stream_stage_driver.hpp"
#include "caf/stream_stage_trait.hpp"
namespace caf {
namespace detail {
template <class Input, class Output, class Process, class Finalize,
class HandshakeData>
class stream_stage_driver_impl;
/// Identifies an unbound sequence of messages.
template <class Input, class Output, class Process, class Finalize, class... Ts>
class stream_stage_driver_impl<Input, Output, Process, Finalize,
std::tuple<Ts...>> final
: public stream_stage_driver<Output, Ts...> {
public:
// -- member types -----------------------------------------------------------
using super = stream_stage_driver<Input, Output, Ts...>;
using input_type = typename super::output_type;
using output_type = typename super::output_type;
using stream_type = stream<output_type>;
using annotated_stream_type = typename super::annotated_stream_type;
using tuple_type = std::tuple<Ts...>;
using handshake_tuple_type = typename super::handshake_tuple_type;
using trait = stream_stage_trait_t<Process>;
using state_type = typename trait::state;
template <class Init, class Tuple>
stream_stage_driver_impl(Init init, Process f, Finalize fin, Tuple&& hs)
: process_(std::move(f)),
fin_(std::move(fin)),
hs_(std::forward<Tuple>(hs)) {
init(state_);
}
handshake_tuple_type make_handshake() override {
return std::tuple_cat(std::make_tuple(none), hs_);
}
void process(std::vector<input_type>&& batch,
downstream<output_type>& out) override {
trait::process::invoke(process_, state_, std::move(batch), out);
}
void finalize() override {
return fin_();
}
private:
state_type state_;
Process process_;
Finalize fin_;
tuple_type hs_;
};
} // namespace detail
} // namespace caf
#endif // CAF_STREAM_STAGE_DRIVER_IMPL_HPP
...@@ -32,30 +32,36 @@ ...@@ -32,30 +32,36 @@
namespace caf { namespace caf {
namespace detail { namespace detail {
template <class Fun, class Cleanup, class DownstreamPolicy> template <class Driver, class Scatterer>
class stream_stage_impl : public stream_manager { class stream_stage_impl : public stream_manager {
public: public:
using trait = stream_stage_trait_t<Fun>; // -- static asserts ---------------------------------------------------------
using state_type = typename trait::state; static_assert(std::is_same<typename Driver::output_type,
typename Scatterer::value_type>::value,
"Driver and Scatterer have incompatible types.");
using input_type = typename trait::input; // -- member types -----------------------------------------------------------
using output_type = typename trait::output; using driver_type = Driver;
stream_stage_impl(local_actor* self, Fun fun, Cleanup cleanup) using scatterer_type = Scatterer;
using output_type = typename driver_type::output_type;
// -- constructors, destructors, and assignment operators --------------------
template <class... Ts>
stream_stage_impl(local_actor* self, Ts&&... xs)
: stream_manager(self), : stream_manager(self),
fun_(std::move(fun)), driver_(std::forward<Ts>(xs)...),
cleanup_(std::move(cleanup)),
out_(self) { out_(self) {
// nop // nop
} }
state_type& state() { // -- implementation of virtual functions ------------------------------------
return state_;
}
DownstreamPolicy& out() override { scatterer_type& out() override {
return out_; return out_;
} }
...@@ -67,10 +73,9 @@ public: ...@@ -67,10 +73,9 @@ public:
CAF_LOG_TRACE(CAF_ARG(x)); CAF_LOG_TRACE(CAF_ARG(x));
using vec_type = std::vector<output_type>; using vec_type = std::vector<output_type>;
if (x.xs.match_elements<vec_type>()) { if (x.xs.match_elements<vec_type>()) {
auto& xs = x.xs.get_as<vec_type>(0); auto& xs = x.xs.get_mutable_as<vec_type>(0);
downstream<typename DownstreamPolicy::value_type> ds{out_.buf()}; downstream<output_type> ds{out_.buf()};
for (auto& x : xs) driver_.process(std::move(xs), ds);
fun_(state_, ds, std::move(x));
return none; return none;
} }
CAF_LOG_ERROR("received unexpected batch type"); CAF_LOG_ERROR("received unexpected batch type");
...@@ -78,8 +83,7 @@ public: ...@@ -78,8 +83,7 @@ public:
} }
message make_output_token(const stream_id&) const override { message make_output_token(const stream_id&) const override {
// TODO: return make_message(stream<output_type>{x}); return make_message_from_tuple(driver_.make_handshake());
return make_message();
} }
bool congested() const override { bool congested() const override {
...@@ -87,19 +91,16 @@ public: ...@@ -87,19 +91,16 @@ public:
} }
private: private:
state_type state_; driver_type driver_;
Fun fun_; scatterer_type out_;
Cleanup cleanup_;
DownstreamPolicy out_;
}; };
template <class Init, class Fun, class Cleanup, class Scatterer> template <class Driver,
stream_manager_ptr make_stream_stage(local_actor* self, Init init, Fun f, class Scatterer = broadcast_scatterer<typename Driver::output_type>,
Cleanup cleanup, policy::arg<Scatterer>) { class... Ts>
using impl = stream_stage_impl<Fun, Cleanup, Scatterer>; stream_manager_ptr make_stream_stage(local_actor* self, Ts&&... xs) {
auto ptr = make_counted<impl>(self, std::move(f), std::move(cleanup)); using impl = stream_stage_impl<Driver, Scatterer>;
init(ptr->state()); return make_counted<impl>(self, std::forward<Ts>(xs)...);
return ptr;
} }
} // namespace detail } // namespace detail
......
...@@ -48,6 +48,11 @@ public: ...@@ -48,6 +48,11 @@ public:
buf_.emplace_back(std::forward<Ts>(xs)...); buf_.emplace_back(std::forward<Ts>(xs)...);
} }
template <class Iterator, class Sentinel>
void append(Iterator first, Sentinel last) {
buf_.insert(buf_.end(), first, last);
}
// @private // @private
queue_type& buf() { queue_type& buf() {
return buf_; return buf_;
......
...@@ -35,6 +35,7 @@ template <class> class intrusive_ptr; ...@@ -35,6 +35,7 @@ template <class> class intrusive_ptr;
template <class> class behavior_type_of; template <class> class behavior_type_of;
template <class> class trivial_match_case; template <class> class trivial_match_case;
template <class> class weak_intrusive_ptr; template <class> class weak_intrusive_ptr;
template <class> class broadcast_scatterer;
template <class> struct timeout_definition; template <class> struct timeout_definition;
......
...@@ -71,16 +71,15 @@ public: ...@@ -71,16 +71,15 @@ public:
// -- downstream communication ----------------------------------------------- // -- downstream communication -----------------------------------------------
/// Sets `open_credit` to `initial_credit` and clears `cached_handshake`.
void handle_ack_open(long initial_credit);
/// Sends a stream handshake. /// Sends a stream handshake.
void emit_open(local_actor* self, strong_actor_ptr origin, static void emit_open(local_actor* self, stream_slot slot,
mailbox_element::forwarding_stack stages, message_id mid, strong_actor_ptr to, message handshake_data,
message handshake_data, stream_priority prio, stream_priority prio, bool is_redeployable);
bool is_redeployable);
/// Sends a `stream_msg::batch` on this path, decrements `open_credit` by /// Sends a `stream_msg::batch` on this path, decrements `open_credit` by
/// Sets `open_credit` to `initial_credit` and clears `cached_handshake`.
void handle_ack_open(long initial_credit);
/// `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);
......
...@@ -26,10 +26,12 @@ ...@@ -26,10 +26,12 @@
#endif // CAF_NO_EXCEPTIONS #endif // CAF_NO_EXCEPTIONS
#include <forward_list> #include <forward_list>
#include <map>
#include <type_traits> #include <type_traits>
#include <unordered_map> #include <unordered_map>
#include "caf/actor_marker.hpp" #include "caf/actor_marker.hpp"
#include "caf/broadcast_scatterer.hpp"
#include "caf/error.hpp" #include "caf/error.hpp"
#include "caf/extend.hpp" #include "caf/extend.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
...@@ -41,13 +43,21 @@ ...@@ -41,13 +43,21 @@
#include "caf/response_handle.hpp" #include "caf/response_handle.hpp"
#include "caf/scheduled_actor.hpp" #include "caf/scheduled_actor.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/stream_manager.hpp"
#include "caf/stream_result.hpp" #include "caf/stream_result.hpp"
#include "caf/stream_result_trait.hpp" #include "caf/stream_result_trait.hpp"
#include "caf/stream_source_trait.hpp"
#include "caf/to_string.hpp" #include "caf/to_string.hpp"
#include "caf/policy/arg.hpp" #include "caf/policy/arg.hpp"
#include "caf/detail/behavior_stack.hpp" #include "caf/detail/behavior_stack.hpp"
#include "caf/detail/stream_sink_driver_impl.hpp"
#include "caf/detail/stream_sink_impl.hpp"
#include "caf/detail/stream_source_driver_impl.hpp"
#include "caf/detail/stream_source_impl.hpp"
#include "caf/detail/stream_stage_driver_impl.hpp"
#include "caf/detail/unordered_flat_map.hpp"
#include "caf/intrusive/fifo_inbox.hpp" #include "caf/intrusive/fifo_inbox.hpp"
...@@ -335,7 +345,6 @@ public: ...@@ -335,7 +345,6 @@ public:
// -- stream management ------------------------------------------------------ // -- stream management ------------------------------------------------------
/*
/// Creates a new stream source and starts streaming to `dest`. /// Creates a new stream source and starts streaming to `dest`.
/// @param dest Actor handle to the stream destination. /// @param dest Actor handle to the stream destination.
/// @param xs User-defined handshake payload. /// @param xs User-defined handshake payload.
...@@ -345,283 +354,313 @@ public: ...@@ -345,283 +354,313 @@ 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 Handle, class... Ts, class Init, class Getter, template <class Driver,
class ClosedPredicate, class ResHandler, class Scatterer = broadcast_scatterer<typename Driver::output_type>,
class Scatterer = broadcast_scatterer< class... Ts>
typename stream_source_trait_t<Getter>::output>> typename Driver::annotated_stream_type make_source(Ts&&... xs) {
annotated_stream<typename stream_source_trait_t<Getter>::output, Ts...> auto slot = next_slot();
make_source(const Handle& dest, std::tuple<Ts...> xs, Init init, auto ptr = detail::make_stream_source<Driver, Scatterer>(
Getter getter, ClosedPredicate pred, ResHandler res_handler, this, std::forward<Ts>(xs)...);
policy::arg<Scatterer> scatterer_type = {}) { ptr->generate_messages();
CAF_IGNORE_UNUSED(scatterer_type); pending_stream_managers_.emplace(slot, ptr);
using type = typename stream_source_trait_t<Getter>::output; return {slot, std::move(ptr)};
using state_type = typename stream_source_trait_t<Getter>::state;
static_assert(std::is_same<
void (state_type&),
typename detail::get_callable_trait<Init>::fun_sig
>::value,
"Expected signature `void (State&)` for init function");
static_assert(std::is_same<
void (state_type&, downstream<type>&, size_t),
typename detail::get_callable_trait<Getter>::fun_sig
>::value,
"Expected signature `void (State&, downstream<T>&, size_t)` "
"for getter function");
static_assert(std::is_same<
bool (const state_type&),
typename detail::get_callable_trait<
ClosedPredicate
>::fun_sig
>::value,
"Expected signature `bool (const State&)` for "
"closed_predicate function");
if (!dest) {
CAF_LOG_ERROR("cannot stream to an invalid actor handle");
return none;
}
// generate new stream ID and manager
auto sid = make_stream_id();
using impl = stream_source_impl<Getter, ClosedPredicate, Scatterer>;
auto ptr = make_counted<impl>(this, std::move(getter), std::move(pred));
auto mid = new_request_id(message_priority::normal);
if (!add_sink<type>(ptr, sid, ctrl(), actor_cast<strong_actor_ptr>(dest),
no_stages, mid, stream_priority::normal, std::move(xs)))
return none;
init(ptr->state());
this->add_multiplexed_response_handler(
mid.response_id(),
stream_result_trait_t<ResHandler>::make_result_handler(res_handler));
streams_.emplace(sid, ptr);
return {std::move(sid), std::move(ptr)};
} }
/// Creates a new stream source and starts streaming to `dest`. template <class Driver, class Input, class... Ts>
/// @param dest Actor handle to the stream destination. stream_result<typename Driver::output_type>
/// @param init Function object for initializing the state of the source. make_sink(const stream<Input>& in, Ts&&... xs) {
/// @param getter Function object for generating messages for the stream. auto slot = next_slot();
/// @param pred Predicate returning `true` when the stream is done. stream_slots id{in.slot(), slot};
/// @param res_handler Function object for receiving the stream result. auto ptr = detail::make_stream_sink<Driver>(this, std::forward<Ts>(xs)...);
/// @param scatterer_type Configures the policy for downstream communication. stream_managers_.emplace(id, ptr);
/// @returns A stream object with a pointer to the generated `stream_manager`. return {slot, std::move(ptr)};
template <class Handle, class Init, class Getter, class ClosedPredicate,
class ResHandler,
class Scatterer = broadcast_scatterer<
typename stream_source_trait_t<Getter>::output>>
stream<typename stream_source_trait_t<Getter>::output>
make_source(const Handle& dest, Init init, Getter getter,
ClosedPredicate pred, ResHandler res_handler,
policy::arg<Scatterer> scatterer_type = {}) {
return make_source(dest, std::make_tuple(), std::move(init),
std::move(getter), std::move(pred),
std::move(res_handler), scatterer_type);
} }
/// Creates a new stream source. template <class Input, class Init, class Fun, class Finalize,
/// @param xs User-defined handshake payload. class Trait = stream_sink_trait_t<Fun, Finalize>>
/// @param init Function object for initializing the state of the source. stream_result<typename Trait::output>
/// @param getter Function object for generating messages for the stream. make_sink(const stream<Input>& in, Init init, Fun fun, Finalize fin) {
/// @param pred Predicate returning `true` when the stream is done. using driver = detail::stream_sink_driver_impl<typename Trait::input,
/// @param scatterer_type Configures the policy for downstream communication. Fun, Finalize>;
/// @returns A stream object with a pointer to the generated `stream_manager`. return make_sink<driver>(in, std::move(init), std::move(fun),
template <class Init, class... Ts, class Getter, class ClosedPredicate, std::move(fin));
class Scatterer = broadcast_scatterer<
typename stream_source_trait_t<Getter>::output>>
annotated_stream<typename stream_source_trait_t<Getter>::output, Ts...>
make_source(std::tuple<Ts...> xs, Init init, Getter getter,
ClosedPredicate pred,
policy::arg<Scatterer> scatterer_type = {}) {
CAF_IGNORE_UNUSED(scatterer_type);
using type = typename stream_source_trait_t<Getter>::output;
using state_type = typename stream_source_trait_t<Getter>::state;
static_assert(std::is_same<
void (state_type&),
typename detail::get_callable_trait<Init>::fun_sig
>::value,
"Expected signature `void (State&)` for init function");
static_assert(std::is_same<
void (state_type&, downstream<type>&, size_t),
typename detail::get_callable_trait<Getter>::fun_sig
>::value,
"Expected signature `void (State&, downstream<T>&, size_t)` "
"for getter function");
static_assert(std::is_same<
bool (const state_type&),
typename detail::get_callable_trait<
ClosedPredicate
>::fun_sig
>::value,
"Expected signature `bool (const State&)` for "
"closed_predicate function");
auto sid = make_stream_id();
using impl = stream_source_impl<Getter, ClosedPredicate, Scatterer>;
auto ptr = make_counted<impl>(this, std::move(getter), std::move(pred));
auto next = take_current_next_stage();
if (!add_sink<type>(ptr, sid, current_sender(), std::move(next),
take_current_forwarding_stack(), current_message_id(),
stream_priority::normal, std::move(xs))) {
CAF_LOG_ERROR("cannot create stream source without sink");
auto rp = make_response_promise();
rp.deliver(sec::no_downstream_stages_defined);
return none;
}
drop_current_message_id();
init(ptr->state());
streams_.emplace(sid, ptr);
return {std::move(sid), std::move(ptr)};
} }
/// Creates a new stream source. /*
/// @param init Function object for initializing the state of the source. /// Creates a new stream source and starts streaming to `dest`.
/// @param getter Function object for generating messages for the stream. /// @param dest Actor handle to the stream destination.
/// @param pred Predicate returning `true` when the stream is done. /// @param xs User-defined handshake payload.
/// @param scatterer_type Configures the policy for downstream communication. /// @param init Function object for initializing the state of the source.
/// @returns A stream object with a pointer to the generated `stream_manager`. /// @param getter Function object for generating messages for the stream.
template <class Init, class Getter, class ClosedPredicate, /// @param pred Predicate returning `true` when the stream is done.
class Scatterer = broadcast_scatterer< /// @param res_handler Function object for receiving the stream result.
typename stream_source_trait_t<Getter>::output>> /// @param scatterer_type Configures the policy for downstream
stream<typename stream_source_trait_t<Getter>::output> communication.
make_source(Init init, Getter getter, ClosedPredicate pred, /// @returns A stream object with a pointer to the generated
policy::arg<Scatterer> scatterer_type = {}) { `stream_manager`. template <class Handle, class... Ts, class Init, class
return make_source(std::make_tuple(), std::move(init), std::move(getter), Getter, class ClosedPredicate, class ResHandler, class Scatterer =
std::move(pred), scatterer_type); broadcast_scatterer< typename stream_source_trait_t<Getter>::output>>
} annotated_stream<typename stream_source_trait_t<Getter>::output, Ts...>
make_source(const Handle& dest, std::tuple<Ts...> xs, Init init,
Getter getter, ClosedPredicate pred, ResHandler res_handler,
policy::arg<Scatterer> scatterer_type = {}) {
CAF_IGNORE_UNUSED(scatterer_type);
using type = typename stream_source_trait_t<Getter>::output;
using state_type = typename stream_source_trait_t<Getter>::state;
static_assert(std::is_same<
void (state_type&),
typename detail::get_callable_trait<Init>::fun_sig
>::value,
"Expected signature `void (State&)` for init function");
static_assert(std::is_same<
void (state_type&, downstream<type>&, size_t),
typename detail::get_callable_trait<Getter>::fun_sig
>::value,
"Expected signature `void (State&, downstream<T>&, size_t)`
" "for getter function"); static_assert(std::is_same< bool (const
state_type&), typename detail::get_callable_trait< ClosedPredicate
>::fun_sig
>::value,
"Expected signature `bool (const State&)` for "
"closed_predicate function");
if (!dest) {
CAF_LOG_ERROR("cannot stream to an invalid actor handle");
return none;
}
// Generate new stream ID and manager
auto slot = next_slot();
auto ptr = detail::make_stream_source(this, std::move(init),
std::move(getter), std::move(pred),
scatterer_type);
outbound_path::emit_open(this, slot, actor_cast<strong_actor_ptr>(dest),
make_handshake<type>(std::move(xs)),
stream_priority::normal, false);
ptr->generate_messages();
pending_stream_managers_.emplace(slot, ptr);
return {slot, std::move(ptr)};
}
/// Creates a new stream stage. /// Creates a new stream source and starts streaming to `dest`.
/// @pre `current_mailbox_element()` is a `stream_msg::open` handshake /// @param dest Actor handle to the stream destination.
/// @param in The input of the stage. /// @param init Function object for initializing the state of the source.
/// @param xs User-defined handshake payload. /// @param getter Function object for generating messages for the stream.
/// @param init Function object for initializing the state of the stage. /// @param pred Predicate returning `true` when the stream is done.
/// @param fun Function object for processing stream elements. /// @param res_handler Function object for receiving the stream result.
/// @param cleanup Function object for clearing the stage of the stage. /// @param scatterer_type Configures the policy for downstream
/// @param policies Sets the policies for up- and downstream communication. communication.
/// @returns A stream object with a pointer to the generated `stream_manager`. /// @returns A stream object with a pointer to the generated
template <class In, class... Ts, class Init, class Fun, class Cleanup, `stream_manager`. template <class Handle, class Init, class Getter, class
class Gatherer = random_gatherer, ClosedPredicate, class ResHandler, class Scatterer = broadcast_scatterer<
class Scatterer = typename stream_source_trait_t<Getter>::output>>
broadcast_scatterer<typename stream_stage_trait_t<Fun>::output>> stream<typename stream_source_trait_t<Getter>::output>
annotated_stream<typename stream_stage_trait_t<Fun>::output, Ts...> make_source(const Handle& dest, Init init, Getter getter,
make_stage(const stream<In>& in, std::tuple<Ts...> xs, Init init, Fun fun, ClosedPredicate pred, ResHandler res_handler,
Cleanup cleanup, policy::arg<Gatherer, Scatterer> policies = {}) { policy::arg<Scatterer> scatterer_type = {}) {
CAF_IGNORE_UNUSED(policies); return make_source(dest, std::make_tuple(), std::move(init),
CAF_ASSERT(current_mailbox_element() != nullptr); std::move(getter), std::move(pred),
CAF_ASSERT(current_mailbox_element()->content().match_elements<stream_msg>()); std::move(res_handler), scatterer_type);
using output_type = typename stream_stage_trait_t<Fun>::output;
using state_type = typename stream_stage_trait_t<Fun>::state;
static_assert(std::is_same<
void (state_type&),
typename detail::get_callable_trait<Init>::fun_sig
>::value,
"Expected signature `void (State&)` for init function");
static_assert(std::is_same<
void (state_type&, downstream<output_type>&, In),
typename detail::get_callable_trait<Fun>::fun_sig
>::value,
"Expected signature `void (State&, downstream<Out>&, In)` "
"for consume function");
using impl = stream_stage_impl<Fun, Cleanup, Gatherer, Scatterer>;
auto ptr = make_counted<impl>(this, in.id(), std::move(fun),
std::move(cleanup));
if (!serve_as_stage<output_type>(ptr, in, std::move(xs))) {
CAF_LOG_ERROR("installing sink and source to the manager failed");
return none;
} }
init(ptr->state());
return {in.id(), std::move(ptr)};
}
/// Creates a new stream stage. /// Creates a new stream source.
/// @pre `current_mailbox_element()` is a `stream_msg::open` handshake /// @param xs User-defined handshake payload.
/// @param in The input of the stage. /// @param init Function object for initializing the state of the source.
/// @param init Function object for initializing the state of the stage. /// @param getter Function object for generating messages for the stream.
/// @param fun Function object for processing stream elements. /// @param pred Predicate returning `true` when the stream is done.
/// @param cleanup Function object for clearing the stage of the stage. /// @param scatterer_type Configures the policy for downstream
/// @param policies Sets the policies for up- and downstream communication. communication.
/// @returns A stream object with a pointer to the generated `stream_manager`. /// @returns A stream object with a pointer to the generated
template <class In, class Init, class Fun, class Cleanup, `stream_manager`. template <class Init, class... Ts, class Getter, class
class Gatherer = random_gatherer, ClosedPredicate, class Scatterer = broadcast_scatterer< typename
class Scatterer = stream_source_trait_t<Getter>::output>> annotated_stream<typename
broadcast_scatterer<typename stream_stage_trait_t<Fun>::output>> stream_source_trait_t<Getter>::output, Ts...> make_source(std::tuple<Ts...>
stream<typename stream_stage_trait_t<Fun>::output> xs, Init init, Getter getter, ClosedPredicate pred, policy::arg<Scatterer>
make_stage(const stream<In>& in, Init init, Fun fun, Cleanup cleanup, scatterer_type = {}) { CAF_IGNORE_UNUSED(scatterer_type); using type =
policy::arg<Gatherer, Scatterer> policies = {}) { typename stream_source_trait_t<Getter>::output; using state_type = typename
return make_stage(in, std::make_tuple(), std::move(init), std::move(fun), stream_source_trait_t<Getter>::state; static_assert(std::is_same< void
std::move(cleanup), policies); (state_type&), typename detail::get_callable_trait<Init>::fun_sig
} >::value,
"Expected signature `void (State&)` for init function");
static_assert(std::is_same<
void (state_type&, downstream<type>&, size_t),
typename detail::get_callable_trait<Getter>::fun_sig
>::value,
"Expected signature `void (State&, downstream<T>&, size_t)`
" "for getter function"); static_assert(std::is_same< bool (const
state_type&), typename detail::get_callable_trait< ClosedPredicate
>::fun_sig
>::value,
"Expected signature `bool (const State&)` for "
"closed_predicate function");
auto sid = make_stream_id();
using impl = stream_source_impl<Getter, ClosedPredicate, Scatterer>;
auto ptr = make_counted<impl>(this, std::move(getter), std::move(pred));
auto next = take_current_next_stage();
if (!add_sink<type>(ptr, sid, current_sender(), std::move(next),
take_current_forwarding_stack(), current_message_id(),
stream_priority::normal, std::move(xs))) {
CAF_LOG_ERROR("cannot create stream source without sink");
auto rp = make_response_promise();
rp.deliver(sec::no_downstream_stages_defined);
return none;
}
drop_current_message_id();
init(ptr->state());
streams_.emplace(sid, ptr);
return {std::move(sid), std::move(ptr)};
}
/// Creates a new stream sink of type T. /// Creates a new stream source.
/// @pre `current_mailbox_element()` is a `stream_msg::open` handshake /// @param init Function object for initializing the state of the source.
/// @param in The input of the sink. /// @param getter Function object for generating messages for the stream.
/// @param f Callback for initializing the object after successful creation. /// @param pred Predicate returning `true` when the stream is done.
/// @param xs Parameter pack for creating the instance of T. /// @param scatterer_type Configures the policy for downstream
/// @returns A stream object with a pointer to the generated `stream_manager`. communication.
template <class T, class In, class SuccessCallback, class... Ts> /// @returns A stream object with a pointer to the generated
stream_result<typename T::output_type> `stream_manager`. template <class Init, class Getter, class ClosedPredicate,
make_sink_impl(const stream<In>& in, SuccessCallback f, Ts&&... xs) { class Scatterer = broadcast_scatterer<
CAF_ASSERT(current_mailbox_element() != nullptr); typename stream_source_trait_t<Getter>::output>>
CAF_ASSERT(current_mailbox_element()->content().match_elements<stream_msg>()); stream<typename stream_source_trait_t<Getter>::output>
auto& sm = current_mailbox_element()->content().get_as<stream_msg>(0); make_source(Init init, Getter getter, ClosedPredicate pred,
CAF_ASSERT(holds_alternative<stream_msg::open>(sm.content)); policy::arg<Scatterer> scatterer_type = {}) {
auto& opn = get<stream_msg::open>(sm.content); return make_source(std::make_tuple(), std::move(init), std::move(getter),
auto sid = in.id(); std::move(pred), scatterer_type);
auto next = take_current_next_stage();
auto ptr = make_counted<T>(this, std::forward<Ts>(xs)...);
auto rp = make_response_promise();
if (!add_source(ptr, sid, std::move(opn.prev_stage),
std::move(opn.original_stage), opn.priority,
opn.redeployable, rp)) {
CAF_LOG_ERROR("cannot create stream stage without source");
rp.deliver(sec::cannot_add_upstream);
return none;
} }
f(*ptr);
streams_.emplace(in.id(), ptr);
return {in.id(), std::move(ptr)};
}
/// Creates a new stream sink. /// Creates a new stream stage.
/// @pre `current_mailbox_element()` is a `stream_msg::open` handshake /// @pre `current_mailbox_element()` is a `stream_msg::open` handshake
/// @param in The input of the sink. /// @param in The input of the stage.
/// @param init Function object for initializing the state of the stage. /// @param xs User-defined handshake payload.
/// @param fun Function object for processing stream elements. /// @param init Function object for initializing the state of the stage.
/// @param finalize Function object for producing the final result. /// @param fun Function object for processing stream elements.
/// @param policies Sets the policies for up- and downstream communication. /// @param cleanup Function object for clearing the stage of the stage.
/// @returns A stream object with a pointer to the generated `stream_manager`. /// @param policies Sets the policies for up- and downstream communication.
template <class In, class Init, class Fun, class Finalize, /// @returns A stream object with a pointer to the generated
class Gatherer = random_gatherer, `stream_manager`. template <class In, class... Ts, class Init, class Fun,
class Scatterer = terminal_stream_scatterer> class Cleanup, class Gatherer = random_gatherer, class Scatterer =
stream_result<typename stream_sink_trait_t<Fun, Finalize>::output> broadcast_scatterer<typename stream_stage_trait_t<Fun>::output>>
make_sink(const stream<In>& in, Init init, Fun fun, Finalize finalize, annotated_stream<typename stream_stage_trait_t<Fun>::output, Ts...>
policy::arg<Gatherer, Scatterer> policies = {}) { make_stage(const stream<In>& in, std::tuple<Ts...> xs, Init init, Fun fun,
CAF_IGNORE_UNUSED(policies); Cleanup cleanup, policy::arg<Gatherer, Scatterer> policies = {})
using state_type = typename stream_sink_trait_t<Fun, Finalize>::state; { CAF_IGNORE_UNUSED(policies); CAF_ASSERT(current_mailbox_element() !=
static_assert(std::is_same< nullptr);
void (state_type&), CAF_ASSERT(current_mailbox_element()->content().match_elements<stream_msg>());
typename detail::get_callable_trait<Init>::fun_sig using output_type = typename stream_stage_trait_t<Fun>::output;
>::value, using state_type = typename stream_stage_trait_t<Fun>::state;
"Expected signature `void (State&)` for init function"); static_assert(std::is_same<
static_assert(std::is_same< void (state_type&),
void (state_type&, In), typename detail::get_callable_trait<Init>::fun_sig
typename detail::get_callable_trait<Fun>::fun_sig >::value,
>::value, "Expected signature `void (State&)` for init function");
"Expected signature `void (State&, Input)` " static_assert(std::is_same<
"for consume function"); void (state_type&, downstream<output_type>&, In),
using impl = stream_sink_impl<Fun, Finalize, Gatherer, Scatterer>; typename detail::get_callable_trait<Fun>::fun_sig
auto initializer = [&](impl& x) { >::value,
init(x.state()); "Expected signature `void (State&, downstream<Out>&, In)` "
}; "for consume function");
return make_sink_impl<impl>(in, initializer, std::move(fun), using impl = stream_stage_impl<Fun, Cleanup, Gatherer, Scatterer>;
std::move(finalize)); auto ptr = make_counted<impl>(this, in.id(), std::move(fun),
} std::move(cleanup));
if (!serve_as_stage<output_type>(ptr, in, std::move(xs))) {
CAF_LOG_ERROR("installing sink and source to the manager failed");
return none;
}
init(ptr->state());
return {in.id(), std::move(ptr)};
}
inline streams_map& streams() { /// Creates a new stream stage.
return streams_; /// @pre `current_mailbox_element()` is a `stream_msg::open` handshake
} /// @param in The input of the stage.
/// @param init Function object for initializing the state of the stage.
/// @param fun Function object for processing stream elements.
/// @param cleanup Function object for clearing the stage of the stage.
/// @param policies Sets the policies for up- and downstream communication.
/// @returns A stream object with a pointer to the generated
`stream_manager`. template <class In, class Init, class Fun, class Cleanup,
class Gatherer = random_gatherer,
class Scatterer =
broadcast_scatterer<typename stream_stage_trait_t<Fun>::output>>
stream<typename stream_stage_trait_t<Fun>::output>
make_stage(const stream<In>& in, Init init, Fun fun, Cleanup cleanup,
policy::arg<Gatherer, Scatterer> policies = {}) {
return make_stage(in, std::make_tuple(), std::move(init), std::move(fun),
std::move(cleanup), policies);
}
/// Tries to send more data on all downstream paths. Use this function to /// Creates a new stream sink of type T.
/// manually trigger batches in a source after receiving more data to send. /// @pre `current_mailbox_element()` is a `stream_msg::open` handshake
void trigger_downstreams(); /// @param in The input of the sink.
/// @param f Callback for initializing the object after successful creation.
/// @param xs Parameter pack for creating the instance of T.
/// @returns A stream object with a pointer to the generated
`stream_manager`. template <class T, class In, class SuccessCallback,
class... Ts> stream_result<typename T::output_type> make_sink_impl(const
stream<In>& in, SuccessCallback f, Ts&&... xs) {
CAF_ASSERT(current_mailbox_element() != nullptr);
CAF_ASSERT(current_mailbox_element()->content().match_elements<stream_msg>());
auto& sm = current_mailbox_element()->content().get_as<stream_msg>(0);
CAF_ASSERT(holds_alternative<stream_msg::open>(sm.content));
auto& opn = get<stream_msg::open>(sm.content);
auto sid = in.id();
auto next = take_current_next_stage();
auto ptr = make_counted<T>(this, std::forward<Ts>(xs)...);
auto rp = make_response_promise();
if (!add_source(ptr, sid, std::move(opn.prev_stage),
std::move(opn.original_stage), opn.priority,
opn.redeployable, rp)) {
CAF_LOG_ERROR("cannot create stream stage without source");
rp.deliver(sec::cannot_add_upstream);
return none;
}
f(*ptr);
streams_.emplace(in.id(), ptr);
return {in.id(), std::move(ptr)};
}
*/ /// Creates a new stream sink.
/// @pre `current_mailbox_element()` is a `stream_msg::open` handshake
/// @param in The input of the sink.
/// @param init Function object for initializing the state of the stage.
/// @param fun Function object for processing stream elements.
/// @param finalize Function object for producing the final result.
/// @param policies Sets the policies for up- and downstream communication.
/// @returns A stream object with a pointer to the generated
`stream_manager`. template <class In, class Init, class Fun, class Finalize,
class Gatherer = random_gatherer,
class Scatterer = terminal_stream_scatterer>
stream_result<typename stream_sink_trait_t<Fun, Finalize>::output>
make_sink(const stream<In>& in, Init init, Fun fun, Finalize finalize,
policy::arg<Gatherer, Scatterer> policies = {}) {
CAF_IGNORE_UNUSED(policies);
using state_type = typename stream_sink_trait_t<Fun, Finalize>::state;
static_assert(std::is_same<
void (state_type&),
typename detail::get_callable_trait<Init>::fun_sig
>::value,
"Expected signature `void (State&)` for init function");
static_assert(std::is_same<
void (state_type&, In),
typename detail::get_callable_trait<Fun>::fun_sig
>::value,
"Expected signature `void (State&, Input)` "
"for consume function");
using impl = stream_sink_impl<Fun, Finalize, Gatherer, Scatterer>;
auto initializer = [&](impl& x) {
init(x.state());
};
return make_sink_impl<impl>(in, initializer, std::move(fun),
std::move(finalize));
}
inline streams_map& streams() {
return streams_;
}
/// Tries to send more data on all downstream paths. Use this function to
/// manually trigger batches in a source after receiving more data to send.
void trigger_downstreams();
*/
/// @cond PRIVATE /// @cond PRIVATE
// -- timeout management ----------------------------------------------------- // -- timeout management -----------------------------------------------------
...@@ -692,8 +731,8 @@ public: ...@@ -692,8 +731,8 @@ public:
} }
template <class T, class... Ts> template <class T, class... Ts>
static message make_handshake(const stream_id& sid, std::tuple<Ts...>& xs) { static message make_handshake(std::tuple<Ts...>&& xs) {
stream<T> token{sid}; stream<T> token;
auto ys = std::tuple_cat(std::forward_as_tuple(token), std::move(xs)); auto ys = std::tuple_cat(std::forward_as_tuple(token), std::move(xs));
return make_message_from_tuple(std::move(ys)); return make_message_from_tuple(std::move(ys));
} }
...@@ -835,6 +874,9 @@ public: ...@@ -835,6 +874,9 @@ public:
protected: protected:
/// @cond PRIVATE /// @cond PRIVATE
/// Returns a currently unused slot.
stream_slot next_slot();
/// Utility function that swaps `f` into a temporary before calling it /// Utility function that swaps `f` into a temporary before calling it
/// and restoring `f` only if it has not been replaced by the user. /// and restoring `f` only if it has not been replaced by the user.
template <class F, class... Ts> template <class F, class... Ts>
...@@ -865,11 +907,11 @@ protected: ...@@ -865,11 +907,11 @@ protected:
swap(g, f); swap(g, f);
} }
bool handle_stream_msg(mailbox_element& x, behavior* active_behavior); //bool handle_stream_msg(mailbox_element& x, behavior* active_behavior);
// -- Member Variables ------------------------------------------------------- // -- member variables -------------------------------------------------------
// used by both event-based and blocking actors /// Stores incoming messages.
mailbox_type mailbox_; mailbox_type mailbox_;
/// Stores user-defined callbacks for message handling. /// Stores user-defined callbacks for message handling.
...@@ -882,7 +924,7 @@ protected: ...@@ -882,7 +924,7 @@ protected:
std::forward_list<pending_response> awaited_responses_; std::forward_list<pending_response> awaited_responses_;
/// Stores callbacks for multiplexed responses. /// Stores callbacks for multiplexed responses.
std::unordered_map<message_id, behavior> multiplexed_responses_; detail::unordered_flat_map<message_id, behavior> multiplexed_responses_;
/// Customization point for setting a default `message` callback. /// Customization point for setting a default `message` callback.
default_handler default_handler_; default_handler default_handler_;
...@@ -896,6 +938,13 @@ protected: ...@@ -896,6 +938,13 @@ protected:
/// Customization point for setting a default `exit_msg` callback. /// Customization point for setting a default `exit_msg` callback.
exit_handler exit_handler_; exit_handler exit_handler_;
/// Stores stream managers for established streams.
std::map<stream_slots, stream_manager_ptr> stream_managers_;
/// Stores stream managers for pending streams, i.e., streams that have not
/// yet received an ACK.
std::map<stream_slot, stream_manager_ptr> pending_stream_managers_;
/// Pointer to a private thread object associated with a detached actor. /// Pointer to a private thread object associated with a detached actor.
detail::private_thread* private_thread_; detail::private_thread* private_thread_;
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2017 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_STREAM_SINK_DRIVER_HPP
#define CAF_STREAM_SINK_DRIVER_HPP
#include <tuple>
#include <vector>
#include "caf/fwd.hpp"
#include "caf/message.hpp"
#include "caf/make_message.hpp"
namespace caf {
/// Identifies an unbound sequence of messages.
template <class Input>
class stream_sink_driver {
public:
// -- member types -----------------------------------------------------------
using input_type = Input;
// -- constructors, destructors, and assignment operators --------------------
virtual ~stream_sink_driver() {
// nop
}
// -- virtual functions ------------------------------------------------------
/// Cleans up any state and produces a result message.
virtual message finalize() {
return make_message();
}
// -- pure virtual functions -------------------------------------------------
/// Processes a single batch.
virtual void process(std::vector<input_type>&& batch) = 0;
};
} // namespace caf
#endif // CAF_STREAM_SINK_DRIVER_HPP
...@@ -26,32 +26,110 @@ ...@@ -26,32 +26,110 @@
namespace caf { namespace caf {
template <class Fun, class Fin> namespace detail {
struct stream_sink_trait;
template <class State, class In, class Out> // -- invoke helper to support element-wise and batch-wise processing ----------
struct stream_sink_trait<void (State&, In), Out (State&)> {
using state = State; struct stream_sink_trait_invoke_one {
using input = In; template <class F, class State, class In>
using output = Out; static void invoke(F& f, State& st, std::vector<In>&& xs) {
template <class F> for (auto& x : xs)
static message make_result(state& st, F f) { f(st, std::move(x));
}
};
struct stream_sink_trait_invoke_all {
template <class F, class State, class In>
static void invoke(F& f, State& st, std::vector<In>&& xs) {
f(st, std::move(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)); return make_message(f(st));
} }
}; };
template <class State, class In> struct stream_sink_trait_void_finalize {
struct stream_sink_trait<void (State&, In), void (State&)> { template <class F, class State>
using state = State; static message invoke(F& f, State& st) {
using input = In;
using output = void;
template <class F>
static message make_result(state& st, F& f) {
f(st); f(st);
return make_message(); return make_message();
} }
}; };
} // namespace detail
// -- trait implementation -----------------------------------------------------
/// Base type for all sink traits.
template <class State, class In, class Out>
struct stream_sink_trait_base {
/// Defines the state element for the function objects.
using state = State;
/// Defines the type of a single stream element.
using input = In;
/// Defines the result type of the sink.
using output = Out;
};
/// Defines required type aliases for stream sinks.
template <class Fun, class Fin>
struct stream_sink_trait;
/// Specializes the trait for element-wise processing with result.
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>
struct stream_sink_trait<void(State&, In), void(State&)>
: stream_sink_trait_base<State, In, void> {
/// 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.
using process = detail::stream_sink_trait_invoke_one;
};
/// Specializes the trait for batch-wise processing with result.
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>
struct stream_sink_trait<void(State&, std::vector<In>&), void(State&)>
: stream_sink_trait_base<State, In, void> {
/// 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.
using process = detail::stream_sink_trait_invoke_all;
};
// -- convenience alias --------------------------------------------------------
/// Derives a sink trait from the signatures of Fun and Fin.
template <class Fun, class Fin> template <class Fun, class Fin>
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,
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2017 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_STREAM_SOURCE_DRIVER_HPP
#define CAF_STREAM_SOURCE_DRIVER_HPP
#include <tuple>
#include "caf/fwd.hpp"
namespace caf {
/// Identifies an unbound sequence of messages.
template <class Output, class... HandshakeData>
class stream_source_driver {
public:
// -- member types -----------------------------------------------------------
using output_type = Output;
using stream_type = stream<output_type>;
using annotated_stream_type = annotated_stream<output_type, HandshakeData...>;
using handshake_tuple_type = std::tuple<stream_type, HandshakeData...>;
// -- constructors, destructors, and assignment operators --------------------
virtual ~stream_source_driver() {
// nop
}
// -- virtual functions ------------------------------------------------------
/// Cleans up any state.
virtual void finalize() {
// nop
}
// -- pure virtual functions -------------------------------------------------
/// Generates handshake data for the next actor in the pipeline.
virtual handshake_tuple_type make_handshake() const = 0;
/// Generates more stream elements.
virtual void pull(downstream<output_type>& out, size_t num) = 0;
/// Returns `true` if the source is done sending, otherwise `false`.
virtual bool done() const noexcept = 0;
};
} // namespace caf
#endif // CAF_STREAM_SOURCE_DRIVER_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2017 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_STREAM_STAGE_DRIVER_HPP
#define CAF_STREAM_STAGE_DRIVER_HPP
#include <tuple>
#include <vector>
#include "caf/fwd.hpp"
namespace caf {
/// Identifies an unbound sequence of messages.
template <class Input, class Output, class... HandshakeData>
class stream_stage_driver {
public:
// -- member types -----------------------------------------------------------
using input_type = Input;
using output_type = Output;
using stream_type = stream<output_type>;
using annotated_stream_type = annotated_stream<output_type, HandshakeData...>;
using handshake_tuple_type = std::tuple<stream_type, HandshakeData...>;
// -- constructors, destructors, and assignment operators --------------------
virtual ~stream_stage_driver() {
// nop
}
// -- virtual functions ------------------------------------------------------
/// Cleans up any state.
virtual void finalize() {
// nop
}
// -- pure virtual functions -------------------------------------------------
/// Generates handshake data for the next actor in the pipeline.
virtual handshake_tuple_type make_handshake() const = 0;
/// Processes a single batch.
virtual void process(std::vector<input_type>&& batch,
downstream<output_type>& out) = 0;
};
} // namespace caf
#endif // CAF_STREAM_STAGE_DRIVER_HPP
...@@ -19,11 +19,39 @@ ...@@ -19,11 +19,39 @@
#ifndef CAF_STREAM_STAGE_TRAIT_HPP #ifndef CAF_STREAM_STAGE_TRAIT_HPP
#define CAF_STREAM_STAGE_TRAIT_HPP #define CAF_STREAM_STAGE_TRAIT_HPP
#include <vector>
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/detail/type_traits.hpp" #include "caf/detail/type_traits.hpp"
namespace caf { namespace caf {
namespace detail {
// -- invoke helper to support element-wise and batch-wise processing ----------
struct stream_stage_trait_invoke_one {
template <class F, class State, class In, class Out>
static void invoke(F& f, State& st, std::vector<In>&& xs,
downstream<Out>& out) {
for (auto& x : xs)
f(st, out, std::move(x));
}
};
struct stream_stage_trait_invoke_all {
template <class F, class State, class In, class Out>
static void invoke(F& f, State& st, std::vector<In>&& xs,
downstream<Out>& out) {
f(st, std::move(xs), out);
}
};
} // namespace detail
// -- trait implementation -----------------------------------------------------
template <class F> template <class F>
struct stream_stage_trait; struct stream_stage_trait;
...@@ -32,8 +60,19 @@ struct stream_stage_trait<void (State&, downstream<Out>&, In)> { ...@@ -32,8 +60,19 @@ struct stream_stage_trait<void (State&, downstream<Out>&, In)> {
using state = State; using state = State;
using input = In; using input = In;
using output = Out; using output = Out;
using invoke = detail::stream_stage_trait_invoke_one;
};
template <class State, class In, class Out>
struct stream_stage_trait<void (State&, std::vector<In>&&, downstream<Out>&)> {
using state = State;
using input = In;
using output = Out;
using invoke = detail::stream_stage_trait_invoke_all;
}; };
// -- convenience alias --------------------------------------------------------
template <class F> template <class F>
using stream_stage_trait_t = using stream_stage_trait_t =
stream_stage_trait<typename detail::get_callable_trait<F>::fun_sig>; stream_stage_trait<typename detail::get_callable_trait<F>::fun_sig>;
......
...@@ -18,10 +18,10 @@ ...@@ -18,10 +18,10 @@
#include "caf/outbound_path.hpp" #include "caf/outbound_path.hpp"
#include "caf/send.hpp" #include "caf/local_actor.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/no_stages.hpp" #include "caf/no_stages.hpp"
#include "caf/local_actor.hpp" #include "caf/send.hpp"
namespace caf { namespace caf {
...@@ -40,6 +40,17 @@ outbound_path::~outbound_path() { ...@@ -40,6 +40,17 @@ outbound_path::~outbound_path() {
// nop // nop
} }
void outbound_path::emit_open(local_actor* self, stream_slot slot,
strong_actor_ptr to, message handshake_data,
stream_priority prio, bool is_redeployable) {
CAF_ASSERT(self != nullptr);
CAF_ASSERT(to != nullptr);
// TODO: attach an aborter to `to`
unsafe_send_as(self, to,
open_stream_msg{slot, std::move(handshake_data), self->ctrl(),
nullptr, prio, is_redeployable});
}
void outbound_path::emit_batch(local_actor* self, long xs_size, message xs) { void outbound_path::emit_batch(local_actor* self, long xs_size, message xs) {
CAF_ASSERT(open_credit >= xs_size); CAF_ASSERT(open_credit >= xs_size);
open_credit -= xs_size; open_credit -= xs_size;
......
...@@ -671,6 +671,18 @@ void scheduled_actor::push_to_cache(mailbox_element_ptr ptr) { ...@@ -671,6 +671,18 @@ void scheduled_actor::push_to_cache(mailbox_element_ptr ptr) {
q->cache().push_back(ptr.release()); q->cache().push_back(ptr.release());
} }
stream_slot scheduled_actor::next_slot() {
stream_slot result = 0;
auto nslot = [](stream_slot x) -> stream_slot {
return x + 1;
};
if (!stream_managers_.empty())
result = std::max(nslot(stream_managers_.rbegin()->first.receiver), result);
if (!pending_stream_managers_.empty())
result = std::max(nslot(pending_stream_managers_.rbegin()->first), result);
return result;
}
/* /*
bool scheduled_actor::handle_stream_msg(mailbox_element& x, bool scheduled_actor::handle_stream_msg(mailbox_element& x,
behavior* active_behavior) { behavior* active_behavior) {
......
...@@ -46,7 +46,10 @@ ...@@ -46,7 +46,10 @@
#include "caf/send.hpp" #include "caf/send.hpp"
#include "caf/stream_manager.hpp" #include "caf/stream_manager.hpp"
#include "caf/stream_scatterer.hpp" #include "caf/stream_scatterer.hpp"
#include "caf/stream_sink_driver.hpp"
#include "caf/stream_slot.hpp" #include "caf/stream_slot.hpp"
#include "caf/stream_source_driver.hpp"
#include "caf/stream_stage_driver.hpp"
#include "caf/system_messages.hpp" #include "caf/system_messages.hpp"
#include "caf/upstream_msg.hpp" #include "caf/upstream_msg.hpp"
#include "caf/variant.hpp" #include "caf/variant.hpp"
...@@ -314,22 +317,32 @@ public: ...@@ -314,22 +317,32 @@ public:
auto slot = next_slot_++; auto slot = next_slot_++;
CAF_MESSAGE(name_ << " starts streaming to " << ref.name() CAF_MESSAGE(name_ << " starts streaming to " << ref.name()
<< " on slot " << slot); << " on slot " << slot);
strong_actor_ptr to = ref.ctrl(); outbound_path::emit_open(this, slot, ref.ctrl(), make_message(),
send(to, open_stream_msg{slot, make_message(), ctrl(), nullptr, stream_priority::normal, false);
stream_priority::normal, false}); struct driver final : public stream_source_driver<int> {
auto init = [](int& x) { public:
x = 0; driver(int sentinel) : x_(0), sentinel_(sentinel) {
}; // nop
auto f = [=](int& x, downstream<int>& out, size_t hint) { }
auto y = std::min(num_messages, x + static_cast<int>(hint));
while (x < y) handshake_tuple_type make_handshake() const override {
out.push(x++); return std::make_tuple(none);
}; }
auto fin = [=](const int& x) {
return x == num_messages; void pull(downstream<int>& out, size_t hint) override {
auto y = std::min(sentinel_, x_ + static_cast<int>(hint));
while (x_ < y)
out.push(x_++);
}
bool done() const noexcept override {
return x_ == sentinel_;
}
private:
int x_;
int sentinel_;
}; };
policy::arg<broadcast_scatterer<int>> token; auto ptr = detail::make_stream_source<driver>(this, num_messages);
auto ptr = detail::make_stream_source(this, init, f, fin, token);
ptr->generate_messages(); ptr->generate_messages();
pending_managers_.emplace(slot, std::move(ptr)); pending_managers_.emplace(slot, std::move(ptr));
} }
...@@ -341,19 +354,24 @@ public: ...@@ -341,19 +354,24 @@ public:
strong_actor_ptr to = ref.ctrl(); strong_actor_ptr to = ref.ctrl();
send(to, open_stream_msg{slot, make_message(), ctrl(), nullptr, send(to, open_stream_msg{slot, make_message(), ctrl(), nullptr,
stream_priority::normal, false}); stream_priority::normal, false});
using log_ptr = vector<int>*; struct driver final : public stream_stage_driver<int, int> {
auto init = [&](log_ptr& ptr) { public:
ptr = &data; driver(vector<int>* log) : log_(log) {
}; // nop
auto f = [](log_ptr& ptr, downstream<int>& out, int x) { }
ptr->push_back(x);
out.push(x); handshake_tuple_type make_handshake() const override {
}; return std::make_tuple(none);
auto cleanup = [](log_ptr&) { }
// nop
void process(vector<int>&& batch, downstream<int>& out) override {
log_->insert(log_->end(), batch.begin(), batch.end());
out.append(batch.begin(), batch.end());
}
private:
vector<int>* log_;
}; };
policy::arg<broadcast_scatterer<int>> token; forwarder = detail::make_stream_stage<driver>(this, &data);
forwarder = detail::make_stream_stage(this, init, f, cleanup, token);
pending_managers_.emplace(slot, forwarder); pending_managers_.emplace(slot, forwarder);
} }
...@@ -367,20 +385,20 @@ public: ...@@ -367,20 +385,20 @@ public:
// was called and we run as a stage. // was called and we run as a stage.
auto mgr = forwarder; auto mgr = forwarder;
if (mgr == nullptr) { if (mgr == nullptr) {
using log_ptr = vector<int>*; struct driver final : public stream_sink_driver<int> {
auto init = [&](log_ptr& ptr) { public:
ptr = &data; driver(std::vector<int>* log) : log_(log) {
}; // nop
auto f = [](log_ptr& ptr, int x) { }
ptr->push_back(x);
}; void process(std::vector<int>&& xs) override {
auto fin = [](log_ptr&) { log_->insert(log_->end(), xs.begin(), xs.end());
// nop }
private:
vector<int>* log_;
}; };
mgr = detail::make_stream_sink(this, std::move(init), std::move(f), mgr = detail::make_stream_sink<driver>(this, &data);
std::move(fin));
} }
// mgr->out().add_path(id, hs.prev_stage);
managers_.emplace(id, mgr); managers_.emplace(id, mgr);
// Create a new queue in the mailbox for incoming traffic. // Create a new queue in the mailbox for incoming traffic.
auto ip = new inbound_path(mgr, id, hs.prev_stage); auto ip = new inbound_path(mgr, id, hs.prev_stage);
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment