Commit d3458b2a authored by Dominik Charousset's avatar Dominik Charousset

Fix producer and consumer handling in SPSC buffer

parent 02ab9698
......@@ -45,6 +45,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.
## [0.19.0-rc.1] - 2022-10-31
......
......@@ -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();
......@@ -164,10 +157,9 @@ public:
if (consumer_)
CAF_RAISE_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();
}
......@@ -175,13 +167,14 @@ public:
void set_producer(producer_ptr producer) {
CAF_ASSERT(producer != nullptr);
lock_type guard{mtx_};
if (producer_)
CAF_RAISE_ERROR("SPSC buffer already has a producer");
if (producer_) {
producer->on_consumer_cancel();
return;
}
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 +340,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();
......@@ -431,6 +424,10 @@ public:
return ctrl_ != nullptr;
}
bool operator!() const noexcept {
return ctrl_ == nullptr;
}
private:
intrusive_ptr<resource_ctrl<T, false>> ctrl_;
};
......@@ -485,10 +482,20 @@ 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;
}
private:
intrusive_ptr<resource_ctrl<T, true>> ctrl_;
};
......
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