Commit c04371ed authored by Dominik Charousset's avatar Dominik Charousset

Merge topic/neverlord/prefix-and-tail

parents 56d43a75 59070cf4
...@@ -26,7 +26,7 @@ ...@@ -26,7 +26,7 @@
#include "caf/flow/op/interval.hpp" #include "caf/flow/op/interval.hpp"
#include "caf/flow/op/merge.hpp" #include "caf/flow/op/merge.hpp"
#include "caf/flow/op/never.hpp" #include "caf/flow/op/never.hpp"
#include "caf/flow/op/prefetch.hpp" #include "caf/flow/op/prefix_and_tail.hpp"
#include "caf/flow/op/publish.hpp" #include "caf/flow/op/publish.hpp"
#include "caf/flow/step/all.hpp" #include "caf/flow/step/all.hpp"
#include "caf/flow/subscription.hpp" #include "caf/flow/subscription.hpp"
...@@ -681,33 +681,19 @@ auto observable<T>::concat_map(F f) { ...@@ -681,33 +681,19 @@ auto observable<T>::concat_map(F f) {
template <class T> template <class T>
observable<cow_tuple<cow_vector<T>, observable<T>>> observable<cow_tuple<cow_vector<T>, observable<T>>>
observable<T>::prefix_and_tail(size_t n) { observable<T>::prefix_and_tail(size_t n) {
using vector_t = cow_vector<T>; using impl_t = op::prefix_and_tail<T>;
CAF_ASSERT(n > 0); return make_observable<impl_t>(ctx(), as_observable(), n);
auto do_prefetch = [](auto in) {
auto ptr = op::prefetch<T>::apply(std::move(in).as_observable().pimpl());
return observable<T>{std::move(ptr)};
};
auto split = share(2);
auto tail = split.skip(n).compose(do_prefetch);
return split //
.take(n)
.to_vector()
.filter([n](const vector_t& xs) { return xs.size() == n; })
.map([tail](const vector_t& xs) { return make_cow_tuple(xs, tail); })
.as_observable();
} }
template <class T> template <class T>
observable<cow_tuple<T, observable<T>>> observable<T>::head_and_tail() { observable<cow_tuple<T, observable<T>>> observable<T>::head_and_tail() {
auto do_prefetch = [](auto in) { using prefix_tuple_t = cow_tuple<cow_vector<T>, observable<T>>;
auto ptr = op::prefetch<T>::apply(std::move(in).as_observable().pimpl()); return prefix_and_tail(1)
return observable<T>{std::move(ptr)}; .map([](const prefix_tuple_t& tup) {
}; auto& [vec, obs] = tup.data();
auto split = share(2); CAF_ASSERT(vec.size() == 1);
auto tail = split.skip(1).compose(do_prefetch); return make_cow_tuple(vec.front(), obs);
return split // })
.take(1)
.map([tail](const T& x) { return make_cow_tuple(x, tail); })
.as_observable(); .as_observable();
} }
......
...@@ -8,6 +8,7 @@ ...@@ -8,6 +8,7 @@
#include "caf/flow/observer.hpp" #include "caf/flow/observer.hpp"
#include "caf/flow/op/empty.hpp" #include "caf/flow/op/empty.hpp"
#include "caf/flow/op/hot.hpp" #include "caf/flow/op/hot.hpp"
#include "caf/flow/op/ucast.hpp"
#include "caf/flow/subscription.hpp" #include "caf/flow/subscription.hpp"
#include "caf/intrusive_ptr.hpp" #include "caf/intrusive_ptr.hpp"
#include "caf/make_counted.hpp" #include "caf/make_counted.hpp"
...@@ -20,114 +21,7 @@ namespace caf::flow::op { ...@@ -20,114 +21,7 @@ namespace caf::flow::op {
/// State shared between one multicast operator and one subscribed observer. /// State shared between one multicast operator and one subscribed observer.
template <class T> template <class T>
class mcast_sub_state : public detail::plain_ref_counted { using mcast_sub_state = ucast_sub_state<T>;
public:
friend void intrusive_ptr_add_ref(const mcast_sub_state* ptr) noexcept {
ptr->ref();
}
friend void intrusive_ptr_release(const mcast_sub_state* ptr) noexcept {
ptr->deref();
}
mcast_sub_state(coordinator* ctx, observer<T> out)
: ctx(ctx), out(std::move(out)) {
// nop
}
coordinator* ctx;
std::deque<T> buf;
size_t demand = 0;
observer<T> out;
bool disposed = false;
bool closed = false;
bool running = false;
error err;
action when_disposed;
action when_consumed_some;
void push(const T& item) {
if (disposed) {
// nop
} else if (demand > 0 && !running) {
CAF_ASSERT(out);
CAF_ASSERT(buf.empty());
--demand;
out.on_next(item);
if (when_consumed_some)
ctx->delay(when_consumed_some);
} else {
buf.push_back(item);
}
}
void close() {
if (!disposed) {
closed = true;
if (!running && buf.empty()) {
disposed = true;
out.on_complete();
out = nullptr;
when_disposed = nullptr;
when_consumed_some = nullptr;
}
}
}
void abort(const error& reason) {
if (!disposed && !err) {
closed = true;
err = reason;
if (!running && buf.empty()) {
disposed = true;
out.on_error(reason);
out = nullptr;
when_disposed = nullptr;
when_consumed_some = nullptr;
}
}
}
void do_dispose() {
if (out) {
out.on_complete();
out = nullptr;
}
if (when_disposed) {
ctx->delay(std::move(when_disposed));
}
if (when_consumed_some) {
when_consumed_some.dispose();
when_consumed_some = nullptr;
}
buf.clear();
demand = 0;
disposed = true;
}
void do_run() {
auto guard = detail::make_scope_guard([this] { running = false; });
if (!disposed) {
auto got_some = demand > 0 && !buf.empty();
for (bool run = got_some; run; run = demand > 0 && !buf.empty()) {
out.on_next(buf.front());
buf.pop_front();
--demand;
}
if (buf.empty() && closed) {
if (err)
out.on_error(err);
else
out.on_complete();
out = nullptr;
do_dispose();
} else if (got_some && when_consumed_some) {
ctx->delay(when_consumed_some);
}
}
}
};
template <class T> template <class T>
using mcast_sub_state_ptr = intrusive_ptr<mcast_sub_state<T>>; using mcast_sub_state_ptr = intrusive_ptr<mcast_sub_state<T>>;
......
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/error.hpp"
#include "caf/flow/observer.hpp"
#include "caf/flow/op/hot.hpp"
#include "caf/flow/subscription.hpp"
#include "caf/sec.hpp"
#include <utility>
namespace caf::flow::op {
/// Allows operators to subscribe to an observable immediately to force an eager
/// subscription while the observable that actually consumes the items
/// subscribes later. May only be subscribed once.
template <class T>
class prefetch : public hot<T>, public observer_impl<T> {
public:
// -- member types -----------------------------------------------------------
using super = hot<T>;
using output_type = T;
// -- constructors, destructors, and assignment operators --------------------
explicit prefetch(coordinator* ctx) : super(ctx) {
// nop
}
// -- ref counting (and disambiguation due to multiple base types) -----------
void ref_coordinated() const noexcept override {
this->ref();
}
void deref_coordinated() const noexcept override {
this->deref();
}
friend void intrusive_ptr_add_ref(const prefetch* ptr) noexcept {
ptr->ref_coordinated();
}
friend void intrusive_ptr_release(const prefetch* ptr) noexcept {
ptr->deref_coordinated();
}
// -- implementation of observable_impl<T> -----------------------------------
disposable subscribe(observer<T> out) override {
if (completed_) {
if (err_)
out.on_error(err_);
else
out.on_complete();
return {};
} else if (!out_ && sub_) {
out_ = std::move(out);
out_.on_subscribe(sub_);
return sub_.as_disposable();
} else {
auto err = make_error(sec::invalid_observable,
"prefetch cannot add more than one subscriber");
out.on_error(err);
return {};
}
}
// -- implementation of observer_impl<T> -------------------------------------
void on_next(const T& item) override {
out_.on_next(item);
}
void on_complete() override {
completed_ = true;
if (out_) {
out_.on_complete();
out_ = nullptr;
sub_ = nullptr;
}
}
void on_error(const error& what) override {
completed_ = true;
err_ = what;
if (out_) {
out_.on_error(what);
out_ = nullptr;
sub_ = nullptr;
}
}
void on_subscribe(subscription sub) override {
if (!sub_) {
sub_ = sub;
} else {
sub.dispose();
}
}
// -- convenience functions --------------------------------------------------
static intrusive_ptr<base<T>> apply(intrusive_ptr<base<T>> src) {
auto ptr = make_counted<prefetch>(src->ctx());
src->subscribe(observer<T>{ptr});
return ptr;
}
private:
bool completed_ = false;
error err_;
observer<T> out_;
subscription sub_;
};
} // namespace caf::flow::op
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/cow_tuple.hpp"
#include "caf/cow_vector.hpp"
#include "caf/flow/coordinator.hpp"
#include "caf/flow/observer.hpp"
#include "caf/flow/op/cold.hpp"
#include "caf/flow/op/ucast.hpp"
#include "caf/flow/subscription.hpp"
#include "caf/intrusive_ptr.hpp"
#include "caf/make_counted.hpp"
#include <memory>
#include <vector>
namespace caf::flow::op {
/// @relates prefix_and_tail
template <class T>
class prefix_and_tail_sub : public detail::plain_ref_counted,
public observer_impl<T>,
public subscription_impl {
public:
using tuple_t = cow_tuple<cow_vector<T>, observable<T>>;
prefix_and_tail_sub(coordinator* ctx, observer<tuple_t> out,
size_t prefix_size)
: ctx_(ctx), out_(std::move(out)), prefix_size_(prefix_size) {
prefix_buf_.reserve(prefix_size);
}
void ref_coordinated() const noexcept override {
ref();
}
void deref_coordinated() const noexcept override {
deref();
}
void on_next(const T& item) override {
if (sink_) {
// We're in tail mode: push to sink.
CAF_ASSERT(in_flight_ > 0);
--in_flight_;
sink_->push(item);
} else if (out_) {
// We're waiting for the prefix.
prefix_buf_.push_back(item);
if (prefix_buf_.size() == prefix_size_) {
// Create the sink to deliver to tail lazily and deliver the prefix.
sink_ = make_counted<ucast<T>>(ctx_);
set_callbacks();
// Force member to be null before calling on_next / on_complete.
auto out = std::move(out_);
auto tup = make_cow_tuple(cow_vector<T>{std::move(prefix_buf_)},
observable<T>{sink_});
out.on_next(tup);
out.on_complete();
}
}
}
void on_error(const error& reason) override {
if (sink_) {
sink_->state().when_demand_changed = nullptr;
sink_->abort(reason);
sub_ = nullptr;
} else if (out_) {
out_.on_error(reason);
out_ = nullptr;
}
}
void on_complete() override {
if (sink_) {
sink_->state().when_demand_changed = nullptr;
sink_->close();
sub_ = nullptr;
} else if (out_) {
out_.on_complete();
out_ = nullptr;
}
}
void on_subscribe(flow::subscription sub) override {
if (!sub_ && out_) {
sub_ = std::move(sub);
if (prefix_demand_ > 0 && !requested_prefix_) {
sub_.request(prefix_size_);
requested_prefix_ = true;
}
} else {
sub.dispose();
}
}
void dispose() override {
if (out_) {
out_ = nullptr;
if (sub_) {
sub_.dispose();
sub_ = nullptr;
}
}
}
bool disposed() const noexcept override {
return !out_ && !sink_;
}
void ref_disposable() const noexcept override {
ref();
}
void deref_disposable() const noexcept override {
deref();
}
void request(size_t demand) override {
// Only called by the out_, never by the sink_. The latter triggers
prefix_demand_ += demand;
if (sub_ && !requested_prefix_) {
sub_.request(prefix_size_);
requested_prefix_ = true;
}
}
private:
intrusive_ptr<prefix_and_tail_sub> strong_this() {
return {this};
}
void set_callbacks() {
auto sptr = strong_this();
auto demand_cb = [sptr] { sptr->on_sink_demand_change(); };
sink_->state().when_demand_changed = make_action(std::move(demand_cb));
auto disposed_cb = [sptr] { sptr->on_sink_dispose(); };
sink_->state().when_disposed = make_action(std::move(disposed_cb));
}
void on_sink_demand_change() {
if (sink_ && sub_) {
auto& st = sink_->state();
auto pending = in_flight_ + st.buf.size();
if (st.demand > pending) {
auto delta = st.demand - pending;
in_flight_ += delta;
sub_.request(delta);
}
}
}
void on_sink_dispose() {
sink_ = nullptr;
if (sub_) {
sub_.dispose();
sub_ = nullptr;
}
}
/// Our scheduling context.
coordinator* ctx_;
/// The observer for the initial prefix-and-tail tuple.
observer<tuple_t> out_;
/// Caches items for the prefix until we can emit them.
std::vector<T> prefix_buf_;
/// Allows us to push to the "tail" observable after emitting the prefix.
ucast_ptr<T> sink_;
/// Pulls data from the decorated observable.
flow::subscription sub_;
/// Stores how much items are currently in-flight while receiving the tail.
size_t in_flight_ = 0;
/// Stores whether we have asked the decorated observable for data yet.
bool requested_prefix_ = false;
/// Keeps track of demand of @p out_ while we receive the prefix.
size_t prefix_demand_ = 0;
/// Stores how many items we need to buffer for the prefix.
size_t prefix_size_;
};
/// @relates prefix_and_tail_sub
template <class T>
void intrusive_ptr_add_ref(prefix_and_tail_sub<T>* ptr) {
ptr->ref();
}
/// @relates prefix_and_tail_sub
template <class T>
void intrusive_ptr_release(prefix_and_tail_sub<T>* ptr) {
ptr->deref();
}
/// Decorates an observable to split its output into a prefix of fixed size plus
/// an observable remainder.
template <class T>
class prefix_and_tail : public cold<cow_tuple<cow_vector<T>, observable<T>>> {
public:
// -- member types -----------------------------------------------------------
using tuple_t = cow_tuple<cow_vector<T>, observable<T>>;
using super = cold<tuple_t>;
// -- constructors, destructors, and assignment operators --------------------
explicit prefix_and_tail(coordinator* ctx, observable<T> decorated,
size_t prefix_size)
: super(ctx), decorated_(std::move(decorated)), prefix_size_(prefix_size) {
// nop
}
disposable subscribe(observer<tuple_t> out) override {
using impl_t = prefix_and_tail_sub<T>;
auto obs = make_counted<impl_t>(super::ctx(), out, prefix_size_);
decorated_.subscribe(observer<T>{obs});
out.on_subscribe(subscription{obs});
return obs->as_disposable();
}
private:
observable<T> decorated_;
size_t prefix_size_;
};
} // namespace caf::flow::op
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/flow/coordinator.hpp"
#include "caf/flow/observer.hpp"
#include "caf/flow/op/empty.hpp"
#include "caf/flow/op/hot.hpp"
#include "caf/flow/subscription.hpp"
#include "caf/intrusive_ptr.hpp"
#include "caf/make_counted.hpp"
#include <deque>
#include <memory>
namespace caf::flow::op {
/// State shared between one multicast operator and one subscribed observer.
template <class T>
class ucast_sub_state : public detail::plain_ref_counted {
public:
friend void intrusive_ptr_add_ref(const ucast_sub_state* ptr) noexcept {
ptr->ref();
}
friend void intrusive_ptr_release(const ucast_sub_state* ptr) noexcept {
ptr->deref();
}
explicit ucast_sub_state(coordinator* ptr) : ctx(ptr) {
// nop
}
ucast_sub_state(coordinator* ctx, observer<T> out)
: ctx(ctx), out(std::move(out)) {
// nop
}
coordinator* ctx;
std::deque<T> buf;
size_t demand = 0;
observer<T> out;
bool disposed = false;
bool closed = false;
bool running = false;
error err;
action when_disposed;
action when_consumed_some;
action when_demand_changed;
void push(const T& item) {
if (disposed) {
// nop
} else if (demand > 0 && !running) {
CAF_ASSERT(out);
CAF_ASSERT(buf.empty());
--demand;
out.on_next(item);
if (when_consumed_some)
ctx->delay(when_consumed_some);
} else {
buf.push_back(item);
}
}
void close() {
if (!disposed) {
closed = true;
if (!running && buf.empty()) {
disposed = true;
if (out) {
out.on_complete();
out = nullptr;
}
when_disposed = nullptr;
when_consumed_some = nullptr;
when_demand_changed = nullptr;
}
}
}
void abort(const error& reason) {
if (!disposed && !err) {
closed = true;
err = reason;
if (!running && buf.empty()) {
disposed = true;
if (out) {
out.on_error(reason);
out = nullptr;
}
when_disposed = nullptr;
when_consumed_some = nullptr;
when_demand_changed = nullptr;
}
}
}
void do_dispose() {
if (out) {
out.on_complete();
out = nullptr;
}
if (when_disposed) {
ctx->delay(std::move(when_disposed));
}
if (when_consumed_some) {
when_consumed_some.dispose();
when_consumed_some = nullptr;
}
when_demand_changed = nullptr;
buf.clear();
demand = 0;
disposed = true;
}
void do_run() {
auto guard = detail::make_scope_guard([this] { running = false; });
if (!disposed) {
auto got_some = demand > 0 && !buf.empty();
for (bool run = got_some; run; run = demand > 0 && !buf.empty()) {
out.on_next(buf.front());
buf.pop_front();
--demand;
}
if (buf.empty() && closed) {
if (err)
out.on_error(err);
else
out.on_complete();
out = nullptr;
do_dispose();
} else if (got_some && when_consumed_some) {
ctx->delay(when_consumed_some);
}
}
}
};
template <class T>
using ucast_sub_state_ptr = intrusive_ptr<ucast_sub_state<T>>;
template <class T>
class ucast_sub : public subscription::impl_base {
public:
// -- constructors, destructors, and assignment operators --------------------
ucast_sub(coordinator* ctx, ucast_sub_state_ptr<T> state)
: ctx_(ctx), state_(std::move(state)) {
// nop
}
// -- implementation of subscription -----------------------------------------
bool disposed() const noexcept override {
return !state_;
}
void dispose() override {
if (state_) {
ctx_->delay_fn([state = std::move(state_)]() { state->do_dispose(); });
}
}
void request(size_t n) override {
state_->demand += n;
if (state_->when_demand_changed)
state_->when_demand_changed.run();
if (!state_->running) {
state_->running = true;
ctx_->delay_fn([state = state_] { state->do_run(); });
}
}
private:
/// Stores the context (coordinator) that runs this flow.
coordinator* ctx_;
/// Stores a handle to the state.
ucast_sub_state_ptr<T> state_;
};
// Base type for *hot* operators that "unicast" data to a subscribed observer.
template <class T>
class ucast : public hot<T> {
public:
// -- member types -----------------------------------------------------------
using super = hot<T>;
using state_type = ucast_sub_state<T>;
using state_ptr_type = ucast_sub_state_ptr<T>;
using observer_type = observer<T>;
// -- constructors, destructors, and assignment operators --------------------
explicit ucast(coordinator* ctx) : super(ctx) {
state_ = make_counted<ucast_sub_state<T>>(ctx);
}
/// Pushes @p item to the subscriber or buffers them until subscribed.
void push(const T& item) {
state_->push(item);
}
/// Closes the operator, eventually emitting on_complete on all observers.
void close() {
state_->close();
}
/// Closes the operator, eventually emitting on_error on all observers.
void abort(const error& reason) {
state_->abort(reason);
}
size_t demand() const noexcept {
return state_->demand;
}
size_t buffered() const noexcept {
return state_->buf.size();
}
/// Queries the current number of subscribed observers.
size_t has_observer() const noexcept {
return state_.out.valid();
}
bool disposed() const noexcept {
return state_->disposed;
}
state_type& state() {
return *state_;
}
state_ptr_type state_ptr() {
return state_;
}
disposable subscribe(observer<T> out) override {
if (state_->closed) {
if (state_->err) {
out.on_error(state_->err);
return disposable{};
}
return make_counted<op::empty<T>>(super::ctx_)->subscribe(out);
}
if (state_->out) {
auto err = make_error(sec::too_many_observers,
"may only subscribe once to an unicast operator");
out.on_error(err);
return disposable{};
}
state_->out = out;
auto sub_ptr = make_counted<ucast_sub<T>>(super::ctx(), state_);
out.on_subscribe(subscription{sub_ptr});
return disposable{sub_ptr};
}
private:
state_ptr_type state_;
};
/// @relates ucast
template <class T>
using ucast_ptr = intrusive_ptr<ucast<T>>;
} // namespace caf::flow::op
...@@ -29,6 +29,15 @@ auto ls(T x, Ts... xs) { ...@@ -29,6 +29,15 @@ auto ls(T x, Ts... xs) {
return std::vector<T>{x, xs...}; return std::vector<T>{x, xs...};
} }
// Note: last is inclusive.
template <class T>
auto ls_range(T first, T last) {
auto result = std::vector<T>{};
for (; first <= last; ++first)
result.push_back(first);
return result;
}
} // namespace } // namespace
BEGIN_FIXTURE_SCOPE(fixture) BEGIN_FIXTURE_SCOPE(fixture)
...@@ -113,6 +122,30 @@ SCENARIO("prefix_and_tail splits off initial elements") { ...@@ -113,6 +122,30 @@ SCENARIO("prefix_and_tail splits off initial elements") {
} }
} }
} }
GIVEN("a generation with 256 values") {
WHEN("calling prefix_and_tail(7)") {
THEN("the observer receives the first 7 elements plus remainder") {
auto snk = flow::make_auto_observer<int>();
auto flat_map_calls = 0;
ctx->make_observable()
.iota(1)
.take(256)
.prefix_and_tail(7)
.flat_map([&](const tuple_t& x) {
++flat_map_calls;
auto& [prefix, tail] = x.data();
CHECK_EQ(prefix, ls(1, 2, 3, 4, 5, 6, 7));
return tail;
})
.subscribe(snk->as_observer());
ctx->run();
CHECK_EQ(flat_map_calls, 1);
CHECK_EQ(snk->buf, ls_range(8, 256));
CHECK_EQ(snk->state, flow::observer_state::completed);
CHECK_EQ(snk->err, error{});
}
}
}
} }
SCENARIO("head_and_tail splits off the first element") { SCENARIO("head_and_tail splits off the first element") {
......
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