Commit 8fa01471 authored by Dominik Charousset's avatar Dominik Charousset

Fix synchronization on the bounded buffer

parent 4c68deb8
...@@ -72,6 +72,7 @@ public: ...@@ -72,6 +72,7 @@ public:
std::unique_lock guard{mtx_}; std::unique_lock guard{mtx_};
CAF_ASSERT(producer_ != nullptr); CAF_ASSERT(producer_ != nullptr);
CAF_ASSERT(!closed_); CAF_ASSERT(!closed_);
CAF_ASSERT(wr_pos_ + items.size() < max_in_flight_ * 2);
std::uninitialized_copy(items.begin(), items.end(), buf_ + wr_pos_); std::uninitialized_copy(items.begin(), items.end(), buf_ + wr_pos_);
wr_pos_ += items.size(); wr_pos_ += items.size();
if (size() == items.size() && consumer_) if (size() == items.size() && consumer_)
...@@ -204,6 +205,7 @@ private: ...@@ -204,6 +205,7 @@ private:
consumer_->on_producer_ready(); consumer_->on_producer_ready();
if (!empty()) if (!empty())
consumer_->on_producer_wakeup(); consumer_->on_producer_wakeup();
signal_demand(max_in_flight_);
} }
size_t empty() const noexcept { size_t empty() const noexcept {
......
...@@ -152,6 +152,24 @@ public: ...@@ -152,6 +152,24 @@ public:
template <class Step> template <class Step>
transformation<Step> transform(Step step); transformation<Step> transform(Step step);
/// Registers a callback for `on_complete` events.
template <class F>
auto do_on_complete(F f) {
return transform(on_complete_step<T, F>{std::move(f)});
}
/// Registers a callback for `on_error` events.
template <class F>
auto do_on_error(F f) {
return transform(on_error_step<T, F>{std::move(f)});
}
/// Registers a callback that runs on `on_complete` or `on_error`.
template <class F>
auto do_finally(F f) {
return transform(finally_step<T, F>{std::move(f)});
}
/// Returns a transformation that selects only the first `n` items. /// Returns a transformation that selects only the first `n` items.
transformation<limit_step<T>> take(size_t n); transformation<limit_step<T>> take(size_t n);
...@@ -969,19 +987,19 @@ public: ...@@ -969,19 +987,19 @@ public:
} }
template <class F> template <class F>
auto do_on_complete(F f) { auto do_on_complete(F f) && {
return std::move(*this) // return std::move(*this) //
.transform(on_complete_step<output_type, F>{std::move(f)}); .transform(on_complete_step<output_type, F>{std::move(f)});
} }
template <class F> template <class F>
auto do_on_error(F f) { auto do_on_error(F f) && {
return std::move(*this) // return std::move(*this) //
.transform(on_error_step<output_type, F>{std::move(f)}); .transform(on_error_step<output_type, F>{std::move(f)});
} }
template <class F> template <class F>
auto do_finally(F f) { auto do_finally(F f) && {
return std::move(*this) // return std::move(*this) //
.transform(finally_step<output_type, F>{std::move(f)}); .transform(finally_step<output_type, F>{std::move(f)});
} }
......
...@@ -21,11 +21,13 @@ void scoped_coordinator::run() { ...@@ -21,11 +21,13 @@ void scoped_coordinator::run() {
}; };
for (;;) { for (;;) {
auto hdl = next(!watched_disposables_.empty()); auto hdl = next(!watched_disposables_.empty());
if (hdl.ptr() != nullptr) if (hdl.ptr() != nullptr) {
hdl.run(); hdl.run();
else drop_disposed_flows();
} else {
return; return;
} }
}
} }
void scoped_coordinator::ref_coordinator() const noexcept { void scoped_coordinator::ref_coordinator() const noexcept {
......
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