Commit 95aeba25 authored by Dominik Charousset's avatar Dominik Charousset

Merge branch 'topic/neverlord/coverage'

parents bdc663a9 a50d4e81
...@@ -9,6 +9,14 @@ is based on [Keep a Changelog](https://keepachangelog.com). ...@@ -9,6 +9,14 @@ is based on [Keep a Changelog](https://keepachangelog.com).
- The new classes `json_value`, `json_array` and `json_object` allow working - The new classes `json_value`, `json_array` and `json_object` allow working
with JSON inputs directly. Actors can also pass around JSON values safely. with JSON inputs directly. Actors can also pass around JSON values safely.
- Fused stages now properly forward errors during the initial subscription to
their observer.
### Fixed
- The SPSC buffer now makes sure that subscribers get informed of a producer has
already left before the subscriber appeared and vice versa. This fixes a race
on the buffer that could cause indefinite hanging of an application.
## [0.19.0-rc.1] - 2022-10-31 ## [0.19.0-rc.1] - 2022-10-31
......
...@@ -41,8 +41,19 @@ public: ...@@ -41,8 +41,19 @@ public:
using lock_type = std::unique_lock<std::mutex>; using lock_type = std::unique_lock<std::mutex>;
/// Packs various status flags for the buffer into a single struct.
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;
};
spsc_buffer(uint32_t capacity, uint32_t min_pull_size) spsc_buffer(uint32_t capacity, uint32_t min_pull_size)
: capacity_(capacity), min_pull_size_(min_pull_size) { : capacity_(capacity), min_pull_size_(min_pull_size) {
memset(&flags_, 0, sizeof(flags));
// Allocate some extra space in the buffer in case the producer goes beyond // Allocate some extra space in the buffer in case the producer goes beyond
// the announced capacity. // the announced capacity.
buf_.reserve(capacity + (capacity / 2)); buf_.reserve(capacity + (capacity / 2));
...@@ -58,7 +69,7 @@ public: ...@@ -58,7 +69,7 @@ public:
size_t push(span<const T> items) { size_t push(span<const T> items) {
lock_type guard{mtx_}; lock_type guard{mtx_};
CAF_ASSERT(producer_ != nullptr); CAF_ASSERT(producer_ != nullptr);
CAF_ASSERT(!closed_); CAF_ASSERT(!flags_.closed);
buf_.insert(buf_.end(), items.begin(), items.end()); buf_.insert(buf_.end(), items.begin(), items.end());
if (buf_.size() == items.size() && consumer_) if (buf_.size() == items.size() && consumer_)
consumer_->on_producer_wakeup(); consumer_->on_producer_wakeup();
...@@ -95,7 +106,7 @@ public: ...@@ -95,7 +106,7 @@ public:
/// closed or aborted the flow. /// closed or aborted the flow.
bool has_consumer_event() const noexcept { bool has_consumer_event() const noexcept {
lock_type guard{mtx_}; lock_type guard{mtx_};
return !buf_.empty() || closed_; return !buf_.empty() || flags_.closed;
} }
/// Returns how many items are currently available. This may be greater than /// Returns how many items are currently available. This may be greater than
...@@ -116,7 +127,7 @@ public: ...@@ -116,7 +127,7 @@ public:
void close() { void close() {
lock_type guard{mtx_}; lock_type guard{mtx_};
if (producer_) { if (producer_) {
closed_ = true; flags_.closed = true;
producer_ = nullptr; producer_ = nullptr;
if (buf_.empty() && consumer_) if (buf_.empty() && consumer_)
consumer_->on_producer_wakeup(); consumer_->on_producer_wakeup();
...@@ -128,7 +139,7 @@ public: ...@@ -128,7 +139,7 @@ public:
void abort(error reason) { void abort(error reason) {
lock_type guard{mtx_}; lock_type guard{mtx_};
if (producer_) { if (producer_) {
closed_ = true; flags_.closed = true;
err_ = std::move(reason); err_ = std::move(reason);
producer_ = nullptr; producer_ = nullptr;
if (buf_.empty() && consumer_) if (buf_.empty() && consumer_)
...@@ -153,8 +164,11 @@ public: ...@@ -153,8 +164,11 @@ public:
if (consumer_) if (consumer_)
CAF_RAISE_ERROR("SPSC buffer already has a consumer"); CAF_RAISE_ERROR("SPSC buffer already has a consumer");
consumer_ = std::move(consumer); consumer_ = std::move(consumer);
flags_.had_consumer = true;
if (producer_) if (producer_)
ready(); ready();
else if (flags_.had_producer)
consumer_->on_producer_wakeup();
} }
/// Producer callback for the initial handshake between producer and consumer. /// Producer callback for the initial handshake between producer and consumer.
...@@ -164,8 +178,11 @@ public: ...@@ -164,8 +178,11 @@ public:
if (producer_) if (producer_)
CAF_RAISE_ERROR("SPSC buffer already has a producer"); CAF_RAISE_ERROR("SPSC buffer already has a producer");
producer_ = std::move(producer); producer_ = std::move(producer);
flags_.had_producer = true;
if (consumer_) if (consumer_)
ready(); ready();
else if (flags_.had_consumer)
producer_->on_consumer_cancel();
} }
/// Returns the capacity as passed to the constructor of the buffer. /// Returns the capacity as passed to the constructor of the buffer.
...@@ -195,7 +212,7 @@ public: ...@@ -195,7 +212,7 @@ public:
/// Blocks until there is at least one item available or the producer stopped. /// Blocks until there is at least one item available or the producer stopped.
/// @pre the consumer calls `cv.notify_all()` in its `on_producer_wakeup` /// @pre the consumer calls `cv.notify_all()` in its `on_producer_wakeup`
void await_consumer_ready(lock_type& guard, std::condition_variable& cv) { void await_consumer_ready(lock_type& guard, std::condition_variable& cv) {
while (!closed_ && buf_.empty()) { while (!flags_.closed && buf_.empty()) {
cv.wait(guard); cv.wait(guard);
} }
} }
...@@ -206,7 +223,7 @@ public: ...@@ -206,7 +223,7 @@ public:
template <class TimePoint> template <class TimePoint>
bool await_consumer_ready(lock_type& guard, std::condition_variable& cv, bool await_consumer_ready(lock_type& guard, std::condition_variable& cv,
TimePoint timeout) { TimePoint timeout) {
while (!closed_ && buf_.empty()) while (!flags_.closed && buf_.empty())
if (cv.wait_until(guard, timeout) == std::cv_status::timeout) if (cv.wait_until(guard, timeout) == std::cv_status::timeout)
return false; return false;
return true; return true;
...@@ -248,7 +265,7 @@ public: ...@@ -248,7 +265,7 @@ public:
guard.lock(); guard.lock();
overflow = buf_.size() <= capacity_ ? 0u : buf_.size() - capacity_; overflow = buf_.size() <= capacity_ ? 0u : buf_.size() - capacity_;
} }
if (!buf_.empty() || !closed_) { if (!buf_.empty() || !flags_.closed) {
return {true, consumed}; return {true, consumed};
} else { } else {
consumer_ = nullptr; consumer_ = nullptr;
...@@ -298,7 +315,7 @@ private: ...@@ -298,7 +315,7 @@ private:
uint32_t demand_ = 0; uint32_t demand_ = 0;
/// Stores whether `close` has been called. /// Stores whether `close` has been called.
bool closed_ = false; flags flags_;
/// Stores the abort reason. /// Stores the abort reason.
error err_; error err_;
......
...@@ -37,7 +37,8 @@ public: ...@@ -37,7 +37,8 @@ public:
if (!val) { if (!val) {
step.on_complete(steps...); step.on_complete(steps...);
return; return;
} else if (!step.on_next(*val, steps...)) }
if (!step.on_next(*val, steps...))
return; return;
} else { } else {
if (!step.on_next(fn_(), steps...)) if (!step.on_next(fn_(), steps...))
......
...@@ -35,8 +35,10 @@ public: ...@@ -35,8 +35,10 @@ public:
} }
~from_resource_sub() { ~from_resource_sub() {
if (buf_) // The buffer points back to this object as consumer, so this cannot be
buf_->cancel(); // 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(); ctx_->deref_execution_context();
} }
...@@ -51,7 +53,7 @@ public: ...@@ -51,7 +53,7 @@ public:
if (!disposed_) { if (!disposed_) {
disposed_ = true; disposed_ = true;
if (!running_) if (!running_)
do_cancel(); do_dispose();
} }
} }
...@@ -114,7 +116,7 @@ private: ...@@ -114,7 +116,7 @@ private:
} }
} }
void do_cancel() { void do_dispose() {
if (buf_) { if (buf_) {
buf_->cancel(); buf_->cancel();
buf_ = nullptr; buf_ = nullptr;
...@@ -129,7 +131,7 @@ private: ...@@ -129,7 +131,7 @@ private:
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
auto guard = detail::make_scope_guard([this] { running_ = false; }); auto guard = detail::make_scope_guard([this] { running_ = false; });
if (disposed_) { if (disposed_) {
do_cancel(); do_dispose();
return; return;
} }
CAF_ASSERT(out_); CAF_ASSERT(out_);
...@@ -142,7 +144,7 @@ private: ...@@ -142,7 +144,7 @@ private:
disposed_ = true; disposed_ = true;
return; return;
} else if (disposed_) { } else if (disposed_) {
do_cancel(); do_dispose();
return; return;
} else if (pulled == 0) { } else if (pulled == 0) {
return; return;
......
...@@ -47,9 +47,10 @@ public: ...@@ -47,9 +47,10 @@ public:
} }
void on_error(const error& what) { void on_error(const error& what) {
CAF_ASSERT(sub->in_.valid()); if (sub->in_) {
sub->in_.dispose(); sub->in_.dispose();
sub->in_ = nullptr; sub->in_ = nullptr;
}
sub->err_ = what; sub->err_ = what;
} }
}; };
...@@ -137,6 +138,19 @@ public: ...@@ -137,6 +138,19 @@ public:
void on_error(const error& what) override { void on_error(const error& what) override {
if (in_) { if (in_) {
if (!err_) {
auto fn = [this, &what](auto& step, auto&... steps) {
term_step term{this};
step.on_error(what, steps..., term);
};
std::apply(fn, steps_);
if (!running_) {
running_ = true;
do_run();
}
}
} else if (out_) {
// This may only happen if subscribing to the input fails.
auto fn = [this, &what](auto& step, auto&... steps) { auto fn = [this, &what](auto& step, auto&... steps) {
term_step term{this}; term_step term{this};
step.on_error(what, steps..., term); step.on_error(what, steps..., term);
...@@ -273,19 +287,10 @@ public: ...@@ -273,19 +287,10 @@ public:
auto ptr = make_counted<sub_t>(super::ctx_, out, steps_); auto ptr = make_counted<sub_t>(super::ctx_, out, steps_);
input_->subscribe(observer<input_type>{ptr}); input_->subscribe(observer<input_type>{ptr});
if (ptr->subscribed()) { if (ptr->subscribed()) {
auto sub = subscription{std::move(ptr)}; out.on_subscribe(subscription{ptr});
out.on_subscribe(sub); return ptr->as_disposable();
return std::move(sub).as_disposable();
} else if (auto& fail_reason = ptr->fail_reason()) {
out.on_error(fail_reason);
return disposable{};
} else {
auto err = make_error(sec::invalid_observable,
"flow operator from_steps failed "
"to subscribe to its input");
out.on_error(err);
return disposable{};
} }
return disposable{};
} }
private: private:
......
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