Commit 1b98e243 authored by Dominik Charousset's avatar Dominik Charousset

Move add_..._path functions to stream mangers

parent e730ad4a
......@@ -176,7 +176,7 @@ public:
/// Calls `(*this)(x.ptr)`.
template <class T>
void operator()(stream_result<T>& x) {
(*this)(x.slot(), 0, x.ptr());
(*this)(x.in(), 0, x.ptr());
}
// -- visit API: return true if T was visited, false if T was skipped --------
......
......@@ -28,19 +28,17 @@ 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> {
template <class Input, class Result, class Process, class Finalize>
class stream_sink_driver_impl final : public stream_sink_driver<Input, Result> {
public:
// -- member types -----------------------------------------------------------
using super = stream_sink_driver<Input>;
using super = stream_sink_driver<Input, Result>;
using input_type = typename super::input_type;
using 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>
......
......@@ -25,6 +25,7 @@
#include "caf/message_id.hpp"
#include "caf/sec.hpp"
#include "caf/stream_manager.hpp"
#include "caf/stream_sink.hpp"
#include "caf/stream_sink_trait.hpp"
#include "caf/terminal_stream_scatterer.hpp"
......@@ -34,9 +35,9 @@ namespace caf {
namespace detail {
template <class Driver>
class stream_sink_impl : public stream_manager {
class stream_sink_impl : public Driver::sink_type {
public:
using super = stream_manager;
using super = typename Driver::sink_type;
using driver_type = Driver;
......@@ -45,6 +46,7 @@ public:
template <class... Ts>
stream_sink_impl(scheduled_actor* self, Ts&&... xs)
: stream_manager(self),
super(self),
driver_(std::forward<Ts>(xs)...),
out_(self) {
// nop
......@@ -84,7 +86,8 @@ private:
};
template <class Driver, class... Ts>
stream_manager_ptr make_stream_sink(scheduled_actor* self, Ts&&... xs) {
typename Driver::sink_ptr_type make_stream_sink(scheduled_actor* self,
Ts&&... xs) {
using impl = stream_sink_impl<Driver>;
return make_counted<impl>(self, std::forward<Ts>(xs)...);
}
......
......@@ -21,7 +21,6 @@
#include "caf/fwd.hpp"
#include "caf/make_source_result.hpp"
#include "caf/make_stage_result.hpp"
#include "caf/stream_slot.hpp"
#include "caf/invalid_stream.hpp"
......@@ -57,14 +56,6 @@ public:
// nop
}
template <class I, class R, class S>
output_stream(make_stage_result<I, R, T, S, Ts...> x)
: in_(x.in()),
out_(x.out()),
ptr_(std::move(x.ptr())) {
// nop
}
output_stream(stream_slot in_slot, stream_slot out_slot,
stream_manager_ptr mgr)
: in_(in_slot),
......
......@@ -413,58 +413,6 @@ public:
// -- stream management ------------------------------------------------------
/// Creates an output path for given source.
template <class Out, class Scatterer, class... Ts>
make_source_result<Out, Scatterer, Ts...>
add_outbound_path(stream_source_ptr<Out, Scatterer, Ts...> mgr) {
auto slot = assign_next_pending_slot(mgr);
return {slot, std::move(mgr)};
}
/// Creates an output path for given stage.
template <class In, class Result, class Out, class Scatterer, class... Ts>
make_source_result<Out, Scatterer, Ts...>
add_outbound_path(stream_stage_ptr<In, Result, Out, Scatterer, Ts...> mgr) {
auto slot = assign_next_pending_slot(mgr);
return {slot, std::move(mgr)};
}
/// Creates an output path for the given stage without any type checking.
/// @private
template <class Out, class... Ts>
output_stream<Out, Ts...> add_unsafe_outbound_path(stream_manager_ptr mgr) {
auto slot = assign_next_pending_slot(mgr);
return {0, slot, std::move(mgr)};
}
/// Creates an output path for the given stage without any type checking.
/// @pre `next != nullptr`
/// @pre `pending_stream_managers_[slot] == mgr`
/// @pre `mgr->out().terminal() == false`
/// @private
void add_unsafe_outbound_path(stream_manager_ptr mgr, strong_actor_ptr next,
stream_slot slot, strong_actor_ptr origin,
mailbox_element::forwarding_stack stages,
message_id mid);
/// Creates an input path for given stage.
template <class In, class Result, class Out, class Scatterer, class... Ts>
make_sink_result<In, Result>
add_inbound_path(const stream<In>&,
stream_stage_ptr<In, Result, Out, Scatterer, Ts...> mgr) {
auto slot = assign_next_slot(mgr);
return {slot, std::move(mgr)};
}
/// Creates an input path for given stage.
/// @private
template <class Result, class In>
stream_result<Result> add_unsafe_inbound_path(const stream<In>&,
stream_manager_ptr mgr) {
auto slot = assign_next_slot(mgr);
return {slot, std::move(mgr)};
}
/// Creates a new stream source from `Driver`.
/// @param dest Actor handle to the stream destination.
/// @param xs User-defined handshake payload.
......@@ -478,8 +426,7 @@ public:
typename Driver::make_source_result_type make_source(Ts&&... xs) {
using detail::make_stream_source;
auto mgr = make_stream_source<Driver>(this, std::forward<Ts>(xs)...);
auto slot = assign_next_pending_slot(mgr);
return {slot, std::move(mgr)};
return mgr->add_outbound_path();
}
template <class... Ts, class Init, class Pull, class Done, class Finalize,
......@@ -552,9 +499,9 @@ public:
auto result = make_source(std::move(xs), std::move(init), std::move(pull),
std::move(done), scatterer_arg);
auto mid = new_request_id(message_priority::normal);
add_unsafe_outbound_path(result.ptr(), actor_cast<strong_actor_ptr>(dest),
result.out(), actor_cast<strong_actor_ptr>(this),
no_stages, mid);
result.ptr()->add_unsafe_outbound_path(
actor_cast<strong_actor_ptr>(dest), result.out(),
actor_cast<strong_actor_ptr>(this), no_stages, mid);
behavior tmp{
[=](handle_res_result& x) mutable { handle_res(std::move(x)); },
[=](error& err) mutable { handle_res(std::move(err)); }
......@@ -580,13 +527,10 @@ public:
}
template <class Driver, class Input, class... Ts>
stream_result<typename Driver::output_type>
make_sink(const stream<Input>&,
Ts&&... xs) {
stream_result<typename Driver::result_type>
make_sink(const stream<Input>& src, Ts&&... xs) {
auto mgr = detail::make_stream_sink<Driver>(this, std::forward<Ts>(xs)...);
auto slot = assign_next_slot(mgr);
return {slot, std::move(mgr)};
return mgr->add_inbound_path(src);
}
template <class Input, class Init, class Fun, class Finalize,
......@@ -594,18 +538,19 @@ public:
stream_result<typename Trait::output>
make_sink(const stream<Input>& in, Init init, Fun fun, Finalize fin) {
using driver = detail::stream_sink_driver_impl<typename Trait::input,
typename Trait::output,
Fun, Finalize>;
return make_sink<driver>(in, std::move(init), std::move(fun),
std::move(fin));
}
template <class Driver, class In, class... Ts>
typename Driver::make_stage_result_type make_stage(const stream<In>&,
typename Driver::make_stage_result_type make_stage(const stream<In>& src,
Ts&&... xs) {
using detail::make_stream_stage;
auto mgr = make_stream_stage<Driver>(this, std::forward<Ts>(xs)...);
auto in = assign_next_slot(mgr);
auto out = assign_next_pending_slot(mgr);
auto in = mgr->add_inbound_path(src).in();
auto out = mgr->add_outbound_path().out();
return {in, out, std::move(mgr)};
}
......@@ -896,15 +841,15 @@ public:
/// Assigns slot `x` to `mgr`, i.e., adds a new entry to `stream_managers_`.
void assign_slot(stream_slot x, stream_manager_ptr mgr);
/// Convenience function for calling `assign_slot(next_slot(), mgr)`.
stream_slot assign_next_slot(stream_manager_ptr mgr);
/// Assigns slot `x` to the pending manager `mgr`, i.e., adds a new entry to
/// `pending_stream_managers_`.
void assign_pending_slot(stream_slot x, stream_manager_ptr mgr);
/// Convenience function for calling `assign_slot(next_slot(), mgr)`.
stream_slot assign_next_pending_slot(stream_manager_ptr mgr);
stream_slot assign_next_slot_to(stream_manager_ptr mgr);
/// Convenience function for calling `assign_slot(next_slot(), mgr)`.
stream_slot assign_next_pending_slot_to(stream_manager_ptr mgr);
/// Adds a new stream manager to the actor and starts cycle management if
/// needed.
......
......@@ -175,7 +175,33 @@ public:
return self_;
}
/// Creates an outbound path to the current sender without any type checking.
/// @pre `out().terminal() == false`
/// @private
template <class Out, class... Ts>
output_stream<Out, Ts...> add_unsafe_outbound_path();
/// Creates an outbound path to `next` without any type checking.
/// @pre `next != nullptr`
/// @pre `self()->pending_stream_managers_[slot] == this`
/// @pre `out().terminal() == false`
/// @private
void add_unsafe_outbound_path(strong_actor_ptr next, stream_slot slot,
strong_actor_ptr origin,
mailbox_element::forwarding_stack stages,
message_id mid);
/// Creates an inbound path to the current sender without any type checking.
template <class Result, class In>
stream_result<Result> add_unsafe_inbound_path(const stream<In>& in);
protected:
// -- modifiers for self -----------------------------------------------------
stream_slot assign_next_slot();
stream_slot assign_next_pending_slot();
// -- implementation hooks ---------------------------------------------------
virtual void finalize(const error& reason);
......@@ -238,4 +264,23 @@ using stream_manager_ptr = intrusive_ptr<stream_manager>;
} // 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
......@@ -20,9 +20,10 @@
#define CAF_STREAM_RESULT_HPP
#include "caf/fwd.hpp"
#include "caf/make_sink_result.hpp"
#include "caf/none.hpp"
#include "caf/stream_slot.hpp"
#include "caf/stream_manager.hpp"
#include "caf/stream_slot.hpp"
#include "caf/meta/type_name.hpp"
......@@ -37,18 +38,18 @@ public:
stream_result& operator=(stream_result&&) = default;
stream_result& operator=(const stream_result&) = default;
stream_result(none_t = none) : slot_(0) {
stream_result(none_t = none) : in_(0) {
// nop
}
stream_result(stream_slot id) : slot_(id) {
stream_result(stream_slot id) : in_(id) {
// nop
}
/// Convenience constructor for returning the result of `self->new_stream_result`
/// and similar functions.
stream_result(stream_slot id, stream_manager_ptr sptr)
: slot_(id),
: in_(id),
ptr_(std::move(sptr)) {
// nop
}
......@@ -56,14 +57,21 @@ public:
/// Convenience constructor for returning the result of `self->new_stream_result`
/// and similar functions.
stream_result(stream_result other, stream_manager_ptr sptr)
: slot_(other.slot_),
: in_(other.in_),
ptr_(std::move(sptr)) {
// nop
}
template <class In>
stream_result(make_sink_result<In, T> other)
: in_(other.in()),
ptr_(other.ptr()) {
// nop
}
/// Returns the unique identifier for this stream_result.
inline stream_slot slot() const noexcept {
return slot_;
inline stream_slot in() const noexcept {
return in_;
}
/// Returns the handler assigned to this stream on this actor.
......@@ -79,11 +87,11 @@ public:
template <class Inspector>
friend typename Inspector::result_type inspect(Inspector& f,
stream_result& x) {
return f(meta::type_name("stream_result"), x.slot_);
return f(meta::type_name("stream_result"), x.in_);
}
private:
stream_slot slot_;
stream_slot in_;
stream_manager_ptr ptr_;
};
......
......@@ -40,6 +40,11 @@ public:
stream_sink(scheduled_actor* self) : stream_manager(self) {
// nop
}
// -- properties -------------------------------------------------------------
/// Creates a new input path to the current sender.
make_sink_result<In, Result> add_inbound_path(const stream<In>& in);
};
template <class In, class Result>
......@@ -47,4 +52,16 @@ using stream_sink_ptr = intrusive_ptr<stream_sink<In, Result>>;
} // namespace caf
#include "caf/make_sink_result.hpp"
namespace caf {
template <class In, class Result>
make_sink_result<In, Result>
stream_sink<In, Result>::add_inbound_path(const stream<In>&) {
return {this->assign_next_slot(), this};
}
} // namespace caf
#endif // CAF_DETAIL_STREAM_SINK_HPP
......@@ -30,13 +30,24 @@
namespace caf {
/// Identifies an unbound sequence of messages.
template <class Input>
template <class Input, class Result>
class stream_sink_driver {
public:
// -- member types -----------------------------------------------------------
using input_type = Input;
using result_type = Result;
/// Implemented `stream_sink` interface.
using sink_type = stream_sink<input_type, result_type>;
/// Smart pointer to the interface type.
using sink_ptr_type = intrusive_ptr<sink_type>;
/// Wrapper type holding a pointer to `sink_type`.
using make_sink_result_type = make_sink_result<input_type, result_type>;
// -- constructors, destructors, and assignment operators --------------------
virtual ~stream_sink_driver() {
......
......@@ -21,6 +21,7 @@
#include <tuple>
#include "caf/fwd.hpp"
#include "caf/intrusive_ptr.hpp"
#include "caf/stream_manager.hpp"
......@@ -45,6 +46,9 @@ public:
return out_;
}
/// Creates a new output path to the current sender.
make_source_result<Out, Scatterer, Ts...> add_outbound_path();
protected:
Scatterer out_;
};
......@@ -60,4 +64,16 @@ using stream_source_ptr_t =
} // namespace caf
#include "caf/make_source_result.hpp"
namespace caf {
template <class Out, class Scatterer, class... Ts>
make_source_result<Out, Scatterer, Ts...>
stream_source<Out, Scatterer, Ts...>::add_outbound_path() {
return {this->assign_next_pending_slot(), this};
}
} // namespace caf
#endif // CAF_DETAIL_STREAM_SOURCE_HPP
......@@ -21,6 +21,7 @@
#include <tuple>
#include "caf/fwd.hpp"
#include "caf/intrusive_ptr.hpp"
#include "caf/stream_sink.hpp"
#include "caf/stream_source.hpp"
......
......@@ -402,19 +402,6 @@ void scheduled_actor::quit(error x) {
// -- stream management --------------------------------------------------------
void scheduled_actor::add_unsafe_outbound_path(
stream_manager_ptr mgr, strong_actor_ptr next, stream_slot slot,
strong_actor_ptr origin, mailbox_element::forwarding_stack stages,
message_id mid) {
CAF_ASSERT(next != nullptr);
CAF_ASSERT(mgr->out().terminal() == false);
CAF_ASSERT(pending_stream_managers_[slot] == mgr);
// Build pipeline by forwarding handshake along the path.
mgr->send_handshake(std::move(next), slot, std::move(origin),
std::move(stages), mid);
mgr->generate_messages();
}
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));
......@@ -444,9 +431,9 @@ sec scheduled_actor::build_pipeline(stream_slot in, stream_slot out,
if (out != 0 && mgr->out().path(out) == nullptr) {
if (next == nullptr)
return fail(sec::no_downstream_stages_defined);
add_unsafe_outbound_path(mgr, std::move(next), out, current_sender(),
take_current_forwarding_stack(),
current_message_id());
mgr->add_unsafe_outbound_path(std::move(next), out, current_sender(),
take_current_forwarding_stack(),
current_message_id());
}
return sec::none;
}
......@@ -940,20 +927,13 @@ stream_slot scheduled_actor::next_slot() {
}
void scheduled_actor::assign_slot(stream_slot x, stream_manager_ptr mgr) {
CAF_LOG_TRACE("");
CAF_LOG_TRACE(CAF_ARG(x));
if (stream_managers_.empty())
stream_ticks_.start(clock().now());
CAF_ASSERT(stream_managers_.count(x) == 0);
stream_managers_.emplace(x, std::move(mgr));
}
stream_slot scheduled_actor::assign_next_slot(stream_manager_ptr mgr) {
CAF_LOG_TRACE("");
auto x = next_slot();
assign_slot(x, std::move(mgr));
return x;
}
void scheduled_actor::assign_pending_slot(stream_slot x,
stream_manager_ptr mgr) {
CAF_LOG_TRACE(CAF_ARG(x));
......@@ -961,7 +941,15 @@ void scheduled_actor::assign_pending_slot(stream_slot x,
pending_stream_managers_.emplace(x, std::move(mgr));
}
stream_slot scheduled_actor::assign_next_pending_slot(stream_manager_ptr mgr) {
stream_slot scheduled_actor::assign_next_slot_to(stream_manager_ptr mgr) {
CAF_LOG_TRACE("");
auto x = next_slot();
assign_slot(x, std::move(mgr));
return x;
}
stream_slot
scheduled_actor::assign_next_pending_slot_to(stream_manager_ptr mgr) {
CAF_LOG_TRACE("");
auto x = next_slot();
assign_pending_slot(x, std::move(mgr));
......
......@@ -197,6 +197,25 @@ void stream_manager::deliver_promises(message x) {
promises_.clear();
}
void stream_manager::add_unsafe_outbound_path(
strong_actor_ptr next, stream_slot slot, strong_actor_ptr origin,
mailbox_element::forwarding_stack stages, message_id mid) {
CAF_ASSERT(next != nullptr);
CAF_ASSERT(out().terminal() == false);
// Build pipeline by forwarding handshake along the path.
send_handshake(std::move(next), slot, std::move(origin),
std::move(stages), mid);
generate_messages();
}
stream_slot stream_manager::assign_next_slot() {
return self_->assign_next_slot_to(this);
}
stream_slot stream_manager::assign_next_pending_slot() {
return self_->assign_next_pending_slot_to(this);
}
void stream_manager::finalize(const error&) {
// nop
}
......
......@@ -127,11 +127,11 @@ TESTEE(stream_multiplexer) {
return {
[=](join_atom) {
CAF_MESSAGE("received 'join' request");
return self->add_outbound_path(self->state.stage);
return self->state.stage->add_outbound_path();
},
[=](const stream<int>& in, std::string& fname) {
CAF_CHECK_EQUAL(fname, "numbers.txt");
return self->add_inbound_path(in, self->state.stage);
return self->state.stage->add_inbound_path(in);
},
};
}
......
......@@ -225,24 +225,24 @@ TESTEE(stream_multiplexer) {
[=](join_atom, ints_atom) {
auto& stg = self->state.stage;
CAF_MESSAGE("received 'join' request for integers");
auto result = self->add_unsafe_outbound_path<int>(stg);
auto result = stg->add_unsafe_outbound_path<int>();
stg->out().assign<int_scatterer>(result.out());
return result;
},
[=](join_atom, strings_atom) {
auto& stg = self->state.stage;
CAF_MESSAGE("received 'join' request for integers");
auto result = self->add_unsafe_outbound_path<string>(stg);
auto result = stg->add_unsafe_outbound_path<string>();
stg->out().assign<string_scatterer>(result.out());
return result;
},
[=](const stream<int>& in) {
CAF_MESSAGE("received handshake for integers");
return self->add_unsafe_inbound_path<void>(in, self->state.stage);
return self->state.stage->add_unsafe_inbound_path<void>(in);
},
[=](const stream<string>& in) {
CAF_MESSAGE("received handshake for strings");
return self->add_unsafe_inbound_path<void>(in, self->state.stage);
return self->state.stage->add_unsafe_inbound_path<void>(in);
}
};
}
......
......@@ -319,7 +319,7 @@ public:
// was called and we run as a stage.
auto mgr = forwarder;
if (mgr == nullptr) {
struct driver final : public stream_sink_driver<int> {
struct driver final : public stream_sink_driver<int, void> {
public:
driver(std::vector<int>* log) : log_(log) {
// nop
......
......@@ -132,12 +132,12 @@ TESTEE(log_dispatcher) {
[=](join_atom, level lvl) {
auto& stg = self->state.stage;
CAF_MESSAGE("received 'join' request");
auto result = self->add_outbound_path(stg);
auto result = stg->add_outbound_path();
stg->out().set_filter(result.out(), lvl);
return result;
},
[=](const stream<value_type>& in) {
return self->add_inbound_path(in, self->state.stage);
return self->state.stage->add_inbound_path(in);
}
};
}
......
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