Commit 3e0bf390 authored by Dominik Charousset's avatar Dominik Charousset

Implement buffer flow operator

parent 5f5d3b7f
......@@ -276,6 +276,7 @@ caf_add_component(
dynamic_spawn
error
expected
flow.buffer
flow.concat
flow.concat_map
flow.defer
......
......@@ -9,6 +9,7 @@
#include "caf/make_counted.hpp"
#include "caf/ref_counted.hpp"
#include <initializer_list>
#include <vector>
namespace caf {
......@@ -41,6 +42,10 @@ public:
impl_ = make_counted<impl>(std::move(std));
}
explicit cow_vector(std::initializer_list<T> values) {
impl_ = make_counted<impl>(std_type{values});
}
cow_vector(cow_vector&&) noexcept = default;
cow_vector(const cow_vector&) noexcept = default;
......
......@@ -19,10 +19,13 @@
#include "caf/flow/observable_state.hpp"
#include "caf/flow/observer.hpp"
#include "caf/flow/op/base.hpp"
#include "caf/flow/op/buffer.hpp"
#include "caf/flow/op/concat.hpp"
#include "caf/flow/op/from_resource.hpp"
#include "caf/flow/op/from_steps.hpp"
#include "caf/flow/op/interval.hpp"
#include "caf/flow/op/merge.hpp"
#include "caf/flow/op/never.hpp"
#include "caf/flow/op/prefetch.hpp"
#include "caf/flow/op/publish.hpp"
#include "caf/flow/step/all.hpp"
......@@ -219,6 +222,15 @@ public:
return add_step(step::take<output_type>{n});
}
/// @copydoc observable::buffer
auto buffer(size_t count) && {
return materialize().buffer(count);
}
auto buffer(size_t count, timespan period) {
return materialize().buffer(count, period);
}
template <class Predicate>
auto filter(Predicate predicate) && {
return add_step(step::filter<Predicate>{std::move(predicate)});
......@@ -566,6 +578,22 @@ observable<T>::take_while(Predicate predicate) {
return transform(step::take_while{std::move(predicate)});
}
template <class T>
observable<cow_vector<T>> observable<T>::buffer(size_t count) {
using trait_t = op::buffer_default_trait<T>;
using impl_t = op::buffer<trait_t>;
return make_observable<impl_t>(ctx(), count, *this,
make_observable<op::never<unit_t>>(ctx()));
}
template <class T>
observable<cow_vector<T>> observable<T>::buffer(size_t count, timespan period) {
using trait_t = op::buffer_interval_trait<T>;
using impl_t = op::buffer<trait_t>;
return make_observable<impl_t>(
ctx(), count, *this, make_observable<op::interval>(ctx(), period, period));
}
// -- observable: combining ----------------------------------------------------
template <class T>
......
......@@ -141,6 +141,13 @@ public:
.filter([](const vector_type& xs) { return !xs.empty(); });
}
/// Emits items in buffers of size @p count.
observable<cow_vector<T>> buffer(size_t count);
/// Emits items in buffers of size up to @p count and forces an item at
/// regular intervals .
observable<cow_vector<T>> buffer(size_t count, timespan period);
// -- combining --------------------------------------------------------------
/// Combines the output of multiple @ref observable objects into one by
......
// 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_vector.hpp"
#include "caf/flow/observable_decl.hpp"
#include "caf/flow/observer.hpp"
#include "caf/flow/op/cold.hpp"
#include "caf/flow/subscription.hpp"
#include "caf/unit.hpp"
#include <vector>
namespace caf::flow::op {
struct buffer_input_t {};
struct buffer_emit_t {};
template <class T>
struct buffer_default_trait {
static constexpr bool skip_empty = false;
using input_type = T;
using output_type = cow_vector<T>;
using select_token_type = unit_t;
output_type operator()(const std::vector<input_type>& xs) {
return output_type{xs};
}
};
template <class T>
struct buffer_interval_trait {
static constexpr bool skip_empty = false;
using input_type = T;
using output_type = cow_vector<T>;
using select_token_type = int64_t;
output_type operator()(const std::vector<input_type>& xs) {
return output_type{xs};
}
};
///
template <class Trait>
class buffer_sub : public subscription::impl_base {
public:
// -- member types -----------------------------------------------------------
using input_type = typename Trait::input_type;
using output_type = typename Trait::output_type;
using select_token_type = typename Trait::select_token_type;
// -- constants --------------------------------------------------------------
static constexpr size_t val_id = 0;
static constexpr size_t ctrl_id = 1;
// -- constructors, destructors, and assignment operators --------------------
buffer_sub(coordinator* ctx, size_t max_buf_size, observer<output_type> out)
: ctx_(ctx), max_buf_size_(max_buf_size), out_(std::move(out)) {
// nop
}
// -- callbacks for the parent -----------------------------------------------
void init(observable<input_type> vals, observable<select_token_type> ctrl) {
using val_fwd_t = forwarder<input_type, buffer_sub, buffer_input_t>;
using ctrl_fwd_t = forwarder<select_token_type, buffer_sub, buffer_emit_t>;
vals.subscribe(
make_counted<val_fwd_t>(this, buffer_input_t{})->as_observer());
ctrl.subscribe(
make_counted<ctrl_fwd_t>(this, buffer_emit_t{})->as_observer());
}
// -- callbacks for the forwarders -------------------------------------------
void fwd_on_subscribe(buffer_input_t, subscription sub) {
if (!value_sub_ && out_) {
value_sub_ = std::move(sub);
if (pending_demand_ > 0) {
value_sub_.request(pending_demand_);
pending_demand_ = 0;
}
} else {
sub.dispose();
}
}
void fwd_on_complete(buffer_input_t) {
CAF_ASSERT(value_sub_.valid());
CAF_ASSERT(out_.valid());
value_sub_ = nullptr;
if (!buf_.empty())
do_emit();
out_.on_complete();
out_ = nullptr;
if (control_sub_) {
control_sub_.dispose();
control_sub_ = nullptr;
}
}
void fwd_on_error(buffer_input_t, const error& what) {
value_sub_ = nullptr;
do_abort(what);
}
void fwd_on_next(buffer_input_t, const input_type& item) {
buf_.push_back(item);
if (buf_.size() == max_buf_size_)
do_emit();
}
void fwd_on_subscribe(buffer_emit_t, subscription sub) {
if (!control_sub_ && out_) {
control_sub_ = std::move(sub);
control_sub_.request(1);
} else {
sub.dispose();
}
}
void fwd_on_complete(buffer_emit_t) {
do_abort(make_error(sec::end_of_stream,
"buffer: unexpected end of the control stream"));
}
void fwd_on_error(buffer_emit_t, const error& what) {
control_sub_ = nullptr;
do_abort(what);
}
void fwd_on_next(buffer_emit_t, select_token_type) {
if constexpr (Trait::skip_empty) {
if (!buf_.empty())
do_emit();
} else {
do_emit();
}
control_sub_.request(1);
}
// -- implementation of subscription -----------------------------------------
bool disposed() const noexcept override {
return !out_;
}
void dispose() override {
if (out_) {
ctx_->delay_fn([strong_this = intrusive_ptr<buffer_sub>{this}] {
strong_this->do_dispose();
});
}
}
void request(size_t n) override {
CAF_ASSERT(out_.valid());
if (value_sub_)
value_sub_.request(n);
else
pending_demand_ += n;
}
private:
void do_emit() {
Trait f;
out_.on_next(f(buf_));
buf_.clear();
}
void do_dispose() {
if (value_sub_) {
value_sub_.dispose();
value_sub_ = nullptr;
}
if (control_sub_) {
control_sub_.dispose();
control_sub_ = nullptr;
}
if (out_) {
out_.on_complete();
}
}
void do_abort(const error& reason) {
if (value_sub_) {
value_sub_.dispose();
value_sub_ = nullptr;
}
if (control_sub_) {
control_sub_.dispose();
control_sub_ = nullptr;
}
if (out_) {
out_.on_error(reason);
}
}
/// Stores the context (coordinator) that runs this flow.
coordinator* ctx_;
/// Stores the maximum buffer size before forcing a batch.
size_t max_buf_size_;
/// Stores the elements until we can emit them.
std::vector<input_type> buf_;
/// Stores a handle to the subscribed observer.
observer<output_type> out_;
/// Our subscription for the values.
subscription value_sub_;
/// Our subscription for the control tokens.
subscription control_sub_;
///
size_t pending_demand_ = 0;
};
template <class Trait>
class buffer : public cold<typename Trait::output_type> {
public:
// -- member types -----------------------------------------------------------
using input_type = typename Trait::input_type;
using output_type = typename Trait::output_type;
using super = cold<output_type>;
using input = observable<input_type>;
using selector = observable<typename Trait::select_token_type>;
// -- constructors, destructors, and assignment operators --------------------
buffer(coordinator* ctx, size_t max_items, input in, selector select)
: super(ctx),
max_items_(max_items),
in_(std::move(in)),
select_(std::move(select)) {
// nop
}
// -- implementation of observable<T> -----------------------------------
disposable subscribe(observer<output_type> out) override {
auto ptr = make_counted<buffer_sub<Trait>>(super::ctx_, max_items_, out);
ptr->init(in_, select_);
out.on_subscribe(subscription{ptr});
return ptr->as_disposable();
}
private:
/// Configures how many items get pushed into one buffer at most.
size_t max_items_;
/// Sequence of input values.
input in_;
/// Sequence of control messages to select previous inputs.
selector select_;
};
} // namespace caf::flow::op
......@@ -25,6 +25,8 @@ public:
void run();
size_t run_some();
// -- reference counting -----------------------------------------------------
void ref_execution_context() const noexcept override;
......
......@@ -26,6 +26,20 @@ void scoped_coordinator::run() {
}
}
size_t scoped_coordinator::run_some() {
size_t result = 0;
for (;;) {
auto f = next(false);
if (f.ptr() != nullptr) {
++result;
f.run();
drop_disposed_flows();
} else {
return result;
}
}
}
// -- reference counting -------------------------------------------------------
void scoped_coordinator::ref_execution_context() const noexcept {
......
// 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.
#define CAF_SUITE flow.buffer
#include "caf/flow/observable.hpp"
#include "core-test.hpp"
#include "caf/flow/coordinator.hpp"
#include "caf/flow/item_publisher.hpp"
#include "caf/flow/merge.hpp"
#include "caf/flow/observable_builder.hpp"
#include "caf/flow/observer.hpp"
#include "caf/flow/scoped_coordinator.hpp"
using namespace caf;
using namespace std::literals;
namespace {
struct fixture : test_coordinator_fixture<> {
flow::scoped_coordinator_ptr ctx = flow::make_scoped_coordinator();
};
} // namespace
BEGIN_FIXTURE_SCOPE(fixture)
SCENARIO("the buffer operator groups items together") {
GIVEN("an observable") {
WHEN("calling .buffer(3)") {
THEN("the observer receives values in groups of three") {
auto inputs = std::vector<int>{1, 2, 4, 8, 16, 32, 64, 128};
auto outputs = std::vector<cow_vector<int>>{};
auto expected = std::vector<cow_vector<int>>{
cow_vector<int>{1, 2, 4},
cow_vector<int>{8, 16, 32},
cow_vector<int>{64, 128},
};
ctx->make_observable()
.from_container(inputs) //
.buffer(3)
.for_each([&outputs](const cow_vector<int>& xs) {
outputs.emplace_back(xs);
});
ctx->run();
CHECK_EQ(outputs, expected);
}
}
}
}
SCENARIO("the buffer operator forces items at regular intervals") {
GIVEN("an observable") {
WHEN("calling .buffer(3, 1s)") {
THEN("the observer receives values in groups of three or after 1s") {
auto outputs = std::make_shared<std::vector<cow_vector<int>>>();
auto expected = std::vector<cow_vector<int>>{
cow_vector<int>{1, 2, 4}, cow_vector<int>{8, 16, 32},
cow_vector<int>{}, cow_vector<int>{64},
cow_vector<int>{}, cow_vector<int>{128, 256, 512},
};
auto pub = flow::item_publisher<int>{ctx.get()};
sys.spawn([&pub, outputs](caf::event_based_actor* self) {
pub //
.as_observable()
.observe_on(self)
.buffer(3, 1s)
.for_each([outputs](const cow_vector<int>& xs) {
outputs->emplace_back(xs);
});
});
sched.run();
MESSAGE("emit the first six items");
pub.push({1, 2, 4, 8, 16, 32});
ctx->run_some();
sched.run();
MESSAGE("force an empty buffer");
advance_time(1s);
sched.run();
MESSAGE("force a buffer with a single element");
pub.push(64);
ctx->run_some();
sched.run();
advance_time(1s);
sched.run();
MESSAGE("force an empty buffer");
advance_time(1s);
sched.run();
MESSAGE("emit the last items and close the source");
pub.push({128, 256, 512});
pub.close();
ctx->run_some();
sched.run();
advance_time(1s);
sched.run();
CHECK_EQ(*outputs, expected);
}
}
}
}
END_FIXTURE_SCOPE()
......@@ -47,7 +47,7 @@ SCENARIO("scheduled actors schedule observable intervals on the actor clock") {
WHEN("an observer subscribes to it") {
THEN("the actor uses the actor clock to schedule flow processing") {
auto outputs = i64_list{};
sys.spawn([&outputs](caf::event_based_actor* self) {
sys.spawn([&outputs](event_based_actor* self) {
self->make_observable()
.interval(50ms, 25ms)
.take(3)
......
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