Commit b4e0d75b authored by Dominik Charousset's avatar Dominik Charousset

Harmonize add_input_path and add_output_path impl

The function `scheduled_actor::build_pipeline` previously treated input
and output paths differently by requiring actors to store new managers
in `stream_manager_` when adding an input path while
`pending_stream_managers_` could remainded unchanged. Making things more
consistent simplifies coding going forward.
parent 9a5d69e7
......@@ -416,7 +416,7 @@ public:
bool cleanup(error&& fail_state, execution_unit* host) override;
sec build_pipeline(stream_manager_ptr);
sec build_pipeline(stream_slot in, stream_slot out, stream_manager_ptr mgr);
/// @endcond
......
......@@ -56,9 +56,10 @@ public:
delegate(x);
}
void operator()(stream_slot, stream_manager_ptr& ptr) override {
void operator()(stream_slot in, stream_slot out,
stream_manager_ptr& mgr) override {
// TODO: error handling
self_->build_pipeline(std::move(ptr));
self_->build_pipeline(in, out, std::move(mgr));
}
private:
......
......@@ -60,9 +60,10 @@ public:
/// Called if the message handler returned any "ordinary" value.
virtual void operator()(message&) = 0;
/// Called if the message handler returned a "stream<T>",
/// `output_stream<T, ...>`, or `stream_result<T>`.
virtual void operator()(stream_slot, stream_manager_ptr&) = 0;
/// Called if the message handler returned a `stream<...>`,
/// `output_stream<...>`, or `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
/// default-constructed `optional<T>`.
......@@ -149,13 +150,13 @@ public:
template <class T, class... Us>
void operator()(output_stream<T, Us...>& x) {
stream_manager_ptr ptr{std::move(x.ptr())};
(*this)(x.slot(), ptr);
(*this)(x.origin_slot(), x.slot(), ptr);
}
/// Calls `(*this)(x.ptr)`.
template <class T>
void operator()(stream_result<T>& x) {
(*this)(x.slot(), x.ptr());
(*this)(x.slot(), 0, x.ptr());
}
// -- visit API: return true if T was visited, false if T was skipped --------
......
......@@ -22,6 +22,7 @@
#include "caf/fwd.hpp"
#include "caf/stream_slot.hpp"
#include "caf/stream_source.hpp"
#include "caf/stream_stage.hpp"
#include "caf/meta/type_name.hpp"
......@@ -45,15 +46,33 @@ public:
// -- constructors, destructors, and assignment operators --------------------
output_stream(output_stream&&) = default;
output_stream(const output_stream&) = default;
output_stream(stream_slot id, source_pointer sptr)
: slot_(id),
: origin_(0),
slot_(id),
ptr_(std::move(sptr)) {
// nop
}
template <class In>
output_stream(stream_slot origin, stream_slot id,
stream_stage_ptr<In, T, Ts...> sptr)
: origin_(origin), slot_(id), ptr_(std::move(sptr)) {
// nop
}
// -- properties -------------------------------------------------------------
/// Returns the actor-specific stream slot ID.
/// Returns the slot of the origin stream if `ptr()` is a stage or 0 if
/// `ptr()` is a source.
inline stream_slot origin_slot() const {
return origin_;
}
/// Returns the output slot.
inline stream_slot slot() const {
return slot_;
}
......@@ -73,12 +92,13 @@ public:
template <class Inspector>
friend typename Inspector::result_type inspect(Inspector& f,
output_stream& x) {
return f(meta::type_name("output_stream"), x.slot_);
return f(meta::type_name("output_stream"), x.origin_, x.slot_);
}
private:
// -- member variables -------------------------------------------------------
stream_slot origin_;
stream_slot slot_;
source_pointer ptr_;
};
......
......@@ -415,27 +415,25 @@ public:
/// Creates an output path for given source.
template <class Out, class... Ts>
output_stream<Out, Ts...>
add_output_path(stream_source_ptr<Out, Ts...> ptr) {
// The actual plumbing is done when returning the `output_stream<>` from a
// message handler.
return {0, std::move(ptr)};
add_output_path(stream_source_ptr<Out, 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 Out, class... Ts>
output_stream<Out, Ts...>
add_output_path(stream_stage_ptr<In, Out, Ts...> ptr) {
// The actual plumbing is done when returning the `output_stream<>` from a
// message handler.
return {0, std::move(ptr)};
add_output_path(stream_stage_ptr<In, Out, Ts...> mgr) {
auto slot = assign_next_pending_slot(mgr);
return {slot, std::move(mgr)};
}
/// Creates an input path for given stage.
template <class In, class Out, class... Ts>
output_stream<Out, Ts...>
add_input_path(const stream<In>&, stream_stage_ptr<In, Out, Ts...> mgr) {
auto slot = assign_new_slot(mgr);
return {slot, std::move(mgr)};
auto slot = assign_next_slot(mgr);
return {slot, 0, std::move(mgr)};
}
/// Creates a new stream source and starts streaming to `dest`.
......@@ -451,9 +449,10 @@ public:
class Scatterer = broadcast_scatterer<typename Driver::output_type>,
class... Ts>
typename Driver::output_stream_type make_source(Ts&&... xs) {
auto ptr = detail::make_stream_source<Driver, Scatterer>(
auto mgr = detail::make_stream_source<Driver, Scatterer>(
this, std::forward<Ts>(xs)...);
return {0, std::move(ptr)};
auto slot = assign_next_pending_slot(mgr);
return {slot, std::move(mgr)};
}
template <class... Ts, class Init, class Pull, class Done, class Finalize,
......@@ -511,7 +510,7 @@ public:
stream_result<typename Driver::output_type> make_sink(const stream<Input>&,
Ts&&... xs) {
auto mgr = detail::make_stream_sink<Driver>(this, std::forward<Ts>(xs)...);
auto slot = assign_new_slot(mgr);
auto slot = assign_next_slot(mgr);
return {slot, std::move(mgr)};
}
......@@ -528,13 +527,14 @@ public:
template <class Driver,
class Scatterer = broadcast_scatterer<typename Driver::output_type>,
class Input = int, class... Ts>
typename Driver::output_stream_type make_stage(const stream<Input>&,
typename Driver::output_stream_type make_stage(const stream<Input>& in,
Ts&&... xs) {
using detail::make_stream_stage;
auto mgr = make_stream_stage<Driver, Scatterer>(this,
std::forward<Ts>(xs)...);
auto slot = assign_new_slot(mgr);
return {slot, std::move(mgr)};
assign_slot(in.slot(), mgr);
auto slot = assign_next_pending_slot(mgr);
return {in.slot(), slot, std::move(mgr)};
}
template <class In, class... Ts, class Init, class Fun, class Cleanup,
......@@ -630,10 +630,8 @@ public:
/// @cond PRIVATE
/// Builds the pipeline after receiving an `open_stream_msg`. Sends a stream
/// handshake to the next actor if `mgr->out().terminal() == false`,
/// otherwise
sec build_pipeline(stream_manager_ptr mgr);
/// 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 -----------------------------------------------------
......@@ -815,9 +813,18 @@ public:
/// Returns a currently unused slot.
stream_slot next_slot();
/// Assigns a new slot to `ptr`, adds a new entry to `stream_managers_`, and
/// returns the slot ID.
stream_slot assign_new_slot(stream_manager_ptr ptr);
/// 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);
/// Adds a new stream manager to the actor and starts cycle management if
/// needed.
......
......@@ -74,7 +74,7 @@ public:
(*this)();
}
void operator()(stream_slot, stream_manager_ptr&) override {
void operator()(stream_slot, stream_slot, stream_manager_ptr&) override {
(*this)();
}
};
......
......@@ -290,8 +290,9 @@ void blocking_actor::varargs_tup_receive(receive_cond& rcc, message_id mid,
}
}
sec blocking_actor::build_pipeline(stream_manager_ptr) {
CAF_LOG_ERROR("blocking_actor::connect_pipeline called");
sec blocking_actor::build_pipeline(stream_slot, stream_slot,
stream_manager_ptr) {
CAF_LOG_ERROR("blocking_actor::build_pipeline called");
return sec::bad_function_call;
}
......
......@@ -409,28 +409,42 @@ void scheduled_actor::trigger_downstreams() {
}
*/
sec scheduled_actor::build_pipeline(stream_manager_ptr mgr) {
auto next = take_current_next_stage();
if (next == nullptr) {
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()) {
// Response will be delivered by the sink after stream completion.
mgr->add_promise(std::move(rp));
return sec::none;
CAF_ASSERT(out == 0);
if (next != nullptr)
return fail(sec::cannot_add_downstream);
mgr->add_promise(make_response_promise());
}
// Manager requires a next stage.
CAF_LOG_INFO("broken stream pipeline");
rp.deliver(sec::no_downstream_stages_defined);
return sec::no_downstream_stages_defined;
} else if (mgr->out().terminal()) {
// Cannot connect a sink to a next stage.
return sec::cannot_add_downstream;
}
auto slot = next_slot();
mgr->send_handshake(std::move(next), slot, current_sender(),
take_current_forwarding_stack(), current_message_id());
mgr->generate_messages();
pending_stream_managers_.emplace(slot, std::move(mgr));
if (out != 0) {
CAF_ASSERT(mgr->out().terminal() == false);
CAF_ASSERT(pending_stream_managers_[out] == mgr);
if (next == nullptr)
return fail(sec::no_downstream_stages_defined);
// Build pipeline by forwarding handshake along the path.
mgr->send_handshake(std::move(next), out, current_sender(),
take_current_forwarding_stack(), current_message_id());
mgr->generate_messages();
}
return sec::none;
}
......@@ -909,7 +923,7 @@ uint64_t scheduled_actor::set_timeout(atom_value type,
}
stream_slot scheduled_actor::next_slot() {
stream_slot result = 0;
stream_slot result = 1;
auto nslot = [](const stream_manager_map& x) -> stream_slot {
return x.rbegin()->first + 1;
};
......@@ -920,23 +934,41 @@ stream_slot scheduled_actor::next_slot() {
return result;
}
stream_slot scheduled_actor::assign_new_slot(stream_manager_ptr ptr) {
void scheduled_actor::assign_slot(stream_slot x, stream_manager_ptr mgr) {
CAF_LOG_TRACE("");
if (stream_managers_.empty())
stream_ticks_.start(clock().now());
auto slot = next_slot();
CAF_ASSERT(stream_managers_.count(slot) == 0);
stream_managers_.emplace(slot, std::move(ptr));
return slot;
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));
CAF_ASSERT(pending_stream_managers_.count(x) == 0);
pending_stream_managers_.emplace(x, std::move(mgr));
}
stream_slot scheduled_actor::assign_next_pending_slot(stream_manager_ptr mgr) {
CAF_LOG_TRACE("");
auto x = next_slot();
assign_pending_slot(x, std::move(mgr));
return x;
}
bool scheduled_actor::add_stream_manager(stream_slot id,
stream_manager_ptr ptr) {
stream_manager_ptr mgr) {
CAF_LOG_TRACE(CAF_ARG(id));
if (stream_managers_.empty())
stream_ticks_.start(clock().now());
auto result = stream_managers_.emplace(id, std::move(ptr)).second;
return result;
return stream_managers_.emplace(id, std::move(mgr)).second;
}
void scheduled_actor::erase_stream_manager(stream_slot id) {
......@@ -963,7 +995,8 @@ scheduled_actor::handle_open_stream_msg(mailbox_element& x) {
// Fetches a stream manger from a behavior.
struct visitor : detail::invoke_result_visitor {
stream_manager_ptr ptr;
stream_slots id;
stream_slot in_slot;
stream_slot out_slot;
void operator()() override {
// nop
}
......@@ -976,9 +1009,11 @@ scheduled_actor::handle_open_stream_msg(mailbox_element& x) {
// nop
}
void operator()(stream_slot slot, stream_manager_ptr& x) override {
id.receiver = slot;
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 {
......@@ -989,10 +1024,10 @@ scheduled_actor::handle_open_stream_msg(mailbox_element& x) {
CAF_ASSERT(x.content().match_elements<open_stream_msg>());
auto& osm = x.content().get_mutable_as<open_stream_msg>(0);
visitor f;
f.id.sender = osm.slot;
// Utility lambda for aborting the stream on error.
auto fail = [&](sec x, const char* reason) {
inbound_path::emit_irregular_shutdown(this, f.id, osm.prev_stage,
stream_slots path_id{osm.slot, 0};
inbound_path::emit_irregular_shutdown(this, path_id, osm.prev_stage,
make_error(x, reason));
};
// Utility for invoking the default handler.
......@@ -1026,11 +1061,12 @@ scheduled_actor::handle_open_stream_msg(mailbox_element& x) {
fail(sec::stream_init_failed, "behavior did not create a manager");
return im_dropped;
}
auto path = make_inbound_path(f.ptr, f.id, std::move(osm.prev_stage));
stream_slots path_id{osm.slot, f.in_slot};
auto path = make_inbound_path(f.ptr, path_id, std::move(osm.prev_stage));
CAF_ASSERT(path != nullptr);
path->emit_ack_open(this, actor_cast<actor_addr>(osm.original_stage));
// Propagate handshake down the pipeline.
build_pipeline(std::move(f.ptr));
build_pipeline(f.in_slot, f.out_slot, std::move(f.ptr));
return im_success;
}
default:
......
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