Commit 335d3a2d authored by Dominik Charousset's avatar Dominik Charousset

Fix subscription and event handling in op::buffer

parent 80cacfe4
......@@ -58,6 +58,7 @@ caf_add_component(
exit_reason
flow.observable_state
flow.observer_state
flow.op.state
intrusive.inbox_result
intrusive.task_result
invoke_message_result
......
......@@ -8,6 +8,7 @@
#include "caf/flow/observable_decl.hpp"
#include "caf/flow/observer.hpp"
#include "caf/flow/op/cold.hpp"
#include "caf/flow/op/state.hpp"
#include "caf/flow/subscription.hpp"
#include "caf/unit.hpp"
......@@ -82,62 +83,62 @@ public:
// -- callbacks for the forwarders -------------------------------------------
void fwd_on_subscribe(buffer_input_t, subscription sub) {
if (!value_sub_ && out_) {
value_sub_ = std::move(sub);
if (demand_ > 0) {
in_flight_ += max_buf_size_;
value_sub_.request(max_buf_size_);
}
} else {
if (state_ != state::idle || value_sub_ || !out_) {
sub.dispose();
return;
}
value_sub_ = std::move(sub);
value_sub_.request(max_buf_size_);
if (control_sub_)
state_ = state::running;
}
void fwd_on_complete(buffer_input_t) {
if (!value_sub_.valid())
return;
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;
}
shutdown();
}
void fwd_on_error(buffer_input_t, const error& what) {
value_sub_ = nullptr;
do_abort(what);
err_ = what;
shutdown();
}
void fwd_on_next(buffer_input_t, const input_type& item) {
CAF_ASSERT(in_flight_ > 0);
--in_flight_;
switch (state_) {
case state::idle:
case state::running:
buf_.push_back(item);
if (buf_.size() == max_buf_size_)
do_emit();
break;
default:
break;
}
}
void fwd_on_subscribe(buffer_emit_t, subscription sub) {
if (!control_sub_ && out_) {
control_sub_ = std::move(sub);
control_sub_.request(1);
} else {
if (state_ != state::idle || control_sub_ || !out_) {
sub.dispose();
return;
}
control_sub_ = std::move(sub);
control_sub_.request(1);
if (value_sub_)
state_ = state::running;
}
void fwd_on_complete(buffer_emit_t) {
do_abort(make_error(sec::end_of_stream,
"buffer: unexpected end of the control stream"));
control_sub_ = nullptr;
err_ = make_error(sec::end_of_stream,
"buffer: unexpected end of the control stream");
shutdown();
}
void fwd_on_error(buffer_emit_t, const error& what) {
control_sub_ = nullptr;
do_abort(what);
err_ = what;
shutdown();
}
void fwd_on_next(buffer_emit_t, select_token_type) {
......@@ -167,54 +168,89 @@ public:
void request(size_t n) override {
CAF_ASSERT(out_.valid());
demand_ += n;
if (value_sub_ && pending() == 0) {
in_flight_ = max_buf_size_;
value_sub_.request(max_buf_size_);
// If we can ship a batch, schedule an event to do so.
if (demand_ == n && can_emit()) {
ctx_->delay_fn([strong_this = intrusive_ptr<buffer_sub>{this}] {
strong_this->on_request();
});
}
}
private:
size_t pending() const noexcept {
return buf_.size() + in_flight_;
bool can_emit() const noexcept {
return buf_.size() == max_buf_size_ || has_shut_down(state_);
}
void do_emit() {
void shutdown() {
value_sub_.dispose();
control_sub_.dispose();
switch (state_) {
case state::idle:
case state::running:
if (!buf_.empty()) {
if (demand_ == 0) {
state_ = err_ ? state::aborted : state::completed;
return;
}
Trait f;
out_.on_next(f(buf_));
auto buffered = buf_.size();
buf_.clear();
if (value_sub_ && buffered > 0) {
in_flight_ += buffered;
value_sub_.request(buffered);
}
if (err_)
out_.on_error(err_);
else
out_.on_complete();
out_ = nullptr;
state_ = state::disposed;
break;
default:
break;
}
void do_dispose() {
if (value_sub_) {
value_sub_.dispose();
value_sub_ = nullptr;
}
if (control_sub_) {
control_sub_.dispose();
control_sub_ = nullptr;
}
if (out_) {
void on_request() {
if (demand_ == 0 || !can_emit())
return;
switch (state_) {
case state::idle:
case state::running:
CAF_ASSERT(buf_.size() == max_buf_size_);
do_emit();
break;
case state::completed:
case state::aborted:
if (!buf_.empty())
do_emit();
if (err_)
out_.on_error(err_);
else
out_.on_complete();
out_ = nullptr;
break;
default:
break;
}
}
void do_abort(const error& reason) {
if (value_sub_) {
value_sub_.dispose();
value_sub_ = nullptr;
void do_emit() {
CAF_ASSERT(demand_ > 0);
Trait f;
--demand_;
out_.on_next(f(buf_));
auto buffered = buf_.size();
buf_.clear();
if (value_sub_ && buffered > 0)
value_sub_.request(buffered);
}
if (control_sub_) {
void do_dispose() {
value_sub_.dispose();
control_sub_.dispose();
control_sub_ = nullptr;
}
if (out_) {
out_.on_error(reason);
out_.on_complete();
out_ = nullptr;
}
state_ = state::disposed;
}
/// Stores the context (coordinator) that runs this flow.
......@@ -223,23 +259,33 @@ private:
/// Stores the maximum buffer size before forcing a batch.
size_t max_buf_size_;
/// Keeps track of how many items we have already requested.
size_t in_flight_ = 0;
/// 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.
/// Our subscription for the values. We request `max_buf_size_` items and
/// whenever we emit a batch, we request whatever amount we have shipped. That
/// way, we should always have enough demand at the source to fill up a batch.
subscription value_sub_;
/// Our subscription for the control tokens.
/// Our subscription for the control tokens. We always request 1 item.
subscription control_sub_;
/// Demand signaled by the observer.
size_t demand_ = 0;
/// Our current state.
/// - idle: until we have received both subscriptions.
/// - running: emitting batches.
/// - completed: on_complete was called but some data is still buffered.
/// - aborted: on_error was called but some data is still buffered.
/// - disposed: inactive.
state state_ = state::idle;
/// Caches the abort reason.
error err_;
};
template <class Trait>
......
// 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/default_enum_inspect.hpp"
#include "caf/detail/core_export.hpp"
namespace caf::flow::op {
/// Represents the state of a flow operator. Some operators only use a subset of
/// the possible states.
enum class state {
idle = 0b0000'0001,
running = 0b0000'0010,
completed = 0b0000'0100,
aborted = 0b0000'1000,
disposed = 0b0001'0000,
};
/// Checks whether `x` is either `completed` or `aborted`.
constexpr bool has_shut_down(state x) {
return (static_cast<int>(x) & 0b0000'1100) != 0;
}
/// @relates state
CAF_CORE_EXPORT std::string to_string(state);
/// @relates state
CAF_CORE_EXPORT bool from_string(std::string_view, state&);
/// @relates state
CAF_CORE_EXPORT bool from_integer(std::underlying_type_t<state>, state&);
/// @relates state
template <class Inspector>
bool insstatet(Inspector& f, state& x) {
return default_enum_inspect(f, x);
}
} // namespace caf::flow::op
......@@ -102,4 +102,54 @@ SCENARIO("the buffer operator forces items at regular intervals") {
}
}
SCENARIO("the buffer operator forwards errors") {
GIVEN("an observable that produces some values followed by an error") {
WHEN("calling .buffer() on it") {
THEN("the observer receives the values and then the error") {
auto outputs = std::make_shared<std::vector<cow_vector<int>>>();
auto err = std::make_shared<error>();
sys.spawn([outputs, err](caf::event_based_actor* self) {
auto obs = self->make_observable();
obs.iota(1)
.take(17)
.concat(obs.fail<int>(make_error(caf::sec::runtime_error)))
.buffer(7, 1s)
.do_on_error([err](const error& what) { *err = what; })
.for_each([outputs](const cow_vector<int>& xs) {
outputs->emplace_back(xs);
});
});
sched.run();
auto expected = std::vector<cow_vector<int>>{
cow_vector<int>{1, 2, 3, 4, 5, 6, 7},
cow_vector<int>{8, 9, 10, 11, 12, 13, 14},
cow_vector<int>{15, 16, 17},
};
CHECK_EQ(*outputs, expected);
CHECK_EQ(*err, caf::sec::runtime_error);
}
}
}
GIVEN("an observable that produces only an error") {
WHEN("calling .buffer() on it") {
THEN("the observer receives the error") {
auto outputs = std::make_shared<std::vector<cow_vector<int>>>();
auto err = std::make_shared<error>();
sys.spawn([outputs, err](caf::event_based_actor* self) {
self->make_observable()
.fail<int>(make_error(caf::sec::runtime_error))
.buffer(3, 1s)
.do_on_error([err](const error& what) { *err = what; })
.for_each([outputs](const cow_vector<int>& xs) {
outputs->emplace_back(xs);
});
});
sched.run();
CHECK(outputs->empty());
CHECK_EQ(*err, caf::sec::runtime_error);
}
}
}
}
END_FIXTURE_SCOPE()
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