Commit 153aa37e authored by Dominik Charousset's avatar Dominik Charousset

Improve testing of the SPSC buffer

parent d3458b2a
...@@ -126,7 +126,7 @@ SCENARIO("SPSC buffers may go past their capacity") { ...@@ -126,7 +126,7 @@ SCENARIO("SPSC buffers may go past their capacity") {
buf->push(2); buf->push(2);
CHECK_EQ(cons->producer_wakeups, 1u); CHECK_EQ(cons->producer_wakeups, 1u);
THEN("excess items are stored but do not trigger demand when consumed") { THEN("excess items are stored but do not trigger demand when consumed") {
std::vector<int> tmp{3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14}; auto tmp = std::vector<int>{3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14};
buf->push(make_span(tmp)); buf->push(make_span(tmp));
prod->demand = 0; prod->demand = 0;
CHECK_EQ(cons->producer_wakeups, 1u); CHECK_EQ(cons->producer_wakeups, 1u);
...@@ -153,8 +153,42 @@ SCENARIO("SPSC buffers may go past their capacity") { ...@@ -153,8 +153,42 @@ SCENARIO("SPSC buffers may go past their capacity") {
} }
} }
SCENARIO("the prioritize_errors policy skips processing of pending items") {
GIVEN("an SPSC buffer with consumer and produer") {
WHEN("pushing into the buffer and then aborting") {
THEN("pulling items with prioritize_errors skips remaining items") {
auto prod = make_counted<dummy_producer>();
auto cons = make_counted<dummy_consumer>();
auto buf = make_counted<async::spsc_buffer<int>>(10, 2);
auto tmp = std::vector<int>{1, 2, 3, 4, 5};
buf->set_producer(prod);
buf->push(make_span(tmp));
buf->set_consumer(cons);
CHECK_EQ(cons->producer_wakeups, 1u);
MESSAGE("consume one element");
{
dummy_observer obs;
auto [ok, consumed] = buf->pull(async::prioritize_errors, 1, obs);
CHECK_EQ(ok, true);
CHECK_EQ(consumed, 1u);
CHECK_EQ(obs.consumed, 1u);
}
MESSAGE("set an error and try consuming remaining elements");
{
buf->abort(sec::runtime_error);
dummy_observer obs;
auto [ok, consumed] = buf->pull(async::prioritize_errors, 1, obs);
CHECK_EQ(ok, false);
CHECK_EQ(consumed, 0u);
CHECK_EQ(obs.err, sec::runtime_error);
}
}
}
}
}
SCENARIO("SPSC buffers moves data between actors") { SCENARIO("SPSC buffers moves data between actors") {
GIVEN("a SPSC buffer resource") { GIVEN("an SPSC buffer resource") {
WHEN("opening the resource from two actors") { WHEN("opening the resource from two actors") {
THEN("data travels through the SPSC buffer") { THEN("data travels through the SPSC buffer") {
using actor_t = event_based_actor; using actor_t = event_based_actor;
...@@ -180,4 +214,322 @@ SCENARIO("SPSC buffers moves data between actors") { ...@@ -180,4 +214,322 @@ SCENARIO("SPSC buffers moves data between actors") {
} }
} }
SCENARIO("SPSC buffers appear empty when only one actor is connected") {
GIVEN("an SPSC buffer resource") {
WHEN("destroying the write end before adding a subscriber") {
THEN("no data arrives through the SPSC buffer") {
using actor_t = event_based_actor;
auto outputs = std::vector<int>{};
auto finalized = false;
{
auto [rd, wr] = async::make_spsc_buffer_resource<int>(6, 2);
sys.spawn([rd{rd}, &outputs, &finalized](actor_t* snk) {
snk
->make_observable() //
.from_resource(rd)
.do_finally([&finalized] { finalized = true; })
.for_each([&outputs](int x) { outputs.emplace_back(x); });
});
}
// At scope exit, `wr` gets destroyed, closing the buffer.
run();
CHECK(finalized);
CHECK(outputs.empty());
}
}
WHEN("destroying the write end after adding a subscriber") {
THEN("no data arrives through the SPSC buffer") {
using actor_t = event_based_actor;
auto outputs = std::vector<int>{};
auto finalized = false;
{
auto [rd, wr] = async::make_spsc_buffer_resource<int>(6, 2);
sys.spawn([rd{rd}, &outputs, &finalized](actor_t* snk) {
snk
->make_observable() //
.from_resource(rd)
.do_finally([&finalized] { finalized = true; })
.for_each([&outputs](int x) { outputs.emplace_back(x); });
});
// Only difference to before: have the actor create the observable
// from `rd` handle before destroying `wr`.
run();
}
// At scope exit, `wr` gets destroyed, closing the buffer.
run();
CHECK(finalized);
CHECK(outputs.empty());
}
}
WHEN("aborting the write end") {
THEN("the observer receives on_error") {
using actor_t = event_based_actor;
auto outputs = std::vector<int>{};
auto on_error_called = false;
auto [rd, wr] = async::make_spsc_buffer_resource<int>(6, 2);
sys.spawn([rd{rd}, &outputs, &on_error_called](actor_t* snk) {
snk
->make_observable() //
.from_resource(rd)
.do_on_error([&on_error_called](const error& err) {
on_error_called = true;
CHECK_EQ(err, sec::runtime_error);
})
.for_each([&outputs](int x) { outputs.emplace_back(x); });
});
wr.abort(sec::runtime_error);
wr.abort(sec::runtime_error); // Calling twice must have no side effect.
run();
CHECK(on_error_called);
CHECK(outputs.empty());
}
}
WHEN("closing the write end") {
THEN("the observer receives on_complete") {
using actor_t = event_based_actor;
auto outputs = std::vector<int>{};
auto on_complete_called = false;
auto [rd, wr] = async::make_spsc_buffer_resource<int>(6, 2);
sys.spawn([rd{rd}, &outputs, &on_complete_called](actor_t* snk) {
snk
->make_observable() //
.from_resource(rd)
.do_on_complete([&on_complete_called] { //
on_complete_called = true;
})
.for_each([&outputs](int x) { outputs.emplace_back(x); });
});
wr.close();
wr.close(); // Calling twice must have no side effect.
run();
CHECK(on_complete_called);
CHECK(outputs.empty());
}
}
}
}
SCENARIO("SPSC buffers drop data when discarding the read end") {
GIVEN("an SPSC buffer resource") {
WHEN("destroying the read end before adding a publisher") {
THEN("the flow of the writing actor gets canceled") {
using actor_t = event_based_actor;
auto outputs = std::vector<int>{};
{
auto [rd, wr] = async::make_spsc_buffer_resource<int>(6, 2);
sys.spawn([wr{wr}](actor_t* src) {
src
->make_observable() //
.iota(1)
.subscribe(wr);
});
}
// At scope exit, `rd` gets destroyed, closing the buffer.
run();
CHECK(outputs.empty());
}
}
WHEN("destroying the read end before adding a publisher") {
THEN("the flow of the writing actor gets canceled") {
using actor_t = event_based_actor;
auto outputs = std::vector<int>{};
{
auto [rd, wr] = async::make_spsc_buffer_resource<int>(6, 2);
sys.spawn([wr{wr}](actor_t* src) {
src
->make_observable() //
.iota(1)
.subscribe(wr);
});
// Only difference to before: have the actor adding an observer that
// writes to `wr` before destroying `rd`.
run();
}
// At scope exit, `rd` gets destroyed, closing the buffer.
run();
CHECK(outputs.empty());
}
}
WHEN("canceling the read end before adding a publisher") {
THEN("the flow of the writing actor gets canceled") {
using actor_t = event_based_actor;
auto outputs = std::vector<int>{};
auto [rd, wr] = async::make_spsc_buffer_resource<int>(6, 2);
sys.spawn([wr{wr}](actor_t* src) {
src
->make_observable() //
.iota(1)
.subscribe(wr);
});
rd.cancel();
rd.cancel(); // Calling twice must have no side effect.
run();
CHECK(outputs.empty());
}
}
}
}
SCENARIO("resources are invalid after calling try_open") {
GIVEN("a producer resource") {
WHEN("opening it twice") {
THEN("the second try_open fails") {
auto [rd, wr] = async::make_spsc_buffer_resource<int>(6, 2);
CHECK(rd);
CHECK_NE(rd.try_open(), nullptr);
CHECK(!rd);
CHECK_EQ(rd.try_open(), nullptr);
auto rd2 = async::consumer_resource<int>{};
rd2 = rd;
CHECK(!rd2);
rd2 = std::move(rd);
CHECK(!rd2);
CHECK(wr);
CHECK_NE(wr.try_open(), nullptr);
CHECK(!wr);
CHECK_EQ(wr.try_open(), nullptr);
auto wr2 = async::producer_resource<int>{};
wr2 = wr;
CHECK(!wr2);
wr2 = std::move(wr);
CHECK(!wr2);
}
}
}
}
SCENARIO("try_open on producer resources succeeds only once") {
GIVEN("a producer resource") {
WHEN("opening it twice") {
THEN("the second try_open fails") {
using actor_t = event_based_actor;
auto outputs = std::vector<int>{};
auto [rd, wr] = async::make_spsc_buffer_resource<int>(6, 2);
auto prod1 = sys.spawn([wr{wr}](actor_t* src) {
src
->make_observable() //
.iota(1)
.subscribe(wr);
});
self->monitor(prod1);
run();
auto prod2 = sys.spawn([wr{wr}](actor_t* src) {
src
->make_observable() //
.iota(1)
.subscribe(wr);
});
self->monitor(prod2);
run();
expect((down_msg), to(self).with(down_msg{prod2.address(), error{}}));
CHECK(self->mailbox().empty());
}
}
}
}
SCENARIO("try_open on consumer resources succeeds only once") {
GIVEN("a consumer resource") {
WHEN("opening it twice") {
THEN("the second try_open fails") {
using actor_t = event_based_actor;
auto outputs = std::vector<int>{};
auto [rd, wr] = async::make_spsc_buffer_resource<int>(6, 2);
auto snk1 = sys.spawn([rd{rd}, &outputs](actor_t* snk) {
snk
->make_observable() //
.from_resource(rd)
.for_each([&outputs](int x) { outputs.emplace_back(x); });
});
self->monitor(snk1);
run();
auto snk2 = sys.spawn([rd{rd}, &outputs](actor_t* snk) {
snk
->make_observable() //
.from_resource(rd)
.for_each([&outputs](int x) { outputs.emplace_back(x); });
});
self->monitor(snk2);
run();
expect((down_msg), to(self).with(down_msg{snk2.address(), error{}}));
CHECK(self->mailbox().empty());
CHECK(outputs.empty());
}
}
}
}
// Note: this basically checks that buffer protects against misuse and is not
// how users should do things.
SCENARIO("SPSC buffers reject multiple producers") {
GIVEN("an SPSC buffer resource") {
WHEN("attaching a second producer") {
THEN("the buffer immediately calls on_consumer_cancel on it") {
// Note: two resources should never point to the same buffer.
using actor_t = event_based_actor;
using buffer_type = async::spsc_buffer<int>;
auto buf = make_counted<buffer_type>(20, 5);
auto rd = async::consumer_resource<int>{buf};
auto wr1 = async::producer_resource<int>{buf};
auto wr2 = async::producer_resource<int>{buf};
auto prod1 = sys.spawn([wr1](actor_t* src) {
src->make_observable().iota(1).subscribe(wr1);
});
self->monitor(prod1);
run();
auto prod2 = sys.spawn([wr2](actor_t* src) {
src->make_observable().iota(1).subscribe(wr2);
});
self->monitor(prod2);
run();
expect((down_msg), to(self).with(down_msg{prod2.address(), error{}}));
CHECK(self->mailbox().empty());
}
}
}
}
#ifdef CAF_ENABLE_EXCEPTIONS
// Note: this basically checks that buffer protects against misuse and is not
// how users should do things.
SCENARIO("SPSC buffers reject multiple consumers") {
GIVEN("an SPSC buffer resource") {
WHEN("attaching a second consumer") {
THEN("the buffer throws an exception") {
// Note: two resources should never point to the same buffer.
using actor_t = event_based_actor;
using buffer_type = async::spsc_buffer<int>;
auto buf = make_counted<buffer_type>(20, 5);
auto rd1 = async::consumer_resource<int>{buf};
auto rd2 = async::consumer_resource<int>{buf};
auto wr = async::producer_resource<int>{buf};
auto outputs = std::vector<int>{};
auto snk1 = sys.spawn([rd1, &outputs](actor_t* snk) {
snk
->make_observable() //
.from_resource(rd1)
.for_each([&outputs](int x) { outputs.emplace_back(x); });
});
self->monitor(snk1);
run();
auto snk2 = sys.spawn([rd2, &outputs](actor_t* snk) {
snk
->make_observable() //
.from_resource(rd2)
.for_each([&outputs](int x) { outputs.emplace_back(x); });
});
self->monitor(snk2);
run();
expect((down_msg),
to(self).with(down_msg{snk2.address(), sec::runtime_error}));
CHECK(self->mailbox().empty());
CHECK(outputs.empty());
}
}
}
}
#endif // CAF_ENABLE_EXCEPTIONS
END_FIXTURE_SCOPE() END_FIXTURE_SCOPE()
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