Commit 7f988bf5 authored by Dominik Charousset's avatar Dominik Charousset

Rename {bounded => spsc}_buffer and allow overflow

Rename the buffer to make it more descriptive and also allow the
producer to push more elements than announced to the constructor to
cover use cases where a producer transforms single inputs into
potentially multiple outputs.
parent 962619c3
...@@ -227,7 +227,7 @@ caf_add_component( ...@@ -227,7 +227,7 @@ caf_add_component(
actor_system_config actor_system_config
actor_termination actor_termination
aout aout
async.bounded_buffer async.spsc_buffer
behavior behavior
binary_deserializer binary_deserializer
binary_serializer binary_serializer
......
...@@ -17,7 +17,7 @@ class producer; ...@@ -17,7 +17,7 @@ class producer;
// -- template classes --------------------------------------------------------- // -- template classes ---------------------------------------------------------
template <class T> template <class T>
class bounded_buffer; class spsc_buffer;
template <class T> template <class T>
class consumer_resource; class consumer_resource;
......
...@@ -23,7 +23,7 @@ ...@@ -23,7 +23,7 @@
namespace caf::async { namespace caf::async {
/// Policy type for having `consume` call `on_error` immediately after the /// Policy type for having `consume` call `on_error` immediately after the
/// producer has aborted even if the buffer still contains events. /// producer has aborted even if the buffer still contains items.
struct prioritize_errors_t { struct prioritize_errors_t {
static constexpr bool calls_on_error = true; static constexpr bool calls_on_error = true;
}; };
...@@ -32,7 +32,7 @@ struct prioritize_errors_t { ...@@ -32,7 +32,7 @@ struct prioritize_errors_t {
constexpr auto prioritize_errors = prioritize_errors_t{}; constexpr auto prioritize_errors = prioritize_errors_t{};
/// Policy type for having `consume` call `on_error` only after processing all /// Policy type for having `consume` call `on_error` only after processing all
/// events from the buffer. /// items from the buffer.
struct delay_errors_t { struct delay_errors_t {
static constexpr bool calls_on_error = true; static constexpr bool calls_on_error = true;
}; };
...@@ -40,16 +40,30 @@ struct delay_errors_t { ...@@ -40,16 +40,30 @@ struct delay_errors_t {
/// @relates delay_errors_t /// @relates delay_errors_t
constexpr auto delay_errors = delay_errors_t{}; constexpr auto delay_errors = delay_errors_t{};
/// A bounded buffer for transmitting events from one producer to one consumer. /// A Single Producer Single Consumer buffer. The buffer uses a "soft bound",
/// which means that the producer announces a desired maximum for in-flight
/// items that the buffer uses for its bookkeeping, but the producer may add
/// more than that number of items. Allowing producers to go "beyond the limit"
/// is intended for producer that transform inputs into outputs where one input
/// event can produce multiple output items.
///
/// Aside from providing storage, this buffer also resumes the consumer if data
/// is available and signals demand to the producer whenever the consumer takes
/// data out of the buffer.
template <class T> template <class T>
class bounded_buffer : public ref_counted { class spsc_buffer : public ref_counted {
public: public:
using value_type = T; using value_type = T;
bounded_buffer(uint32_t max_in_flight, uint32_t min_pull_size) spsc_buffer(uint32_t capacity, uint32_t min_pull_size)
: max_in_flight_(max_in_flight), min_pull_size_(min_pull_size) { : capacity_(capacity), min_pull_size_(min_pull_size) {
buf_.reserve(max_in_flight); // Allocate some extra space in the buffer in case the producer goes beyond
consumer_buf_.reserve(max_in_flight); // the announced capacity.
buf_.reserve(capacity + (capacity / 2));
// Note: this buffer can never go above its limit since it's a short-term
// buffer for the consumer that cannot ask for more than capacity
// items.
consumer_buf_.reserve(capacity);
} }
/// Appends to the buffer and calls `on_producer_wakeup` on the consumer if /// Appends to the buffer and calls `on_producer_wakeup` on the consumer if
...@@ -59,11 +73,13 @@ public: ...@@ -59,11 +73,13 @@ 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(buf_.size() + items.size() <= max_in_flight_);
buf_.insert(buf_.end(), items.begin(), items.end()); buf_.insert(buf_.end(), items.begin(), items.end());
if (buf_.size() == items.size() && consumer_) if (buf_.size() == items.size() && consumer_)
consumer_->on_producer_wakeup(); consumer_->on_producer_wakeup();
return capacity() - buf_.size(); if (capacity_ >= buf_.size())
return capacity_ - buf_.size();
else
return 0;
} }
size_t push(const T& item) { size_t push(const T& item) {
...@@ -72,10 +88,10 @@ public: ...@@ -72,10 +88,10 @@ public:
/// Consumes up to `demand` items from the buffer. /// Consumes up to `demand` items from the buffer.
/// @tparam Policy Either `instant_error_t`, `delay_error_t` or /// @tparam Policy Either `instant_error_t`, `delay_error_t` or
/// `ignore_errors_t`. The former two policies require also /// `ignore_errors_t`.
/// passing an `on_error` handler. /// @returns a tuple indicating whether the consumer may call pull again and
/// @returns true if the consumer may call `pull` again, otherwise `false`. /// how many items were consumed. When returning `false` for the
/// When returning `false`, the function has called `on_complete` or /// first tuple element, the function has called `on_complete` or
/// `on_error` on the observer. /// `on_error` on the observer.
template <class Policy, class Observer> template <class Policy, class Observer>
std::pair<bool, size_t> pull(Policy, size_t demand, Observer& dst) { std::pair<bool, size_t> pull(Policy, size_t demand, Observer& dst) {
...@@ -89,6 +105,10 @@ public: ...@@ -89,6 +105,10 @@ public:
return {false, 0}; return {false, 0};
} }
} }
// We must not signal demand to the producer when reading excess elements
// from the buffer. Otherwise, we end up generating more demand than
// capacity_ allows us to.
auto overflow = buf_.size() <= capacity_ ? 0u : buf_.size() - capacity_;
auto next_n = [this, &demand] { return std::min(demand, buf_.size()); }; auto next_n = [this, &demand] { return std::min(demand, buf_.size()); };
size_t consumed = 0; size_t consumed = 0;
for (auto n = next_n(); n > 0; n = next_n()) { for (auto n = next_n(); n > 0; n = next_n()) {
...@@ -96,7 +116,14 @@ public: ...@@ -96,7 +116,14 @@ public:
consumer_buf_.assign(make_move_iterator(buf_.begin()), consumer_buf_.assign(make_move_iterator(buf_.begin()),
make_move_iterator(buf_.begin() + n)); make_move_iterator(buf_.begin() + n));
buf_.erase(buf_.begin(), buf_.begin() + n); buf_.erase(buf_.begin(), buf_.begin() + n);
if (overflow == 0) {
signal_demand(n); signal_demand(n);
} else if (n <= overflow) {
overflow -= n;
} else {
signal_demand(n - overflow);
overflow = 0;
}
guard.unlock(); guard.unlock();
dst.on_next(span<const T>{consumer_buf_.data(), n}); dst.on_next(span<const T>{consumer_buf_.data(), n});
demand -= n; demand -= n;
...@@ -122,14 +149,15 @@ public: ...@@ -122,14 +149,15 @@ public:
return !buf_.empty(); return !buf_.empty();
} }
/// Checks whether the there is data available or the producer has closed or /// Checks whether the there is data available or whether the producer has
/// aborted the flow. /// closed or aborted the flow.
bool has_consumer_event() const noexcept { bool has_consumer_event() const noexcept {
std::unique_lock guard{mtx_}; std::unique_lock guard{mtx_};
return !buf_.empty() || closed_; return !buf_.empty() || closed_;
} }
/// Returns how many items are currently available. /// Returns how many items are currently available. This may be greater than
/// the `capacity`.
size_t available() const noexcept { size_t available() const noexcept {
std::unique_lock guard{mtx_}; std::unique_lock guard{mtx_};
return buf_.size(); return buf_.size();
...@@ -146,7 +174,8 @@ public: ...@@ -146,7 +174,8 @@ public:
} }
} }
/// Closes the buffer and signals an error by request of the producer. /// Closes the buffer by request of the producer and signals an error to the
/// consumer.
void abort(error reason) { void abort(error reason) {
std::unique_lock guard{mtx_}; std::unique_lock guard{mtx_};
if (producer_) { if (producer_) {
...@@ -168,28 +197,31 @@ public: ...@@ -168,28 +197,31 @@ public:
} }
} }
/// Consumer callback for the initial handshake between producer and consumer.
void set_consumer(consumer_ptr consumer) { void set_consumer(consumer_ptr consumer) {
CAF_ASSERT(consumer != nullptr); CAF_ASSERT(consumer != nullptr);
std::unique_lock guard{mtx_}; std::unique_lock guard{mtx_};
if (consumer_) if (consumer_)
CAF_RAISE_ERROR("producer-consumer queue already has a consumer"); CAF_RAISE_ERROR("SPSC buffer already has a consumer");
consumer_ = std::move(consumer); consumer_ = std::move(consumer);
if (producer_) if (producer_)
ready(); ready();
} }
/// Producer callback for the initial handshake between producer and consumer.
void set_producer(producer_ptr producer) { void set_producer(producer_ptr producer) {
CAF_ASSERT(producer != nullptr); CAF_ASSERT(producer != nullptr);
std::unique_lock guard{mtx_}; std::unique_lock guard{mtx_};
if (producer_) if (producer_)
CAF_RAISE_ERROR("producer-consumer queue already has a producer"); CAF_RAISE_ERROR("SPSC buffer already has a producer");
producer_ = std::move(producer); producer_ = std::move(producer);
if (consumer_) if (consumer_)
ready(); ready();
} }
/// Returns the capacity as passed to the constructor of the buffer.
size_t capacity() const noexcept { size_t capacity() const noexcept {
return max_in_flight_; return capacity_;
} }
/// Returns the mutex for this object. /// Returns the mutex for this object.
...@@ -209,7 +241,7 @@ private: ...@@ -209,7 +241,7 @@ private:
consumer_->on_producer_ready(); consumer_->on_producer_ready();
if (!buf_.empty()) if (!buf_.empty())
consumer_->on_producer_wakeup(); consumer_->on_producer_wakeup();
signal_demand(max_in_flight_); signal_demand(capacity_);
} }
void signal_demand(uint32_t new_demand) { void signal_demand(uint32_t new_demand) {
...@@ -223,13 +255,11 @@ private: ...@@ -223,13 +255,11 @@ private:
/// Guards access to all other member variables. /// Guards access to all other member variables.
mutable std::mutex mtx_; mutable std::mutex mtx_;
/// Allocated to max_in_flight_ * 2, but at most holds max_in_flight_ /// Caches in-flight items.
/// elements at any point in time. We dynamically shift elements into the
/// first half of the buffer whenever rd_pos_ crosses the midpoint.
std::vector<T> buf_; std::vector<T> buf_;
/// Stores how many items the buffer may hold at any time. /// Stores how many items the buffer may hold at any time.
uint32_t max_in_flight_; uint32_t capacity_;
/// Configures the minimum amount of free buffer slots that we signal to the /// Configures the minimum amount of free buffer slots that we signal to the
/// producer. /// producer.
...@@ -250,18 +280,18 @@ private: ...@@ -250,18 +280,18 @@ private:
/// Callback handle to the producer. /// Callback handle to the producer.
producer_ptr producer_; producer_ptr producer_;
/// Caches elements before passing them to the consumer (without lock). /// Caches items before passing them to the consumer (without lock).
std::vector<T> consumer_buf_; std::vector<T> consumer_buf_;
}; };
/// @relates bounded_buffer /// @relates spsc_buffer
template <class T> template <class T>
using bounded_buffer_ptr = intrusive_ptr<bounded_buffer<T>>; using spsc_buffer_ptr = intrusive_ptr<spsc_buffer<T>>;
/// @relates bounded_buffer /// @relates spsc_buffer
template <class T, bool IsProducer> template <class T, bool IsProducer>
struct resource_ctrl : ref_counted { struct resource_ctrl : ref_counted {
using buffer_ptr = bounded_buffer_ptr<T>; using buffer_ptr = spsc_buffer_ptr<T>;
explicit resource_ctrl(buffer_ptr ptr) : buf(std::move(ptr)) { explicit resource_ctrl(buffer_ptr ptr) : buf(std::move(ptr)) {
// nop // nop
...@@ -296,15 +326,15 @@ struct resource_ctrl : ref_counted { ...@@ -296,15 +326,15 @@ struct resource_ctrl : ref_counted {
/// Grants read access to the first consumer that calls `open` on the resource. /// Grants read access to the first consumer that calls `open` on the resource.
/// Cancels consumption of items on the buffer if the resources gets destroyed /// Cancels consumption of items on the buffer if the resources gets destroyed
/// before opening it. /// before opening it.
/// @relates bounded_buffer /// @relates spsc_buffer
template <class T> template <class T>
class consumer_resource { class consumer_resource {
public: public:
using value_type = T; using value_type = T;
using buffer_type = bounded_buffer<T>; using buffer_type = spsc_buffer<T>;
using buffer_ptr = bounded_buffer_ptr<T>; using buffer_ptr = spsc_buffer_ptr<T>;
explicit consumer_resource(buffer_ptr buf) { explicit consumer_resource(buffer_ptr buf) {
ctrl_.emplace(std::move(buf)); ctrl_.emplace(std::move(buf));
...@@ -335,15 +365,15 @@ private: ...@@ -335,15 +365,15 @@ private:
/// Grants access to a buffer to the first producer that calls `open`. Aborts /// Grants access to a buffer to the first producer that calls `open`. Aborts
/// writes on the buffer if the resources gets destroyed before opening it. /// writes on the buffer if the resources gets destroyed before opening it.
/// @relates bounded_buffer /// @relates spsc_buffer
template <class T> template <class T>
class producer_resource { class producer_resource {
public: public:
using value_type = T; using value_type = T;
using buffer_type = bounded_buffer<T>; using buffer_type = spsc_buffer<T>;
using buffer_ptr = bounded_buffer_ptr<T>; using buffer_ptr = spsc_buffer_ptr<T>;
explicit producer_resource(buffer_ptr buf) { explicit producer_resource(buffer_ptr buf) {
ctrl_.emplace(std::move(buf)); ctrl_.emplace(std::move(buf));
...@@ -372,20 +402,20 @@ private: ...@@ -372,20 +402,20 @@ private:
intrusive_ptr<resource_ctrl<T, true>> ctrl_; intrusive_ptr<resource_ctrl<T, true>> ctrl_;
}; };
/// Creates bounded buffer and returns two resources connected by that buffer. /// Creates spsc buffer and returns two resources connected by that buffer.
template <class T> template <class T>
std::pair<consumer_resource<T>, producer_resource<T>> std::pair<consumer_resource<T>, producer_resource<T>>
make_bounded_buffer_resource(size_t buffer_size, size_t min_request_size) { make_spsc_buffer_resource(size_t buffer_size, size_t min_request_size) {
using buffer_type = bounded_buffer<T>; using buffer_type = spsc_buffer<T>;
auto buf = make_counted<buffer_type>(buffer_size, min_request_size); auto buf = make_counted<buffer_type>(buffer_size, min_request_size);
return {async::consumer_resource<T>{buf}, async::producer_resource<T>{buf}}; return {async::consumer_resource<T>{buf}, async::producer_resource<T>{buf}};
} }
/// Creates bounded buffer and returns two resources connected by that buffer. /// Creates spsc buffer and returns two resources connected by that buffer.
template <class T> template <class T>
std::pair<consumer_resource<T>, producer_resource<T>> std::pair<consumer_resource<T>, producer_resource<T>>
make_bounded_buffer_resource() { make_spsc_buffer_resource() {
return make_bounded_buffer_resource<T>(defaults::flow::buffer_size, return make_spsc_buffer_resource<T>(defaults::flow::buffer_size,
defaults::flow::min_demand); defaults::flow::min_demand);
} }
......
...@@ -9,9 +9,9 @@ ...@@ -9,9 +9,9 @@
#include <type_traits> #include <type_traits>
#include <vector> #include <vector>
#include "caf/async/bounded_buffer.hpp"
#include "caf/async/consumer.hpp" #include "caf/async/consumer.hpp"
#include "caf/async/producer.hpp" #include "caf/async/producer.hpp"
#include "caf/async/spsc_buffer.hpp"
#include "caf/defaults.hpp" #include "caf/defaults.hpp"
#include "caf/detail/core_export.hpp" #include "caf/detail/core_export.hpp"
#include "caf/disposable.hpp" #include "caf/disposable.hpp"
...@@ -209,7 +209,7 @@ public: ...@@ -209,7 +209,7 @@ public:
auto concat_map(F f); auto concat_map(F f);
/// Creates an asynchronous resource that makes emitted items available in a /// Creates an asynchronous resource that makes emitted items available in a
/// bounded buffer. /// spsc buffer.
async::consumer_resource<T> to_resource(size_t buffer_size, async::consumer_resource<T> to_resource(size_t buffer_size,
size_t min_request_size); size_t min_request_size);
...@@ -1662,7 +1662,7 @@ private: ...@@ -1662,7 +1662,7 @@ private:
template <class T> template <class T>
async::consumer_resource<T> async::consumer_resource<T>
observable<T>::to_resource(size_t buffer_size, size_t min_request_size) { observable<T>::to_resource(size_t buffer_size, size_t min_request_size) {
using buffer_type = async::bounded_buffer<T>; using buffer_type = async::spsc_buffer<T>;
auto buf = make_counted<buffer_type>(buffer_size, min_request_size); auto buf = make_counted<buffer_type>(buffer_size, min_request_size);
auto up = make_counted<buffer_writer_impl<buffer_type>>(pimpl_->ctx(), buf); auto up = make_counted<buffer_writer_impl<buffer_type>>(pimpl_->ctx(), buf);
buf->set_producer(up); buf->set_producer(up);
...@@ -1675,7 +1675,7 @@ observable<T>::to_resource(size_t buffer_size, size_t min_request_size) { ...@@ -1675,7 +1675,7 @@ observable<T>::to_resource(size_t buffer_size, size_t min_request_size) {
template <class T> template <class T>
observable<T> observable<T>::observe_on(coordinator* other, size_t buffer_size, observable<T> observable<T>::observe_on(coordinator* other, size_t buffer_size,
size_t min_request_size) { size_t min_request_size) {
using buffer_type = async::bounded_buffer<T>; using buffer_type = async::spsc_buffer<T>;
auto buf = make_counted<buffer_type>(buffer_size, min_request_size); auto buf = make_counted<buffer_type>(buffer_size, min_request_size);
auto up = make_counted<buffer_writer_impl<buffer_type>>(pimpl_->ctx(), buf); auto up = make_counted<buffer_writer_impl<buffer_type>>(pimpl_->ctx(), buf);
auto down = make_counted<observable_buffer_impl<buffer_type>>(other, buf); auto down = make_counted<observable_buffer_impl<buffer_type>>(other, buf);
......
...@@ -4,7 +4,7 @@ ...@@ -4,7 +4,7 @@
#pragma once #pragma once
#include "caf/async/bounded_buffer.hpp" #include "caf/async/spsc_buffer.hpp"
#include "caf/defaults.hpp" #include "caf/defaults.hpp"
#include "caf/flow/coordinator.hpp" #include "caf/flow/coordinator.hpp"
#include "caf/flow/observable.hpp" #include "caf/flow/observable.hpp"
......
// 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 async.bounded_buffer
#include "caf/async/bounded_buffer.hpp"
#include "core-test.hpp"
#include <memory>
#include "caf/flow/coordinator.hpp"
#include "caf/flow/merge.hpp"
#include "caf/flow/observable_builder.hpp"
#include "caf/flow/observer.hpp"
#include "caf/scheduled_actor/flow.hpp"
using namespace caf;
BEGIN_FIXTURE_SCOPE(test_coordinator_fixture<>)
SCENARIO("bounded buffers moves data between actors") {
GIVEN("a bounded buffer resource") {
WHEN("opening the resource from two actors") {
THEN("data travels through the bounded buffer") {
using actor_t = event_based_actor;
auto [rd, wr] = async::make_bounded_buffer_resource<int>(6, 2);
auto inputs = std::vector<int>{1, 2, 4, 8, 16, 32, 64, 128};
auto outputs = std::vector<int>{};
sys.spawn([wr{wr}, &inputs](actor_t* src) {
src->make_observable()
.from_container(inputs)
.filter([](int) { return true; })
.subscribe(wr);
});
sys.spawn([rd{rd}, &outputs](actor_t* snk) {
snk
->make_observable() //
.from_resource(rd)
.for_each([&outputs](int x) { outputs.emplace_back(x); });
});
run();
CHECK_EQ(inputs, outputs);
}
}
}
}
END_FIXTURE_SCOPE()
// 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 async.spsc_buffer
#include "caf/async/spsc_buffer.hpp"
#include "core-test.hpp"
#include <memory>
#include "caf/flow/coordinator.hpp"
#include "caf/flow/merge.hpp"
#include "caf/flow/observable_builder.hpp"
#include "caf/flow/observer.hpp"
#include "caf/scheduled_actor/flow.hpp"
using namespace caf;
namespace {
class dummy_producer : public async::producer {
public:
dummy_producer() = default;
dummy_producer(const dummy_producer&) = delete;
dummy_producer& operator=(const dummy_producer&) = delete;
void on_consumer_ready() {
consumer_ready = true;
}
void on_consumer_cancel() {
consumer_cancel = true;
}
void on_consumer_demand(size_t new_demand) {
demand += new_demand;
}
void ref_producer() const noexcept {
++rc;
}
void deref_producer() const noexcept {
if (--rc == 0)
delete this;
}
mutable size_t rc = 1;
bool consumer_ready = false;
bool consumer_cancel = false;
size_t demand = 0;
};
class dummy_consumer : public async::consumer {
public:
dummy_consumer() = default;
dummy_consumer(const dummy_consumer&) = delete;
dummy_consumer& operator=(const dummy_consumer&) = delete;
void on_producer_ready() {
producer_ready = true;
}
void on_producer_wakeup() {
++producer_wakeups;
}
void ref_consumer() const noexcept {
++rc;
}
void deref_consumer() const noexcept {
if (--rc == 0)
delete this;
}
mutable size_t rc = 1;
bool producer_ready = false;
size_t producer_wakeups = 0;
};
struct dummy_observer {
template <class T>
void on_next(span<const T> items) {
consumed += items.size();
}
void on_error(error what) {
on_error_called = true;
err = std::move(what);
}
void on_complete() {
on_complete_called = true;
}
size_t consumed = 0;
bool on_error_called = false;
bool on_complete_called = false;
error err;
};
} // namespace
BEGIN_FIXTURE_SCOPE(test_coordinator_fixture<>)
SCENARIO("SPSC buffers may go past their capacity") {
GIVEN("an SPSC buffer with consumer and produer") {
auto prod = make_counted<dummy_producer>();
auto cons = make_counted<dummy_consumer>();
auto buf = make_counted<async::spsc_buffer<int>>(10, 2);
buf->set_producer(prod);
buf->set_consumer(cons);
CHECK_EQ(prod->consumer_ready, true);
CHECK_EQ(prod->consumer_cancel, false);
CHECK_EQ(prod->demand, 10u);
CHECK_EQ(cons->producer_ready, true);
CHECK_EQ(cons->producer_wakeups, 0u);
WHEN("pushing into the buffer") {
buf->push(1);
CHECK_EQ(cons->producer_wakeups, 1u);
buf->push(2);
CHECK_EQ(cons->producer_wakeups, 1u);
THEN("excess items are stored but do not trigger demand when consumed") {
std::vector<int> tmp{3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14};
buf->push(make_span(tmp));
prod->demand = 0;
CHECK_EQ(cons->producer_wakeups, 1u);
MESSAGE("consume one element");
{
dummy_observer obs;
auto [ok, consumed] = buf->pull(async::prioritize_errors, 1, obs);
CHECK_EQ(ok, true);
CHECK_EQ(consumed, 1u);
CHECK_EQ(obs.consumed, 1u);
CHECK_EQ(prod->demand, 0u);
}
MESSAGE("consume all remaining elements");
{
dummy_observer obs;
auto [ok, consumed] = buf->pull(async::prioritize_errors, 20, obs);
CHECK_EQ(ok, true);
CHECK_EQ(consumed, 13u);
CHECK_EQ(obs.consumed, 13u);
CHECK_EQ(prod->demand, 10u);
}
}
}
}
}
SCENARIO("SPSC buffers moves data between actors") {
GIVEN("a SPSC buffer resource") {
WHEN("opening the resource from two actors") {
THEN("data travels through the SPSC buffer") {
using actor_t = event_based_actor;
auto [rd, wr] = async::make_spsc_buffer_resource<int>(6, 2);
auto inputs = std::vector<int>{1, 2, 4, 8, 16, 32, 64, 128};
auto outputs = std::vector<int>{};
sys.spawn([wr{wr}, &inputs](actor_t* src) {
src->make_observable()
.from_container(inputs)
.filter([](int) { return true; })
.subscribe(wr);
});
sys.spawn([rd{rd}, &outputs](actor_t* snk) {
snk
->make_observable() //
.from_resource(rd)
.for_each([&outputs](int x) { outputs.emplace_back(x); });
});
run();
CHECK_EQ(inputs, outputs);
}
}
}
}
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