Commit 53152a06 authored by Dominik Charousset's avatar Dominik Charousset

Fix stream abortion across remote actors

parent 091cdf1e
...@@ -45,36 +45,28 @@ public: ...@@ -45,36 +45,28 @@ public:
void operator()(stream_msg& x); void operator()(stream_msg& x);
/// @pre `self_->current_sender() != nullptr` /// @pre `self_->current_sender() != nullptr && current_stream_state_valid()`
/// @pre `current_stream_state_ == streams_.end()`
void operator()(stream_msg::open&); void operator()(stream_msg::open&);
/// @pre `self_->current_sender() != nullptr` /// @pre `self_->current_sender() != nullptr && current_stream_state_valid()`
/// @pre `current_stream_state_ != streams_.end()`
void operator()(stream_msg::ack_open&); void operator()(stream_msg::ack_open&);
/// @pre `self_->current_sender() != nullptr` /// @pre `self_->current_sender() != nullptr && current_stream_state_valid()`
/// @pre `current_stream_state_ != streams_.end()`
void operator()(stream_msg::batch&); void operator()(stream_msg::batch&);
/// @pre `self_->current_sender() != nullptr` /// @pre `self_->current_sender() != nullptr && current_stream_state_valid()`
/// @pre `current_stream_state_ != streams_.end()`
void operator()(stream_msg::ack_batch&); void operator()(stream_msg::ack_batch&);
/// @pre `self_->current_sender() != nullptr` /// @pre `self_->current_sender() != nullptr && current_stream_state_valid()`
/// @pre `current_stream_state_ != streams_.end()`
void operator()(stream_msg::close&); void operator()(stream_msg::close&);
/// @pre `self_->current_sender() != nullptr` /// @pre `self_->current_sender() != nullptr && current_stream_state_valid()`
/// @pre `current_stream_state_ != streams_.end()`
void operator()(stream_msg::abort&); void operator()(stream_msg::abort&);
/// @pre `self_->current_sender() != nullptr` /// @pre `self_->current_sender() != nullptr && current_stream_state_valid()`
/// @pre `current_stream_state_ != streams_.end()`
void operator()(stream_msg::downstream_failed&); void operator()(stream_msg::downstream_failed&);
/// @pre `self_->current_sender() != nullptr` /// @pre `self_->current_sender() != nullptr && current_stream_state_valid()`
/// @pre `current_stream_state_ != streams_.end()`
void operator()(stream_msg::upstream_failed&); void operator()(stream_msg::upstream_failed&);
private: private:
......
...@@ -45,20 +45,28 @@ public: ...@@ -45,20 +45,28 @@ public:
void operator()(stream_msg& x); void operator()(stream_msg& x);
/// @pre `self_->current_sender() != nullptr && current_stream_state_valid()`
void operator()(stream_msg::open&); void operator()(stream_msg::open&);
/// @pre `self_->current_sender() != nullptr && current_stream_state_valid()`
void operator()(stream_msg::ack_open&); void operator()(stream_msg::ack_open&);
/// @pre `self_->current_sender() != nullptr && current_stream_state_valid()`
void operator()(stream_msg::batch&); void operator()(stream_msg::batch&);
/// @pre `self_->current_sender() != nullptr && current_stream_state_valid()`
void operator()(stream_msg::ack_batch&); void operator()(stream_msg::ack_batch&);
/// @pre `self_->current_sender() != nullptr && current_stream_state_valid()`
void operator()(stream_msg::close&); void operator()(stream_msg::close&);
/// @pre `self_->current_sender() != nullptr && current_stream_state_valid()`
void operator()(stream_msg::abort&); void operator()(stream_msg::abort&);
/// @pre `self_->current_sender() != nullptr && current_stream_state_valid()`
void operator()(stream_msg::downstream_failed&); void operator()(stream_msg::downstream_failed&);
/// @pre `self_->current_sender() != nullptr && current_stream_state_valid()`
void operator()(stream_msg::upstream_failed&); void operator()(stream_msg::upstream_failed&);
private: private:
......
...@@ -95,13 +95,13 @@ public: ...@@ -95,13 +95,13 @@ public:
} }
/// Queries whether `x` is a known remote node. /// Queries whether `x` is a known remote node.
bool has_remote_path(const node_id& x) const { inline bool has_remote_path(const node_id& x) const {
return remotes().count(x) > 0; return remotes().count(x) > 0;
} }
/// Adds `ptr` as remote stream serv on `x`. This is a no-op if `x` already /// Adds `ptr` as remote stream serv on `x`. This is a no-op if `x` already
/// has a known path. /// has a known path.
void add_remote_path(const node_id& x, strong_actor_ptr ptr) { inline void add_remote_path(const node_id& x, strong_actor_ptr ptr) {
remotes().emplace(x, std::move(ptr)); remotes().emplace(x, std::move(ptr));
} }
...@@ -137,6 +137,14 @@ public: ...@@ -137,6 +137,14 @@ public:
/// @pre `self != nullptr && basp != nullptr` /// @pre `self != nullptr && basp != nullptr`
stream_multiplexer(local_actor* self, backend& service); stream_multiplexer(local_actor* self, backend& service);
/// Adds stream state for `current_stream_msg_->sid`.
stream_states::iterator add_stream(strong_actor_ptr prev,
strong_actor_ptr next,
remote_path* current_path);
/// Removes `current_stream_state_`.
void remove_current_stream();
/// Queries whether stream `x` is managed by this multiplexer. /// Queries whether stream `x` is managed by this multiplexer.
inline bool has_stream(const stream_id& x) const { inline bool has_stream(const stream_id& x) const {
return streams_.count(x) > 0; return streams_.count(x) > 0;
...@@ -148,6 +156,10 @@ public: ...@@ -148,6 +156,10 @@ public:
} }
protected: protected:
inline bool current_stream_state_valid() const {
return current_stream_state_ != streams_.end();
}
// Dispatches `x` on the subtype `T`. // Dispatches `x` on the subtype `T`.
template <class T> template <class T>
static void dispatch(T& derived, stream_msg& x) { static void dispatch(T& derived, stream_msg& x) {
...@@ -257,6 +269,7 @@ protected: ...@@ -257,6 +269,7 @@ protected:
// The remoting backend. // The remoting backend.
backend& service_; backend& service_;
private:
// Open streams. // Open streams.
stream_states streams_; stream_states streams_;
}; };
......
...@@ -358,7 +358,7 @@ bool actor_system::has_middleman() const { ...@@ -358,7 +358,7 @@ bool actor_system::has_middleman() const {
io::middleman& actor_system::middleman() { io::middleman& actor_system::middleman() {
auto& clptr = modules_[module::middleman]; auto& clptr = modules_[module::middleman];
if (!clptr) if (!clptr)
CAF_RAISE_ERROR("cannot access opencl manager: module not loaded"); CAF_RAISE_ERROR("cannot access middleman: module not loaded");
return *reinterpret_cast<io::middleman*>(clptr->subtype_ptr()); return *reinterpret_cast<io::middleman*>(clptr->subtype_ptr());
} }
......
...@@ -58,10 +58,8 @@ void incoming_stream_multiplexer::operator()(stream_msg::open& x) { ...@@ -58,10 +58,8 @@ void incoming_stream_multiplexer::operator()(stream_msg::open& x) {
cme->stages.pop_back(); cme->stages.pop_back();
// Our prev always is the remote stream server proxy. // Our prev always is the remote stream server proxy.
auto current_remote_path = remotes().emplace(prev->node(), prev).first; auto current_remote_path = remotes().emplace(prev->node(), prev).first;
current_stream_state_ = streams_.emplace(current_stream_msg_->sid, current_stream_state_ = add_stream(std::move(prev), successor,
stream_state{std::move(prev), &current_remote_path->second);
successor,
&current_remote_path->second}).first;
// Rewrite handshake and forward it to the next stage. // Rewrite handshake and forward it to the next stage.
x.prev_stage = self_->ctrl(); x.prev_stage = self_->ctrl();
auto ptr = make_mailbox_element( auto ptr = make_mailbox_element(
...@@ -76,54 +74,54 @@ void incoming_stream_multiplexer::operator()(stream_msg::open& x) { ...@@ -76,54 +74,54 @@ void incoming_stream_multiplexer::operator()(stream_msg::open& x) {
void incoming_stream_multiplexer::operator()(stream_msg::ack_open&) { void incoming_stream_multiplexer::operator()(stream_msg::ack_open&) {
CAF_ASSERT(current_stream_msg_ != nullptr); CAF_ASSERT(current_stream_msg_ != nullptr);
CAF_ASSERT(current_stream_state_ != streams_.end()); CAF_ASSERT(current_stream_state_valid());
forward_to_upstream(); forward_to_upstream();
} }
void incoming_stream_multiplexer::operator()(stream_msg::batch&) { void incoming_stream_multiplexer::operator()(stream_msg::batch&) {
CAF_ASSERT(current_stream_msg_ != nullptr); CAF_ASSERT(current_stream_msg_ != nullptr);
CAF_ASSERT(current_stream_state_ != streams_.end()); CAF_ASSERT(current_stream_state_valid());
forward_to_downstream(); forward_to_downstream();
} }
void incoming_stream_multiplexer::operator()(stream_msg::ack_batch&) { void incoming_stream_multiplexer::operator()(stream_msg::ack_batch&) {
CAF_ASSERT(current_stream_msg_ != nullptr); CAF_ASSERT(current_stream_msg_ != nullptr);
CAF_ASSERT(current_stream_state_ != streams_.end()); CAF_ASSERT(current_stream_state_valid());
forward_to_upstream(); forward_to_upstream();
} }
void incoming_stream_multiplexer::operator()(stream_msg::close&) { void incoming_stream_multiplexer::operator()(stream_msg::close&) {
CAF_ASSERT(current_stream_msg_ != nullptr); CAF_ASSERT(current_stream_msg_ != nullptr);
CAF_ASSERT(current_stream_state_ != streams_.end()); CAF_ASSERT(current_stream_state_valid());
forward_to_downstream(); forward_to_downstream();
streams_.erase(current_stream_state_); remove_current_stream();
} }
void incoming_stream_multiplexer::operator()(stream_msg::abort& x) { void incoming_stream_multiplexer::operator()(stream_msg::abort& x) {
CAF_ASSERT(current_stream_msg_ != nullptr); CAF_ASSERT(current_stream_msg_ != nullptr);
CAF_ASSERT(current_stream_state_ != streams_.end()); CAF_ASSERT(current_stream_state_valid());
if (current_stream_state_->second.prev_stage == self_->current_sender()) if (current_stream_state_->second.prev_stage == self_->current_sender())
fail(x.reason, nullptr, current_stream_state_->second.next_stage); fail(x.reason, nullptr, current_stream_state_->second.next_stage);
else else
fail(x.reason, current_stream_state_->second.prev_stage); fail(x.reason, current_stream_state_->second.prev_stage);
streams_.erase(current_stream_state_); remove_current_stream();
} }
void incoming_stream_multiplexer::operator()(stream_msg::downstream_failed&) { void incoming_stream_multiplexer::operator()(stream_msg::downstream_failed&) {
CAF_ASSERT(current_stream_msg_ != nullptr); CAF_ASSERT(current_stream_msg_ != nullptr);
CAF_ASSERT(current_stream_state_ != streams_.end()); CAF_ASSERT(current_stream_state_valid());
// TODO: implement me // TODO: implement me
} }
void incoming_stream_multiplexer::operator()(stream_msg::upstream_failed&) { void incoming_stream_multiplexer::operator()(stream_msg::upstream_failed&) {
CAF_ASSERT(current_stream_msg_ != nullptr); CAF_ASSERT(current_stream_msg_ != nullptr);
CAF_ASSERT(current_stream_state_ != streams_.end()); CAF_ASSERT(current_stream_state_valid());
// TODO: implement me // TODO: implement me
} }
void incoming_stream_multiplexer::forward_to_upstream() { void incoming_stream_multiplexer::forward_to_upstream() {
CAF_ASSERT(current_stream_msg_ != nullptr); CAF_ASSERT(current_stream_msg_ != nullptr);
CAF_ASSERT(current_stream_state_ != streams_.end()); CAF_ASSERT(current_stream_state_valid());
send_remote(*current_stream_state_->second.rpath, send_remote(*current_stream_state_->second.rpath,
std::move(*current_stream_msg_)); std::move(*current_stream_msg_));
} }
......
...@@ -49,7 +49,7 @@ void outgoing_stream_multiplexer::operator()(stream_msg::open& x) { ...@@ -49,7 +49,7 @@ void outgoing_stream_multiplexer::operator()(stream_msg::open& x) {
return fail(sec::invalid_upstream, nullptr); return fail(sec::invalid_upstream, nullptr);
} }
// Make sure we don't receive a handshake for an already open stream. // Make sure we don't receive a handshake for an already open stream.
if (streams_.count(current_stream_msg_->sid) > 0) { if (has_stream(current_stream_msg_->sid)) {
CAF_LOG_WARNING("received stream_msg::open twice"); CAF_LOG_WARNING("received stream_msg::open twice");
return fail(sec::upstream_already_exists, std::move(predecessor)); return fail(sec::upstream_already_exists, std::move(predecessor));
} }
...@@ -68,9 +68,7 @@ void outgoing_stream_multiplexer::operator()(stream_msg::open& x) { ...@@ -68,9 +68,7 @@ void outgoing_stream_multiplexer::operator()(stream_msg::open& x) {
} }
// Update state and send handshake to remote stream_serv (via // Update state and send handshake to remote stream_serv (via
// middleman/basp_broker). // middleman/basp_broker).
streams_.emplace(current_stream_msg_->sid, add_stream(std::move(predecessor), path->hdl, &(*path));
stream_state{std::move(predecessor), path->hdl,
&(*path)});
// Send handshake to remote stream_serv (via middleman/basp_broker). We need // Send handshake to remote stream_serv (via middleman/basp_broker). We need
// to send this message as `current_sender`. We have do bypass the queue // to send this message as `current_sender`. We have do bypass the queue
// since `send_remote` always sends the message from `self_`. // since `send_remote` always sends the message from `self_`.
...@@ -84,49 +82,55 @@ void outgoing_stream_multiplexer::operator()(stream_msg::open& x) { ...@@ -84,49 +82,55 @@ void outgoing_stream_multiplexer::operator()(stream_msg::open& x) {
} }
void outgoing_stream_multiplexer::operator()(stream_msg::ack_open&) { void outgoing_stream_multiplexer::operator()(stream_msg::ack_open&) {
CAF_ASSERT(current_stream_msg_ != nullptr);
CAF_ASSERT(current_stream_state_valid());
forward_to_upstream(); forward_to_upstream();
} }
void outgoing_stream_multiplexer::operator()(stream_msg::batch&) { void outgoing_stream_multiplexer::operator()(stream_msg::batch&) {
CAF_ASSERT(current_stream_msg_ != nullptr);
CAF_ASSERT(current_stream_state_valid());
forward_to_downstream(); forward_to_downstream();
} }
void outgoing_stream_multiplexer::operator()(stream_msg::ack_batch&) { void outgoing_stream_multiplexer::operator()(stream_msg::ack_batch&) {
CAF_ASSERT(current_stream_msg_ != nullptr);
CAF_ASSERT(current_stream_state_valid());
forward_to_upstream(); forward_to_upstream();
} }
void outgoing_stream_multiplexer::operator()(stream_msg::close&) { void outgoing_stream_multiplexer::operator()(stream_msg::close&) {
CAF_ASSERT(current_stream_msg_ != nullptr); CAF_ASSERT(current_stream_msg_ != nullptr);
CAF_ASSERT(current_stream_state_ != streams_.end()); CAF_ASSERT(current_stream_state_valid());
forward_to_downstream(); forward_to_downstream();
streams_.erase(current_stream_state_); remove_current_stream();
} }
void outgoing_stream_multiplexer::operator()(stream_msg::abort& x) { void outgoing_stream_multiplexer::operator()(stream_msg::abort& x) {
CAF_ASSERT(current_stream_msg_ != nullptr); CAF_ASSERT(current_stream_msg_ != nullptr);
CAF_ASSERT(current_stream_state_ != streams_.end()); CAF_ASSERT(current_stream_state_valid());
if (current_stream_state_->second.prev_stage == self_->current_sender()) if (current_stream_state_->second.prev_stage == self_->current_sender())
fail(x.reason, nullptr, current_stream_state_->second.next_stage); fail(x.reason, nullptr, current_stream_state_->second.next_stage);
else else
fail(x.reason, current_stream_state_->second.prev_stage); fail(x.reason, current_stream_state_->second.prev_stage);
streams_.erase(current_stream_state_); remove_current_stream();
} }
void outgoing_stream_multiplexer::operator()(stream_msg::downstream_failed&) { void outgoing_stream_multiplexer::operator()(stream_msg::downstream_failed&) {
CAF_ASSERT(current_stream_msg_ != nullptr); CAF_ASSERT(current_stream_msg_ != nullptr);
CAF_ASSERT(current_stream_state_ != streams_.end()); CAF_ASSERT(current_stream_state_valid());
// TODO: implement me // TODO: implement me
} }
void outgoing_stream_multiplexer::operator()(stream_msg::upstream_failed&) { void outgoing_stream_multiplexer::operator()(stream_msg::upstream_failed&) {
CAF_ASSERT(current_stream_msg_ != nullptr); CAF_ASSERT(current_stream_msg_ != nullptr);
CAF_ASSERT(current_stream_state_ != streams_.end()); CAF_ASSERT(current_stream_state_valid());
// TODO: implement me // TODO: implement me
} }
void outgoing_stream_multiplexer::forward_to_upstream() { void outgoing_stream_multiplexer::forward_to_upstream() {
CAF_ASSERT(current_stream_msg_ != nullptr); CAF_ASSERT(current_stream_msg_ != nullptr);
CAF_ASSERT(current_stream_state_ != streams_.end()); CAF_ASSERT(current_stream_state_valid());
manage_credit(); manage_credit();
send_local(current_stream_state_->second.prev_stage, send_local(current_stream_state_->second.prev_stage,
std::move(*current_stream_msg_)); std::move(*current_stream_msg_));
...@@ -134,7 +138,7 @@ void outgoing_stream_multiplexer::forward_to_upstream() { ...@@ -134,7 +138,7 @@ void outgoing_stream_multiplexer::forward_to_upstream() {
void outgoing_stream_multiplexer::forward_to_downstream() { void outgoing_stream_multiplexer::forward_to_downstream() {
CAF_ASSERT(current_stream_msg_ != nullptr); CAF_ASSERT(current_stream_msg_ != nullptr);
CAF_ASSERT(current_stream_state_ != streams_.end()); CAF_ASSERT(current_stream_state_valid());
send_remote(*current_stream_state_->second.rpath, send_remote(*current_stream_state_->second.rpath,
std::move(*current_stream_msg_)); std::move(*current_stream_msg_));
} }
......
...@@ -23,6 +23,7 @@ ...@@ -23,6 +23,7 @@
#include "caf/variant.hpp" #include "caf/variant.hpp"
#include "caf/to_string.hpp" #include "caf/to_string.hpp"
#include "caf/local_actor.hpp" #include "caf/local_actor.hpp"
#include "caf/stream_aborter.hpp"
namespace caf { namespace caf {
namespace detail { namespace detail {
...@@ -63,6 +64,27 @@ stream_multiplexer::stream_multiplexer(local_actor* self, backend& service) ...@@ -63,6 +64,27 @@ stream_multiplexer::stream_multiplexer(local_actor* self, backend& service)
CAF_ASSERT(self_ != nullptr); CAF_ASSERT(self_ != nullptr);
} }
stream_multiplexer::stream_states::iterator
stream_multiplexer::add_stream(strong_actor_ptr prev, strong_actor_ptr next,
remote_path* path) {
CAF_LOG_TRACE(CAF_ARG(prev) << CAF_ARG(next) << CAF_ARG(path));
auto sid = current_stream_msg_->sid;
auto x = streams_.emplace(sid, stream_state{prev, next, path});
if (x.second) {
stream_aborter::add(prev, self_->address(), sid);
stream_aborter::add(next, self_->address(), sid);
}
return x.first;
}
void stream_multiplexer::remove_current_stream() {
auto& sid = current_stream_state_->first;
auto& st = current_stream_state_->second;
stream_aborter::del(st.prev_stage, self_->address(), sid);
stream_aborter::del(st.next_stage, self_->address(), sid);
streams_.erase(current_stream_state_);
}
optional<stream_multiplexer::remote_path&> optional<stream_multiplexer::remote_path&>
stream_multiplexer::get_remote_or_try_connect(const node_id& nid) { stream_multiplexer::get_remote_or_try_connect(const node_id& nid) {
auto i = remotes().find(nid); auto i = remotes().find(nid);
......
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