Commit e7a0c077 authored by Dominik Charousset's avatar Dominik Charousset

Integrate review feedback

parent 153aa37e
...@@ -155,7 +155,7 @@ public: ...@@ -155,7 +155,7 @@ public:
CAF_ASSERT(consumer != nullptr); CAF_ASSERT(consumer != nullptr);
lock_type guard{mtx_}; lock_type guard{mtx_};
if (consumer_) if (consumer_)
CAF_RAISE_ERROR("SPSC buffer already has a consumer"); CAF_RAISE_ERROR(std::logic_error, "SPSC buffer already has a consumer");
consumer_ = std::move(consumer); consumer_ = std::move(consumer);
if (producer_) if (producer_)
ready(); ready();
...@@ -167,10 +167,8 @@ public: ...@@ -167,10 +167,8 @@ public:
void set_producer(producer_ptr producer) { void set_producer(producer_ptr producer) {
CAF_ASSERT(producer != nullptr); CAF_ASSERT(producer != nullptr);
lock_type guard{mtx_}; lock_type guard{mtx_};
if (producer_) { if (producer_)
producer->on_consumer_cancel(); CAF_RAISE_ERROR(std::logic_error, "SPSC buffer already has a producer");
return;
}
producer_ = std::move(producer); producer_ = std::move(producer);
if (consumer_) if (consumer_)
ready(); ready();
...@@ -349,13 +347,12 @@ struct resource_ctrl : ref_counted { ...@@ -349,13 +347,12 @@ struct resource_ctrl : ref_counted {
} }
buffer_ptr try_open() { buffer_ptr try_open() {
auto res = buffer_ptr{};
std::unique_lock guard{mtx}; std::unique_lock guard{mtx};
if (buf) { if (buf) {
auto res = buffer_ptr{};
res.swap(buf); res.swap(buf);
return res;
} }
return nullptr; return res;
} }
mutable std::mutex mtx; mutable std::mutex mtx;
...@@ -428,6 +425,16 @@ public: ...@@ -428,6 +425,16 @@ public:
return ctrl_ == nullptr; return ctrl_ == nullptr;
} }
friend bool operator==(const consumer_resource& lhs,
const consumer_resource& rhs) {
return lhs.ctrl_ == rhs.ctrl_;
}
friend bool operator!=(const consumer_resource& lhs,
const consumer_resource& rhs) {
return lhs.ctrl_ != rhs.ctrl_;
}
private: private:
intrusive_ptr<resource_ctrl<T, false>> ctrl_; intrusive_ptr<resource_ctrl<T, false>> ctrl_;
}; };
...@@ -496,6 +503,16 @@ public: ...@@ -496,6 +503,16 @@ public:
return ctrl_ == nullptr; return ctrl_ == nullptr;
} }
friend bool operator==(const producer_resource& lhs,
const producer_resource& rhs) {
return lhs.ctrl_ == rhs.ctrl_;
}
friend bool operator!=(const producer_resource& lhs,
const producer_resource& rhs) {
return lhs.ctrl_ != rhs.ctrl_;
}
private: private:
intrusive_ptr<resource_ctrl<T, true>> ctrl_; intrusive_ptr<resource_ctrl<T, true>> ctrl_;
}; };
......
...@@ -477,8 +477,8 @@ disposable observable<T>::subscribe(async::producer_resource<T> resource) { ...@@ -477,8 +477,8 @@ disposable observable<T>::subscribe(async::producer_resource<T> resource) {
using adapter_type = buffer_writer_impl<buffer_type>; using adapter_type = buffer_writer_impl<buffer_type>;
if (auto buf = resource.try_open()) { if (auto buf = resource.try_open()) {
CAF_LOG_DEBUG("subscribe producer resource to flow"); CAF_LOG_DEBUG("subscribe producer resource to flow");
auto adapter = make_counted<adapter_type>(pimpl_->ctx(), buf); auto adapter = make_counted<adapter_type>(pimpl_->ctx());
buf->set_producer(adapter); adapter->init(buf);
auto obs = adapter->as_observer(); auto obs = adapter->as_observer();
auto sub = subscribe(std::move(obs)); auto sub = subscribe(std::move(obs));
pimpl_->ctx()->watch(sub); pimpl_->ctx()->watch(sub);
...@@ -754,8 +754,8 @@ async::consumer_resource<T> ...@@ -754,8 +754,8 @@ async::consumer_resource<T>
observable<T>::to_resource(size_t buffer_size, size_t min_request_size) { observable<T>::to_resource(size_t buffer_size, size_t min_request_size) {
using buffer_type = async::spsc_buffer<T>; using buffer_type = async::spsc_buffer<T>;
auto buf = make_counted<buffer_type>(buffer_size, min_request_size); auto buf = make_counted<buffer_type>(buffer_size, min_request_size);
auto up = make_counted<buffer_writer_impl<buffer_type>>(pimpl_->ctx(), buf); auto up = make_counted<buffer_writer_impl<buffer_type>>(pimpl_->ctx());
buf->set_producer(up); up->init(buf);
subscribe(up->as_observer()); subscribe(up->as_observer());
return async::consumer_resource<T>{std::move(buf)}; return async::consumer_resource<T>{std::move(buf)};
} }
......
...@@ -287,10 +287,8 @@ public: ...@@ -287,10 +287,8 @@ public:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
buffer_writer_impl(coordinator* ctx, buffer_ptr buf) buffer_writer_impl(coordinator* ctx) : ctx_(ctx) {
: ctx_(ctx), buf_(std::move(buf)) {
CAF_ASSERT(ctx_ != nullptr); CAF_ASSERT(ctx_ != nullptr);
CAF_ASSERT(buf_ != nullptr);
} }
~buffer_writer_impl() { ~buffer_writer_impl() {
...@@ -298,6 +296,15 @@ public: ...@@ -298,6 +296,15 @@ public:
buf_->close(); buf_->close();
} }
void init(buffer_ptr buf) {
// This step is a bit subtle. Basically, buf->set_producer might throw, in
// which case we must not set buf_ to avoid closing a buffer that we don't
// actually own.
CAF_ASSERT(buf != nullptr);
buf->set_producer(this);
buf_ = std::move(buf);
}
// -- intrusive_ptr interface ------------------------------------------------ // -- intrusive_ptr interface ------------------------------------------------
friend void intrusive_ptr_add_ref(const buffer_writer_impl* ptr) noexcept { friend void intrusive_ptr_add_ref(const buffer_writer_impl* ptr) noexcept {
......
...@@ -108,6 +108,44 @@ struct dummy_observer { ...@@ -108,6 +108,44 @@ struct dummy_observer {
BEGIN_FIXTURE_SCOPE(test_coordinator_fixture<>) BEGIN_FIXTURE_SCOPE(test_coordinator_fixture<>)
TEST_CASE("resources may be copied") {
auto [rd, wr] = async::make_spsc_buffer_resource<int>(6, 2);
// Test copy constructor.
async::consumer_resource<int> rd2{rd};
CHECK_EQ(rd, rd2);
async::producer_resource<int> wr2{wr};
CHECK_EQ(wr, wr2);
// Test copy-assignment.
async::consumer_resource<int> rd3;
CHECK_NE(rd2, rd3);
rd3 = rd2;
CHECK_EQ(rd2, rd3);
async::producer_resource<int> wr3;
CHECK_NE(wr2, wr3);
wr3 = wr2;
CHECK_EQ(wr2, wr3);
}
TEST_CASE("resources may be moved") {
auto [rd, wr] = async::make_spsc_buffer_resource<int>(6, 2);
CHECK(rd);
CHECK(wr);
// Test move constructor.
async::consumer_resource<int> rd2{std::move(rd)};
CHECK(!rd);
CHECK(rd2);
async::producer_resource<int> wr2{std::move(wr)};
CHECK(!wr);
CHECK(wr2);
// Test move-assignment.
async::consumer_resource<int> rd3{std::move(rd2)};
CHECK(!rd2);
CHECK(rd3);
async::producer_resource<int> wr3{std::move(wr2)};
CHECK(!wr2);
CHECK(wr3);
}
SCENARIO("SPSC buffers may go past their capacity") { SCENARIO("SPSC buffers may go past their capacity") {
GIVEN("an SPSC buffer with consumer and produer") { GIVEN("an SPSC buffer with consumer and produer") {
auto prod = make_counted<dummy_producer>(); auto prod = make_counted<dummy_producer>();
...@@ -379,29 +417,15 @@ SCENARIO("resources are invalid after calling try_open") { ...@@ -379,29 +417,15 @@ SCENARIO("resources are invalid after calling try_open") {
CHECK_NE(rd.try_open(), nullptr); CHECK_NE(rd.try_open(), nullptr);
CHECK(!rd); CHECK(!rd);
CHECK_EQ(rd.try_open(), nullptr); 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") { SCENARIO("producer resources may be subscribed to flows only once") {
GIVEN("a producer resource") { GIVEN("a producer resource") {
WHEN("opening it twice") { WHEN("subscribing it to a flow twice") {
THEN("the second try_open fails") { THEN("the second attempt results in a canceled subscription") {
using actor_t = event_based_actor; using actor_t = event_based_actor;
auto outputs = std::vector<int>{}; auto outputs = std::vector<int>{};
auto [rd, wr] = async::make_spsc_buffer_resource<int>(6, 2); auto [rd, wr] = async::make_spsc_buffer_resource<int>(6, 2);
...@@ -428,10 +452,10 @@ SCENARIO("try_open on producer resources succeeds only once") { ...@@ -428,10 +452,10 @@ SCENARIO("try_open on producer resources succeeds only once") {
} }
} }
SCENARIO("try_open on consumer resources succeeds only once") { SCENARIO("consumer resources may be converted to flows only once") {
GIVEN("a consumer resource") { GIVEN("a consumer resource") {
WHEN("opening it twice") { WHEN("making an observable from the resource") {
THEN("the second try_open fails") { THEN("the second attempt results in an empty observable") {
using actor_t = event_based_actor; using actor_t = event_based_actor;
auto outputs = std::vector<int>{}; auto outputs = std::vector<int>{};
auto [rd, wr] = async::make_spsc_buffer_resource<int>(6, 2); auto [rd, wr] = async::make_spsc_buffer_resource<int>(6, 2);
...@@ -459,6 +483,8 @@ SCENARIO("try_open on consumer resources succeeds only once") { ...@@ -459,6 +483,8 @@ SCENARIO("try_open on consumer resources succeeds only once") {
} }
} }
#ifdef CAF_ENABLE_EXCEPTIONS
// Note: this basically checks that buffer protects against misuse and is not // Note: this basically checks that buffer protects against misuse and is not
// how users should do things. // how users should do things.
SCENARIO("SPSC buffers reject multiple producers") { SCENARIO("SPSC buffers reject multiple producers") {
...@@ -482,15 +508,15 @@ SCENARIO("SPSC buffers reject multiple producers") { ...@@ -482,15 +508,15 @@ SCENARIO("SPSC buffers reject multiple producers") {
}); });
self->monitor(prod2); self->monitor(prod2);
run(); run();
expect((down_msg), to(self).with(down_msg{prod2.address(), error{}})); // prod2 dies immediately to the exception.
expect((down_msg),
to(self).with(down_msg{prod2.address(), sec::runtime_error}));
CHECK(self->mailbox().empty()); CHECK(self->mailbox().empty());
} }
} }
} }
} }
#ifdef CAF_ENABLE_EXCEPTIONS
// Note: this basically checks that buffer protects against misuse and is not // Note: this basically checks that buffer protects against misuse and is not
// how users should do things. // how users should do things.
SCENARIO("SPSC buffers reject multiple consumers") { SCENARIO("SPSC buffers reject multiple consumers") {
...@@ -521,6 +547,7 @@ SCENARIO("SPSC buffers reject multiple consumers") { ...@@ -521,6 +547,7 @@ SCENARIO("SPSC buffers reject multiple consumers") {
}); });
self->monitor(snk2); self->monitor(snk2);
run(); run();
// snk2 dies immediately to the exception.
expect((down_msg), expect((down_msg),
to(self).with(down_msg{snk2.address(), sec::runtime_error})); to(self).with(down_msg{snk2.address(), sec::runtime_error}));
CHECK(self->mailbox().empty()); CHECK(self->mailbox().empty());
......
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