Unverified Commit 5042ac89 authored by Dominik Charousset's avatar Dominik Charousset Committed by GitHub

Merge pull request #760

Close #753
parents b055e589 600bc956
...@@ -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();
......
...@@ -52,6 +52,11 @@ public: ...@@ -52,6 +52,11 @@ public:
// -- implementation of virtual functions ------------------------------------ // -- implementation of virtual functions ------------------------------------
void shutdown() override {
super::shutdown();
at_end_ = true;
}
bool done() const override { bool done() const override {
return this->pending_handshakes_ == 0 && at_end_ && this->out_.clean(); return this->pending_handshakes_ == 0 && at_end_ && this->out_.clean();
} }
......
...@@ -690,19 +690,14 @@ public: ...@@ -690,19 +690,14 @@ public:
// -- behavior management ---------------------------------------------------- // -- behavior management ----------------------------------------------------
/// Returns whether `true` if the behavior stack is not empty or /// Returns `true` if the behavior stack is not empty.
/// if outstanding responses exist, `false` otherwise. inline bool has_behavior() const noexcept {
inline bool has_behavior() const { return !bhvr_stack_.empty();
return !bhvr_stack_.empty()
|| !awaited_responses_.empty()
|| !multiplexed_responses_.empty()
|| !stream_managers_.empty()
|| !pending_stream_managers_.empty();
} }
inline behavior& current_behavior() { inline behavior& current_behavior() {
return !awaited_responses_.empty() ? awaited_responses_.front().second return !awaited_responses_.empty() ? awaited_responses_.front().second
: bhvr_stack_.back(); : bhvr_stack_.back();
} }
/// Installs a new behavior without performing any type checks. /// Installs a new behavior without performing any type checks.
...@@ -769,7 +764,7 @@ public: ...@@ -769,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;
} }
...@@ -783,13 +778,15 @@ public: ...@@ -783,13 +778,15 @@ public:
return; return;
} }
CAF_ASSERT(i->second != nullptr); CAF_ASSERT(i->second != nullptr);
i->second->handle(slots, x); auto ptr = i->second;
if (i->second->done()) { ptr->handle(slots, x);
CAF_LOG_INFO("done sending:" << CAF_ARG(slots)); if (ptr->done()) {
i->second->stop(); CAF_LOG_DEBUG("done sending:" << CAF_ARG(slots));
stream_managers_.erase(i); ptr->stop();
if (stream_managers_.empty()) erase_stream_manager(ptr);
stream_ticks_.stop(); } else if (ptr->out().path(slots.receiver) == nullptr) {
CAF_LOG_DEBUG("done sending on path:" << CAF_ARG(slots.receiver));
erase_stream_manager(slots.receiver);
} }
} }
...@@ -871,6 +868,19 @@ public: ...@@ -871,6 +868,19 @@ public:
/// this function again. /// this function again.
actor_clock::time_point advance_streams(actor_clock::time_point now); actor_clock::time_point advance_streams(actor_clock::time_point now);
// -- properties -------------------------------------------------------------
/// Returns `true` if the actor has a behavior, awaits responses, or
/// participates in streams.
/// @private
inline bool alive() const noexcept {
return !bhvr_stack_.empty()
|| !awaited_responses_.empty()
|| !multiplexed_responses_.empty()
|| !stream_managers_.empty()
|| !pending_stream_managers_.empty();
}
/// @endcond /// @endcond
protected: protected:
......
...@@ -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`.
......
...@@ -73,7 +73,7 @@ public: ...@@ -73,7 +73,7 @@ public:
this->setf(abstract_actor::is_initialized_flag); this->setf(abstract_actor::is_initialized_flag);
auto bhvr = make_behavior(); auto bhvr = make_behavior();
CAF_LOG_DEBUG_IF(!bhvr, "make_behavior() did not return a behavior:" CAF_LOG_DEBUG_IF(!bhvr, "make_behavior() did not return a behavior:"
<< CAF_ARG(this->has_behavior())); << CAF_ARG2("alive", this->alive()));
if (bhvr) { if (bhvr) {
// make_behavior() did return a behavior instead of using become() // make_behavior() did return a behavior instead of using become()
CAF_LOG_DEBUG("make_behavior() did return a valid behavior"); CAF_LOG_DEBUG("make_behavior() did return a valid behavior");
......
...@@ -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;
......
...@@ -38,7 +38,7 @@ void event_based_actor::initialize() { ...@@ -38,7 +38,7 @@ void event_based_actor::initialize() {
setf(is_initialized_flag); setf(is_initialized_flag);
auto bhvr = make_behavior(); auto bhvr = make_behavior();
CAF_LOG_DEBUG_IF(!bhvr, "make_behavior() did not return a behavior:" CAF_LOG_DEBUG_IF(!bhvr, "make_behavior() did not return a behavior:"
<< CAF_ARG(has_behavior())); << CAF_ARG2("alive", alive()));
if (bhvr) { if (bhvr) {
// make_behavior() did return a behavior instead of using become() // make_behavior() did return a behavior instead of using become()
CAF_LOG_DEBUG("make_behavior() did return a valid behavior"); CAF_LOG_DEBUG("make_behavior() did return a valid behavior");
......
...@@ -348,13 +348,7 @@ void logger::render_time_diff(std::ostream& out, timestamp t0, timestamp tn) { ...@@ -348,13 +348,7 @@ void logger::render_time_diff(std::ostream& out, timestamp t0, timestamp tn) {
} }
void logger::render_date(std::ostream& out, timestamp x) { void logger::render_date(std::ostream& out, timestamp x) {
auto y = std::chrono::time_point_cast<timestamp::clock::duration>(x); out << deep_to_string(x);
auto z = timestamp::clock::to_time_t(y);
// strftime workaround: std::put_time not available on GCC 4.8
// out << std::put_time(std::localtime(&z), "%F %T");
char buf[50];
if (strftime(buf, sizeof(buf), "%Y-%m-%d %H:%M:%S", std::localtime(&z)))
out << buf;
} }
void logger::render(std::ostream& out, const line_format& lf, void logger::render(std::ostream& out, const line_format& lf,
......
...@@ -48,9 +48,10 @@ void monitorable_actor::attach(attachable_ptr ptr) { ...@@ -48,9 +48,10 @@ void monitorable_actor::attach(attachable_ptr ptr) {
attach_impl(ptr); attach_impl(ptr);
return true; return true;
}); });
CAF_LOG_DEBUG("cannot attach functor to terminated actor: call immediately"); if (!attached) {
if (!attached) CAF_LOG_DEBUG("cannot attach functor to terminated actor: call immediately");
ptr->actor_exited(fail_state, nullptr); ptr->actor_exited(fail_state, nullptr);
}
} }
size_t monitorable_actor::detach(const attachable::token& what) { size_t monitorable_actor::detach(const attachable::token& what) {
......
...@@ -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();
...@@ -274,7 +284,8 @@ operator()(size_t, upstream_queue&, mailbox_element& x) { ...@@ -274,7 +284,8 @@ operator()(size_t, upstream_queue&, mailbox_element& x) {
auto& um = x.content().get_mutable_as<upstream_msg>(0); auto& um = x.content().get_mutable_as<upstream_msg>(0);
upstream_msg_visitor f{self, um}; upstream_msg_visitor f{self, um};
visit(f, um.content); visit(f, um.content);
return intrusive::task_result::resume; return ++handled_msgs < max_throughput ? intrusive::task_result::resume
: intrusive::task_result::stop_all;
} }
namespace { namespace {
...@@ -333,7 +344,9 @@ operator()(size_t, downstream_queue& qs, stream_slot, ...@@ -333,7 +344,9 @@ 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};
return visit(f, dm.content); auto res = visit(f, dm.content);
return ++handled_msgs < max_throughput ? res
: intrusive::task_result::stop_all;
} }
intrusive::task_result intrusive::task_result
...@@ -412,8 +425,40 @@ proxy_registry* scheduled_actor::proxy_registry_ptr() { ...@@ -412,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 -------------------------------------------------------
...@@ -539,11 +584,10 @@ scheduled_actor::categorize(mailbox_element& x) { ...@@ -539,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>(): {
...@@ -680,12 +724,8 @@ bool scheduled_actor::activate(execution_unit* ctx) { ...@@ -680,12 +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) if (getf(is_initialized_flag) && !alive()) {
&& (!has_behavior() || getf(is_terminated_flag))) { CAF_LOG_ERROR("activate called on a terminated actor");
CAF_LOG_DEBUG_IF(!has_behavior(),
"resume called on an actor without behavior");
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
...@@ -694,7 +734,7 @@ bool scheduled_actor::activate(execution_unit* ctx) { ...@@ -694,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()));
...@@ -762,6 +802,10 @@ auto scheduled_actor::reactivate(mailbox_element& x) -> activation_result { ...@@ -762,6 +802,10 @@ 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 | is_shutting_down_flag)) {
CAF_LOG_WARNING("called become() on a terminated actor");
return;
}
if (discard_old && !bhvr_stack_.empty()) if (discard_old && !bhvr_stack_.empty())
bhvr_stack_.pop_back(); bhvr_stack_.pop_back();
// request_timeout simply resets the timeout when it's invalid // request_timeout simply resets the timeout when it's invalid
...@@ -777,11 +821,11 @@ bool scheduled_actor::finalize() { ...@@ -777,11 +821,11 @@ 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 (has_behavior() && !getf(is_terminated_flag)) if (alive())
return false; return false;
CAF_LOG_DEBUG("actor either has no behavior or has set an exit reason"); 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));
...@@ -973,6 +1017,7 @@ void scheduled_actor::erase_stream_manager(stream_slot id) { ...@@ -973,6 +1017,7 @@ void scheduled_actor::erase_stream_manager(stream_slot id) {
CAF_LOG_TRACE(CAF_ARG(id)); CAF_LOG_TRACE(CAF_ARG(id));
if (stream_managers_.erase(id) != 0 && stream_managers_.empty()) if (stream_managers_.erase(id) != 0 && stream_managers_.empty())
stream_ticks_.stop(); stream_ticks_.stop();
CAF_LOG_DEBUG(CAF_ARG2("stream_managers_.size", stream_managers_.size()));
} }
void scheduled_actor::erase_pending_stream_manager(stream_slot id) { void scheduled_actor::erase_pending_stream_manager(stream_slot id) {
...@@ -981,7 +1026,8 @@ void scheduled_actor::erase_pending_stream_manager(stream_slot id) { ...@@ -981,7 +1026,8 @@ void scheduled_actor::erase_pending_stream_manager(stream_slot id) {
} }
void scheduled_actor::erase_stream_manager(const stream_manager_ptr& mgr) { void scheduled_actor::erase_stream_manager(const stream_manager_ptr& mgr) {
{ // Lifetime scope of first iterator pair. CAF_LOG_TRACE("");
if (!stream_managers_.empty()) {
auto i = stream_managers_.begin(); auto i = stream_managers_.begin();
auto e = stream_managers_.end(); auto e = stream_managers_.end();
while (i != e) while (i != e)
...@@ -989,6 +1035,8 @@ void scheduled_actor::erase_stream_manager(const stream_manager_ptr& mgr) { ...@@ -989,6 +1035,8 @@ void scheduled_actor::erase_stream_manager(const stream_manager_ptr& mgr) {
i = stream_managers_.erase(i); i = stream_managers_.erase(i);
else else
++i; ++i;
if (stream_managers_.empty())
stream_ticks_.stop();
} }
{ // Lifetime scope of second iterator pair. { // Lifetime scope of second iterator pair.
auto i = pending_stream_managers_.begin(); auto i = pending_stream_managers_.begin();
...@@ -999,8 +1047,9 @@ void scheduled_actor::erase_stream_manager(const stream_manager_ptr& mgr) { ...@@ -999,8 +1047,9 @@ void scheduled_actor::erase_stream_manager(const stream_manager_ptr& mgr) {
else else
++i; ++i;
} }
if (stream_managers_.empty()) CAF_LOG_DEBUG(CAF_ARG2("stream_managers_.size", stream_managers_.size())
stream_ticks_.stop(); << CAF_ARG2("pending_stream_managers_.size",
pending_stream_managers_.size()));
} }
invoke_message_result invoke_message_result
......
...@@ -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,34 @@ void stream_manager::handle(stream_slots slots, upstream_msg::ack_batch& x) { ...@@ -104,21 +104,34 @@ 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;
CAF_LOG_DEBUG("emit shutdown messages on" << inbound_paths_.size()
<< "inbound paths;" << CAF_ARG2("out.clean", out().clean())
<< CAF_ARG2("out.paths", out().num_paths()));
for (auto ipath : inbound_paths_)
ipath->emit_regular_shutdown(self_);
} }
void stream_manager::advance() { void stream_manager::advance() {
...@@ -144,13 +157,6 @@ void stream_manager::advance() { ...@@ -144,13 +157,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 {
......
...@@ -97,7 +97,10 @@ void stringification_inspector::consume(timestamp& x) { ...@@ -97,7 +97,10 @@ void stringification_inspector::consume(timestamp& x) {
// time_t has no milliseconds, so we need to insert them manually. // time_t has no milliseconds, so we need to insert them manually.
auto ms = (x.time_since_epoch().count() / 1000000) % 1000; auto ms = (x.time_since_epoch().count() / 1000000) % 1000;
result_ += '.'; result_ += '.';
result_ += std::to_string(ms); auto frac = std::to_string(ms);
if (frac.size() < 3)
frac.insert(0, 3 - frac.size(), '0');
result_ += frac;
} }
} // namespace detail } // namespace detail
......
...@@ -106,7 +106,7 @@ CAF_TEST(rendering) { ...@@ -106,7 +106,7 @@ CAF_TEST(rendering) {
time_t t0_t = 0; time_t t0_t = 0;
char t0_buf[50]; char t0_buf[50];
CAF_REQUIRE(strftime(t0_buf, sizeof(t0_buf), CAF_REQUIRE(strftime(t0_buf, sizeof(t0_buf),
"%Y-%m-%d %H:%M:%S", localtime(&t0_t))); "%Y-%m-%dT%H:%M:%S.000", localtime(&t0_t)));
CAF_CHECK_EQUAL(render(logger::render_date, t0), t0_buf); CAF_CHECK_EQUAL(render(logger::render_date, t0), t0_buf);
// Rendering of events. // Rendering of events.
logger::event e{ logger::event e{
......
...@@ -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() {
......
...@@ -74,6 +74,28 @@ std::function<void (T&, const error&)> fin(scheduled_actor* self) { ...@@ -74,6 +74,28 @@ std::function<void (T&, const error&)> fin(scheduled_actor* self) {
}; };
} }
TESTEE(infinite_source) {
return {
[=](string& fname) -> output_stream<int> {
CAF_CHECK_EQUAL(fname, "numbers.txt");
CAF_CHECK_EQUAL(self->mailbox().empty(), true);
return self->make_source(
[](int& x) {
x = 0;
},
[](int& x, downstream<int>& out, size_t num) {
for (size_t i = 0; i < num; ++i)
out.push(x++);
},
[](const int&) {
return false;
},
fin<int>(self)
);
}
};
}
VARARGS_TESTEE(file_reader, size_t buf_size) { VARARGS_TESTEE(file_reader, size_t buf_size) {
return { return {
[=](string& fname) -> output_stream<int> { [=](string& fname) -> output_stream<int> {
...@@ -215,6 +237,12 @@ struct fixture : test_coordinator_fixture<> { ...@@ -215,6 +237,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 +367,8 @@ CAF_TEST(depth_2_pipeline_error_at_source) { ...@@ -339,10 +367,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 +382,8 @@ CAF_TEST(depth_2_pipelin_error_at_sink) { ...@@ -356,10 +382,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));
} }
...@@ -419,4 +443,41 @@ CAF_TEST(depth_4_pipeline_500_items) { ...@@ -419,4 +443,41 @@ CAF_TEST(depth_4_pipeline_500_items) {
CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.x, 125000); CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.x, 125000);
} }
CAF_TEST(depth_3_pipeline_graceful_shutdown) {
auto src = sys.spawn(file_reader, 50u);
auto stg = sys.spawn(filter);
auto snk = sys.spawn(sum_up);
CAF_MESSAGE(CAF_ARG(self) << CAF_ARG(src) << CAF_ARG(stg) << CAF_ARG(snk));
CAF_MESSAGE("initiate stream handshake");
self->send(snk * stg * src, "numbers.txt");
expect((string), from(self).to(src).with("numbers.txt"));
expect((open_stream_msg), from(self).to(stg));
expect((open_stream_msg), from(self).to(snk));
expect((upstream_msg::ack_open), from(snk).to(stg));
expect((upstream_msg::ack_open), from(stg).to(src));
CAF_MESSAGE("start data transmission (a single batch) and stop the stage");
anon_send_exit(stg, exit_reason::user_shutdown);
CAF_MESSAGE("expect the stage to still transfer pending items to the sink");
run();
CAF_MESSAGE("check sink result");
CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.x, 625);
}
CAF_TEST(depth_3_pipeline_infinite_source) {
auto src = sys.spawn(infinite_source);
auto stg = sys.spawn(filter);
auto snk = sys.spawn(sum_up);
CAF_MESSAGE(CAF_ARG(self) << CAF_ARG(src) << CAF_ARG(stg) << CAF_ARG(snk));
CAF_MESSAGE("initiate stream handshake");
self->send(snk * stg * src, "numbers.txt");
expect((string), from(self).to(src).with("numbers.txt"));
expect((open_stream_msg), from(self).to(stg));
expect((open_stream_msg), from(self).to(snk));
expect((upstream_msg::ack_open), from(snk).to(stg));
expect((upstream_msg::ack_open), from(stg).to(src));
CAF_MESSAGE("send exit to the source and expect the stream to terminate");
anon_send_exit(src, exit_reason::user_shutdown);
run();
}
CAF_TEST_FIXTURE_SCOPE_END() CAF_TEST_FIXTURE_SCOPE_END()
...@@ -92,7 +92,8 @@ protected: ...@@ -92,7 +92,8 @@ protected:
invoke_mailbox_element_impl(ctx, value_); invoke_mailbox_element_impl(ctx, value_);
// only consume an activity token if actor did not produce them now // only consume an activity token if actor did not produce them now
if (prev && activity_tokens_ && --(*activity_tokens_) == 0) { if (prev && activity_tokens_ && --(*activity_tokens_) == 0) {
if (this->parent()->getf(abstract_actor::is_terminated_flag)) if (this->parent()->getf(abstract_actor::is_shutting_down_flag
| abstract_actor::is_terminated_flag))
return false; return false;
// tell broker it entered passive mode, this can result in // tell broker it entered passive mode, this can result in
// producing, why we check the condition again afterwards // producing, why we check the condition again afterwards
......
...@@ -96,7 +96,7 @@ public: ...@@ -96,7 +96,7 @@ public:
this->init_broker(); this->init_broker();
auto bhvr = make_behavior(); auto bhvr = make_behavior();
CAF_LOG_DEBUG_IF(!bhvr, "make_behavior() did not return a behavior:" CAF_LOG_DEBUG_IF(!bhvr, "make_behavior() did not return a behavior:"
<< CAF_ARG(this->has_behavior())); << CAF_ARG2("alive", this->alive()));
if (bhvr) { if (bhvr) {
// make_behavior() did return a behavior instead of using become() // make_behavior() did return a behavior instead of using become()
CAF_LOG_DEBUG("make_behavior() did return a valid behavior"); CAF_LOG_DEBUG("make_behavior() did return a valid behavior");
......
...@@ -37,7 +37,7 @@ void broker::initialize() { ...@@ -37,7 +37,7 @@ void broker::initialize() {
init_broker(); init_broker();
auto bhvr = make_behavior(); auto bhvr = make_behavior();
CAF_LOG_DEBUG_IF(!bhvr, "make_behavior() did not return a behavior:" CAF_LOG_DEBUG_IF(!bhvr, "make_behavior() did not return a behavior:"
<< CAF_ARG(has_behavior())); << CAF_ARG2("alive", alive()));
if (bhvr) { if (bhvr) {
// make_behavior() did return a behavior instead of using become() // make_behavior() did return a behavior instead of using become()
CAF_LOG_DEBUG("make_behavior() did return a valid behavior"); CAF_LOG_DEBUG("make_behavior() did return a valid behavior");
......
...@@ -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