Commit 77f06bcd authored by Dominik Charousset's avatar Dominik Charousset

Implement error handling for broken sinks

parent 85c4d35d
......@@ -408,10 +408,6 @@ public:
/// Silently closes incoming stream traffic on `slot`.
virtual void erase_inbound_path_later(stream_slot slot);
/// Closes incoming stream traffic on `slot`. Emits a drop message if `reason
/// == none` and a `forced_drop` message otherwise.
virtual void erase_inbound_path_later(stream_slot slot, error reason);
/// Silently closes all inbound paths for `mgr`.
virtual void erase_inbound_paths_later(const stream_manager* mgr);
......
......@@ -42,6 +42,9 @@ public:
response_promise(none_t);
response_promise(strong_actor_ptr self, strong_actor_ptr source,
forwarding_stack stages, message_id id);
response_promise(strong_actor_ptr self, mailbox_element& src);
response_promise(response_promise&&) = default;
......
......@@ -35,6 +35,7 @@
#include "caf/error.hpp"
#include "caf/extend.hpp"
#include "caf/fwd.hpp"
#include "caf/inbound_path.hpp"
#include "caf/invoke_message_result.hpp"
#include "caf/local_actor.hpp"
#include "caf/logger.hpp"
......@@ -954,8 +955,6 @@ public:
void erase_inbound_path_later(stream_slot slot) override;
void erase_inbound_path_later(stream_slot slot, error reason) override;
void erase_inbound_paths_later(const stream_manager* mgr) override;
void erase_inbound_paths_later(const stream_manager* mgr,
......@@ -963,8 +962,36 @@ public:
// -- handling of upstream message -------------------------------------------
void handle(stream_slots slots, actor_addr& sender,
upstream_msg::ack_open& x);
void handle_upstream_msg(stream_slots slots, actor_addr& sender,
upstream_msg::ack_open& x);
template <class T>
void handle_upstream_msg(stream_slots slots, actor_addr&, T& x) {
auto i = stream_managers_.find(slots);
if (i == stream_managers_.end()) {
auto j = pending_stream_managers_.find(slots.receiver);
if (j != pending_stream_managers_.end()) {
j->second->abort(sec::stream_init_failed);
pending_stream_managers_.erase(j);
return;
}
CAF_LOG_INFO("no manager found:" << CAF_ARG(slots));
// TODO: replace with `if constexpr` when switching to C++17
if (std::is_same<T, upstream_msg::ack_batch>::value) {
// Make sure the other actor does not falsely believe us a source.
inbound_path::emit_irregular_shutdown(this, slots, current_sender(),
sec::invalid_upstream);
}
return;
}
CAF_ASSERT(i->second != nullptr);
i->second->handle(slots, x);
if (i->second->done()) {
CAF_LOG_INFO("done sending:" << CAF_ARG(slots));
i->second->close();
stream_managers_.erase(i);
}
}
protected:
/// @cond PRIVATE
......
......@@ -136,6 +136,9 @@ public:
/// @pre `out().terminal() == true`
void add_promise(response_promise x);
/// Calls `x.deliver()`
void deliver_promises(message x);
// -- inline functions -------------------------------------------------------
inline local_actor* self() {
......@@ -185,6 +188,10 @@ protected:
/// Stores response promises for dellivering the final result.
std::vector<response_promise> promises_;
/// Stores promises while a handshake is active. The sink at the associated
/// key becomes responsible for the promise after receiving `ack_open`.
std::map<stream_slot, response_promise> in_flight_promises_;
};
/// A reference counting pointer to a `stream_manager`.
......
......@@ -134,10 +134,6 @@ void local_actor::erase_inbound_path_later(stream_slot) {
// nop
}
void local_actor::erase_inbound_path_later(stream_slot, error) {
// nop
}
void local_actor::erase_inbound_paths_later(const stream_manager*) {
// nop
}
......
......@@ -34,17 +34,27 @@ response_promise::response_promise(none_t) : response_promise() {
// nop
}
response_promise::response_promise(strong_actor_ptr self, mailbox_element& src)
response_promise::response_promise(strong_actor_ptr self,
strong_actor_ptr source,
forwarding_stack stages, message_id mid)
: self_(std::move(self)),
id_(src.mid) {
source_(std::move(source)),
stages_(std::move(stages)),
id_(mid) {
// form an invalid request promise when initialized from a
// response ID, since CAF always drops messages in this case
if (!src.mid.is_response()) {
source_ = std::move(src.sender);
stages_ = std::move(src.stages);
if (mid.is_response()) {
source_ = nullptr;
stages_.clear();
}
}
response_promise::response_promise(strong_actor_ptr self, mailbox_element& src)
: response_promise(std::move(self), std::move(src.sender),
std::move(src.stages), src.mid) {
// nop
}
response_promise response_promise::deliver(error x) {
//if (id_.valid())
return deliver_impl(make_message(std::move(x)));
......
......@@ -237,40 +237,26 @@ void scheduled_actor::intrusive_ptr_release_impl() {
intrusive_ptr_release(ctrl());
}
namespace {
// TODO: replace with generic lambda when switching to C++14
struct upstream_msg_visitor {
scheduled_actor* selfptr;
upstream_msg& um;
template <class T>
void operator()(T& x) {
selfptr->handle_upstream_msg(um.slots, um.sender, x);
}
};
} // namespace <anonymous>
intrusive::task_result scheduled_actor::mailbox_visitor::
operator()(size_t, upstream_queue&, mailbox_element& x) {
CAF_ASSERT(x.content().type_token() == make_type_token<upstream_msg>());
struct visitor {
scheduled_actor* selfptr;
upstream_msg& um;
void operator()(upstream_msg::ack_open& y) {
selfptr->handle(um.slots, um.sender, y);
}
void operator()(upstream_msg::ack_batch& x) {
// Dispatch to the responsible manager.
auto slots = um.slots;
auto& managers = selfptr->stream_managers_;
auto i = managers.find(slots);
if (i == managers.end()) {
CAF_LOG_INFO("no manager available for ack_batch:" << CAF_ARG(slots));
return;
}
i->second->handle(um.slots, x);
if (i->second->done()) {
CAF_LOG_INFO("done sending:" << CAF_ARG(slots));
i->second->close();
managers.erase(i);
}
}
void operator()(upstream_msg::drop&) {
CAF_LOG_ERROR("implement me");
}
void operator()(upstream_msg::forced_drop&) {
CAF_LOG_ERROR("implement me");
}
};
auto& um = x.content().get_mutable_as<upstream_msg>(0);
visitor f{self, um};
upstream_msg_visitor f{self, um};
visit(f, um.content);
return intrusive::task_result::resume;
}
......@@ -835,44 +821,21 @@ inbound_path* scheduled_actor::make_inbound_path(stream_manager_ptr mgr,
}
void scheduled_actor::erase_inbound_path_later(stream_slot slot) {
constexpr size_t id = mailbox_policy::downstream_queue_index;
get<id>(mailbox_.queue().queues()).erase_later(slot);
}
void scheduled_actor::erase_inbound_path_later(stream_slot, error) {
CAF_LOG_ERROR("implement me");
/*
using fn = void (*)(local_actor*, inbound_path&, error&);
fn regular = [](local_actor* self, inbound_path& in, error&) {
in.emit_regular_shutdown(self);
};
fn irregular = [](local_actor* self, inbound_path& in, error& rsn) {
in.emit_irregular_shutdown(self, rsn);
};
auto f = reason == none ? regular : irregular;
for (auto& kvp : get<2>(mbox.queues()).queues()) {
auto& path = kvp.second.policy().handler;
if (path != nullptr && path->mgr == mgr) {
f(this, *path, reason);
erase_inbound_path_later(kvp.first);
}
}
*/
constexpr size_t queue_id = mailbox_policy::downstream_queue_index;
get<queue_id>(mailbox_.queue().queues()).erase_later(slot);
}
void scheduled_actor::erase_inbound_paths_later(const stream_manager* ptr) {
constexpr size_t id = mailbox_policy::downstream_queue_index;
for (auto& kvp : get<id>(mailbox_.queue().queues()).queues()) {
constexpr size_t queue_id = mailbox_policy::downstream_queue_index;
for (auto& kvp : get<queue_id>(mailbox_.queue().queues()).queues()) {
auto& path = kvp.second.policy().handler;
if (path != nullptr && path->mgr == ptr)
erase_inbound_path_later(kvp.first);
}
}
void scheduled_actor::erase_inbound_paths_later(const stream_manager*,
error) {
CAF_LOG_ERROR("implement me");
/*
void scheduled_actor::erase_inbound_paths_later(const stream_manager* ptr,
error reason) {
using fn = void (*)(local_actor*, inbound_path&, error&);
fn regular = [](local_actor* self, inbound_path& in, error&) {
in.emit_regular_shutdown(self);
......@@ -881,18 +844,19 @@ void scheduled_actor::erase_inbound_paths_later(const stream_manager*,
in.emit_irregular_shutdown(self, rsn);
};
auto f = reason == none ? regular : irregular;
for (auto& kvp : get<2>(mbox.queues()).queues()) {
static constexpr size_t queue_id = mailbox_policy::downstream_queue_index;
for (auto& kvp : get<queue_id>(mailbox_.queue().queues()).queues()) {
auto& path = kvp.second.policy().handler;
if (path != nullptr && path->mgr == mgr) {
if (path != nullptr && path->mgr == ptr) {
f(this, *path, reason);
erase_inbound_path_later(kvp.first);
}
}
*/
}
void scheduled_actor::handle(stream_slots slots, actor_addr& sender,
upstream_msg::ack_open& x) {
void scheduled_actor::handle_upstream_msg(stream_slots slots,
actor_addr& sender,
upstream_msg::ack_open& x) {
CAF_LOG_TRACE(CAF_ARG(slots) << CAF_ARG(sender) << CAF_ARG(x));
CAF_ASSERT(sender == x.rebind_to);
// Get the manager for that stream, move it from `pending_managers_` to
......
......@@ -63,6 +63,7 @@ void stream_manager::handle(stream_slots slots, upstream_msg::ack_open& x) {
path->desired_batch_size = x.desired_batch_size;
--pending_handshakes_;
push();
in_flight_promises_.erase(slots.sender);
}
void stream_manager::handle(stream_slots slots, upstream_msg::ack_batch& x) {
......@@ -87,15 +88,19 @@ void stream_manager::handle(stream_slots slots, upstream_msg::forced_drop& x) {
void stream_manager::close() {
out().close();
self_->erase_inbound_paths_later(this);
if (!promises_.empty()) {
auto msg = make_final_result();
for (auto& x : promises_)
x.deliver(msg);
}
if (!promises_.empty())
deliver_promises(make_final_result());
}
void stream_manager::abort(error reason) {
out().abort(std::move(reason));
if (!promises_.empty() || !in_flight_promises_.empty()) {
auto msg = make_message(reason);
deliver_promises(msg);
for (auto& kvp : in_flight_promises_)
kvp.second.deliver(msg);
in_flight_promises_.clear();
}
out().abort(reason);
self_->erase_inbound_paths_later(this, std::move(reason));
}
......@@ -116,6 +121,8 @@ void stream_manager::send_handshake(strong_actor_ptr dest, stream_slot slot,
message_id mid) {
CAF_ASSERT(dest != nullptr);
++pending_handshakes_;
in_flight_promises_.emplace(
slot, response_promise{self()->ctrl(), client, fwd_stack, mid});
dest->enqueue(
make_mailbox_element(
std::move(client), mid, std::move(fwd_stack),
......@@ -157,6 +164,13 @@ void stream_manager::add_promise(response_promise x) {
promises_.emplace_back(std::move(x));
}
void stream_manager::deliver_promises(message x) {
CAF_LOG_TRACE(CAF_ARG(x));
for (auto& p : promises_)
p.deliver(x);
promises_.clear();
}
message stream_manager::make_final_result() {
return none;
}
......
......@@ -119,9 +119,8 @@ TESTEE(delayed_sum_up) {
x = 0;
},
// processing step
[](int& x, std::vector<int>&& ys) {
for (auto y : ys)
x += y;
[](int& x, int y) {
x += y;
},
// cleanup and produce result message
[](int& x) -> int {
......@@ -250,28 +249,12 @@ CAF_TEST(broken_pipeline) {
auto src = sys.spawn(file_reader, 50);
auto snk = sys.spawn(broken_sink);
auto pipeline = snk * src;
sched.run();
CAF_MESSAGE("initiate stream handshake");
self->send(pipeline, "test.txt");
expect((std::string), from(self).to(src).with("test.txt"));
expect((open_stream_msg), from(self).to(snk));
expect((upstream_msg::forced_drop), from(snk).to(src));
expect((error), from(snk).to(self).with(sec::stream_init_failed));
}
CAF_TEST(delayed_pipeline) {
CAF_MESSAGE("streams must abort if a stage fails to initialize its state");
auto src = sys.spawn(file_reader, 50);
auto snk = sys.spawn(broken_sink);
auto pipeline = snk * src;
sched.run();
CAF_MESSAGE("initiate stream handshake");
self->send(pipeline, "test.txt");
expect((std::string), from(self).to(src).with("test.txt"));
self->send(pipeline, "numbers.txt");
expect((std::string), from(self).to(src).with("numbers.txt"));
expect((open_stream_msg), from(self).to(snk));
expect((upstream_msg::forced_drop), from(snk).to(src));
expect((error), from(snk).to(self).with(sec::stream_init_failed));
}
CAF_TEST_FIXTURE_SCOPE_END()
......@@ -389,10 +389,6 @@ public:
get<2>(mbox.queues()).erase_later(slot);
}
void erase_inbound_path_later(stream_slot, error) override {
CAF_FAIL("unexpected function call");
}
void erase_inbound_paths_later(const stream_manager* mgr) override {
for (auto& kvp : get<2>(mbox.queues()).queues()) {
auto& path = kvp.second.policy().handler;
......
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