Commit 80ad81ec authored by Dominik Charousset's avatar Dominik Charousset

Use response promises for stream handshakes

parent 4ef3a353
......@@ -86,8 +86,7 @@ public:
/// Sends a handshake to `dest`.
/// @pre `dest != nullptr`
virtual void send_handshake(strong_actor_ptr next, stream_slot slot,
mailbox_element::forwarding_stack stages,
virtual void deliver_handshake(response_promise& rp, stream_slot slot,
message handshake);
// -- implementation hooks for sources ---------------------------------------
......@@ -212,17 +211,21 @@ public:
return {add_unsafe_inbound_path_impl(), this};
}
/// Adds `next` as a new outbound path.
/// Adds a new outbound path to `rp.next()`.
/// @private
stream_slot
add_unsafe_outbound_path_impl(strong_actor_ptr next, message handshake,
mailbox_element::forwarding_stack stages = {});
stream_slot add_unsafe_outbound_path_impl(response_promise& rp,
message handshake);
/// Adds a new outbound path to `next`.
/// @private
stream_slot add_unsafe_outbound_path_impl(strong_actor_ptr next,
message handshake);
/// Adds `self_->current_sender()` as outbound path.
/// Calls `add_unsafe_outbound_path_impl(make_response_promise(), handshake)`.
/// @private
stream_slot add_unsafe_outbound_path_impl(message handshake);
/// Adds
/// Adds the current sender as an inbound path.
/// @pre Current message is an `open_stream_msg`.
stream_slot add_unsafe_inbound_path_impl();
......
......@@ -990,6 +990,8 @@ scheduled_actor::handle_open_stream_msg(mailbox_element& x) {
stream_slots path_id{osm.slot, 0};
inbound_path::emit_irregular_shutdown(this, path_id, osm.prev_stage,
make_error(x, reason));
auto rp = make_response_promise();
rp.deliver(sec::stream_init_failed);
};
// Utility for invoking the default handler.
auto fallback = [&] {
......
......@@ -129,17 +129,15 @@ bool stream_manager::congested() const noexcept {
return false;
}
void stream_manager::send_handshake(strong_actor_ptr next, stream_slot slot,
mailbox_element::forwarding_stack stages,
void stream_manager::deliver_handshake(response_promise& rp, stream_slot slot,
message handshake) {
CAF_ASSERT(next != nullptr);
CAF_LOG_TRACE(CAF_ARG(rp) << CAF_ARG(slot) << CAF_ARG(handshake));
CAF_ASSERT(rp.pending());
CAF_ASSERT(slot != invalid_stream_slot);
++pending_handshakes_;
next->enqueue(
make_mailbox_element(self_->ctrl(), make_message_id(), std::move(stages),
open_stream_msg{slot, std::move(handshake),
self_->ctrl(), next, priority_}),
self_->context());
auto next = rp.next();
rp.deliver(open_stream_msg{slot, std::move(handshake), self_->ctrl(), next,
priority_});
}
bool stream_manager::generate_messages() {
......@@ -178,38 +176,36 @@ void stream_manager::remove_input_path(stream_slot slot, error reason,
self_->erase_inbound_path_later(slot, std::move(reason));
}
stream_slot stream_manager::add_unsafe_outbound_path_impl(
strong_actor_ptr next, message handshake,
mailbox_element::forwarding_stack stages) {
CAF_LOG_TRACE(CAF_ARG(next) << CAF_ARG(handshake) << CAF_ARG(stages));
stream_slot stream_manager::add_unsafe_outbound_path_impl(response_promise& rp,
message handshake) {
CAF_LOG_TRACE(CAF_ARG(rp) << CAF_ARG(handshake));
CAF_ASSERT(out().terminal() == false);
if (next == nullptr) {
if (!rp.pending()) {
CAF_LOG_WARNING("add_outbound_path called with next == nullptr");
auto rp = self_->make_response_promise();
rp.deliver(sec::no_downstream_stages_defined);
return invalid_stream_slot;
}
auto slot = self_->assign_next_pending_slot_to(this);
auto path = out().add_path(slot, next);
auto path = out().add_path(slot, rp.next());
CAF_IGNORE_UNUSED(path);
CAF_ASSERT(path != nullptr);
// Build pipeline by forwarding handshake along the path.
send_handshake(std::move(next), slot, std::move(stages),
std::move(handshake));
deliver_handshake(rp, slot, std::move(handshake));
generate_messages();
return slot;
}
stream_slot stream_manager::add_unsafe_outbound_path_impl(strong_actor_ptr next,
message handshake) {
CAF_LOG_TRACE(CAF_ARG(next) << CAF_ARG(handshake));
response_promise rp{self_->ctrl(), nullptr, {next}, make_message_id()};
return add_unsafe_outbound_path_impl(rp, std::move(handshake));
}
stream_slot stream_manager::add_unsafe_outbound_path_impl(message handshake) {
CAF_LOG_TRACE(CAF_ARG(handshake));
// Call take_current_next_stage explicitly before calling
// take_current_forwarding_stack to avoid UB due to undefined evaluation
// order.
auto next = self_->take_current_next_stage();
auto stages = self_->take_current_forwarding_stack();
// Sources simply forward the original request.
return add_unsafe_outbound_path_impl(std::move(next), std::move(handshake),
std::move(stages));
auto rp = self_->make_response_promise();
return add_unsafe_outbound_path_impl(rp, std::move(handshake));
}
stream_slot stream_manager::add_unsafe_inbound_path_impl() {
......@@ -221,17 +217,17 @@ stream_slot stream_manager::add_unsafe_inbound_path_impl() {
return invalid_stream_slot;
}
auto& osm = x->content().get_mutable_as<open_stream_msg>(0);
auto fail = [&](sec code) {
if (out().terminal() && !self_->current_forwarding_stack().empty()) {
// Sinks must always terminate the stream.
CAF_LOG_WARNING("add_unsafe_inbound_path called in a sink, but the "
"handshake has further stages");
stream_slots path_id{osm.slot, 0};
auto code = sec::cannot_add_downstream;
inbound_path::emit_irregular_shutdown(self_, path_id,
std::move(osm.prev_stage), code);
auto rp = self_->make_response_promise();
rp.deliver(code);
return invalid_stream_slot;
};
if (out().terminal() && self_->current_next_stage() != nullptr) {
// Sinks must always terminate the stream.
CAF_LOG_WARNING("add_unsafe_inbound_path called in a sink, but the "
"handshake has further stages");
return fail(sec::cannot_add_downstream);
}
auto slot = assign_next_slot();
stream_slots path_id{osm.slot, slot};
......
......@@ -235,7 +235,7 @@ CAF_TEST(depth_2_pipeline_50_items) {
CAF_MESSAGE("initiate stream handshake");
self->send(snk * src, "numbers.txt");
expect((string), from(self).to(src).with("numbers.txt"));
expect((open_stream_msg), from(src).to(snk));
expect((open_stream_msg), from(self).to(snk));
expect((upstream_msg::ack_open), from(snk).to(src));
CAF_MESSAGE("start data transmission (a single batch)");
expect((downstream_msg::batch), from(src).to(snk));
......@@ -256,7 +256,7 @@ CAF_TEST(depth_2_pipeline_setup2_50_items) {
CAF_MESSAGE("initiate stream handshake");
self->send(src, "numbers.txt", snk);
expect((string, actor), from(self).to(src).with("numbers.txt", snk));
expect((open_stream_msg), from(src).to(snk));
expect((open_stream_msg), from(strong_actor_ptr{nullptr}).to(snk));
expect((upstream_msg::ack_open), from(snk).to(src));
CAF_MESSAGE("start data transmission (a single batch)");
expect((downstream_msg::batch), from(src).to(snk));
......@@ -277,13 +277,13 @@ CAF_TEST(delayed_depth_2_pipeline_50_items) {
CAF_MESSAGE("initiate stream handshake");
self->send(snk * src, "numbers.txt");
expect((string), from(self).to(src).with("numbers.txt"));
expect((open_stream_msg), from(src).to(snk));
expect((open_stream_msg), from(self).to(snk));
disallow((upstream_msg::ack_open), from(snk).to(src));
disallow((upstream_msg::forced_drop), from(snk).to(src));
CAF_MESSAGE("send 'ok' to trigger sink to handle open_stream_msg");
self->send(snk, ok_atom::value);
expect((ok_atom), from(self).to(snk));
expect((open_stream_msg), from(src).to(snk));
expect((open_stream_msg), from(self).to(snk));
expect((upstream_msg::ack_open), from(snk).to(src));
CAF_MESSAGE("start data transmission (a single batch)");
expect((downstream_msg::batch), from(src).to(snk));
......@@ -304,7 +304,7 @@ CAF_TEST(depth_2_pipeline_500_items) {
CAF_MESSAGE("initiate stream handshake");
self->send(snk * src, "numbers.txt");
expect((string), from(self).to(src).with("numbers.txt"));
expect((open_stream_msg), from(src).to(snk));
expect((open_stream_msg), from(self).to(snk));
expect((upstream_msg::ack_open), from(snk).to(src));
CAF_MESSAGE("start data transmission (loop until src sends 'close')");
do {
......@@ -332,9 +332,9 @@ CAF_TEST(depth_2_pipeline_error_during_handshake) {
CAF_MESSAGE("initiate stream handshake");
self->send(snk * src, "numbers.txt");
expect((std::string), from(self).to(src).with("numbers.txt"));
expect((open_stream_msg), from(src).to(snk));
expect((open_stream_msg), from(self).to(snk));
expect((upstream_msg::forced_drop), from(snk).to(src));
// TODO: expect((error), from(snk).to(self).with(sec::stream_init_failed));
expect((error), from(snk).to(self).with(sec::stream_init_failed));
}
CAF_TEST(depth_2_pipeline_error_at_source) {
......@@ -345,7 +345,7 @@ CAF_TEST(depth_2_pipeline_error_at_source) {
CAF_MESSAGE("initiate stream handshake");
self->send(snk * src, "numbers.txt");
expect((string), from(self).to(src).with("numbers.txt"));
expect((open_stream_msg), from(src).to(snk));
expect((open_stream_msg), from(self).to(snk));
expect((upstream_msg::ack_open), from(snk).to(src));
CAF_MESSAGE("start data transmission (and abort source)");
self->send_exit(src, exit_reason::kill);
......@@ -353,7 +353,6 @@ CAF_TEST(depth_2_pipeline_error_at_source) {
expect((exit_msg), from(self).to(src));
CAF_MESSAGE("expect close message from src and then result from snk");
expect((downstream_msg::forced_close), from(src).to(snk));
// TODO: expect((error), from(snk).to(self));
}
CAF_TEST(depth_2_pipelin_error_at_sink) {
......@@ -364,14 +363,13 @@ CAF_TEST(depth_2_pipelin_error_at_sink) {
CAF_MESSAGE("initiate stream handshake");
self->send(snk * src, "numbers.txt");
expect((string), from(self).to(src).with("numbers.txt"));
expect((open_stream_msg), from(src).to(snk));
expect((open_stream_msg), from(self).to(snk));
CAF_MESSAGE("start data transmission (and abort sink)");
self->send_exit(snk, exit_reason::kill);
expect((upstream_msg::ack_open), from(snk).to(src));
expect((exit_msg), from(self).to(snk));
CAF_MESSAGE("expect close and result messages from snk");
expect((upstream_msg::forced_drop), from(snk).to(src));
// TODO: expect((error), from(snk).to(self));
}
CAF_TEST(depth_3_pipeline_50_items) {
......@@ -389,8 +387,8 @@ CAF_TEST(depth_3_pipeline_50_items) {
CAF_MESSAGE("initiate stream handshake");
self->send(snk * stg * src, "numbers.txt");
expect((string), from(self).to(src).with("numbers.txt"));
expect((open_stream_msg), from(src).to(stg));
expect((open_stream_msg), from(stg).to(snk));
expect((open_stream_msg), from(self).to(stg));
expect((open_stream_msg), from(self).to(snk));
expect((upstream_msg::ack_open), from(snk).to(stg));
expect((upstream_msg::ack_open), from(stg).to(src));
CAF_MESSAGE("start data transmission (a single batch)");
......@@ -419,9 +417,9 @@ CAF_TEST(depth_4_pipeline_500_items) {
CAF_MESSAGE("initiate stream handshake");
self->send(snk * stg2 * stg1 * src, "numbers.txt");
expect((string), from(self).to(src).with("numbers.txt"));
expect((open_stream_msg), from(src).to(stg1));
expect((open_stream_msg), from(stg1).to(stg2));
expect((open_stream_msg), from(stg2).to(snk));
expect((open_stream_msg), from(self).to(stg1));
expect((open_stream_msg), from(self).to(stg2));
expect((open_stream_msg), from(self).to(snk));
expect((upstream_msg::ack_open), from(snk).to(stg2));
expect((upstream_msg::ack_open), from(stg2).to(stg1));
expect((upstream_msg::ack_open), from(stg1).to(src));
......
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