Commit 175574fa authored by Dominik Charousset's avatar Dominik Charousset

Implement error handling for failing sources

parent 77f06bcd
......@@ -69,9 +69,10 @@ public:
}
protected:
void about_to_erase(typename super::map_type::iterator i, bool,
error*) override {
void about_to_erase(typename super::map_type::iterator i, bool silent,
error* reason) override {
caches_.erase(i->second->slots);
super::about_to_erase(i, silent, reason);
}
private:
......
......@@ -114,11 +114,20 @@ public:
~inbound_path();
/// Updates `last_batch_id` and `assigned_credit`.
//void handle_batch(long batch_size, int64_t batch_id);
/// Updates `last_batch_id` and `assigned_credit` before calling
/// `mgr->handle(this, x)`.
void handle(downstream_msg::batch& x);
/// Calls `mgr->handle(this, x)`.
inline void handle(downstream_msg::close& x) {
mgr->handle(this, x);
}
/// Calls `mgr->handle(this, x)`.
inline void handle(downstream_msg::forced_close& x) {
mgr->handle(this, x);
}
/// Emits a `stream_msg::ack_batch` on this path and sets `assigned_credit`
/// to `initial_demand`.
void emit_ack_open(local_actor* self, actor_addr rebind_from);
......
......@@ -130,14 +130,20 @@ public:
}
}
}
cleanup();
return {result, stopped};
}
/// Erases all keys previously marked via `erase_later`.
void cleanup() {
if (!erase_list_.empty()) {
for (auto& k : erase_list_)
qs_.erase(k);
erase_list_.clear();
}
return {result, stopped};
}
/// Marks the key `k` for erasing from the map later.
void erase_later(key_type k) {
erase_list_.emplace_back(std::move(k));
}
......
......@@ -948,6 +948,18 @@ public:
return mailbox_;
}
/// @private
default_queue& get_default_queue();
/// @private
upstream_queue& get_upstream_queue();
/// @private
downstream_queue& get_downstream_queue();
/// @private
urgent_queue& get_urgent_queue();
// -- inbound_path management ------------------------------------------------
inbound_path* make_inbound_path(stream_manager_ptr mgr, stream_slots slots,
......@@ -960,7 +972,7 @@ public:
void erase_inbound_paths_later(const stream_manager* mgr,
error reason) override;
// -- handling of upstream message -------------------------------------------
// -- handling of stream message ---------------------------------------------
void handle_upstream_msg(stream_slots slots, actor_addr& sender,
upstream_msg::ack_open& x);
......@@ -988,12 +1000,11 @@ public:
i->second->handle(slots, x);
if (i->second->done()) {
CAF_LOG_INFO("done sending:" << CAF_ARG(slots));
i->second->close();
i->second->stop();
stream_managers_.erase(i);
}
}
protected:
/// @cond PRIVATE
// -- utility functions for invoking default handler -------------------------
......@@ -1052,6 +1063,7 @@ protected:
/// this function again.
actor_clock::time_point advance_streams(actor_clock::time_point now);
protected:
// -- member variables -------------------------------------------------------
/// Stores incoming messages.
......
......@@ -74,9 +74,9 @@ public:
virtual void handle(stream_slots slots, upstream_msg::forced_drop& x);
/// Closes the stream when the parent terminates with default exit reason or
/// the stream reached its end.
virtual void close();
/// Closes all output and input paths and sends the final result to the
/// client.
virtual void stop();
/// Aborts a stream after any stream message handler returned a non-default
/// constructed error `reason` or the parent actor terminates with a
......@@ -125,9 +125,13 @@ public:
// -- input path management --------------------------------------------------
/// Informs the manager that a new input path opens.
/// @note The lifetime of inbound paths is managed by the downstream queue.
/// This function is called from the constructor of `inbound_path`.
virtual void register_input_path(inbound_path* x);
/// Informs the manager that an input path closes.
/// @note The lifetime of inbound paths is managed by the downstream queue.
/// This function is called from the destructor of `inbound_path`.
virtual void deregister_input_path(inbound_path* x) noexcept;
// -- mutators ---------------------------------------------------------------
......
......@@ -195,26 +195,35 @@ void scheduled_actor::launch(execution_unit* eu, bool lazy, bool hide) {
}
bool scheduled_actor::cleanup(error&& fail_state, execution_unit* host) {
CAF_LOG_TRACE(CAF_ARG(fail_state));
// Shutdown hosting thread when running detached.
if (getf(is_detached_flag)) {
CAF_ASSERT(private_thread_ != nullptr);
private_thread_->shutdown();
}
// Clear all state.
// Clear state for open requests.
awaited_responses_.clear();
multiplexed_responses_.clear();
/*
if (fail_state != none)
for (auto& kvp : streams_)
// Clear state for open streams.
if (fail_state == none) {
for (auto& kvp : stream_managers_)
kvp.second->stop();
for (auto& kvp : pending_stream_managers_)
kvp.second->stop();
} else {
for (auto& kvp : stream_managers_)
kvp.second->abort(fail_state);
else
for (auto& kvp : streams_)
kvp.second->close();
streams_.clear();
*/
for (auto& kvp : pending_stream_managers_)
kvp.second->abort(fail_state);
}
stream_managers_.clear();
pending_stream_managers_.clear();
get_downstream_queue().cleanup();
// Clear mailbox.
if (!mailbox_.closed()) {
mailbox_.close();
// TODO: messages that are stuck in the cache can get lost
get_default_queue().flush_cache();
get_urgent_queue().flush_cache();
detail::sync_request_bouncer bounce{fail_state};
while (mailbox_.queue().new_round(1000, bounce).consumed_items)
; // nop
......@@ -261,67 +270,47 @@ operator()(size_t, upstream_queue&, mailbox_element& x) {
return intrusive::task_result::resume;
}
namespace {
// TODO: replace with generic lambda when switching to C++14
struct downstream_msg_visitor {
scheduled_actor* selfptr;
scheduled_actor::downstream_queue& qs_ref;
policy::downstream_messages::nested_queue_type& q_ref;
downstream_msg& dm;
template <class T>
intrusive::task_result operator()(T& x) {
auto& inptr = q_ref.policy().handler;
if (inptr == nullptr)
return intrusive::task_result::stop;
auto mgr = inptr->mgr;
inptr->handle(x);
// TODO: replace with `if constexpr` when switching to C++17
if (std::is_same<T, downstream_msg::close>::value
|| std::is_same<T, downstream_msg::forced_close>::value) {
inptr.reset();
qs_ref.erase_later(dm.slots.receiver);
}
if (mgr->done()) {
CAF_LOG_DEBUG("path is done receiving and closes its manager");
mgr->stop();
selfptr->remove_stream_manager(dm.slots);
return intrusive::task_result::stop;
}
return intrusive::task_result::resume;
}
};
} // namespace <anonymous>
intrusive::task_result scheduled_actor::mailbox_visitor::
operator()(size_t, downstream_queue& qs, stream_slot,
policy::downstream_messages::nested_queue_type& q,
mailbox_element& x) {
CAF_ASSERT(x.content().type_token() == make_type_token<downstream_msg>());
struct visitor {
scheduled_actor* thisptr;
downstream_queue& qs_ref;
policy::downstream_messages::nested_queue_type& q_ref;
inbound_path* inptr;
downstream_msg& dm;
intrusive::task_result operator()(downstream_msg::batch& y) {
inptr->handle(y);
if (inptr->mgr->done()) {
CAF_LOG_DEBUG("path is done receiving and closes its manager");
inptr->mgr->close();
}
return intrusive::task_result::resume;
}
intrusive::task_result operator()(downstream_msg::close& y) {
CAF_LOG_TRACE(CAF_ARG(y));
auto& managers = thisptr->stream_managers_;
auto slots = dm.slots;
auto i = managers.find(slots);
if (i == managers.end()) {
CAF_LOG_DEBUG("no manager found for dispatching close message:"
<< CAF_ARG(slots));
return intrusive::task_result::resume;
}
i->second->handle(inptr, y);
q_ref.policy().handler.reset();
qs_ref.erase_later(slots.receiver);
if (!i->second->done()) {
CAF_LOG_DEBUG("inbound path closed:" << CAF_ARG(slots));
managers.erase(i);
} else {
CAF_LOG_DEBUG("manager closed its final inbound path:"
<< CAF_ARG(slots));
// Close the manager and remove it on all registered slots.
auto mgr = i->second;
mgr->close();
auto j = managers.begin();
while (j != managers.end()) {
if (j->second == mgr)
j = managers.erase(j);
else
++j;
}
}
return intrusive::task_result::resume;
}
intrusive::task_result operator()(downstream_msg::forced_close&) {
CAF_LOG_ERROR("implement me");
return intrusive::task_result::stop;
}
};
auto inptr = q.policy().handler.get();
if (inptr == nullptr)
return intrusive::task_result::stop;
auto& dm = x.content().get_mutable_as<downstream_msg>(0);
visitor f{self, qs, q, inptr, dm};
downstream_msg_visitor f{self, qs, q, dm};
return visit(f, dm.content);
}
......@@ -808,6 +797,26 @@ void scheduled_actor::push_to_cache(mailbox_element_ptr ptr) {
}
}
scheduled_actor::default_queue& scheduled_actor::get_default_queue() {
constexpr size_t queue_id = mailbox_policy::default_queue_index;
return get<queue_id>(mailbox_.queue().queues());
}
scheduled_actor::upstream_queue& scheduled_actor::get_upstream_queue() {
constexpr size_t queue_id = mailbox_policy::upstream_queue_index;
return get<queue_id>(mailbox_.queue().queues());
}
scheduled_actor::downstream_queue& scheduled_actor::get_downstream_queue() {
constexpr size_t queue_id = mailbox_policy::downstream_queue_index;
return get<queue_id>(mailbox_.queue().queues());
}
scheduled_actor::urgent_queue& scheduled_actor::get_urgent_queue() {
constexpr size_t queue_id = mailbox_policy::urgent_queue_index;
return get<queue_id>(mailbox_.queue().queues());
}
inbound_path* scheduled_actor::make_inbound_path(stream_manager_ptr mgr,
stream_slots slots,
strong_actor_ptr sender) {
......@@ -821,13 +830,13 @@ inbound_path* scheduled_actor::make_inbound_path(stream_manager_ptr mgr,
}
void scheduled_actor::erase_inbound_path_later(stream_slot slot) {
constexpr size_t queue_id = mailbox_policy::downstream_queue_index;
get<queue_id>(mailbox_.queue().queues()).erase_later(slot);
CAF_LOG_TRACE(CAF_ARG(slot));
get_downstream_queue().erase_later(slot);
}
void scheduled_actor::erase_inbound_paths_later(const stream_manager* ptr) {
constexpr size_t queue_id = mailbox_policy::downstream_queue_index;
for (auto& kvp : get<queue_id>(mailbox_.queue().queues()).queues()) {
CAF_LOG_TRACE("");
for (auto& kvp : get_downstream_queue().queues()) {
auto& path = kvp.second.policy().handler;
if (path != nullptr && path->mgr == ptr)
erase_inbound_path_later(kvp.first);
......@@ -836,6 +845,7 @@ void scheduled_actor::erase_inbound_paths_later(const stream_manager* ptr) {
void scheduled_actor::erase_inbound_paths_later(const stream_manager* ptr,
error reason) {
CAF_LOG_TRACE(CAF_ARG(reason));
using fn = void (*)(local_actor*, inbound_path&, error&);
fn regular = [](local_actor* self, inbound_path& in, error&) {
in.emit_regular_shutdown(self);
......@@ -844,8 +854,7 @@ void scheduled_actor::erase_inbound_paths_later(const stream_manager* ptr,
in.emit_irregular_shutdown(self, rsn);
};
auto f = reason == none ? regular : irregular;
static constexpr size_t queue_id = mailbox_policy::downstream_queue_index;
for (auto& kvp : get<queue_id>(mailbox_.queue().queues()).queues()) {
for (auto& kvp : get_downstream_queue().queues()) {
auto& path = kvp.second.policy().handler;
if (path != nullptr && path->mgr == ptr) {
f(this, *path, reason);
......
......@@ -85,7 +85,8 @@ void stream_manager::handle(stream_slots slots, upstream_msg::forced_drop& x) {
abort(std::move(x.reason));
}
void stream_manager::close() {
void stream_manager::stop() {
CAF_LOG_TRACE("");
out().close();
self_->erase_inbound_paths_later(this);
if (!promises_.empty())
......@@ -93,6 +94,7 @@ void stream_manager::close() {
}
void stream_manager::abort(error reason) {
CAF_LOG_TRACE(CAF_ARG(reason));
if (!promises_.empty() || !in_flight_promises_.empty()) {
auto msg = make_message(reason);
deliver_promises(msg);
......@@ -145,10 +147,15 @@ void stream_manager::cycle_timeout(size_t) {
}
void stream_manager::register_input_path(inbound_path* ptr) {
CAF_ASSERT(ptr != nullptr);
CAF_LOG_TRACE(CAF_ARG2("path", *ptr));
inbound_paths_.emplace_back(ptr);
}
void stream_manager::deregister_input_path(inbound_path* ptr) noexcept {
CAF_ASSERT(ptr != nullptr);
CAF_LOG_TRACE(CAF_ARG2("path", *ptr));
CAF_ASSERT(inbound_paths_.size() > 0);
using std::swap;
if (ptr != inbound_paths_.back()) {
auto i = std::find(inbound_paths_.begin(), inbound_paths_.end(), ptr);
......
......@@ -94,16 +94,17 @@ bool stream_scatterer::clean() const noexcept {
}
void stream_scatterer::close() {
for (auto i = paths_.begin(); i != paths_.end(); ++i) {
CAF_LOG_TRACE(CAF_ARG(paths_));
if (paths_.empty())
return;
for (auto i = paths_.begin(); i != paths_.end(); ++i)
about_to_erase(i, false, nullptr);
i->second->emit_regular_shutdown(self_);
}
paths_.clear();
}
void stream_scatterer::abort(error reason) {
auto& vec = paths_.container();
if (vec.empty())
CAF_LOG_TRACE(CAF_ARG(reason) << CAF_ARG(paths_));
if (paths_.empty())
return;
auto i = paths_.begin();
auto s = paths_.end() - 1;
......@@ -112,7 +113,7 @@ void stream_scatterer::abort(error reason) {
about_to_erase(i, false, &tmp);
}
about_to_erase(i, false, &reason);
vec.clear();
paths_.clear();
}
namespace {
......
......@@ -244,8 +244,8 @@ CAF_TEST(depth_2_pipeline_500_items) {
CAF_CHECK_EQUAL(fail_state(src), exit_reason::normal);
}
CAF_TEST(broken_pipeline) {
CAF_MESSAGE("streams must abort if a stage fails to initialize its state");
CAF_TEST(depth_2_pipelin_error_during_handshake) {
CAF_MESSAGE("streams must abort if a sink fails to initialize its state");
auto src = sys.spawn(file_reader, 50);
auto snk = sys.spawn(broken_sink);
auto pipeline = snk * src;
......@@ -257,4 +257,26 @@ CAF_TEST(broken_pipeline) {
expect((error), from(snk).to(self).with(sec::stream_init_failed));
}
CAF_TEST(depth_2_pipelin_error_at_source) {
CAF_MESSAGE("streams must abort if a source fails at runtime");
auto src = sys.spawn(file_reader, 500);
auto snk = sys.spawn(sum_up);
auto pipeline = snk * src;
CAF_MESSAGE(CAF_ARG(self) << CAF_ARG(src) << CAF_ARG(snk));
CAF_MESSAGE("initiate stream handshake");
self->send(pipeline, "numbers.txt");
expect((string), from(self).to(src).with("numbers.txt"));
expect((open_stream_msg), from(self).to(snk));
expect((upstream_msg::ack_open), from(snk).to(src));
CAF_MESSAGE("start data transmission (and abort source)");
self->send_exit(src, exit_reason::kill);
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(src).to(snk));
expect((error), from(snk).to(self));
CAF_CHECK_EQUAL(fail_state(snk), exit_reason::normal);
CAF_CHECK_EQUAL(fail_state(src), exit_reason::kill);
}
CAF_TEST_FIXTURE_SCOPE_END()
......@@ -348,7 +348,7 @@ public:
i->second->handle(slots, x);
if (i->second->done()) {
CAF_MESSAGE(name_ << " is done sending batches");
i->second->close();
i->second->stop();
managers_.erase(i);
}
}
......@@ -483,7 +483,7 @@ struct msg_visitor {
if (inptr->mgr->done()) {
CAF_MESSAGE(self->name()
<< " is done receiving and closes its manager");
inptr->mgr->close();
inptr->mgr->stop();
}
return intrusive::task_result::resume;
},
......@@ -500,7 +500,7 @@ struct msg_visitor {
} else {
// Close the manager and remove it on all registered slots.
auto mgr = i->second;
mgr->close();
mgr->stop();
auto j = self->managers_.begin();
while (j != self->managers_.end()) {
if (j->second == mgr)
......
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