Commit d9639b60 authored by Dominik Charousset's avatar Dominik Charousset

Merge pull request #1382

parents e6f6dab9 e7a0c077
......@@ -47,6 +47,9 @@ is based on [Keep a Changelog](https://keepachangelog.com).
could end up in a long-running read loop. To avoid potentially starving other
actors or activities, scheduled actors now limit the amount of actions that
may run in one iteration (#1364).
- Destroying a consumer or producer resource before opening it lead to a stall
of the consumer / producer. The buffer now keeps track of whether `close` or
`abort` were called prior to consumers or producers attaching.
### Deprecated
......
......@@ -45,10 +45,8 @@ public:
struct flags {
/// Stores whether `close` has been called.
bool closed : 1;
/// Stores whether the buffer had a consumer at some point.
bool had_consumer : 1;
/// Stores whether the buffer had a producer at some point.
bool had_producer : 1;
/// Stores whether `cancel` has been called.
bool canceled : 1;
};
spsc_buffer(uint32_t capacity, uint32_t min_pull_size)
......@@ -125,20 +123,14 @@ public:
/// Closes the buffer by request of the producer.
void close() {
lock_type guard{mtx_};
if (producer_) {
flags_.closed = true;
producer_ = nullptr;
if (buf_.empty() && consumer_)
consumer_->on_producer_wakeup();
}
abort(error{});
}
/// Closes the buffer by request of the producer and signals an error to the
/// consumer.
void abort(error reason) {
lock_type guard{mtx_};
if (producer_) {
if (!flags_.closed) {
flags_.closed = true;
err_ = std::move(reason);
producer_ = nullptr;
......@@ -150,7 +142,8 @@ public:
/// Closes the buffer by request of the consumer.
void cancel() {
lock_type guard{mtx_};
if (consumer_) {
if (!flags_.canceled) {
flags_.canceled = true;
consumer_ = nullptr;
if (producer_)
producer_->on_consumer_cancel();
......@@ -162,12 +155,11 @@ public:
CAF_ASSERT(consumer != nullptr);
lock_type guard{mtx_};
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);
flags_.had_consumer = true;
if (producer_)
ready();
else if (flags_.had_producer)
else if (flags_.closed)
consumer_->on_producer_wakeup();
}
......@@ -176,12 +168,11 @@ public:
CAF_ASSERT(producer != nullptr);
lock_type guard{mtx_};
if (producer_)
CAF_RAISE_ERROR("SPSC buffer already has a producer");
CAF_RAISE_ERROR(std::logic_error, "SPSC buffer already has a producer");
producer_ = std::move(producer);
flags_.had_producer = true;
if (consumer_)
ready();
else if (flags_.had_consumer)
else if (flags_.canceled)
producer_->on_consumer_cancel();
}
......@@ -347,7 +338,7 @@ struct resource_ctrl : ref_counted {
if (buf) {
if constexpr (IsProducer) {
auto err = make_error(sec::disposed,
"producer_resource destroyed without opening it");
"destroyed producer_resource without opening it");
buf->abort(err);
} else {
buf->cancel();
......@@ -356,13 +347,12 @@ struct resource_ctrl : ref_counted {
}
buffer_ptr try_open() {
auto res = buffer_ptr{};
std::unique_lock guard{mtx};
if (buf) {
auto res = buffer_ptr{};
res.swap(buf);
return res;
}
return nullptr;
return res;
}
mutable std::mutex mtx;
......@@ -431,6 +421,20 @@ public:
return ctrl_ != nullptr;
}
bool operator!() const noexcept {
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:
intrusive_ptr<resource_ctrl<T, false>> ctrl_;
};
......@@ -485,10 +489,30 @@ public:
buf->close();
}
/// Calls `try_open` and on success immediately calls `abort` on the buffer.
void abort(error reason) {
if (auto buf = try_open())
buf->abort(std::move(reason));
}
explicit operator bool() const noexcept {
return ctrl_ != nullptr;
}
bool operator!() const noexcept {
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:
intrusive_ptr<resource_ctrl<T, true>> ctrl_;
};
......
......@@ -477,8 +477,8 @@ disposable observable<T>::subscribe(async::producer_resource<T> resource) {
using adapter_type = buffer_writer_impl<buffer_type>;
if (auto buf = resource.try_open()) {
CAF_LOG_DEBUG("subscribe producer resource to flow");
auto adapter = make_counted<adapter_type>(pimpl_->ctx(), buf);
buf->set_producer(adapter);
auto adapter = make_counted<adapter_type>(pimpl_->ctx());
adapter->init(buf);
auto obs = adapter->as_observer();
auto sub = subscribe(std::move(obs));
pimpl_->ctx()->watch(sub);
......@@ -754,8 +754,8 @@ async::consumer_resource<T>
observable<T>::to_resource(size_t buffer_size, size_t min_request_size) {
using buffer_type = async::spsc_buffer<T>;
auto buf = make_counted<buffer_type>(buffer_size, min_request_size);
auto up = make_counted<buffer_writer_impl<buffer_type>>(pimpl_->ctx(), buf);
buf->set_producer(up);
auto up = make_counted<buffer_writer_impl<buffer_type>>(pimpl_->ctx());
up->init(buf);
subscribe(up->as_observer());
return async::consumer_resource<T>{std::move(buf)};
}
......
......@@ -287,10 +287,8 @@ public:
// -- constructors, destructors, and assignment operators --------------------
buffer_writer_impl(coordinator* ctx, buffer_ptr buf)
: ctx_(ctx), buf_(std::move(buf)) {
buffer_writer_impl(coordinator* ctx) : ctx_(ctx) {
CAF_ASSERT(ctx_ != nullptr);
CAF_ASSERT(buf_ != nullptr);
}
~buffer_writer_impl() {
......@@ -298,6 +296,15 @@ public:
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 ------------------------------------------------
friend void intrusive_ptr_add_ref(const buffer_writer_impl* ptr) noexcept {
......
......@@ -35,10 +35,6 @@ public:
}
~from_resource_sub() {
// The buffer points back to this object as consumer, so this cannot be
// destroyed unless we have called buf_->cancel(). All code paths that do
// call cancel() on the buffer also must set the variable to `nullptr`.
CAF_ASSERT(buf_ == nullptr);
ctx_->deref_execution_context();
}
......
This diff is collapsed.
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