Commit d3ea583a authored by Dominik Charousset's avatar Dominik Charousset

Re-implement streaming stages

parent 71b47d40
......@@ -50,18 +50,17 @@ public:
void emit_batches() override {
CAF_LOG_TRACE("");
if (!this->delay_partial_batches()) {
force_emit_batches();
} else {
auto f = [&](outbound_path&, cache_type&) {
// Do nothing, i.e., leave the cache untouched.
};
emit_batches_impl(f);
}
// Handler for the last (underful) batch that does nothing, i.e., leaves
// the cache untouched.
auto f = [&](outbound_path&, cache_type&) {
// nop
};
emit_batches_impl(f);
}
void force_emit_batches() override {
CAF_LOG_TRACE("");
// Handler for the last (underful) batch that sends it anyway.
auto f = [&](outbound_path& p, cache_type& c) {
p.emit_batch(this->self_, c.size(), make_message(std::move(c)));
};
......
......@@ -35,7 +35,7 @@ class stream_stage_driver_impl;
template <class Input, class Output, class Process, class Finalize, class... Ts>
class stream_stage_driver_impl<Input, Output, Process, Finalize,
std::tuple<Ts...>> final
: public stream_stage_driver<Output, Ts...> {
: public stream_stage_driver<Input, Output, Ts...> {
public:
// -- member types -----------------------------------------------------------
......@@ -65,7 +65,7 @@ public:
init(state_);
}
handshake_tuple_type make_handshake() override {
handshake_tuple_type make_handshake() const override {
return std::tuple_cat(std::make_tuple(none), hs_);
}
......@@ -75,7 +75,7 @@ public:
}
void finalize() override {
return fin_();
return fin_(state_);
}
private:
......
......@@ -47,6 +47,8 @@ public:
using scatterer_type = Scatterer;
using input_type = typename driver_type::input_type;
using output_type = typename driver_type::output_type;
// -- constructors, destructors, and assignment operators --------------------
......@@ -72,11 +74,12 @@ public:
void handle(inbound_path*, downstream_msg::batch& x) override {
CAF_LOG_TRACE(CAF_ARG(x));
using vec_type = std::vector<output_type>;
using vec_type = std::vector<input_type>;
if (x.xs.match_elements<vec_type>()) {
auto& xs = x.xs.get_mutable_as<vec_type>(0);
downstream<output_type> ds{out_.buf()};
driver_.process(std::move(xs), ds);
return;
}
CAF_LOG_ERROR("received unexpected batch type (dropped)");
}
......
......@@ -57,12 +57,14 @@
#include "caf/policy/urgent_messages.hpp"
#include "caf/detail/behavior_stack.hpp"
#include "caf/detail/tick_emitter.hpp"
#include "caf/detail/stream_sink_driver_impl.hpp"
#include "caf/detail/stream_sink_impl.hpp"
#include "caf/detail/stream_source_driver_impl.hpp"
#include "caf/detail/stream_source_impl.hpp"
#include "caf/detail/stream_stage_driver_impl.hpp"
#include "caf/detail/stream_stage_driver_impl.hpp"
#include "caf/detail/stream_stage_impl.hpp"
#include "caf/detail/tick_emitter.hpp"
#include "caf/detail/unordered_flat_map.hpp"
#include "caf/intrusive/drr_cached_queue.hpp"
......@@ -463,6 +465,54 @@ public:
std::move(fin));
}
template <class Driver,
class Scatterer = broadcast_scatterer<typename Driver::output_type>,
class Input = int, class... Ts>
typename Driver::output_stream_type make_stage(const stream<Input>& in,
Ts&&... xs) {
auto slot = next_slot();
stream_slots id{in.slot(), slot};
auto ptr = detail::make_stream_stage<Driver, Scatterer>(
this, std::forward<Ts>(xs)...);
if (!add_stream_manager(id, ptr)) {
CAF_LOG_WARNING("unable to add a stream manager for a stage");
return {0, nullptr};
}
return {slot, std::move(ptr)};
}
template <class In, class... Ts, class Init, class Fun, class Cleanup,
class Scatterer =
broadcast_scatterer<typename stream_stage_trait_t<Fun>::output>,
class Trait = stream_stage_trait_t<Fun>>
output_stream<typename Trait::output, detail::decay_t<Ts>...>
make_stage(const stream<In>& in, std::tuple<Ts...> xs, Init init, Fun fun,
Cleanup cleanup, policy::arg<Scatterer> scatterer_type = {}) {
CAF_IGNORE_UNUSED(scatterer_type);
CAF_ASSERT(current_mailbox_element() != nullptr);
CAF_ASSERT(
current_mailbox_element()->content().match_elements<open_stream_msg>());
using output_type = typename stream_stage_trait_t<Fun>::output;
using state_type = typename stream_stage_trait_t<Fun>::state;
static_assert(std::is_same<
void (state_type&),
typename detail::get_callable_trait<Init>::fun_sig
>::value,
"Expected signature `void (State&)` for init function");
static_assert(std::is_same<
void (state_type&, downstream<output_type>&, In),
typename detail::get_callable_trait<Fun>::fun_sig
>::value,
"Expected signature `void (State&, downstream<Out>&, In)` "
"for consume function");
using driver =
detail::stream_stage_driver_impl<typename Trait::input, output_type, Fun,
Cleanup,
std::tuple<detail::decay_t<Ts>...>>;
return make_stage<driver, Scatterer>(in, std::move(init), std::move(fun),
std::move(cleanup), std::move(xs));
}
/*
/// Creates a new stream source and starts streaming to `dest`.
/// @param dest Actor handle to the stream destination.
......
......@@ -102,20 +102,6 @@ public:
/// all individual credits.
size_t total_credit() const noexcept;
// -- configuration parameters -----------------------------------------------
/// Forces to actor to emit a batch even if the minimum batch size was not
/// reached.
inline duration max_batch_delay() const noexcept {
return max_batch_delay_;
}
/// Forces to actor to emit a batch even if the minimum batch size was not
/// reached.
inline void max_batch_delay(duration x) noexcept {
max_batch_delay_ = x;
}
// -- state access -----------------------------------------------------------
local_actor* self() const {
......@@ -152,10 +138,6 @@ public:
/// Convenience function for calling `find(x, actor_cast<actor_addr>(x))`.
path_ptr find(stream_slot slot, const strong_actor_ptr& x);
inline bool delay_partial_batches() const {
return max_batch_delay().count != 0;
}
protected:
// -- customization points ---------------------------------------------------
......@@ -168,7 +150,6 @@ protected:
map_type paths_;
local_actor* self_;
duration max_batch_delay_;
};
} // namespace caf
......
......@@ -44,7 +44,7 @@ struct stream_stage_trait_invoke_all {
template <class F, class State, class In, class Out>
static void invoke(F& f, State& st, std::vector<In>&& xs,
downstream<Out>& out) {
f(st, std::move(xs), out);
f(st, out, std::move(xs));
}
};
......@@ -60,15 +60,15 @@ struct stream_stage_trait<void (State&, downstream<Out>&, In)> {
using state = State;
using input = In;
using output = Out;
using invoke = detail::stream_stage_trait_invoke_one;
using process = detail::stream_stage_trait_invoke_one;
};
template <class State, class In, class Out>
struct stream_stage_trait<void (State&, std::vector<In>&&, downstream<Out>&)> {
struct stream_stage_trait<void (State&, downstream<Out>&, std::vector<In>&&)> {
using state = State;
using input = In;
using output = Out;
using invoke = detail::stream_stage_trait_invoke_all;
using process = detail::stream_stage_trait_invoke_all;
};
// -- convenience alias --------------------------------------------------------
......
......@@ -99,6 +99,8 @@ void inbound_path::emit_ack_open(local_actor* self, actor_addr rebind_from) {
void inbound_path::emit_ack_batch(local_actor* self, long queued_items,
timespan cycle, timespan complexity) {
CAF_LOG_TRACE(CAF_ARG(queued_items) << CAF_ARG(cycle) << CAF_ARG(complexity));
if (last_acked_batch_id == last_batch_id)
return;
auto x = stats.calculate(cycle, complexity);
// Hand out enough credit to fill our queue for 2 cycles.
auto credit = std::max((x.max_throughput * 2)
......@@ -112,6 +114,7 @@ void inbound_path::emit_ack_batch(local_actor* self, long queued_items,
make<upstream_msg::ack_batch>(slots.invert(), self->address(),
static_cast<int32_t>(credit),
batch_size, last_batch_id));
last_acked_batch_id = last_batch_id;
}
void inbound_path::emit_regular_shutdown(local_actor* self) {
......
......@@ -28,10 +28,10 @@ namespace caf {
outbound_path::outbound_path(stream_slots id, strong_actor_ptr ptr)
: slots(id),
hdl(std::move(ptr)),
next_batch_id(0),
next_batch_id(1),
open_credit(0),
desired_batch_size(0),
next_ack_id(0) {
next_ack_id(1) {
// nop
}
......
......@@ -50,9 +50,7 @@ public:
cstr_name(cstr),
bs(this),
next_slot(0) {
// Make sure emit_batches() does not send partially filled batches.
bs.max_batch_delay(duration{std::chrono::milliseconds(100)});
CAF_CHECK_EQUAL(bs.delay_partial_batches(), true);
// nop
}
void enqueue(mailbox_element_ptr what, execution_unit*) override {
......
......@@ -61,7 +61,7 @@ VARARGS_TESTEE(file_reader, size_t buf_size) {
// initialize state
[&](buf& xs) {
xs.resize(buf_size);
std::iota(xs.begin(), xs.end(), 0);
std::iota(xs.begin(), xs.end(), 1);
},
// get next element
[=](buf& xs, downstream<int>& out, size_t num) {
......@@ -142,8 +142,37 @@ TESTEE(broken_sink) {
};
}
TESTEE(filter) {
CAF_IGNORE_UNUSED(self);
return {
[=](stream<int>& in, std::string& fname) -> stream<int> {
CAF_CHECK_EQUAL(fname, "numbers.txt");
return self->make_stage(
// input stream
in,
// forward file name in handshake to next stage
std::forward_as_tuple(std::move(fname)),
// initialize state
[=](unit_t&) {
// nop
},
// processing step
[=](unit_t&, downstream<int>& out, int x) {
if ((x & 0x01) != 0)
out.push(x);
},
// cleanup
[=](unit_t&) {
// nop
}
);
}
};
}
struct fixture : test_coordinator_fixture<> {
fixture() {
std::chrono::microseconds cycle;
fixture() : cycle(cfg.streaming_credit_round_interval_us) {
// Configure the clock to measure each batch item with 1us.
sched.clock().time_per_unit.emplace(atom("batch"), timespan{1000});
}
......@@ -160,10 +189,9 @@ error fail_state(const actor& x) {
CAF_TEST_FIXTURE_SCOPE(local_streaming_tests, fixture)
CAF_TEST(depth_2_pipeline_10_items) {
std::chrono::microseconds cycle{cfg.streaming_credit_round_interval_us};
CAF_TEST(depth_2_pipeline_50_items) {
sched.clock().current_time += cycle;
auto src = sys.spawn(file_reader, 10);
auto src = sys.spawn(file_reader, 50);
auto snk = sys.spawn(sum_up);
auto pipeline = snk * src;
CAF_MESSAGE(CAF_ARG(self) << CAF_ARG(src) << CAF_ARG(snk));
......@@ -181,15 +209,14 @@ CAF_TEST(depth_2_pipeline_10_items) {
expect((upstream_msg::ack_batch), from(snk).to(src));
CAF_MESSAGE("expect close message from src and then result from snk");
expect((downstream_msg::close), from(src).to(snk));
expect((int), from(snk).to(self).with(45));
expect((int), from(snk).to(self).with(1275));
CAF_CHECK_EQUAL(fail_state(snk), exit_reason::normal);
CAF_CHECK_EQUAL(fail_state(src), exit_reason::normal);
}
CAF_TEST(delayed_depth_2_pipeline_10_items) {
std::chrono::microseconds cycle{cfg.streaming_credit_round_interval_us};
CAF_TEST(delayed_depth_2_pipeline_50_items) {
sched.clock().current_time += cycle;
auto src = sys.spawn(file_reader, 10);
auto src = sys.spawn(file_reader, 50);
auto snk = sys.spawn(delayed_sum_up);
auto pipeline = snk * src;
CAF_MESSAGE(CAF_ARG(self) << CAF_ARG(src) << CAF_ARG(snk));
......@@ -213,13 +240,12 @@ CAF_TEST(delayed_depth_2_pipeline_10_items) {
expect((upstream_msg::ack_batch), from(snk).to(src));
CAF_MESSAGE("expect close message from src and then result from snk");
expect((downstream_msg::close), from(src).to(snk));
expect((int), from(snk).to(self).with(45));
expect((int), from(snk).to(self).with(1275));
CAF_CHECK_EQUAL(fail_state(snk), exit_reason::normal);
CAF_CHECK_EQUAL(fail_state(src), exit_reason::normal);
}
CAF_TEST(depth_2_pipeline_500_items) {
std::chrono::microseconds cycle{cfg.streaming_credit_round_interval_us};
sched.clock().current_time += cycle;
auto src = sys.spawn(file_reader, 500);
auto snk = sys.spawn(sum_up);
......@@ -246,7 +272,7 @@ CAF_TEST(depth_2_pipeline_500_items) {
} while (!received<downstream_msg::close>(snk));
CAF_MESSAGE("expect close message from src and then result from snk");
expect((downstream_msg::close), from(src).to(snk));
expect((int), from(snk).to(self).with(124750));
expect((int), from(snk).to(self).with(125250));
CAF_CHECK_EQUAL(fail_state(snk), exit_reason::normal);
CAF_CHECK_EQUAL(fail_state(src), exit_reason::normal);
}
......@@ -307,4 +333,44 @@ CAF_TEST(depth_2_pipelin_error_at_sink) {
CAF_CHECK_EQUAL(fail_state(snk), exit_reason::kill);
}
CAF_TEST(depth_3_pipeline_50_items) {
sched.clock().current_time += cycle;
auto src = sys.spawn(file_reader, 50);
auto stg = sys.spawn(filter);
auto snk = sys.spawn(sum_up);
auto next_cycle = [&] {
sched.clock().current_time += cycle;
sched.dispatch();
expect((timeout_msg), from(snk).to(snk));
expect((timeout_msg), from(stg).to(stg));
expect((timeout_msg), from(src).to(src));
};
auto pipeline = snk * stg * src;
CAF_MESSAGE(CAF_ARG(self) << CAF_ARG(src) << CAF_ARG(stg) << CAF_ARG(snk));
CAF_MESSAGE("initiate stream handshake");
self->send(pipeline, "numbers.txt");
expect((string), from(self).to(src).with("numbers.txt"));
expect((open_stream_msg), from(self).to(stg));
expect((open_stream_msg), from(self).to(snk));
expect((upstream_msg::ack_open), from(snk).to(stg));
expect((upstream_msg::ack_open), from(stg).to(src));
CAF_MESSAGE("start data transmission (a single batch)");
expect((downstream_msg::batch), from(src).to(stg));
CAF_MESSAGE("the stage should delay its first batch since its underfull");
disallow((downstream_msg::batch), from(stg).to(snk));
next_cycle();
CAF_MESSAGE("expect close message from src and batch from stage to sink");
expect((upstream_msg::ack_batch), from(stg).to(src));
expect((downstream_msg::close), from(src).to(stg));
expect((downstream_msg::batch), from(stg).to(snk));
next_cycle();
CAF_MESSAGE("expect close message from stage to sink");
expect((upstream_msg::ack_batch), from(snk).to(stg));
expect((downstream_msg::close), from(stg).to(snk));
expect((int), from(snk).to(self).with(625));
CAF_CHECK_EQUAL(fail_state(snk), exit_reason::normal);
CAF_CHECK_EQUAL(fail_state(stg), exit_reason::normal);
CAF_CHECK_EQUAL(fail_state(src), exit_reason::normal);
}
CAF_TEST_FIXTURE_SCOPE_END()
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