Commit a52cde49 authored by Dominik Charousset's avatar Dominik Charousset

Switch to single-item on_next design

While micro-batching inside flow operators seemed like a good idea at
first, it is actually more efficient to pass items directly down the
chain in order emit items as soon as possible to concurrently running
pipelines.
parent 2db16d67
...@@ -244,7 +244,9 @@ public: ...@@ -244,7 +244,9 @@ public:
overflow = 0; overflow = 0;
} }
guard.unlock(); guard.unlock();
dst.on_next(span<const T>{consumer_buf_.data(), n}); auto items = span<const T>{consumer_buf_.data(), n};
for (auto& item : items)
dst.on_next(item);
demand -= n; demand -= n;
consumed += n; consumed += n;
consumer_buf_.clear(); consumer_buf_.clear();
......
...@@ -28,7 +28,6 @@ ...@@ -28,7 +28,6 @@
#include "caf/make_counted.hpp" #include "caf/make_counted.hpp"
#include "caf/ref_counted.hpp" #include "caf/ref_counted.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/span.hpp"
namespace caf::flow { namespace caf::flow {
...@@ -661,13 +660,10 @@ public: ...@@ -661,13 +660,10 @@ public:
sub_ = std::move(sub); sub_ = std::move(sub);
} }
void on_next(span<const input_type> items) override { void on_next(const input_type& item) override {
if (!term_.finalized()) { if (!term_.finalized()) {
auto f = [this, items](auto& step, auto&... steps) { auto f = [this, &item](auto& step, auto&... steps) {
for (auto&& item : items) return step.on_next(item, steps..., term_);
if (!step.on_next(item, steps..., term_))
return false;
return true;
}; };
auto still_running = std::apply(f, steps_); auto still_running = std::apply(f, steps_);
term_.push(); term_.push();
...@@ -1111,8 +1107,10 @@ private: ...@@ -1111,8 +1107,10 @@ private:
drop_if_empty(i); drop_if_empty(i);
} else { } else {
auto& in = *i->second; auto& in = *i->second;
if (!in.buf.empty()) if (!in.buf.empty()) {
term_.on_next(in.buf); for (auto& item : in.buf)
term_.on_next(item);
}
term_.fin(); term_.fin();
for (auto j = inputs_.begin(); j != inputs_.end(); ++j) for (auto j = inputs_.begin(); j != inputs_.end(); ++j)
if (j != i && j->second->sub) if (j != i && j->second->sub)
...@@ -1125,19 +1123,18 @@ private: ...@@ -1125,19 +1123,18 @@ private:
} }
} }
void fwd_on_next(input_key key, span<const T> items) { void fwd_on_next(input_key key, const T& item) {
if (auto i = inputs_.find(key); i != inputs_.end()) { if (auto i = inputs_.find(key); i != inputs_.end()) {
auto& in = *i->second; auto& in = *i->second;
if (!term_.finalized()) { if (!term_.finalized()) {
if (auto n = std::min(term_.min_demand(), items.size()); n > 0) { if (term_.min_demand() > 0) {
term_.on_next(items.subspan(0, n)); term_.on_next(item);
term_.push(); term_.push();
if (in.sub) if (in.sub)
in.sub.request(n); in.sub.request(1);
items = items.subspan(n); } else {
in.buf.emplace_back(item);
} }
if (!items.empty())
in.buf.insert(in.buf.end(), items.begin(), items.end());
} }
} }
} }
...@@ -1158,7 +1155,8 @@ private: ...@@ -1158,7 +1155,8 @@ private:
if (auto m = std::min(in.buf.size(), n); m > 0) { if (auto m = std::min(in.buf.size(), n); m > 0) {
n -= m; n -= m;
auto items = make_span(in.buf.data(), m); auto items = make_span(in.buf.data(), m);
term_.on_next(items); for (auto& item : items)
term_.on_next(item);
in.buf.erase(in.buf.begin(), in.buf.begin() + m); in.buf.erase(in.buf.begin(), in.buf.begin() + m);
if (in.sub) { if (in.sub) {
in.sub.request(m); in.sub.request(m);
...@@ -1278,11 +1276,10 @@ public: ...@@ -1278,11 +1276,10 @@ public:
} }
} }
void on_next(span<const T> observables) override { void on_next(const T& src) override {
if (sub_) { if (sub_) {
for (const auto& x : observables) merger_->add(map_(src).as_observable());
merger_->add(map_(x).as_observable()); sub_.request(1);
sub_.request(observables.size());
} }
} }
...@@ -1474,11 +1471,11 @@ private: ...@@ -1474,11 +1471,11 @@ private:
} }
} }
void fwd_on_next(input_key key, span<const T> items) { void fwd_on_next(input_key key, const T& item) {
if (in_key_ == key && !term_.finalized()) { if (in_key_ == key && !term_.finalized()) {
CAF_ASSERT(in_flight_ >= items.size()); CAF_ASSERT(in_flight_ >= 0);
in_flight_ -= items.size(); --in_flight_;
term_.on_next(items); term_.on_next(item);
term_.push(); term_.push();
} }
} }
...@@ -1576,11 +1573,10 @@ public: ...@@ -1576,11 +1573,10 @@ public:
} }
} }
void on_next(span<const T> observables) override { void on_next(const T& src) override {
if (sub_) { if (sub_) {
for (const auto& x : observables) concat_->add(map_(src).as_observable());
concat_->add(map_(x).as_observable()); sub_.request(1);
sub_.request(observables.size());
} }
} }
...@@ -1740,18 +1736,17 @@ public: ...@@ -1740,18 +1736,17 @@ public:
} }
} }
void on_next(span<const in_t> items) override { void on_next(const in_t& item) override {
if (tail_) { if (tail_) {
tail_.on_next(items); tail_.on_next(item);
} else if (obs_) { } else if (obs_) {
CAF_ASSERT(prefix_.size() + items.size() <= prefix_size_); prefix_.emplace_back(item);
prefix_.insert(prefix_.end(), items.begin(), items.end());
if (prefix_.size() >= prefix_size_) { if (prefix_.size() >= prefix_size_) {
auto tptr = make_broadcaster_impl<in_t>(ctx_); auto tptr = make_broadcaster_impl<in_t>(ctx_);
tail_ = tptr->as_observer(); tail_ = tptr->as_observer();
static_cast<observable_impl<in_t>*>(this)->subscribe(tail_); static_cast<observable_impl<in_t>*>(this)->subscribe(tail_);
auto item = make_cow_tuple(std::move(prefix_), tptr->as_observable()); auto tup = make_cow_tuple(std::move(prefix_), tptr->as_observable());
obs_.on_next(make_span(&item, 1)); obs_.on_next(tup);
obs_.on_complete(); obs_.on_complete();
obs_ = nullptr; obs_ = nullptr;
} }
...@@ -1950,13 +1945,11 @@ private: ...@@ -1950,13 +1945,11 @@ private:
struct decorator { struct decorator {
size_t* demand; size_t* demand;
typename observer<value_type>::impl* dst; typename observer<value_type>::impl* dst;
void on_next(span<const value_type> items) { void on_next(const value_type& item) {
CAF_LOG_TRACE(CAF_ARG(items)); CAF_LOG_TRACE(CAF_ARG(item));
CAF_ASSERT(!items.empty()); CAF_ASSERT(*demand > 0);
CAF_ASSERT(*demand >= items.empty()); --*demand;
*demand -= items.size(); dst->on_next(item);
CAF_LOG_DEBUG("got" << items.size() << "items");
dst->on_next(items);
} }
void on_complete() { void on_complete() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
...@@ -2207,11 +2200,15 @@ public: ...@@ -2207,11 +2200,15 @@ public:
if (items.size() > demand) if (items.size() > demand)
CAF_RAISE_ERROR("observables must not emit more items than requested"); CAF_RAISE_ERROR("observables must not emit more items than requested");
demand -= items.size(); demand -= items.size();
out.on_next(items); for (auto& item : items)
out.on_next(item);
} }
void push(const output_type& item) { void push(const output_type& item) {
push(make_span(&item, 1)); if (demand == 0)
CAF_RAISE_ERROR("observables must not emit more items than requested");
--demand;
out.on_next(item);
} }
template <class... Ts> template <class... Ts>
......
...@@ -17,7 +17,6 @@ ...@@ -17,7 +17,6 @@
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/make_counted.hpp" #include "caf/make_counted.hpp"
#include "caf/ref_counted.hpp" #include "caf/ref_counted.hpp"
#include "caf/span.hpp"
#include "caf/unit.hpp" #include "caf/unit.hpp"
namespace caf::flow { namespace caf::flow {
...@@ -33,7 +32,7 @@ public: ...@@ -33,7 +32,7 @@ public:
virtual void on_subscribe(subscription sub) = 0; virtual void on_subscribe(subscription sub) = 0;
virtual void on_next(span<const T> items) = 0; virtual void on_next(const T& item) = 0;
virtual void on_complete() = 0; virtual void on_complete() = 0;
...@@ -91,8 +90,8 @@ public: ...@@ -91,8 +90,8 @@ public:
} }
/// @pre `valid()` /// @pre `valid()`
void on_next(span<const T> items) { void on_next(const T& item) {
pimpl_->on_next(items); pimpl_->on_next(item);
} }
/// Creates a new observer from `Impl`. /// Creates a new observer from `Impl`.
...@@ -151,33 +150,11 @@ struct on_next_trait; ...@@ -151,33 +150,11 @@ struct on_next_trait;
template <class T> template <class T>
struct on_next_trait<void(T)> { struct on_next_trait<void(T)> {
using value_type = T; using value_type = T;
template <class F>
static void apply(F& f, span<const T> items) {
for (auto&& item : items)
f(item);
}
}; };
template <class T> template <class T>
struct on_next_trait<void(const T&)> { struct on_next_trait<void(const T&)> {
using value_type = T; using value_type = T;
template <class F>
static void apply(F& f, span<const T> items) {
for (auto&& item : items)
f(item);
}
};
template <class T>
struct on_next_trait<void(span<const T>)> {
using value_type = T;
template <class F>
static void apply(F& f, span<const T> items) {
f(items);
}
}; };
template <class F> template <class F>
...@@ -226,10 +203,10 @@ public: ...@@ -226,10 +203,10 @@ public:
this->deref(); this->deref();
} }
void on_next(span<const input_type> items) override { void on_next(const input_type& item) override {
if (!completed_) { if (!completed_) {
on_next_trait_t<OnNext>::apply(on_next_, items); on_next_(item);
sub_.request(items.size()); sub_.request(1);
} }
} }
...@@ -384,10 +361,10 @@ public: ...@@ -384,10 +361,10 @@ public:
// -- implementation of observer<T>::impl ------------------------------------ // -- implementation of observer<T>::impl ------------------------------------
void on_next(span<const value_type> items) override { void on_next(const value_type& item) override {
CAF_LOG_TRACE(CAF_ARG(items)); CAF_LOG_TRACE(CAF_ARG(item));
if (buf_) if (buf_)
buf_->push(items); buf_->push(item);
} }
void on_complete() override { void on_complete() override {
...@@ -509,9 +486,9 @@ public: ...@@ -509,9 +486,9 @@ public:
} }
} }
void on_next(span<const T> items) override { void on_next(const T& item) override {
if (parent) { if (parent) {
parent->fwd_on_next(token, items); parent->fwd_on_next(token, item);
} }
} }
...@@ -607,8 +584,8 @@ public: ...@@ -607,8 +584,8 @@ public:
} }
} }
void on_next(span<const T> items) override { void on_next(const T& item) override {
buf.insert(buf.end(), items.begin(), items.end()); buf.emplace_back(item);
} }
// -- member variables ------------------------------------------------------- // -- member variables -------------------------------------------------------
......
...@@ -53,7 +53,7 @@ public: ...@@ -53,7 +53,7 @@ public:
auto f = detail::make_overload( // auto f = detail::make_overload( //
[i, n](none_t) { i->second += n; }, [i, n](none_t) { i->second += n; },
[this, i](const T& val) { [this, i](const T& val) {
i->first.on_next(make_span(&val, 1)); i->first.on_next(val);
i->first.on_complete(); i->first.on_complete();
observers_.erase(i); observers_.erase(i);
}, },
...@@ -90,7 +90,7 @@ public: ...@@ -90,7 +90,7 @@ public:
pred); pred);
first != observers_.end()) { first != observers_.end()) {
for (auto i = first; i != observers_.end(); ++i) { for (auto i = first; i != observers_.end(); ++i) {
i->first.on_next(make_span(&ref, 1)); i->first.on_next(ref);
i->first.on_complete(); i->first.on_complete();
} }
observers_.erase(first, observers_.end()); observers_.erase(first, observers_.end());
...@@ -151,11 +151,8 @@ public: ...@@ -151,11 +151,8 @@ public:
template <class OnSuccess, class OnError> template <class OnSuccess, class OnError>
void subscribe(OnSuccess on_success, OnError on_error) { void subscribe(OnSuccess on_success, OnError on_error) {
static_assert(std::is_invocable_v<OnSuccess, const T&>); static_assert(std::is_invocable_v<OnSuccess, const T&>);
as_observable().for_each( as_observable().for_each([f{std::move(on_success)}](
[f{std::move(on_success)}](span<const T> items) mutable { const T& item) mutable { f(item); },
CAF_ASSERT(items.size() == 1);
f(items[0]);
},
std::move(on_error)); std::move(on_error));
} }
......
...@@ -461,7 +461,8 @@ public: ...@@ -461,7 +461,8 @@ public:
auto items = span<output_type>{buf_.data(), n}; auto items = span<output_type>{buf_.data(), n};
for (auto& out : outputs_) { for (auto& out : outputs_) {
out.demand -= n; out.demand -= n;
out.sink.on_next(items); for (auto& item : items)
out.sink.on_next(item);
} }
buf_.erase(buf_.begin(), buf_.begin() + n); buf_.erase(buf_.begin(), buf_.begin() + n);
} }
...@@ -515,15 +516,6 @@ public: ...@@ -515,15 +516,6 @@ public:
return true; return true;
} }
void on_next(span<const output_type> items) {
// Note: we may receive more data than what we have requested.
if (in_flight_ >= items.size())
in_flight_ -= items.size();
else
in_flight_ = 0;
buf_.insert(buf_.end(), items.begin(), items.end());
}
void fin() { void fin() {
if (is_active(state_)) { if (is_active(state_)) {
if (idle()) { if (idle()) {
......
...@@ -37,6 +37,12 @@ struct zipper_input { ...@@ -37,6 +37,12 @@ struct zipper_input {
observable<T> in; observable<T> in;
subscription sub; subscription sub;
std::vector<T> buf; std::vector<T> buf;
bool broken = false;
/// Returns whether the input can no longer produce additional items.
bool at_end() const noexcept {
return broken && buf.empty();
}
}; };
/// Combines items from any number of observables using a zip function. /// Combines items from any number of observables using a zip function.
...@@ -66,11 +72,11 @@ public: ...@@ -66,11 +72,11 @@ public:
// -- implementation of disposable::impl ------------------------------------- // -- implementation of disposable::impl -------------------------------------
void dispose() override { void dispose() override {
broken_ = true;
if (buffered() == 0) { if (buffered() == 0) {
fin(); fin();
} else { } else {
for_each_input([](auto, auto& input) { for_each_input([](auto, auto& input) {
input.broken = true;
input.in = nullptr; input.in = nullptr;
if (input.sub) { if (input.sub) {
input.sub.cancel(); input.sub.cancel();
...@@ -152,6 +158,11 @@ private: ...@@ -152,6 +158,11 @@ private:
return fold([](auto&... x) { return std::min({x.buf.size()...}); }); return fold([](auto&... x) { return std::min({x.buf.size()...}); });
} }
// A zipper reached the end if any of its inputs reached the end.
bool at_end() {
return fold([](auto&... inputs) { return (inputs.at_end() || ...); });
}
template <size_t I> template <size_t I>
void fwd_on_subscribe(zipper_index<I> index, subscription sub) { void fwd_on_subscribe(zipper_index<I> index, subscription sub) {
if (!term_.finalized()) { if (!term_.finalized()) {
...@@ -170,31 +181,32 @@ private: ...@@ -170,31 +181,32 @@ private:
template <size_t I> template <size_t I>
void fwd_on_complete(zipper_index<I> index) { void fwd_on_complete(zipper_index<I> index) {
if (!broken_) { auto& input = at(index);
broken_ = true; if (!input.broken) {
if (at(index).buf.empty()) input.broken = true;
input.sub = nullptr;
if (input.buf.empty())
fin(); fin();
} }
at(index).sub = nullptr;
} }
template <size_t I> template <size_t I>
void fwd_on_error(zipper_index<I> index, const error& what) { void fwd_on_error(zipper_index<I> index, const error& what) {
if (!term_.err()) auto& input = at(index);
if (!input.broken) {
if (term_.active() && !term_.err())
term_.err(what); term_.err(what);
if (!broken_) { input.broken = true;
broken_ = true; input.sub = nullptr;
if (at(index).buf.empty()) if (input.buf.empty())
fin(); fin();
} }
at(index).sub = nullptr;
} }
template <size_t I, class T> template <size_t I, class T>
void fwd_on_next(zipper_index<I> index, span<const T> items) { void fwd_on_next(zipper_index<I> index, const T& item) {
if (!term_.finalized()) { if (term_.active()) {
auto& buf = at(index).buf; at(index).buf.push_back(item);
buf.insert(buf.end(), items.begin(), items.end());
push(); push();
} }
} }
...@@ -211,9 +223,9 @@ private: ...@@ -211,9 +223,9 @@ private:
x.buf.erase(x.buf.begin(), x.buf.begin() + n); x.buf.erase(x.buf.begin(), x.buf.begin() + n);
}); });
term_.push(); term_.push();
if (broken_ && buffered() == 0)
fin();
} }
if (at_end())
fin();
} }
void fin() { void fin() {
...@@ -232,9 +244,6 @@ private: ...@@ -232,9 +244,6 @@ private:
F fn_; F fn_;
std::tuple<zipper_input<Ts>...> inputs_; std::tuple<zipper_input<Ts>...> inputs_;
/// A zipper breaks as soon as one of its inputs signals completion or error.
bool broken_ = false;
broadcast_step<output_type> term_; broadcast_step<output_type> term_;
}; };
......
...@@ -117,7 +117,7 @@ disposable interval_impl::subscribe(observer<int64_t> sink) { ...@@ -117,7 +117,7 @@ disposable interval_impl::subscribe(observer<int64_t> sink) {
void interval_impl::fire(interval_action* act) { void interval_impl::fire(interval_action* act) {
if (obs_) { if (obs_) {
--demand_; --demand_;
obs_.on_next(make_span(&val_, 1)); obs_.on_next(val_);
if (++val_ == max_) { if (++val_ == max_) {
obs_.on_complete(); obs_.on_complete();
obs_ = nullptr; obs_ = nullptr;
......
...@@ -83,8 +83,8 @@ public: ...@@ -83,8 +83,8 @@ public:
struct dummy_observer { struct dummy_observer {
template <class T> template <class T>
void on_next(span<const T> items) { void on_next(const T&) {
consumed += items.size(); ++consumed;
} }
void on_error(error what) { void on_error(error what) {
......
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