Unverified Commit 52697cba authored by Dominik Charousset's avatar Dominik Charousset Committed by GitHub

Merge pull request #954

Avoid repeated calls to stream finalizers
parents fddc1747 dfd900c8
......@@ -47,10 +47,14 @@ public:
/// outbound paths exist.
static constexpr int is_continuous_flag = 0x0001;
/// Denotes whether the stream is about to stop, only sending already
/// buffered elements.
/// Denotes whether the stream is about to stop, only sending buffered
/// elements.
static constexpr int is_shutting_down_flag = 0x0002;
/// Denotes whether the manager has stopped. Calling member functions such as
/// stop() or abort() on it no longer has any effect.
static constexpr int is_stopped_flag = 0x0004;
// -- member types -----------------------------------------------------------
using inbound_paths_list = std::vector<inbound_path*>;
......@@ -143,22 +147,22 @@ public:
// -- properties -------------------------------------------------------------
/// Returns whether this stream is shutting down.
bool shutting_down() const noexcept {
return getf(is_shutting_down_flag);
/// Returns whether this stream is neither shutting down nor has stopped.
bool running() const noexcept {
return getf(is_shutting_down_flag | is_stopped_flag) == 0;
}
/// 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
/// point where its driver returns `done() == true`.
inline bool continuous() const noexcept {
bool continuous() const noexcept {
return getf(is_continuous_flag);
}
/// Sets whether this stream remains open even if no in- or outbound paths
/// exist.
inline void continuous(bool x) noexcept {
if (!shutting_down()) {
void continuous(bool x) noexcept {
if (running()) {
if (x)
setf(is_continuous_flag);
else
......@@ -167,7 +171,7 @@ public:
}
/// Returns the list of inbound paths.
inline const inbound_paths_list& inbound_paths() const noexcept{
const inbound_paths_list& inbound_paths() const noexcept {
return inbound_paths_;
}
......@@ -179,7 +183,7 @@ public:
bool inbound_paths_idle() const noexcept;
/// Returns the parent actor.
inline scheduled_actor* self() {
scheduled_actor* self() {
return self_;
}
......@@ -265,7 +269,7 @@ public:
/// Adds the current sender as an inbound path.
/// @pre Current message is an `open_stream_msg`.
stream_slot add_unchecked_inbound_path_impl(rtti_pair rtti);
stream_slot add_unchecked_inbound_path_impl(rtti_pair rtti);
protected:
// -- modifiers for self -----------------------------------------------------
......
......@@ -305,6 +305,7 @@ struct downstream_msg_visitor {
template <class T>
intrusive::task_result operator()(T& x) {
CAF_LOG_TRACE(CAF_ARG(x));
auto& inptr = q_ref.policy().handler;
if (inptr == nullptr)
return intrusive::task_result::stop;
......@@ -323,6 +324,7 @@ struct downstream_msg_visitor {
qs_ref.erase_later(dm.slots.receiver);
selfptr->erase_stream_manager(dm.slots.receiver);
if (mgr->done()) {
CAF_LOG_DEBUG("path is done receiving and closes its manager");
selfptr->erase_stream_manager(mgr);
mgr->stop();
}
......
......@@ -37,10 +37,7 @@
namespace caf {
stream_manager::stream_manager(scheduled_actor* selfptr, stream_priority prio)
: self_(selfptr),
pending_handshakes_(0),
priority_(prio),
flags_(0) {
: self_(selfptr), pending_handshakes_(0), priority_(prio), flags_(0) {
// nop
}
......@@ -116,15 +113,21 @@ void stream_manager::handle(stream_slots slots, upstream_msg::ack_batch& x) {
}
void stream_manager::handle(stream_slots slots, upstream_msg::drop&) {
CAF_LOG_TRACE(CAF_ARG(slots));
out().close(slots.receiver);
}
void stream_manager::handle(stream_slots slots, upstream_msg::forced_drop& x) {
CAF_LOG_TRACE(CAF_ARG(slots) << CAF_ARG(x));
if (out().remove_path(slots.receiver, x.reason, true))
stop(std::move(x.reason));
}
void stream_manager::stop(error reason) {
CAF_LOG_TRACE(CAF_ARG(reason));
if (getf(is_stopped_flag))
return;
flags_ = is_stopped_flag;
if (reason)
out().abort(reason);
else
......@@ -135,12 +138,12 @@ void stream_manager::stop(error reason) {
void stream_manager::shutdown() {
CAF_LOG_TRACE("");
// Mark as shutting down and reset other flags.
if (shutting_down())
if (!running())
return;
flags_ = is_shutting_down_flag;
CAF_LOG_DEBUG("emit shutdown messages on" << inbound_paths_.size()
<< "inbound paths;" << CAF_ARG2("out.clean", out().clean())
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_);
......@@ -233,15 +236,12 @@ void stream_manager::remove_input_path(stream_slot slot, error reason,
}
inbound_path* stream_manager::get_inbound_path(stream_slot x) const noexcept {
auto pred = [=](inbound_path* ptr) {
return ptr->slots.receiver == x;
};
auto pred = [=](inbound_path* ptr) { return ptr->slots.receiver == x; };
auto e = inbound_paths_.end();
auto i = std::find_if(inbound_paths_.begin(), e, pred);
return i != e ? *i : nullptr;
}
bool stream_manager::inbound_paths_idle() const noexcept {
auto f = [](inbound_path* x) {
return x->up_to_date() && x->assigned_credit > 0;
......
......@@ -65,9 +65,10 @@ std::function<bool(const buf&)> is_done(scheduled_actor* self) {
};
}
template <class T>
std::function<void(T&, const error&)> fin(scheduled_actor* self) {
template <class T, class Self>
std::function<void(T&, const error&)> fin(Self* self) {
return [=](T&, const error& err) {
self->state.fin_called += 1;
if (err == none) {
CAF_MESSAGE(self->name() << " is done");
} else {
......@@ -76,8 +77,13 @@ std::function<void(T&, const error&)> fin(scheduled_actor* self) {
};
}
TESTEE_STATE(infinite_source) {
int fin_called = 0;
};
TESTEE(infinite_source) {
return {[=](string& fname) -> result<stream<int>> {
return {
[=](string& fname) -> result<stream<int>> {
CAF_CHECK_EQUAL(fname, "numbers.txt");
CAF_CHECK_EQUAL(self->mailbox().empty(), true);
return attach_stream_source(
......@@ -87,11 +93,17 @@ TESTEE(infinite_source) {
out.push(x++);
},
[](const int&) { return false; }, fin<int>(self));
}};
},
};
}
TESTEE_STATE(file_reader) {
int fin_called = 0;
};
VARARGS_TESTEE(file_reader, size_t buf_size) {
return {[=](string& fname) -> result<stream<int>> {
return {
[=](string& fname) -> result<stream<int>> {
CAF_CHECK_EQUAL(fname, "numbers.txt");
CAF_CHECK_EQUAL(self->mailbox().empty(), true);
return attach_stream_source(self, init(buf_size), push_from_buf,
......@@ -102,16 +114,19 @@ VARARGS_TESTEE(file_reader, size_t buf_size) {
CAF_CHECK_EQUAL(self->mailbox().empty(), true);
attach_stream_source(self, next, init(buf_size), push_from_buf,
is_done(self), fin<buf>(self));
}};
},
};
}
TESTEE_STATE(sum_up) {
int x = 0;
int fin_called = 0;
};
TESTEE(sum_up) {
using intptr = int*;
return {[=](stream<int>& in) {
return {
[=](stream<int>& in) {
return attach_stream_sink(
self,
// input stream
......@@ -120,17 +135,20 @@ TESTEE(sum_up) {
[=](intptr& x) { x = &self->state.x; },
// processing step
[](intptr& x, int y) { *x += y; }, fin<intptr>(self));
}};
},
};
}
TESTEE_STATE(delayed_sum_up) {
int x = 0;
int fin_called = 0;
};
TESTEE(delayed_sum_up) {
using intptr = int*;
self->set_default_handler(skip);
return {[=](ok_atom) {
return {
[=](ok_atom) {
self->become([=](stream<int>& in) {
self->set_default_handler(print_and_drop);
return attach_stream_sink(
......@@ -144,19 +162,31 @@ TESTEE(delayed_sum_up) {
// cleanup
fin<intptr>(self));
});
}};
},
};
}
TESTEE_STATE(broken_sink) {
int fin_called = 0;
};
TESTEE(broken_sink) {
CAF_IGNORE_UNUSED(self);
return {[=](stream<int>&, const actor&) {
return {
[=](stream<int>&, const actor&) {
// nop
}};
},
};
}
TESTEE_STATE(filter) {
int fin_called = 0;
};
TESTEE(filter) {
CAF_IGNORE_UNUSED(self);
return {[=](stream<int>& in) {
return {
[=](stream<int>& in) {
return attach_stream_stage(
self,
// input stream
......@@ -172,12 +202,18 @@ TESTEE(filter) {
},
// cleanup
fin<unit_t>(self));
}};
},
};
}
TESTEE_STATE(doubler) {
int fin_called = 0;
};
TESTEE(doubler) {
CAF_IGNORE_UNUSED(self);
return {[=](stream<int>& in) {
return {
[=](stream<int>& in) {
return attach_stream_stage(
self,
// input stream
......@@ -190,7 +226,8 @@ TESTEE(doubler) {
[](unit_t&, downstream<int>& out, int x) { out.push(x * 2); },
// cleanup
fin<unit_t>(self));
}};
},
};
}
struct fixture : test_coordinator_fixture<> {
......@@ -229,6 +266,9 @@ CAF_TEST(depth_2_pipeline_50_items) {
CAF_MESSAGE("expect close message from src and then result from snk");
expect((downstream_msg::close), from(src).to(snk));
CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.x, 1275);
CAF_MESSAGE("verify that each actor called its finalizer once");
CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.fin_called, 1);
CAF_CHECK_EQUAL(deref<file_reader_actor>(src).state.fin_called, 1);
}
CAF_TEST(depth_2_pipeline_setup2_50_items) {
......@@ -249,6 +289,9 @@ CAF_TEST(depth_2_pipeline_setup2_50_items) {
CAF_MESSAGE("expect close message from src and then result from snk");
expect((downstream_msg::close), from(src).to(snk));
CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.x, 1275);
CAF_MESSAGE("verify that each actor called its finalizer once");
CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.fin_called, 1);
CAF_CHECK_EQUAL(deref<file_reader_actor>(src).state.fin_called, 1);
}
CAF_TEST(delayed_depth_2_pipeline_50_items) {
......@@ -275,6 +318,9 @@ CAF_TEST(delayed_depth_2_pipeline_50_items) {
CAF_MESSAGE("expect close message from src and then result from snk");
expect((downstream_msg::close), from(src).to(snk));
CAF_CHECK_EQUAL(deref<delayed_sum_up_actor>(snk).state.x, 1275);
CAF_MESSAGE("verify that each actor called its finalizer once");
CAF_CHECK_EQUAL(deref<delayed_sum_up_actor>(snk).state.fin_called, 1);
CAF_CHECK_EQUAL(deref<file_reader_actor>(src).state.fin_called, 1);
}
CAF_TEST(depth_2_pipeline_500_items) {
......@@ -302,6 +348,9 @@ CAF_TEST(depth_2_pipeline_500_items) {
CAF_MESSAGE("expect close message from src and then result from snk");
expect((downstream_msg::close), from(src).to(snk));
CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.x, 125250);
CAF_MESSAGE("verify that each actor called its finalizer once");
CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.fin_called, 1);
CAF_CHECK_EQUAL(deref<file_reader_actor>(src).state.fin_called, 1);
}
CAF_TEST(depth_2_pipeline_error_during_handshake) {
......@@ -314,6 +363,9 @@ CAF_TEST(depth_2_pipeline_error_during_handshake) {
expect((open_stream_msg), from(self).to(snk));
expect((upstream_msg::forced_drop), from(_).to(src));
expect((error), from(snk).to(self).with(sec::stream_init_failed));
run();
CAF_MESSAGE("verify that the file reader called its finalizer once");
CAF_CHECK_EQUAL(deref<file_reader_actor>(src).state.fin_called, 1);
}
CAF_TEST(depth_2_pipeline_error_at_source) {
......@@ -330,6 +382,8 @@ CAF_TEST(depth_2_pipeline_error_at_source) {
hard_kill(src);
expect((downstream_msg::batch), from(src).to(snk));
expect((downstream_msg::forced_close), from(_).to(snk));
CAF_MESSAGE("verify that the sink called its finalizer once");
CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.fin_called, 1);
}
CAF_TEST(depth_2_pipelin_error_at_sink) {
......@@ -345,6 +399,8 @@ CAF_TEST(depth_2_pipelin_error_at_sink) {
hard_kill(snk);
expect((upstream_msg::ack_open), from(snk).to(src));
expect((upstream_msg::forced_drop), from(_).to(src));
CAF_MESSAGE("verify that the source called its finalizer once");
CAF_CHECK_EQUAL(deref<file_reader_actor>(src).state.fin_called, 1);
}
CAF_TEST(depth_3_pipeline_50_items) {
......@@ -379,6 +435,10 @@ CAF_TEST(depth_3_pipeline_50_items) {
expect((upstream_msg::ack_batch), from(snk).to(stg));
expect((downstream_msg::close), from(stg).to(snk));
CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.x, 625);
CAF_MESSAGE("verify that each actor called its finalizer once");
CAF_CHECK_EQUAL(deref<file_reader_actor>(src).state.fin_called, 1);
CAF_CHECK_EQUAL(deref<filter_actor>(stg).state.fin_called, 1);
CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.fin_called, 1);
}
CAF_TEST(depth_4_pipeline_500_items) {
......@@ -401,6 +461,11 @@ CAF_TEST(depth_4_pipeline_500_items) {
run();
CAF_MESSAGE("check sink result");
CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.x, 125000);
CAF_MESSAGE("verify that each actor called its finalizer once");
CAF_CHECK_EQUAL(deref<file_reader_actor>(src).state.fin_called, 1);
CAF_CHECK_EQUAL(deref<filter_actor>(stg1).state.fin_called, 1);
CAF_CHECK_EQUAL(deref<doubler_actor>(stg2).state.fin_called, 1);
CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.fin_called, 1);
}
CAF_TEST(depth_3_pipeline_graceful_shutdown) {
......@@ -421,6 +486,10 @@ CAF_TEST(depth_3_pipeline_graceful_shutdown) {
run();
CAF_MESSAGE("check sink result");
CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.x, 625);
CAF_MESSAGE("verify that each actor called its finalizer once");
CAF_CHECK_EQUAL(deref<file_reader_actor>(src).state.fin_called, 1);
CAF_CHECK_EQUAL(deref<filter_actor>(stg).state.fin_called, 1);
CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.fin_called, 1);
}
CAF_TEST(depth_3_pipeline_infinite_source) {
......@@ -438,6 +507,9 @@ CAF_TEST(depth_3_pipeline_infinite_source) {
CAF_MESSAGE("send exit to the source and expect the stream to terminate");
anon_send_exit(src, exit_reason::user_shutdown);
run();
CAF_MESSAGE("verify that each actor called its finalizer once");
CAF_CHECK_EQUAL(deref<filter_actor>(stg).state.fin_called, 1);
CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.fin_called, 1);
}
CAF_TEST_FIXTURE_SCOPE_END()
......@@ -632,12 +632,12 @@ public:
// -- constructors, destructors, and assignment operators --------------------
static Config& init_config(Config& cfg) {
cfg.set("logger.file-verbosity", caf::atom("quiet"));
if (auto err = cfg.parse(caf::test::engine::argc(),
caf::test::engine::argv()))
CAF_FAIL("failed to parse config: " << to_string(err));
cfg.set("scheduler.policy", caf::atom("testing"));
cfg.set("logger.inline-output", true);
cfg.set("logger.file-verbosity", caf::atom("quiet"));
cfg.set("middleman.network-backend", caf::atom("testing"));
cfg.set("middleman.manual-multiplexing", true);
cfg.set("middleman.workers", size_t{0});
......
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