Commit 95af3963 authored by Dominik Charousset's avatar Dominik Charousset Committed by Dominik Charousset

Move open/ack_open logic to stream manager

parent 6d4d62e5
...@@ -65,7 +65,7 @@ public: ...@@ -65,7 +65,7 @@ public:
virtual error handle(inbound_path* from, downstream_msg::forced_close& x); virtual error handle(inbound_path* from, downstream_msg::forced_close& x);
virtual error handle(outbound_path* from, upstream_msg::ack_open& x); virtual error handle(stream_slots, upstream_msg::ack_open& x);
virtual error handle(outbound_path* from, upstream_msg::ack_batch& x); virtual error handle(outbound_path* from, upstream_msg::ack_batch& x);
...@@ -98,6 +98,11 @@ public: ...@@ -98,6 +98,11 @@ public:
mailbox_element::forwarding_stack fwd_stack, mailbox_element::forwarding_stack fwd_stack,
message_id handshake_mid, stream_priority prio); message_id handshake_mid, stream_priority prio);
/// Sends a handshake to `dest`.
/// @pre `dest != nullptr`
void send_handshake(strong_actor_ptr dest, stream_slot slot,
stream_priority prio);
// -- implementation hooks for sources --------------------------------------- // -- implementation hooks for sources ---------------------------------------
/// Tries to generate new messages for the stream. This member function does /// Tries to generate new messages for the stream. This member function does
......
...@@ -57,7 +57,13 @@ error stream_manager::handle(inbound_path*, downstream_msg::forced_close&) { ...@@ -57,7 +57,13 @@ error stream_manager::handle(inbound_path*, downstream_msg::forced_close&) {
return none; return none;
} }
error stream_manager::handle(outbound_path*, upstream_msg::ack_open&) { error stream_manager::handle(stream_slots slots, upstream_msg::ack_open& x) {
auto path = out().add_path(slots.invert(), x.rebind_to);
path->open_credit = x.initial_demand;
path->desired_batch_size = x.desired_batch_size;
--pending_handshakes_;
generate_messages();
push();
return none; return none;
} }
...@@ -107,6 +113,13 @@ void stream_manager::send_handshake(strong_actor_ptr dest, stream_slot slot, ...@@ -107,6 +113,13 @@ void stream_manager::send_handshake(strong_actor_ptr dest, stream_slot slot,
self_->context()); self_->context());
} }
void stream_manager::send_handshake(strong_actor_ptr dest, stream_slot slot,
stream_priority prio) {
mailbox_element::forwarding_stack fwd_stack;
send_handshake(std::move(dest), slot, nullptr, std::move(fwd_stack),
make_message_id(), prio);
}
bool stream_manager::generate_messages() { bool stream_manager::generate_messages() {
return false; return false;
} }
......
...@@ -317,8 +317,6 @@ public: ...@@ -317,8 +317,6 @@ 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);
outbound_path::emit_open(this, slot, ref.ctrl(), make_message(),
stream_priority::normal);
struct driver final : public stream_source_driver<int> { struct driver final : public stream_source_driver<int> {
public: public:
driver(int sentinel) : x_(0), sentinel_(sentinel) { driver(int sentinel) : x_(0), sentinel_(sentinel) {
...@@ -343,6 +341,7 @@ public: ...@@ -343,6 +341,7 @@ public:
int sentinel_; int sentinel_;
}; };
auto ptr = detail::make_stream_source<driver>(this, num_messages); auto ptr = detail::make_stream_source<driver>(this, num_messages);
ptr->send_handshake(ref.ctrl(), slot, stream_priority::normal);
ptr->generate_messages(); ptr->generate_messages();
pending_managers_.emplace(slot, std::move(ptr)); pending_managers_.emplace(slot, std::move(ptr));
} }
...@@ -351,9 +350,6 @@ public: ...@@ -351,9 +350,6 @@ public:
auto slot = next_slot_++; auto slot = next_slot_++;
CAF_MESSAGE(name_ << " starts forwarding to " << ref.name() CAF_MESSAGE(name_ << " starts forwarding to " << ref.name()
<< " on slot " << slot); << " on slot " << slot);
strong_actor_ptr to = ref.ctrl();
send(to, open_stream_msg{slot, make_message(), ctrl(), nullptr,
stream_priority::normal});
struct driver final : public stream_stage_driver<int, int> { struct driver final : public stream_stage_driver<int, int> {
public: public:
driver(vector<int>* log) : log_(log) { driver(vector<int>* log) : log_(log) {
...@@ -372,6 +368,7 @@ public: ...@@ -372,6 +368,7 @@ public:
vector<int>* log_; vector<int>* log_;
}; };
forwarder = detail::make_stream_stage<driver>(this, &data); forwarder = detail::make_stream_stage<driver>(this, &data);
forwarder->send_handshake(ref.ctrl(), slot, stream_priority::normal);
pending_managers_.emplace(slot, forwarder); pending_managers_.emplace(slot, forwarder);
} }
...@@ -412,20 +409,23 @@ public: ...@@ -412,20 +409,23 @@ public:
upstream_msg::ack_open& x) { upstream_msg::ack_open& x) {
TRACE(name_, ack_open, CAF_ARG(slots), TRACE(name_, ack_open, CAF_ARG(slots),
CAF_ARG2("sender", name_of(x.rebind_to)), CAF_ARG(x)); CAF_ARG2("sender", name_of(x.rebind_to)), CAF_ARG(x));
// Get the manager for that stream. CAF_REQUIRE_EQUAL(sender, x.rebind_to);
// Get the manager for that stream, move it from `pending_managers_` to
// `managers_`, and handle `x`.
auto i = pending_managers_.find(slots.receiver); auto i = pending_managers_.find(slots.receiver);
CAF_REQUIRE_NOT_EQUAL(i, pending_managers_.end()); CAF_REQUIRE_NOT_EQUAL(i, pending_managers_.end());
// Create a new queue in the mailbox for incoming traffic. auto res = managers_.emplace(slots, std::move(i->second));
// Swap the buddy/receiver perspective to generate the ID we are using. pending_managers_.erase(i);
managers_.emplace(slots, i->second); CAF_REQUIRE(res.second);
res.first->second->handle(slots, x);
/*
auto to = actor_cast<strong_actor_ptr>(sender); auto to = actor_cast<strong_actor_ptr>(sender);
CAF_REQUIRE_NOT_EQUAL(to, nullptr); CAF_REQUIRE_NOT_EQUAL(to, nullptr);
auto out = i->second->out().add_path(slots.invert(), to); auto out = i->second->out().add_path(slots.invert(), to);
out->open_credit = x.initial_demand; i->second->handle(out, x);
out->desired_batch_size = x.desired_batch_size;
i->second->generate_messages(); i->second->generate_messages();
i->second->push(); i->second->push();
pending_managers_.erase(i); */
} }
void operator()(stream_slots input_slots, actor_addr& sender, void operator()(stream_slots input_slots, actor_addr& sender,
......
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