Commit 3d32db6c authored by Dominik Charousset's avatar Dominik Charousset

Implement graceful shutdown of streams after quit

parent 2c5bd815
...@@ -131,6 +131,7 @@ public: ...@@ -131,6 +131,7 @@ public:
static constexpr int has_used_aout_flag = 0x0400; // local_actor static constexpr int has_used_aout_flag = 0x0400; // local_actor
static constexpr int is_terminated_flag = 0x0800; // local_actor static constexpr int is_terminated_flag = 0x0800; // local_actor
static constexpr int is_cleaned_up_flag = 0x1000; // monitorable_actor static constexpr int is_cleaned_up_flag = 0x1000; // monitorable_actor
static constexpr int is_shutting_down_flag = 0x2000; // scheduled_actor
inline void setf(int flag) { inline void setf(int flag) {
auto x = flags(); auto x = flags();
......
...@@ -764,7 +764,7 @@ public: ...@@ -764,7 +764,7 @@ public:
if (i == stream_managers_.end()) { if (i == stream_managers_.end()) {
auto j = pending_stream_managers_.find(slots.receiver); auto j = pending_stream_managers_.find(slots.receiver);
if (j != pending_stream_managers_.end()) { if (j != pending_stream_managers_.end()) {
j->second->abort(sec::stream_init_failed); j->second->stop(sec::stream_init_failed);
pending_stream_managers_.erase(j); pending_stream_managers_.erase(j);
return; return;
} }
......
...@@ -41,6 +41,16 @@ namespace caf { ...@@ -41,6 +41,16 @@ namespace caf {
/// Manages a single stream with any number of in- and outbound paths. /// Manages a single stream with any number of in- and outbound paths.
class stream_manager : public ref_counted { class stream_manager : public ref_counted {
public: public:
// -- constants --------------------------------------------------------------
/// Configures whether this stream shall remain open even if no in- or
/// outbound paths exist.
static constexpr int is_continuous_flag = 0x0001;
/// Denotes whether the stream is about to stop, only sending already
/// buffered elements.
static constexpr int is_shutting_down_flag = 0x0002;
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
using inbound_paths_list = std::vector<inbound_path*>; using inbound_paths_list = std::vector<inbound_path*>;
...@@ -66,18 +76,16 @@ public: ...@@ -66,18 +76,16 @@ public:
/// Closes all output and input paths and sends the final result to the /// Closes all output and input paths and sends the final result to the
/// client. /// client.
virtual void stop(); virtual void stop(error reason = none);
/// Mark this stream as shutting down, only allowing flushing all related
/// buffers of in- and outbound paths.
virtual void shutdown();
/// Tries to advance the stream by generating more credit or by sending /// Tries to advance the stream by generating more credit or by sending
/// batches. /// batches.
void advance(); void advance();
/// Aborts a stream after any stream message handler returned a non-default
/// constructed error `reason` or the parent actor terminates with a
/// non-default error.
/// @param reason Previous error or non-default exit reason of the parent.
virtual void abort(error reason);
/// Pushes new data to downstream actors by sending batches. The amount of /// Pushes new data to downstream actors by sending batches. The amount of
/// pushed data is limited by the available credit. /// pushed data is limited by the available credit.
virtual void push(); virtual void push();
...@@ -132,17 +140,27 @@ public: ...@@ -132,17 +140,27 @@ public:
// -- properties ------------------------------------------------------------- // -- properties -------------------------------------------------------------
/// Returns whether this stream is shutting down.
bool shutting_down() const noexcept {
return getf(is_shutting_down_flag);
}
/// Returns whether this stream remains open even if no in- or outbound paths /// Returns whether this stream remains open even if no in- or outbound paths
/// exist. The default is `false`. Does not keep a source alive past the /// exist. The default is `false`. Does not keep a source alive past the
/// point where its driver returns `done() == true`. /// point where its driver returns `done() == true`.
inline bool continuous() const noexcept { inline bool continuous() const noexcept {
return continuous_; return getf(is_continuous_flag);
} }
/// Sets whether this stream remains open even if no in- or outbound paths /// Sets whether this stream remains open even if no in- or outbound paths
/// exist. /// exist.
inline void continuous(bool x) noexcept { inline void continuous(bool x) noexcept {
continuous_ = x; if (!shutting_down()) {
if (x)
setf(is_continuous_flag);
else
unsetf(is_continuous_flag);
}
} }
/// Returns the list of inbound paths. /// Returns the list of inbound paths.
...@@ -287,9 +305,23 @@ protected: ...@@ -287,9 +305,23 @@ protected:
/// Configures the importance of outgoing traffic. /// Configures the importance of outgoing traffic.
stream_priority priority_; stream_priority priority_;
/// Stores whether this stream shall remain open even if no in- or outbound /// Stores individual flags, for continuous streaming or when shutting down.
/// paths exist. int flags_;
bool continuous_;
private:
void setf(int flag) noexcept {
auto x = flags_;
flags_ = x | flag;
}
void unsetf(int flag) noexcept {
auto x = flags_;
flags_ = x & ~flag;
}
bool getf(int flag) const noexcept {
return (flags_ & flag) != 0;
}
}; };
/// A reference counting pointer to a `stream_manager`. /// A reference counting pointer to a `stream_manager`.
......
...@@ -78,7 +78,8 @@ struct upstream_msg : tag::boxing_type { ...@@ -78,7 +78,8 @@ struct upstream_msg : tag::boxing_type {
int64_t acknowledged_id; int64_t acknowledged_id;
}; };
/// Informs a source that a sink orderly drops out of a stream. /// Asks the source to discard any remaining credit and close this path
/// after receiving an ACK for the last batch.
struct drop { struct drop {
/// Allows the testing DSL to unbox this type automagically. /// Allows the testing DSL to unbox this type automagically.
using outer_type = upstream_msg; using outer_type = upstream_msg;
......
...@@ -57,6 +57,23 @@ result<message> drop(scheduled_actor*, message_view&) { ...@@ -57,6 +57,23 @@ result<message> drop(scheduled_actor*, message_view&) {
return sec::unexpected_message; return sec::unexpected_message;
} }
// -- implementation details ---------------------------------------------------
namespace {
template <class T>
void silently_ignore(scheduled_actor*, T&) {
// nop
}
result<message> drop_after_quit(scheduled_actor* self, message_view&) {
if (self->current_message_id().is_request())
return make_error(sec::request_receiver_down);
return make_message();
}
} // namespace
// -- static helper functions -------------------------------------------------- // -- static helper functions --------------------------------------------------
void scheduled_actor::default_error_handler(scheduled_actor* ptr, error& x) { void scheduled_actor::default_error_handler(scheduled_actor* ptr, error& x) {
...@@ -210,17 +227,10 @@ bool scheduled_actor::cleanup(error&& fail_state, execution_unit* host) { ...@@ -210,17 +227,10 @@ bool scheduled_actor::cleanup(error&& fail_state, execution_unit* host) {
awaited_responses_.clear(); awaited_responses_.clear();
multiplexed_responses_.clear(); multiplexed_responses_.clear();
// Clear state for open streams. // Clear state for open streams.
if (fail_state == none) { for (auto& kvp : stream_managers_)
for (auto& kvp : stream_managers_) kvp.second->stop(fail_state);
kvp.second->stop(); for (auto& kvp : pending_stream_managers_)
for (auto& kvp : pending_stream_managers_) kvp.second->stop(fail_state);
kvp.second->stop();
} else {
for (auto& kvp : stream_managers_)
kvp.second->abort(fail_state);
for (auto& kvp : pending_stream_managers_)
kvp.second->abort(fail_state);
}
stream_managers_.clear(); stream_managers_.clear();
pending_stream_managers_.clear(); pending_stream_managers_.clear();
get_downstream_queue().cleanup(); get_downstream_queue().cleanup();
...@@ -334,7 +344,7 @@ operator()(size_t, downstream_queue& qs, stream_slot, ...@@ -334,7 +344,7 @@ operator()(size_t, downstream_queue& qs, stream_slot,
CAF_ASSERT(x.content().type_token() == make_type_token<downstream_msg>()); CAF_ASSERT(x.content().type_token() == make_type_token<downstream_msg>());
auto& dm = x.content().get_mutable_as<downstream_msg>(0); auto& dm = x.content().get_mutable_as<downstream_msg>(0);
downstream_msg_visitor f{self, qs, q, dm}; downstream_msg_visitor f{self, qs, q, dm};
auto res = visit(f, dm.content); auto res = visit(f, dm.content);
return ++handled_msgs < max_throughput ? res return ++handled_msgs < max_throughput ? res
: intrusive::task_result::stop_all; : intrusive::task_result::stop_all;
} }
...@@ -415,8 +425,40 @@ proxy_registry* scheduled_actor::proxy_registry_ptr() { ...@@ -415,8 +425,40 @@ proxy_registry* scheduled_actor::proxy_registry_ptr() {
void scheduled_actor::quit(error x) { void scheduled_actor::quit(error x) {
CAF_LOG_TRACE(CAF_ARG(x)); CAF_LOG_TRACE(CAF_ARG(x));
// Make sure repeated calls to quit don't do anything.
if (getf(is_shutting_down_flag))
return;
// Mark this actor as about-to-die.
setf(is_shutting_down_flag);
// Store shutdown reason.
fail_state_ = std::move(x); fail_state_ = std::move(x);
setf(is_terminated_flag); // Clear state for handling regular messages.
bhvr_stack_.clear();
awaited_responses_.clear();
multiplexed_responses_.clear();
// Ignore future exit, down and error messages.
set_exit_handler(silently_ignore<exit_msg>);
set_down_handler(silently_ignore<down_msg>);
set_error_handler(silently_ignore<error>);
// Drop future messages and produce sec::request_receiver_down for requests.
set_default_handler(drop_after_quit);
// Tell all streams to shut down.
std::vector<stream_manager_ptr> managers;
for (auto& smm : {stream_managers_, pending_stream_managers_})
for (auto& kvp : smm)
managers.emplace_back(kvp.second);
// Make sure we shutdown each manager exactly once.
std::sort(managers.begin(), managers.end());
auto e = std::unique(managers.begin(), managers.end());
for (auto i = managers.begin(); i != e; ++i) {
auto& mgr = *i;
mgr->shutdown();
// Managers can become done after calling quit if they were continous.
if (mgr->done()) {
mgr->stop();
erase_stream_manager(mgr);
}
}
} }
// -- timeout management ------------------------------------------------------- // -- timeout management -------------------------------------------------------
...@@ -542,11 +584,10 @@ scheduled_actor::categorize(mailbox_element& x) { ...@@ -542,11 +584,10 @@ scheduled_actor::categorize(mailbox_element& x) {
// make sure to get rid of attachables if they're no longer needed // make sure to get rid of attachables if they're no longer needed
unlink_from(em.source); unlink_from(em.source);
// exit_reason::kill is always fatal // exit_reason::kill is always fatal
if (em.reason == exit_reason::kill) { if (em.reason == exit_reason::kill)
quit(std::move(em.reason)); quit(std::move(em.reason));
} else { else
call_handler(exit_handler_, this, em); call_handler(exit_handler_, this, em);
}
return message_category::internal; return message_category::internal;
} }
case make_type_token<down_msg>(): { case make_type_token<down_msg>(): {
...@@ -683,10 +724,8 @@ bool scheduled_actor::activate(execution_unit* ctx) { ...@@ -683,10 +724,8 @@ bool scheduled_actor::activate(execution_unit* ctx) {
CAF_ASSERT(ctx != nullptr); CAF_ASSERT(ctx != nullptr);
CAF_ASSERT(!getf(is_blocking_flag)); CAF_ASSERT(!getf(is_blocking_flag));
context(ctx); context(ctx);
if (getf(is_initialized_flag) && (!alive() || getf(is_terminated_flag))) { if (getf(is_initialized_flag) && !alive()) {
CAF_LOG_DEBUG_IF(!alive(), "resume called on an actor without behavior"); CAF_LOG_ERROR("activate called on a terminated actor");
CAF_LOG_DEBUG_IF(getf(is_terminated_flag),
"resume called on a terminated actor");
return false; return false;
} }
# ifndef CAF_NO_EXCEPTIONS # ifndef CAF_NO_EXCEPTIONS
...@@ -695,7 +734,7 @@ bool scheduled_actor::activate(execution_unit* ctx) { ...@@ -695,7 +734,7 @@ bool scheduled_actor::activate(execution_unit* ctx) {
if (!getf(is_initialized_flag)) { if (!getf(is_initialized_flag)) {
initialize(); initialize();
if (finalize()) { if (finalize()) {
CAF_LOG_DEBUG("actor_done() returned true right after make_behavior()"); CAF_LOG_DEBUG("finalize() returned true right after make_behavior()");
return false; return false;
} }
CAF_LOG_DEBUG("initialized actor:" << CAF_ARG(name())); CAF_LOG_DEBUG("initialized actor:" << CAF_ARG(name()));
...@@ -763,7 +802,7 @@ auto scheduled_actor::reactivate(mailbox_element& x) -> activation_result { ...@@ -763,7 +802,7 @@ auto scheduled_actor::reactivate(mailbox_element& x) -> activation_result {
// -- behavior management ---------------------------------------------------- // -- behavior management ----------------------------------------------------
void scheduled_actor::do_become(behavior bhvr, bool discard_old) { void scheduled_actor::do_become(behavior bhvr, bool discard_old) {
if (getf(is_terminated_flag)) { if (getf(is_terminated_flag | is_shutting_down_flag)) {
CAF_LOG_WARNING("called become() on a terminated actor"); CAF_LOG_WARNING("called become() on a terminated actor");
return; return;
} }
...@@ -782,11 +821,12 @@ bool scheduled_actor::finalize() { ...@@ -782,11 +821,12 @@ bool scheduled_actor::finalize() {
return true; return true;
// An actor is considered alive as long as it has a behavior and didn't set // An actor is considered alive as long as it has a behavior and didn't set
// the terminated flag. // the terminated flag.
if (alive() && !getf(is_terminated_flag)) if (alive())
return false; return false;
CAF_LOG_DEBUG("actor either has no behavior or has set an exit reason"); setf(is_terminated_flag);
CAF_LOG_DEBUG("actor has no behavior and is ready for cleanup");
CAF_ASSERT(!has_behavior());
on_exit(); on_exit();
bhvr_stack_.clear();
bhvr_stack_.cleanup(); bhvr_stack_.cleanup();
cleanup(std::move(fail_state_), context()); cleanup(std::move(fail_state_), context());
CAF_ASSERT(getf(is_cleaned_up_flag)); CAF_ASSERT(getf(is_cleaned_up_flag));
......
...@@ -39,7 +39,7 @@ stream_manager::stream_manager(scheduled_actor* selfptr, stream_priority prio) ...@@ -39,7 +39,7 @@ stream_manager::stream_manager(scheduled_actor* selfptr, stream_priority prio)
: self_(selfptr), : self_(selfptr),
pending_handshakes_(0), pending_handshakes_(0),
priority_(prio), priority_(prio),
continuous_(false) { flags_(0) {
// nop // nop
} }
...@@ -60,7 +60,7 @@ void stream_manager::handle(inbound_path* in, downstream_msg::forced_close& x) { ...@@ -60,7 +60,7 @@ void stream_manager::handle(inbound_path* in, downstream_msg::forced_close& x) {
CAF_LOG_TRACE(CAF_ARG2("slots", in->slots) << CAF_ARG(x)); CAF_LOG_TRACE(CAF_ARG2("slots", in->slots) << CAF_ARG(x));
// Reset the actor handle to make sure no further message travels upstream. // Reset the actor handle to make sure no further message travels upstream.
in->hdl = nullptr; in->hdl = nullptr;
abort(std::move(x.reason)); stop(std::move(x.reason));
} }
bool stream_manager::handle(stream_slots slots, upstream_msg::ack_open& x) { bool stream_manager::handle(stream_slots slots, upstream_msg::ack_open& x) {
...@@ -104,21 +104,31 @@ void stream_manager::handle(stream_slots slots, upstream_msg::ack_batch& x) { ...@@ -104,21 +104,31 @@ void stream_manager::handle(stream_slots slots, upstream_msg::ack_batch& x) {
} }
void stream_manager::handle(stream_slots slots, upstream_msg::drop&) { void stream_manager::handle(stream_slots slots, upstream_msg::drop&) {
error tmp; out().close(slots.receiver);
out().remove_path(slots.receiver, std::move(tmp), true);
} }
void stream_manager::handle(stream_slots slots, upstream_msg::forced_drop& x) { void stream_manager::handle(stream_slots slots, upstream_msg::forced_drop& x) {
if (out().remove_path(slots.receiver, x.reason, true)) if (out().remove_path(slots.receiver, x.reason, true))
abort(std::move(x.reason)); stop(std::move(x.reason));
} }
void stream_manager::stop() { void stream_manager::stop(error reason) {
if (reason)
out().abort(reason);
else
out().close();
finalize(reason);
self_->erase_inbound_paths_later(this, std::move(reason));
}
void stream_manager::shutdown() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
out().close(); // Mark as shutting down and reset other flags.
error tmp; if (shutting_down())
finalize(tmp); return;
self_->erase_inbound_paths_later(this); flags_ = is_shutting_down_flag;
for (auto ipath : inbound_paths_)
ipath->emit_regular_shutdown(self_);
} }
void stream_manager::advance() { void stream_manager::advance() {
...@@ -144,13 +154,6 @@ void stream_manager::advance() { ...@@ -144,13 +154,6 @@ void stream_manager::advance() {
push(); push();
} }
void stream_manager::abort(error reason) {
CAF_LOG_TRACE(CAF_ARG(reason));
out().abort(reason);
finalize(reason);
self_->erase_inbound_paths_later(this, std::move(reason));
}
void stream_manager::push() { void stream_manager::push() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
do { do {
......
...@@ -371,8 +371,10 @@ public: ...@@ -371,8 +371,10 @@ public:
} }
} }
void erase_inbound_paths_later(const stream_manager*, error) override { void erase_inbound_paths_later(const stream_manager* mgr,
CAF_FAIL("unexpected function call"); error err) override {
CAF_REQUIRE_EQUAL(err, none);
erase_inbound_paths_later(mgr);
} }
time_point now() { time_point now() {
......
...@@ -215,6 +215,12 @@ struct fixture : test_coordinator_fixture<> { ...@@ -215,6 +215,12 @@ struct fixture : test_coordinator_fixture<> {
void tick() { void tick() {
advance_time(cfg.stream_credit_round_interval); advance_time(cfg.stream_credit_round_interval);
} }
/// Simulate a hard error on an actor such as an uncaught exception or a
/// disconnect from a remote actor.
void hard_kill(const actor& x) {
deref(x).cleanup(exit_reason::kill, nullptr);
}
}; };
} // namespace <anonymous> } // namespace <anonymous>
...@@ -339,10 +345,8 @@ CAF_TEST(depth_2_pipeline_error_at_source) { ...@@ -339,10 +345,8 @@ CAF_TEST(depth_2_pipeline_error_at_source) {
expect((open_stream_msg), from(self).to(snk)); expect((open_stream_msg), from(self).to(snk));
expect((upstream_msg::ack_open), from(snk).to(src)); expect((upstream_msg::ack_open), from(snk).to(src));
CAF_MESSAGE("start data transmission (and abort source)"); CAF_MESSAGE("start data transmission (and abort source)");
self->send_exit(src, exit_reason::kill); hard_kill(src);
expect((downstream_msg::batch), from(src).to(snk)); expect((downstream_msg::batch), from(src).to(snk));
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(_).to(snk)); expect((downstream_msg::forced_close), from(_).to(snk));
} }
...@@ -356,10 +360,8 @@ CAF_TEST(depth_2_pipelin_error_at_sink) { ...@@ -356,10 +360,8 @@ CAF_TEST(depth_2_pipelin_error_at_sink) {
expect((string), from(self).to(src).with("numbers.txt")); expect((string), from(self).to(src).with("numbers.txt"));
expect((open_stream_msg), from(self).to(snk)); expect((open_stream_msg), from(self).to(snk));
CAF_MESSAGE("start data transmission (and abort sink)"); CAF_MESSAGE("start data transmission (and abort sink)");
self->send_exit(snk, exit_reason::kill); hard_kill(snk);
expect((upstream_msg::ack_open), from(snk).to(src)); 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(_).to(src)); expect((upstream_msg::forced_drop), from(_).to(src));
} }
......
...@@ -356,7 +356,7 @@ void middleman::stop() { ...@@ -356,7 +356,7 @@ void middleman::stop() {
auto ptr = static_cast<broker*>(actor_cast<abstract_actor*>(hdl)); auto ptr = static_cast<broker*>(actor_cast<abstract_actor*>(hdl));
if (!ptr->getf(abstract_actor::is_terminated_flag)) { if (!ptr->getf(abstract_actor::is_terminated_flag)) {
ptr->context(&backend()); ptr->context(&backend());
ptr->setf(abstract_actor::is_terminated_flag); ptr->quit();
ptr->finalize(); ptr->finalize();
} }
} }
......
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