Commit 6293a3c4 authored by Dominik Charousset's avatar Dominik Charousset

Add stream aborters when adding new paths

parent 2cc85236
...@@ -38,9 +38,6 @@ namespace caf { ...@@ -38,9 +38,6 @@ namespace caf {
/// State for a path to an upstream actor (source). /// State for a path to an upstream actor (source).
class inbound_path { class inbound_path {
public: public:
/// Stream aborter flag to monitor a path.
static constexpr const auto aborter_type = stream_aborter::source_aborter;
/// Message type for propagating graceful shutdowns. /// Message type for propagating graceful shutdowns.
using regular_shutdown = upstream_msg::drop; using regular_shutdown = upstream_msg::drop;
......
...@@ -35,13 +35,13 @@ public: ...@@ -35,13 +35,13 @@ public:
struct token { struct token {
const actor_addr& observer; const actor_addr& observer;
stream_slots slots; stream_slot slot;
mode m; mode m;
static constexpr size_t token_type = attachable::token::stream_aborter; static constexpr size_t token_type = attachable::token::stream_aborter;
}; };
stream_aborter(actor_addr&& observed, actor_addr&& observer, stream_aborter(actor_addr&& observed, actor_addr&& observer,
stream_slots slots, mode m); stream_slot slot, mode m);
~stream_aborter() override; ~stream_aborter() override;
...@@ -51,21 +51,21 @@ public: ...@@ -51,21 +51,21 @@ public:
/// Adds a stream aborter to `observed`. /// Adds a stream aborter to `observed`.
static void add(strong_actor_ptr observed, actor_addr observer, static void add(strong_actor_ptr observed, actor_addr observer,
stream_slots slots, mode m); stream_slot slot, mode m);
/// Removes a stream aborter from `observed`. /// Removes a stream aborter from `observed`.
static void del(strong_actor_ptr observed, const actor_addr& observer, static void del(strong_actor_ptr observed, const actor_addr& observer,
stream_slots slots, mode m); stream_slot slot, mode m);
private: private:
actor_addr observed_; actor_addr observed_;
actor_addr observer_; actor_addr observer_;
stream_slots slots_; stream_slot slot_;
mode mode_; mode mode_;
}; };
attachable_ptr make_stream_aborter(actor_addr observed, actor_addr observer, attachable_ptr make_stream_aborter(actor_addr observed, actor_addr observer,
stream_slots slot, stream_aborter::mode m); stream_slot slot, stream_aborter::mode m);
} // namespace caf } // namespace caf
......
...@@ -89,7 +89,12 @@ void inbound_path::handle(downstream_msg::batch& x) { ...@@ -89,7 +89,12 @@ void inbound_path::handle(downstream_msg::batch& x) {
void inbound_path::emit_ack_open(local_actor* self, actor_addr rebind_from) { void inbound_path::emit_ack_open(local_actor* self, actor_addr rebind_from) {
CAF_LOG_TRACE(CAF_ARG(slots) << CAF_ARG(rebind_from)); CAF_LOG_TRACE(CAF_ARG(slots) << CAF_ARG(rebind_from));
// Update state.
assigned_credit = initial_credit; assigned_credit = initial_credit;
// Make sure we receive errors from this point on.
stream_aborter::add(hdl, self->address(), slots.receiver,
stream_aborter::source_aborter);
// Send message.
unsafe_send_as(self, hdl, unsafe_send_as(self, hdl,
make<upstream_msg::ack_open>( make<upstream_msg::ack_open>(
slots.invert(), self->address(), std::move(rebind_from), slots.invert(), self->address(), std::move(rebind_from),
......
...@@ -47,7 +47,10 @@ void outbound_path::emit_open(local_actor* self, stream_slot slot, ...@@ -47,7 +47,10 @@ void outbound_path::emit_open(local_actor* self, stream_slot slot,
<< CAF_ARG(prio)); << CAF_ARG(prio));
CAF_ASSERT(self != nullptr); CAF_ASSERT(self != nullptr);
CAF_ASSERT(to != nullptr); CAF_ASSERT(to != nullptr);
// TODO: attach an aborter to `to` // Make sure we receive errors from this point on.
stream_aborter::add(to, self->address(), slot,
stream_aborter::sink_aborter);
// Send message.
unsafe_send_as(self, to, unsafe_send_as(self, to,
open_stream_msg{slot, std::move(handshake_data), self->ctrl(), open_stream_msg{slot, std::move(handshake_data), self->ctrl(),
nullptr, prio}); nullptr, prio});
......
...@@ -291,7 +291,11 @@ struct downstream_msg_visitor { ...@@ -291,7 +291,11 @@ struct downstream_msg_visitor {
// Do *not* store a reference here since we potentially reset `inptr`. // Do *not* store a reference here since we potentially reset `inptr`.
auto mgr = inptr->mgr; auto mgr = inptr->mgr;
inptr->handle(x); inptr->handle(x);
CAF_ASSERT(inptr->slots == dm.slots); // The sender slot can be 0. This is the case for forced_close or
// forced_drop messages from stream aborters.
CAF_ASSERT(inptr->slots == dm.slots
|| (dm.slots.sender == 0
&& dm.slots.receiver == inptr->slots.receiver));
// TODO: replace with `if constexpr` when switching to C++17 // TODO: replace with `if constexpr` when switching to C++17
if (std::is_same<T, downstream_msg::close>::value if (std::is_same<T, downstream_msg::close>::value
|| std::is_same<T, downstream_msg::forced_close>::value) { || std::is_same<T, downstream_msg::forced_close>::value) {
......
...@@ -23,6 +23,7 @@ ...@@ -23,6 +23,7 @@
#include "caf/downstream_msg.hpp" #include "caf/downstream_msg.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/message.hpp" #include "caf/message.hpp"
#include "caf/no_stages.hpp"
#include "caf/system_messages.hpp" #include "caf/system_messages.hpp"
#include "caf/upstream_msg.hpp" #include "caf/upstream_msg.hpp"
...@@ -35,18 +36,19 @@ stream_aborter::~stream_aborter() { ...@@ -35,18 +36,19 @@ stream_aborter::~stream_aborter() {
void stream_aborter::actor_exited(const error& rsn, execution_unit* host) { void stream_aborter::actor_exited(const error& rsn, execution_unit* host) {
CAF_ASSERT(observed_ != observer_); CAF_ASSERT(observed_ != observer_);
auto observer = actor_cast<strong_actor_ptr>(observer_); auto observer = actor_cast<strong_actor_ptr>(observer_);
if (observer != nullptr) if (observer != nullptr) {
{ stream_slots slots{0, slot_};
if (mode_ == source_aborter) mailbox_element_ptr ptr;
observer->enqueue(nullptr, make_message_id(), if (mode_ == source_aborter) {
make_message(caf::make<downstream_msg::forced_close>( using msg_type = downstream_msg::forced_close;
slots_, observed_, rsn)), ptr = make_mailbox_element(nullptr, make_message_id(), no_stages,
host); caf::make<msg_type>(slots, observed_, rsn));
else } else {
observer->enqueue(nullptr, make_message_id(), using msg_type = upstream_msg::forced_drop;
make_message(caf::make<upstream_msg::forced_drop>( ptr = make_mailbox_element(nullptr, make_message_id(), no_stages,
slots_, observed_, rsn)), caf::make<msg_type>(slots, observed_, rsn));
host); }
observer->enqueue(std::move(ptr), host);
} }
} }
...@@ -54,37 +56,37 @@ bool stream_aborter::matches(const attachable::token& what) { ...@@ -54,37 +56,37 @@ bool stream_aborter::matches(const attachable::token& what) {
if (what.subtype != attachable::token::stream_aborter) if (what.subtype != attachable::token::stream_aborter)
return false; return false;
auto& ot = *reinterpret_cast<const token*>(what.ptr); auto& ot = *reinterpret_cast<const token*>(what.ptr);
return ot.observer == observer_ && ot.slots == slots_; return ot.observer == observer_ && ot.slot == slot_;
} }
stream_aborter::stream_aborter(actor_addr&& observed, actor_addr&& observer, stream_aborter::stream_aborter(actor_addr&& observed, actor_addr&& observer,
stream_slots slots, mode m) stream_slot slot, mode m)
: observed_(std::move(observed)), : observed_(std::move(observed)),
observer_(std::move(observer)), observer_(std::move(observer)),
slots_(slots), slot_(slot),
mode_(m) { mode_(m) {
// nop // nop
} }
void stream_aborter::add(strong_actor_ptr observed, actor_addr observer, void stream_aborter::add(strong_actor_ptr observed, actor_addr observer,
stream_slots slots, mode m) { stream_slot slot, mode m) {
CAF_LOG_TRACE(CAF_ARG(observed) << CAF_ARG(observer) << CAF_ARG(slots)); CAF_LOG_TRACE(CAF_ARG(observed) << CAF_ARG(observer) << CAF_ARG(slot));
auto ptr = make_stream_aborter(observed->address(), std::move(observer), auto ptr = make_stream_aborter(observed->address(), std::move(observer),
slots, m); slot, m);
observed->get()->attach(std::move(ptr)); observed->get()->attach(std::move(ptr));
} }
void stream_aborter::del(strong_actor_ptr observed, const actor_addr& observer, void stream_aborter::del(strong_actor_ptr observed, const actor_addr& observer,
stream_slots slots, mode m) { stream_slot slot, mode m) {
CAF_LOG_TRACE(CAF_ARG(observed) << CAF_ARG(observer) << CAF_ARG(slots)); CAF_LOG_TRACE(CAF_ARG(observed) << CAF_ARG(observer) << CAF_ARG(slot));
token tk{observer, slots, m}; token tk{observer, slot, m};
observed->get()->detach(tk); observed->get()->detach(tk);
} }
attachable_ptr make_stream_aborter(actor_addr observed, actor_addr observer, attachable_ptr make_stream_aborter(actor_addr observed, actor_addr observer,
stream_slots slots, stream_aborter::mode m) { stream_slot slot, stream_aborter::mode m) {
return attachable_ptr{ return attachable_ptr{
new stream_aborter(std::move(observed), std::move(observer), slots, m)}; new stream_aborter(std::move(observed), std::move(observer), slot, m)};
} }
} // namespace caf } // namespace caf
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