Commit 3b70a5bd authored by Dominik Charousset's avatar Dominik Charousset Committed by Dominik Charousset

Reimplement sink/source/stage impl classes

parent 50601aa4
......@@ -32,6 +32,8 @@ public:
void emit_batches() override;
void force_emit_batches() override;
size_t capacity() const noexcept override;
size_t buffered() const noexcept override;
......
......@@ -104,6 +104,10 @@ public:
/// pushed data is limited by the available credit.
virtual void push();
/// Returns true if the handler is not able to process any further batches
/// since it is unable to make progress sending on its own.
virtual bool congested() const;
// -- implementation hooks for sources ---------------------------------------
/// Tries to generate new messages for the stream. This member function does
......
......@@ -19,15 +19,20 @@
#ifndef CAF_STREAM_SINK_IMPL_HPP
#define CAF_STREAM_SINK_IMPL_HPP
#include "caf/sec.hpp"
#include "caf/config.hpp"
#include "caf/logger.hpp"
#include "caf/make_counted.hpp"
#include "caf/message_id.hpp"
#include "caf/sec.hpp"
#include "caf/stream_manager.hpp"
#include "caf/stream_sink_trait.hpp"
#include "caf/terminal_stream_scatterer.hpp"
#include "caf/policy/arg.hpp"
namespace caf {
template <class Fun, class Finalize, class Gatherer, class Scatterer>
template <class Fun, class Finalize>
class stream_sink_impl : public stream_manager {
public:
using super = stream_manager;
......@@ -41,10 +46,11 @@ public:
using output_type = typename trait::output;
stream_sink_impl(local_actor* self, Fun fun, Finalize fin)
: fun_(std::move(fun)),
: stream_manager(self),
fun_(std::move(fun)),
fin_(std::move(fin)),
out_(self),
in_(self, out_) {
open_inputs_(0) {
// nop
}
......@@ -52,32 +58,40 @@ public:
return state_;
}
Gatherer& in() override {
return in_;
}
Scatterer& out() override {
terminal_stream_scatterer& out() override {
return out_;
}
bool done() const override {
return in_.closed();
return open_inputs_ == 0;
}
protected:
error process_batch(message& msg) override {
error handle(inbound_path*, downstream_msg::batch& x) override {
CAF_LOG_TRACE(CAF_ARG(msg));
using vec_type = std::vector<input_type>;
if (msg.match_elements<vec_type>()) {
auto& xs = msg.get_as<vec_type>(0);
if (x.xs.match_elements<vec_type>()) {
auto& xs = x.xs.get_as<vec_type>(0);
for (auto& x : xs)
fun_(state_, x);
fun_(state_, std::move(x));
return none;
}
CAF_LOG_ERROR("received unexpected batch type");
return sec::unexpected_message;
}
void register_input_path(inbound_path* x) override {
CAF_LOG_TRACE(CAF_ARG(*x));
CAF_IGNORE_UNUSED(x);
++open_inputs_;
}
void deregister_input_path(inbound_path* x) noexcept override {
CAF_LOG_TRACE(CAF_ARG(*x));
CAF_IGNORE_UNUSED(x);
--open_inputs_;
}
protected:
message make_final_result() override {
return trait::make_result(state_, fin_);
}
......@@ -86,10 +100,19 @@ private:
state_type state_;
Fun fun_;
Finalize fin_;
Scatterer out_;
Gatherer in_;
terminal_stream_scatterer out_;
long open_inputs_;
};
template <class Init, class Fun, class Finalize>
stream_manager_ptr make_receiving_manager(local_actor* self, Init init, Fun f,
Finalize fin) {
using impl = stream_sink_impl<Fun, Finalize>;
auto ptr = make_counted<impl>(self, std::move(f), std::move(fin));
init(ptr->state());
return ptr;
}
} // namespace caf
#endif // CAF_STREAM_SINK_IMPL_HPP
......@@ -19,12 +19,15 @@
#ifndef CAF_STREAM_SOURCE_IMPL_HPP
#define CAF_STREAM_SOURCE_IMPL_HPP
#include "caf/logger.hpp"
#include "caf/downstream.hpp"
#include "caf/logger.hpp"
#include "caf/make_counted.hpp"
#include "caf/outbound_path.hpp"
#include "caf/stream_manager.hpp"
#include "caf/stream_source_trait.hpp"
#include "caf/policy/arg.hpp"
namespace caf {
template <class Fun, class Predicate, class DownstreamPolicy>
......@@ -37,14 +40,16 @@ public:
using output_type = typename trait::output;
stream_source_impl(local_actor* self, Fun fun, Predicate pred)
: fun_(std::move(fun)),
: stream_manager(self),
at_end_(false),
fun_(std::move(fun)),
pred_(std::move(pred)),
out_(self) {
// nop
}
bool done() const override {
return at_end() && out_.paths_clean();
return at_end_ && out_.clean();
}
......@@ -52,49 +57,45 @@ public:
return out_;
}
bool generate_messages() override {
// Produce new elements. If we have less elements buffered (bsize) than fit
// into a single batch (dbs = desired batch size) then we fill up the
// buffer up to that point. Otherwise, we fill up the buffer up to its
// capacity since we are currently waiting for downstream demand and can
// use the delay to store batches upfront.
size_t size_hint;
auto bsize = out_.buffered();
auto dbs = out_.desired_batch_size();
if (bsize < dbs) {
size_hint= static_cast<size_t>(dbs - bsize);
} else {
auto bmax = out_.min_buffer_size();
if (bsize < bmax)
size_hint = static_cast<size_t>(bmax - bsize);
else
return false;
}
if (at_end_)
return false;
auto hint = out_.capacity();
if (hint == 0)
return false;
downstream<typename DownstreamPolicy::value_type> ds{out_.buf()};
fun_(state_, ds, size_hint);
return out_.buffered() != bsize;
fun_(state_, ds, hint);
if (pred_(state_))
at_end_ = true;
return hint != out_.capacity();
}
state_type& state() {
return state_;
}
protected:
bool at_end() const {
return pred_(state_);
}
void downstream_demand(outbound_path*, long) override {
// TODO: implment me
bool at_end() const noexcept {
return at_end_;;
}
private:
bool at_end_;
state_type state_;
Fun fun_;
Predicate pred_;
DownstreamPolicy out_;
};
template <class Init, class Fun, class Predicate, class Scatterer>
stream_manager_ptr make_stream_source(local_actor* self, Init init, Fun f,
Predicate p, policy::arg<Scatterer>) {
using impl = stream_source_impl < Fun, Predicate, Scatterer>;
auto ptr = make_counted<impl>(self, std::move(f), std::move(p));
init(ptr->state());
return ptr;
}
} // namespace caf
#endif // CAF_STREAM_SOURCE_IMPL_HPP
......@@ -19,17 +19,19 @@
#ifndef CAF_STREAM_STAGE_IMPL_HPP
#define CAF_STREAM_STAGE_IMPL_HPP
#include "caf/sec.hpp"
#include "caf/logger.hpp"
#include "caf/downstream.hpp"
#include "caf/logger.hpp"
#include "caf/make_counted.hpp"
#include "caf/outbound_path.hpp"
#include "caf/sec.hpp"
#include "caf/stream_manager.hpp"
#include "caf/stream_stage_trait.hpp"
#include "caf/policy/arg.hpp"
namespace caf {
template <class Fun, class Cleanup,
class UpstreamPolicy, class DownstreamPolicy>
template <class Fun, class Cleanup, class DownstreamPolicy>
class stream_stage_impl : public stream_manager {
public:
using trait = stream_stage_trait_t<Fun>;
......@@ -40,12 +42,12 @@ public:
using output_type = typename trait::output;
stream_stage_impl(local_actor* self, const stream_id&,
Fun fun, Cleanup cleanup)
: fun_(std::move(fun)),
stream_stage_impl(local_actor* self, Fun fun, Cleanup cleanup)
: stream_manager(self),
fun_(std::move(fun)),
cleanup_(std::move(cleanup)),
out_(self),
in_(self, out_) {
open_inputs_(0) {
// nop
}
......@@ -53,36 +55,34 @@ public:
return state_;
}
UpstreamPolicy& in() override {
return in_;
}
DownstreamPolicy& out() override {
return out_;
}
bool done() const override {
return in_.closed() && out_.closed();
return open_inputs_ == 0 && out_.clean();
}
protected:
void input_closed(error reason) override {
if (reason == none) {
if (out_.buffered() == 0)
out_.close();
} else {
out_.abort(std::move(reason));
}
void register_input_path(inbound_path* x) override {
CAF_LOG_TRACE(CAF_ARG(*x));
CAF_IGNORE_UNUSED(x);
++open_inputs_;
}
error process_batch(message& msg) override {
void deregister_input_path(inbound_path* x) noexcept override {
CAF_LOG_TRACE(CAF_ARG(*x));
CAF_IGNORE_UNUSED(x);
--open_inputs_;
}
error handle(inbound_path*, downstream_msg::batch& x) override {
CAF_LOG_TRACE(CAF_ARG(msg));
using vec_type = std::vector<output_type>;
if (msg.match_elements<vec_type>()) {
auto& xs = msg.get_as<vec_type>(0);
if (x.xs.match_elements<vec_type>()) {
auto& xs = x.xs.get_as<vec_type>(0);
downstream<typename DownstreamPolicy::value_type> ds{out_.buf()};
for (auto& x : xs)
fun_(state_, ds, x);
fun_(state_, ds, std::move(x));
return none;
}
CAF_LOG_ERROR("received unexpected batch type");
......@@ -90,11 +90,12 @@ protected:
}
message make_output_token(const stream_id& x) const override {
return make_message(stream<output_type>{x});
// TODO: return make_message(stream<output_type>{x});
return make_message();
}
void downstream_demand(outbound_path*, long) override {
// TODO: implement me
bool congested() const override {
return out_.buffered() >= 30;
}
private:
......@@ -102,9 +103,18 @@ private:
Fun fun_;
Cleanup cleanup_;
DownstreamPolicy out_;
UpstreamPolicy in_;
long open_inputs_;
};
template <class Init, class Fun, class Cleanup, class Scatterer>
stream_manager_ptr make_stream_stage(local_actor* self, Init init, Fun f,
Cleanup cleanup, policy::arg<Scatterer>) {
using impl = stream_stage_impl<Fun, Cleanup, Scatterer>;
auto ptr = make_counted<impl>(self, std::move(f), std::move(cleanup));
init(ptr->state());
return ptr;
}
} // namespace caf
#endif // CAF_STREAM_STAGE_IMPL_HPP
......@@ -35,6 +35,10 @@ void invalid_stream_scatterer::emit_batches() {
// nop
}
void invalid_stream_scatterer::force_emit_batches() {
// nop
}
size_t invalid_stream_scatterer::capacity() const noexcept {
return 0u;
}
......
......@@ -87,6 +87,10 @@ void stream_manager::push() {
out().emit_batches();
}
bool stream_manager::congested() const {
return false;
}
bool stream_manager::generate_messages() {
return false;
}
......
......@@ -25,6 +25,9 @@
// The setup is a fixed WDRR queue with three nestes queues. The first nested
// queue stores asynchronous messages, the second one upstream messages, and
// the last queue is a dynamic WDRR queue storing downstream messages.
//
// We mock just enough of an actor to use the streaming classes and put them to
// work in a pipeline with 2 or 3 stages.
#define CAF_SUITE streaming_classes
......@@ -42,11 +45,16 @@
#include "caf/send.hpp"
#include "caf/stream_manager.hpp"
#include "caf/stream_scatterer.hpp"
#include "caf/stream_sink_impl.hpp"
#include "caf/stream_slot.hpp"
#include "caf/stream_source_impl.hpp"
#include "caf/stream_stage_impl.hpp"
#include "caf/system_messages.hpp"
#include "caf/upstream_msg.hpp"
#include "caf/variant.hpp"
#include "caf/policy/arg.hpp"
#include "caf/mixin/sender.hpp"
#include "caf/test/unit_test.hpp"
......@@ -112,93 +120,6 @@ const char* name_of(const actor_addr& x) {
return name_of(actor_cast<strong_actor_ptr>(x));
}
// -- manager and path handlers ------------------------------------------------
class mock_manager : public stream_manager {
public:
mock_manager(local_actor* selfptr)
: stream_manager(selfptr),
open_inputs(0),
out_(selfptr) {
// nop
}
bool done() const override {
return open_inputs == 0 && out_.clean();
}
void register_input_path(inbound_path* x) override {
CAF_MESSAGE(out_.self()->name() << " receives stream input from "
<< name_of(x->hdl));
++open_inputs;
}
void deregister_input_path(inbound_path* x) noexcept override {
CAF_MESSAGE(out_.self()->name() << " no longer receives stream input from "
<< name_of(x->hdl));
--open_inputs;
}
stream_scatterer& out() override {
return out_;
}
protected:
int open_inputs;
broadcast_scatterer<int> out_;
};
class sending_manager : public mock_manager {
public:
sending_manager(local_actor* selfptr, int num_messages)
: mock_manager(selfptr) {
for (int i = 0; i < num_messages; ++i)
out_.push(i);
}
};
class receiving_manager : public mock_manager {
public:
receiving_manager(local_actor* selfptr, vector<int>& log)
: mock_manager(selfptr),
log_(log) {
// nop
}
error handle(inbound_path*, downstream_msg::batch& x) override {
CAF_REQUIRE(x.xs.match_elements<vector<int>>());
auto& xs = x.xs.get_as<vector<int>>(0);
for (auto x : xs)
log_.push_back(x);
return none;
}
private:
vector<int>& log_;
};
class forwarding_manager : public mock_manager {
public:
forwarding_manager(local_actor* selfptr, vector<int>& log)
: mock_manager(selfptr),
log_(log) {
// nop
}
error handle(inbound_path*, downstream_msg::batch& x) override {
CAF_REQUIRE(x.xs.match_elements<vector<int>>());
auto& xs = x.xs.get_as<vector<int>>(0);
for (auto x : xs) {
log_.push_back(x);
out_.push(x);
}
return none;
}
private:
vector<int>& log_;
};
// -- policies and queues ------------------------------------------------------
struct policy_base {
......@@ -222,8 +143,8 @@ struct default_queue_policy : policy_base {
using default_queue = drr_queue<default_queue_policy>;
struct umsg_queue_policy : policy_base {
sending_manager* mgr;
umsg_queue_policy(sending_manager* ptr) : mgr(ptr) {
stream_manager* mgr;
umsg_queue_policy(stream_manager* ptr) : mgr(ptr) {
// nop
}
static inline task_size_type task_size(const mailbox_element&) {
......@@ -270,8 +191,8 @@ struct dmsg_queue_policy : policy_base {
}
template <class Queue>
static inline bool enabled(const Queue&) {
return true;
static inline bool enabled(const Queue& q) {
return !q.policy().handler->mgr->congested();
}
template <class Queue>
......@@ -368,7 +289,20 @@ public:
strong_actor_ptr to = ref.ctrl();
send(to, open_stream_msg{slot, make_message(), ctrl(), nullptr,
stream_priority::normal, false});
auto ptr = make_counted<sending_manager>(this, num_messages);
auto init = [](int& x) {
x = 0;
};
auto f = [=](int& x, downstream<int>& out, size_t hint) {
auto y = std::min(num_messages, x + static_cast<int>(hint));
while (x < y)
out.push(x++);
};
auto fin = [=](const int& x) {
return x == num_messages;
};
auto ptr = make_stream_source(this, init, f, fin,
policy::arg<broadcast_scatterer<int>>::value);
ptr->generate_messages();
pending_managers_.emplace(slot, std::move(ptr));
}
......@@ -379,7 +313,19 @@ public:
strong_actor_ptr to = ref.ctrl();
send(to, open_stream_msg{slot, make_message(), ctrl(), nullptr,
stream_priority::normal, false});
forwarder = make_counted<forwarding_manager>(this, data);
using log_ptr = vector<int>*;
auto init = [&](log_ptr& ptr) {
ptr = &data;
};
auto f = [](log_ptr& ptr, downstream<int>& out, int x) {
ptr->push_back(x);
out.push(x);
};
auto cleanup = [](log_ptr&) {
// nop
};
forwarder = make_stream_stage(this, init, f, cleanup,
policy::arg<broadcast_scatterer<int>>::value);
pending_managers_.emplace(slot, forwarder);
}
......@@ -392,8 +338,20 @@ public:
// Create required state if no forwarder exists yet, otherwise `forward_to`
// was called and we run as a stage.
auto mgr = forwarder;
if (mgr == nullptr)
mgr = make_counted<receiving_manager>(this, data);
if (mgr == nullptr) {
using log_ptr = vector<int>*;
auto init = [&](log_ptr& ptr) {
ptr = &data;
};
auto f = [](log_ptr& ptr, int x) {
ptr->push_back(x);
};
auto fin = [](log_ptr&) {
// nop
};
mgr = make_receiving_manager(this, std::move(init), std::move(f),
std::move(fin));
}
// mgr->out().add_path(id, hs.prev_stage);
managers_.emplace(id, mgr);
// Create a new queue in the mailbox for incoming traffic.
......@@ -422,6 +380,7 @@ public:
auto out = i->second->out().add_path(slots.invert(), to);
out->open_credit = x.initial_demand;
out->desired_batch_size = x.desired_batch_size;
i->second->generate_messages();
i->second->push();
pending_managers_.erase(i);
}
......@@ -440,6 +399,7 @@ public:
out->open_credit += x.new_capacity;
out->desired_batch_size = x.desired_batch_size;
out->next_ack_id = x.acknowledged_id + 1;
i->second->generate_messages();
i->second->push();
if (i->second->done()) {
CAF_MESSAGE(name_ << " is done sending batches");
......@@ -519,6 +479,8 @@ struct msg_visitor {
[&](downstream_msg::batch& y) {
TRACE(self->name(), batch, CAF_ARG(y.xs));
inptr->mgr->handle(inptr, y);
inptr->mgr->generate_messages();
inptr->mgr->push();
if (!inptr->mgr->done()) {
auto to = inptr->hdl->get();
to->eq_impl(make_message_id(), self->ctrl(), nullptr,
......@@ -534,7 +496,6 @@ struct msg_visitor {
[&](downstream_msg::close& y) {
TRACE(self->name(), close, CAF_ARG(dm.slots));
auto slots = dm.slots;
TRACE(self->name(), close, CAF_ARG(slots));
auto i = self->managers_.find(slots);
CAF_REQUIRE_NOT_EQUAL(i, self->managers_.end());
i->second->handle(inptr, y);
......@@ -628,9 +589,20 @@ CAF_TEST(depth_2_pipeline) {
CAF_TEST(depth_3_pipeline) {
bob.forward_to(carl);
alice.start_streaming(bob, 100);
alice.start_streaming(bob, 110);
CAF_MESSAGE("loop over alice and bob until bob is congested");
loop(alice, bob);
CAF_CHECK_EQUAL(bob.data, make_iota(0, 100));
CAF_CHECK_EQUAL(bob.data, make_iota(0, 30));
CAF_MESSAGE("loop over bob and carl until bob finsihed sending");
// bob has one batch from alice in its mailbox that bob will read when
// becoming uncongested again
loop(bob, carl);
CAF_CHECK_EQUAL(bob.data, make_iota(0, 40));
CAF_CHECK_EQUAL(carl.data, make_iota(0, 40));
CAF_MESSAGE("loop over all until done");
loop(alice ,bob, carl);
CAF_CHECK_EQUAL(bob.data, make_iota(0, 110));
CAF_CHECK_EQUAL(carl.data, make_iota(0, 110));
}
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