Commit 08dd5a3e authored by Dominik Charousset's avatar Dominik Charousset

Automagically abort streams on actor termination

parent 2aab0671
...@@ -92,8 +92,9 @@ set (LIBCAF_CORE_SRCS ...@@ -92,8 +92,9 @@ set (LIBCAF_CORE_SRCS
src/skip.cpp src/skip.cpp
src/splitter.cpp src/splitter.cpp
src/stream.cpp src/stream.cpp
src/stream_id.cpp src/stream_aborter.cpp
src/stream_handler.cpp src/stream_handler.cpp
src/stream_id.cpp
src/stream_msg_visitor.cpp src/stream_msg_visitor.cpp
src/stream_multiplexer.cpp src/stream_multiplexer.cpp
src/stream_priority.cpp src/stream_priority.cpp
......
...@@ -51,6 +51,9 @@ public: ...@@ -51,6 +51,9 @@ public:
/// Identifies `default_attachable::observe_token`. /// Identifies `default_attachable::observe_token`.
static constexpr size_t observer = 2; static constexpr size_t observer = 2;
/// Identifies `stream_aborter::token`.
static constexpr size_t stream_aborter = 3;
template <class T> template <class T>
token(const T& tk) : subtype(T::token_type), ptr(&tk) { token(const T& tk) : subtype(T::token_type), ptr(&tk) {
// nop // nop
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2017 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_STREAM_ABORTER_HPP
#define CAF_STREAM_ABORTER_HPP
#include "caf/fwd.hpp"
#include "caf/stream_id.hpp"
#include "caf/actor_addr.hpp"
#include "caf/attachable.hpp"
namespace caf {
class stream_aborter : public attachable {
public:
struct token {
const actor_addr& observer;
const stream_id& sid;
static constexpr size_t token_type = attachable::token::stream_aborter;
};
~stream_aborter() override;
void actor_exited(const error& rsn, execution_unit* host) override;
bool matches(const attachable::token& what) override;
inline static attachable_ptr make(actor_addr observed, actor_addr observer,
const stream_id& sid) {
return attachable_ptr{
new stream_aborter(std::move(observed), std::move(observer), sid)};
}
/// Adds a stream aborter to `observed`.
static void add(strong_actor_ptr observed, actor_addr observer,
const stream_id& sid);
/// Removes a stream aborter from `observed`.
static void del(strong_actor_ptr observed, const actor_addr& observer,
const stream_id& sid);
private:
stream_aborter(actor_addr&& observed, actor_addr&& observer,
const stream_id& type);
actor_addr observed_;
actor_addr observer_;
stream_id sid_;
};
} // namespace caf
#endif // CAF_STREAM_ABORTER_HPP
...@@ -23,6 +23,7 @@ ...@@ -23,6 +23,7 @@
#include "caf/send.hpp" #include "caf/send.hpp"
#include "caf/atom.hpp" #include "caf/atom.hpp"
#include "caf/stream_aborter.hpp"
#include "caf/downstream_path.hpp" #include "caf/downstream_path.hpp"
#include "caf/downstream_policy.hpp" #include "caf/downstream_policy.hpp"
...@@ -60,6 +61,7 @@ bool downstream_policy::add_path(strong_actor_ptr ptr) { ...@@ -60,6 +61,7 @@ bool downstream_policy::add_path(strong_actor_ptr ptr) {
auto predicate = [&](const path_uptr& x) { return x->hdl == ptr; }; auto predicate = [&](const path_uptr& x) { return x->hdl == ptr; };
if (std::none_of(paths_.begin(), paths_.end(), predicate)) { if (std::none_of(paths_.begin(), paths_.end(), predicate)) {
CAF_LOG_DEBUG("added new downstream path" << CAF_ARG(ptr)); CAF_LOG_DEBUG("added new downstream path" << CAF_ARG(ptr));
stream_aborter::add(ptr, self_->address(), sid_);
paths_.emplace_back(new downstream_path(std::move(ptr), false)); paths_.emplace_back(new downstream_path(std::move(ptr), false));
return true; return true;
} }
...@@ -95,7 +97,7 @@ bool downstream_policy::remove_path(strong_actor_ptr& ptr) { ...@@ -95,7 +97,7 @@ bool downstream_policy::remove_path(strong_actor_ptr& ptr) {
auto x = std::move(paths_.back()); auto x = std::move(paths_.back());
paths_.pop_back(); paths_.pop_back();
unsafe_send_as(self_, x->hdl, make<stream_msg::close>(sid_)); unsafe_send_as(self_, x->hdl, make<stream_msg::close>(sid_));
//policy_->reclaim(this, x); stream_aborter::del(x->hdl, self_->address(), sid_);
return true; return true;
} }
return false; return false;
......
...@@ -190,15 +190,9 @@ bool scheduled_actor::cleanup(error&& fail_state, execution_unit* host) { ...@@ -190,15 +190,9 @@ bool scheduled_actor::cleanup(error&& fail_state, execution_unit* host) {
CAF_ASSERT(private_thread_ != nullptr); CAF_ASSERT(private_thread_ != nullptr);
private_thread_->shutdown(); private_thread_->shutdown();
} }
// Discard any state for pending `request` messages. // Clear all state.
awaited_responses_.clear(); awaited_responses_.clear();
multiplexed_responses_.clear(); multiplexed_responses_.clear();
// Abort any open stream.
for (auto& kvp : streams_) {
strong_actor_ptr dummy;
auto err = fail_state;
kvp.second->abort(dummy, err);
}
streams_.clear(); streams_.clear();
// Dispatch to parent's `cleanup` function. // Dispatch to parent's `cleanup` function.
return local_actor::cleanup(std::move(fail_state), host); return local_actor::cleanup(std::move(fail_state), host);
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2017 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/stream_aborter.hpp"
#include "caf/actor.hpp"
#include "caf/logger.hpp"
#include "caf/message.hpp"
#include "caf/actor_cast.hpp"
#include "caf/stream_msg.hpp"
#include "caf/system_messages.hpp"
namespace caf {
stream_aborter::~stream_aborter() {
// nop
}
void stream_aborter::actor_exited(const error& rsn, execution_unit* host) {
CAF_ASSERT(observed_ != observer_);
auto observer = actor_cast<strong_actor_ptr>(observer_);
auto observed = actor_cast<strong_actor_ptr>(observed_);
if (observer != nullptr)
observer->enqueue(std::move(observed), message_id::make(),
make_message(caf::make<stream_msg::abort>(sid_, rsn)),
host);
}
bool stream_aborter::matches(const attachable::token& what) {
if (what.subtype != attachable::token::stream_aborter)
return false;
auto& ot = *reinterpret_cast<const token*>(what.ptr);
return ot.observer == observer_ && ot.sid == sid_;
}
stream_aborter::stream_aborter(actor_addr&& observed, actor_addr&& observer,
const stream_id& sid)
: observed_(std::move(observed)),
observer_(std::move(observer)),
sid_(sid) {
// nop
}
void stream_aborter::add(strong_actor_ptr observed, actor_addr observer,
const stream_id& sid) {
CAF_LOG_TRACE(CAF_ARG(observed) << CAF_ARG(observer) << CAF_ARG(sid));
observed->get()->attach(make(observed->address(), std::move(observer), sid));
}
void stream_aborter::del(strong_actor_ptr observed, const actor_addr& observer,
const stream_id& sid) {
CAF_LOG_TRACE(CAF_ARG(observed) << CAF_ARG(observer) << CAF_ARG(sid));
token tk{observer, sid};
observed->get()->detach(tk);
}
} // namespace caf
...@@ -22,6 +22,7 @@ ...@@ -22,6 +22,7 @@
#include "caf/send.hpp" #include "caf/send.hpp"
#include "caf/stream_msg.hpp" #include "caf/stream_msg.hpp"
#include "caf/upstream_path.hpp" #include "caf/upstream_path.hpp"
#include "caf/stream_aborter.hpp"
namespace caf { namespace caf {
...@@ -83,6 +84,7 @@ expected<long> upstream_policy::add_path(strong_actor_ptr hdl, ...@@ -83,6 +84,7 @@ expected<long> upstream_policy::add_path(strong_actor_ptr hdl,
auto ptr = new upstream_path(std::move(hdl), sid, prio); auto ptr = new upstream_path(std::move(hdl), sid, prio);
paths_.emplace_back(ptr); paths_.emplace_back(ptr);
assignment_vec_.emplace_back(ptr, 0); assignment_vec_.emplace_back(ptr, 0);
stream_aborter::add(ptr->hdl, self_->address(), ptr->sid);
if (downstream_credit > 0) if (downstream_credit > 0)
ptr->assigned_credit = initial_credit(ptr, downstream_credit); ptr->assigned_credit = initial_credit(ptr, downstream_credit);
CAF_LOG_DEBUG("add new upstraem path" << ptr->hdl CAF_LOG_DEBUG("add new upstraem path" << ptr->hdl
...@@ -115,6 +117,7 @@ bool upstream_policy::remove_path(const strong_actor_ptr& hdl, error err) { ...@@ -115,6 +117,7 @@ bool upstream_policy::remove_path(const strong_actor_ptr& hdl, error err) {
// Drop path from list. // Drop path from list.
if (i != e - 1) if (i != e - 1)
std::swap(*i, paths_.back()); std::swap(*i, paths_.back());
stream_aborter::del(hdl, self_->address(), paths_.back()->sid);
if (err != none) if (err != none)
unsafe_send_as(self_, hdl, make<stream_msg::abort>(paths_.back()->sid, unsafe_send_as(self_, hdl, make<stream_msg::abort>(paths_.back()->sid,
std::move(err))); std::move(err)));
......
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