Commit bec3a24c authored by Dominik Charousset's avatar Dominik Charousset

Keep paths map protected in stream scatterer

parent a9036d34
......@@ -58,15 +58,13 @@ public:
// -- path management --------------------------------------------------------
/// Returns the container that stores all paths.
inline const map_type& paths() const noexcept {
return paths_;
}
/// Adds a path to `target` to the scatterer.
/// @returns The added path on success, `nullptr` otherwise.
virtual path_ptr add_path(stream_slots slots, strong_actor_ptr target);
/// Returns the current number of paths.
size_t num_paths() const noexcept;
/// Removes a path from the scatterer and returns it.
path_unique_ptr take_path(stream_slot slots) noexcept;
......
......@@ -56,6 +56,10 @@ pointer stream_scatterer::add_path(stream_slots slots,
return nullptr;
}
size_t stream_scatterer::num_paths() const noexcept {
return paths_.size();
}
unique_pointer stream_scatterer::take_path(stream_slot slot) noexcept {
unique_pointer result;
auto i = paths_.find(slot);
......
......@@ -49,7 +49,7 @@ public:
: super(cfg),
cstr_name(cstr),
bs(this),
next_slot(0) {
next_slot(1) {
// nop
}
......@@ -86,39 +86,24 @@ public:
}
void add_path_to(entity& x, int desired_batch_size) {
auto p = bs.add_path({next_slot++, 0}, x.ctrl());
p->desired_batch_size = desired_batch_size;
}
void give_credit(int) {
// end of recursion
}
template <class... Ts>
void give_credit(int num, entity& x, Ts&... xs) {
auto pred = [&](const stream_scatterer::map_type::value_type& kvp) {
return actor_cast<abstract_actor*>(kvp.second->hdl) == &x;
};
auto& ps = bs.paths();
auto i = std::find_if(ps.begin(), ps.end(), pred);
CAF_REQUIRE(i != ps.end());
i->second->open_credit += num;
give_credit(num, xs...);
auto ptr = bs.add_path({next_slot++, 0}, x.ctrl());
CAF_REQUIRE(ptr != nullptr);
ptr->desired_batch_size = desired_batch_size;
paths.emplace_back(ptr);
}
size_t credit_for(entity& x) {
auto pred = [&](const stream_scatterer::map_type::value_type& kvp) {
return actor_cast<abstract_actor*>(kvp.second->hdl) == &x;
auto pred = [&](outbound_path* ptr) {
return ptr->hdl == &x;
};
auto& ps = bs.paths();
auto i = std::find_if(ps.begin(), ps.end(), pred);
CAF_REQUIRE(i != ps.end());
return i->second->open_credit;
auto i = std::find_if(paths.begin(), paths.end(), pred);
CAF_REQUIRE(i != paths.end());
return (*i)->open_credit;
}
void new_round(int num, bool force_emit) {
for (auto& kvp : bs.paths())
kvp.second->open_credit += num;
for (auto& ptr : paths)
ptr->open_credit += num;
if (force_emit)
bs.force_emit_batches();
else
......@@ -133,10 +118,16 @@ public:
const char* cstr_name;
/// Scatterer-under-test.
broadcast_scatterer<int> bs;
/// Dummy mailbox.
std::vector<message> mbox;
/// All outbound paths managed by `bs`.
std::vector<outbound_path*> paths;
/// Next free ID.
stream_slot next_slot;
};
......
......@@ -163,17 +163,17 @@ CAF_TEST(depth_3_pipeline_with_fork) {
self->send(snk1, join_atom::value, stg);
self->send(snk2, join_atom::value, stg);
sched.run();
CAF_CHECK_EQUAL(st.stage->out().paths().size(), 2u);
CAF_CHECK_EQUAL(st.stage->out().num_paths(), 2u);
CAF_MESSAGE("connect source to the stage (fork)");
self->send(stg * src, "numbers.txt");
sched.run();
CAF_CHECK_EQUAL(st.stage->out().paths().size(), 2u);
CAF_CHECK_EQUAL(st.stage->out().num_paths(), 2u);
CAF_CHECK_EQUAL(st.stage->inbound_paths().size(), 1u);
auto predicate = [&] {
return st.stage->inbound_paths().empty() && st.stage->out().clean();
};
sched.run_dispatch_loop(predicate, cycle);
CAF_CHECK_EQUAL(st.stage->out().paths().size(), 2u);
CAF_CHECK_EQUAL(st.stage->out().num_paths(), 2u);
CAF_CHECK_EQUAL(st.stage->inbound_paths().size(), 0u);
CAF_CHECK_EQUAL(deref<sum_up_actor>(snk1).state.x, 1275);
CAF_CHECK_EQUAL(deref<sum_up_actor>(snk2).state.x, 1275);
......@@ -189,18 +189,18 @@ CAF_TEST(depth_3_pipeline_with_join) {
CAF_MESSAGE("connect sink to the stage");
self->send(snk, join_atom::value, stg);
sched.run();
CAF_CHECK_EQUAL(st.stage->out().paths().size(), 1u);
CAF_CHECK_EQUAL(st.stage->out().num_paths(), 1u);
CAF_MESSAGE("connect sources to the stage (join)");
self->send(stg * src1, "numbers.txt");
self->send(stg * src2, "numbers.txt");
sched.run();
CAF_CHECK_EQUAL(st.stage->out().paths().size(), 1u);
CAF_CHECK_EQUAL(st.stage->out().num_paths(), 1u);
CAF_CHECK_EQUAL(st.stage->inbound_paths().size(), 2u);
auto predicate = [&] {
return st.stage->inbound_paths().empty() && st.stage->out().clean();
};
sched.run_dispatch_loop(predicate, cycle);
CAF_CHECK_EQUAL(st.stage->out().paths().size(), 1u);
CAF_CHECK_EQUAL(st.stage->out().num_paths(), 1u);
CAF_CHECK_EQUAL(st.stage->inbound_paths().size(), 0u);
CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.x, 2550);
self->send_exit(stg, exit_reason::kill);
......
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