Commit c15cedb5 authored by Dominik Charousset's avatar Dominik Charousset

Pass parent stream manager to downstream managers

parent e7aa36c2
......@@ -58,7 +58,7 @@ public:
// -- constructors, destructors, and assignment operators --------------------
broadcast_downstream_manager(scheduled_actor* selfptr) : super(selfptr) {
broadcast_downstream_manager(stream_manager* parent) : super(parent) {
// nop
}
......@@ -242,7 +242,7 @@ private:
if (chunk.empty()) {
auto g = [&](typename map_type::value_type& x,
typename state_map_type::value_type& y) {
x.second->emit_batches(this->self_, y.second.buf, force_underfull);
x.second->emit_batches(this->self(), y.second.buf, force_underfull);
};
detail::zip_foreach(g, this->paths_.container(), state_map_.container());
} else {
......@@ -257,7 +257,7 @@ private:
if (select_(st.filter, piece))
st.buf.emplace_back(piece);
}
x.second->emit_batches(this->self_, st.buf, force_underfull);
x.second->emit_batches(this->self(), st.buf, force_underfull);
};
detail::zip_foreach(g, this->paths_.container(), state_map_.container());
}
......
......@@ -45,7 +45,7 @@ public:
// -- constructors, destructors, and assignment operators --------------------
buffered_downstream_manager(scheduled_actor* self) : super(self) {
explicit buffered_downstream_manager(stream_manager* parent) : super(parent) {
// nop
}
......
......@@ -59,7 +59,7 @@ public:
template <class... Ts>
stream_distribution_tree(scheduled_actor* selfptr, Ts&&... xs)
: super(selfptr),
out_(selfptr),
out_(this),
policy_(this, std::forward<Ts>(xs)...) {
continuous(true);
}
......
......@@ -63,15 +63,15 @@ public:
// -- constructors, destructors, and assignment operators --------------------
explicit downstream_manager(scheduled_actor* self);
explicit downstream_manager(stream_manager* parent);
virtual ~downstream_manager();
// -- properties -------------------------------------------------------------
scheduled_actor* self() const {
return self_;
}
scheduled_actor* self() const noexcept;
stream_manager* parent() const noexcept;
/// Returns `true` if this manager belongs to a sink, i.e., terminates the
/// stream and never has outbound paths.
......@@ -212,7 +212,7 @@ protected:
// -- member variables -------------------------------------------------------
scheduled_actor* self_;
stream_manager* parent_;
};
} // namespace caf
......
......@@ -41,7 +41,7 @@ public:
// -- constructors, destructors, and assignment operators --------------------
explicit downstream_manager_base(scheduled_actor* self);
explicit downstream_manager_base(stream_manager* parent);
~downstream_manager_base() override;
......
......@@ -130,9 +130,9 @@ public:
// -- constructors, destructors, and assignment operators --------------------
fused_downstream_manager(scheduled_actor* self)
: super(self),
nested_(self, detail::pack_repeat<Ts>(self)...) {
fused_downstream_manager(stream_manager* parent)
: super(parent),
nested_(parent, detail::pack_repeat<Ts>(parent)...) {
detail::init_ptr_array<0, sizeof...(Ts) + 1>::apply(ptrs_, nested_);
}
......
......@@ -38,7 +38,7 @@ public:
// -- constructors, destructors, and assignment operators --------------------
stream_sink(scheduled_actor* self) : stream_manager(self), dummy_out_(self) {
stream_sink(scheduled_actor* self) : stream_manager(self), dummy_out_(this) {
// nop
}
......
......@@ -39,7 +39,7 @@ public:
// -- constructors, destructors, and assignment operators --------------------
stream_source(scheduled_actor* self) : stream_manager(self), out_(self) {
stream_source(scheduled_actor* self) : stream_manager(self), out_(this) {
// nop
}
......
......@@ -39,7 +39,8 @@ downstream_manager::downstream_manager::path_predicate::~path_predicate() {
// nop
}
downstream_manager::downstream_manager(scheduled_actor* self) : self_(self) {
downstream_manager::downstream_manager(stream_manager* parent)
: parent_(parent) {
// nop
}
......@@ -49,6 +50,14 @@ downstream_manager::~downstream_manager() {
// -- properties ---------------------------------------------------------------
scheduled_actor* downstream_manager::self() const noexcept {
return parent_->self();
}
stream_manager* downstream_manager::parent() const noexcept {
return parent_;
}
bool downstream_manager::terminal() const noexcept {
return true;
}
......@@ -177,9 +186,9 @@ void downstream_manager::about_to_erase(outbound_path* ptr, bool silent,
CAF_LOG_TRACE(CAF_ARG(ptr) << CAF_ARG(silent) << CAF_ARG(reason));
if (!silent) {
if (reason == nullptr)
ptr->emit_regular_shutdown(self_);
ptr->emit_regular_shutdown(self());
else
ptr->emit_irregular_shutdown(self_, std::move(*reason));
ptr->emit_irregular_shutdown(self(), std::move(*reason));
}
}
......
......@@ -25,8 +25,8 @@
namespace caf {
downstream_manager_base::downstream_manager_base(scheduled_actor* self)
: super(self) {
downstream_manager_base::downstream_manager_base(stream_manager* parent)
: super(parent) {
// nop
}
......
......@@ -35,6 +35,33 @@ using namespace caf;
namespace {
using bcast_manager = broadcast_downstream_manager<int>;
// Mocks just enough of a stream manager to serve our entity.
class mock_stream_manager : public stream_manager {
public:
using super = stream_manager;
mock_stream_manager(scheduled_actor* self) : super(self), out_(this) {
// nop
}
bcast_manager& out() override {
return out_;
}
bool done() const override {
return false;
}
bool idle() const noexcept override {
return false;
}
private:
bcast_manager out_;
};
// Mocks just enough of an actor to receive and send batches.
class entity : public scheduled_actor {
public:
......@@ -49,7 +76,7 @@ public:
entity(actor_config& cfg, const char* cstr)
: super(cfg),
cstr_name(cstr),
bs(this),
mgr(this),
next_slot(1) {
// nop
}
......@@ -87,7 +114,7 @@ public:
}
void add_path_to(entity& x, int32_t desired_batch_size) {
auto ptr = bs.add_path(next_slot++, x.ctrl());
auto ptr = mgr.out().add_path(next_slot++, x.ctrl());
CAF_REQUIRE(ptr != nullptr);
ptr->desired_batch_size = desired_batch_size;
ptr->slots.receiver = x.next_slot++;
......@@ -107,9 +134,9 @@ public:
for (auto& ptr : paths)
ptr->open_credit += num;
if (force_emit)
bs.force_emit_batches();
mgr.out().force_emit_batches();
else
bs.emit_batches();
mgr.out().emit_batches();
}
const char* name() const override {
......@@ -121,7 +148,7 @@ public:
const char* cstr_name;
/// Manager-under-test.
broadcast_downstream_manager<int> bs;
mock_stream_manager mgr;
/// Dummy mailbox.
std::vector<message> mbox;
......@@ -309,7 +336,7 @@ receive_checker<F> operator<<(receive_checker<F> xs, not_empty_t) {
<<
#define TOTAL \
CAF_CHECK_EQUAL(CONCAT(who, __LINE__).bs.total_credit(), \
CAF_CHECK_EQUAL(CONCAT(who, __LINE__).mgr.out().total_credit(), \
CONCAT(amount, __LINE__))
#define BATCH(first, last) make_batch(first, last)
......@@ -325,7 +352,7 @@ CAF_TEST(one_path_force) {
// of 10.
alice.add_path_to(bob, 10);
for (int i = 1; i <= 100; ++i)
alice.bs.push(i);
alice.mgr.out().push(i);
// Give 3 credit (less than 10).
AFTER ENTITY alice TRIED FORCE_SENDING 3 ELEMENTS {
ENTITY bob RECEIVED BATCH(1, 3);
......@@ -364,7 +391,7 @@ CAF_TEST(one_path_without_force) {
// of 10.
alice.add_path_to(bob, 10);
for (int i = 1; i <= 100; ++i)
alice.bs.push(i);
alice.mgr.out().push(i);
// Give 3 credit (less than 10).
AFTER ENTITY alice TRIED SENDING 3 ELEMENTS {
ENTITY bob RECEIVED none;
......@@ -404,7 +431,7 @@ CAF_TEST(two_paths_different_sizes_force) {
alice.add_path_to(bob, 10);
alice.add_path_to(carl, 7);
for (int i = 1; i <= 100; ++i)
alice.bs.push(i);
alice.mgr.out().push(i);
// Give 3 credit (less than 10).
AFTER ENTITY alice TRIED FORCE_SENDING 3 ELEMENTS {
ENTITY bob RECEIVED BATCH(1, 3);
......@@ -455,7 +482,7 @@ CAF_TEST(two_paths_different_sizes_without_force) {
alice.add_path_to(bob, 10);
alice.add_path_to(carl, 7);
for (int i = 1; i <= 100; ++i)
alice.bs.push(i);
alice.mgr.out().push(i);
// Give 3 credit (less than 10).
AFTER ENTITY alice TRIED SENDING 3 ELEMENTS {
ENTITY bob RECEIVED none;
......
......@@ -174,7 +174,7 @@ class fused_stage : public stream_manager {
public:
using super = stream_manager;
fused_stage(scheduled_actor* self) : stream_manager(self), out_(self) {
fused_stage(scheduled_actor* self) : stream_manager(self), out_(this) {
continuous(true);
}
......
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