Commit 4c136929 authored by Dominik Charousset's avatar Dominik Charousset

Allow broadcast scatterer to filter its output

parent 71deb342
......@@ -96,8 +96,6 @@
#include "caf/typed_actor_pointer.hpp"
#include "caf/scoped_execution_unit.hpp"
#include "caf/typed_response_promise.hpp"
#include "caf/random_topic_scatterer.hpp"
#include "caf/broadcast_topic_scatterer.hpp"
#include "caf/typed_event_based_actor.hpp"
#include "caf/abstract_composable_behavior.hpp"
......
......@@ -23,10 +23,12 @@
#include "caf/outbound_path.hpp"
#include "caf/detail/algorithms.hpp"
#include "caf/detail/path_state.hpp"
#include "caf/detail/select_all.hpp"
namespace caf {
template <class T>
template <class T, class Filter = unit_t, class Select = detail::select_all>
class broadcast_scatterer : public buffered_scatterer<T> {
public:
// -- member types -----------------------------------------------------------
......@@ -35,24 +37,43 @@ public:
using super = buffered_scatterer<T>;
/// Type of `paths_`.
using map_type = typename super::map_type;
using typename super::map_type;
/// Container for caching `T`s per path.
using cache_type = std::vector<T>;
/// Enables or disables output per path.
using filter_type = Filter;
/// Function object for evaluating filters.
using select_type = Select;
/// Container for caching `T`s per path with active filter.
using path_state = detail::path_state<Filter, T>;
/// Maps slot IDs to caches.
using cache_map_type = detail::unordered_flat_map<stream_slot, cache_type>;
using state_map_type = detail::unordered_flat_map<stream_slot, path_state>;
typename super::path_ptr add_path(stream_slots slots,
strong_actor_ptr target) override {
auto res = caches_.emplace(slots.sender, cache_type{});
return res.second ? super::add_path(slots, target) : nullptr;
}
// -- constructors, destructors, and assignment operators --------------------
broadcast_scatterer(local_actor* selfptr) : super(selfptr) {
// nop
}
// -- properties -------------------------------------------------------------
/// Sets the filter for `x` to `f` and inserts `x` into the appropriate lane.
/// @pre `x` is not registered on *any* lane
void set_filter(stream_slot slot, filter_type filter) {
CAF_LOG_TRACE(CAF_ARG(slot) << CAF_ARG(filter));
state_map_[slot].filter = std::move(filter);
}
// -- overridden functions ---------------------------------------------------
typename super::path_ptr add_path(stream_slots slots,
strong_actor_ptr target) override {
state_map_.emplace(slots.sender, path_state{});
return super::add_path(slots, target);
}
void emit_batches() override {
CAF_LOG_TRACE(CAF_ARG2("buffered", this->buffered())
<< CAF_ARG2("paths", this->paths_.size()));
......@@ -69,13 +90,13 @@ protected:
void about_to_erase(typename super::map_type::iterator i, bool silent,
error* reason) override {
CAF_LOG_DEBUG("remove cache:" << CAF_ARG2("slot", i->second->slots.sender));
caches_.erase(i->second->slots.sender);
state_map_.erase(i->second->slots.sender);
super::about_to_erase(i, silent, reason);
}
private:
void emit_batches_impl(bool force_underfull) {
CAF_ASSERT(this->paths_.size() == caches_.size());
CAF_ASSERT(this->paths_.size() == state_map_.size());
if (this->paths_.empty())
return;
// Calculate the chunk size, i.e., how many more items we can put to our
......@@ -86,8 +107,8 @@ private:
}
};
struct get_cache_size {
inline size_t operator()(typename cache_map_type::value_type& x) {
return x.second.size();
inline size_t operator()(typename state_map_type::value_type& x) {
return x.second.buf.size();
}
};
auto f = [](size_t x, size_t credit, size_t cache_size) {
......@@ -97,26 +118,34 @@ private:
auto chunk_size = detail::zip_fold(
f, std::numeric_limits<size_t>::max(),
detail::make_container_view<get_credit>(this->paths_.container()),
detail::make_container_view<get_cache_size>(caches_.container()));
detail::make_container_view<get_cache_size>(state_map_.container()));
auto chunk = this->get_chunk(chunk_size);
if (chunk.empty()) {
auto g = [&](typename map_type::value_type& x,
typename cache_map_type::value_type& y) {
x.second->emit_batches(this->self_, y.second, force_underfull);
typename state_map_type::value_type& y) {
x.second->emit_batches(this->self_, y.second.buf, force_underfull);
};
detail::zip_foreach(g, this->paths_.container(), caches_.container());
detail::zip_foreach(g, this->paths_.container(), state_map_.container());
} else {
auto g = [&](typename map_type::value_type& x,
typename cache_map_type::value_type& y) {
auto& c = y.second;
c.insert(c.end(), chunk.begin(), chunk.end());
x.second->emit_batches(this->self_, c, force_underfull);
typename state_map_type::value_type& y) {
// TODO: replace with `if constexpr` when switching to C++17
auto& st = y.second;
if (std::is_same<Select, detail::select_all>::value) {
st.buf.insert(st.buf.end(), chunk.begin(), chunk.end());
} else {
Select select;
for (auto& piece : chunk)
if (select(st.filter, piece))
st.buf.emplace_back(piece);
}
x.second->emit_batches(this->self_, st.buf, force_underfull);
};
detail::zip_foreach(g, this->paths_.container(), caches_.container());
detail::zip_foreach(g, this->paths_.container(), state_map_.container());
}
}
cache_map_type caches_;
state_map_type state_map_;
};
} // namespace caf
......
......@@ -16,62 +16,62 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_RANDOM_TOPIC_SCATTERER_HPP
#define CAF_RANDOM_TOPIC_SCATTERER_HPP
#ifndef CAF_DETAIL_PATH_STATE_HPP
#define CAF_DETAIL_PATH_STATE_HPP
#include <map>
#include <tuple>
#include <deque>
#include <new>
#include <vector>
#include <functional>
#include "caf/topic_scatterer.hpp"
#include "caf/unit.hpp"
namespace caf {
namespace detail {
/// A topic scatterer that delivers data to sinks in random order.
template <class T, class Filter, class Select>
class random_topic_scatterer
: public topic_scatterer<T, Filter, Select> {
public:
using super = topic_scatterer<T, Filter, Select>;
/// Bundles a filter and a buffer.
template <class Filter, class T>
struct path_state {
Filter filter;
std::vector<T> buf;
};
/// Compress path_state if `Filter` is `unit`.
template <class T>
struct path_state<unit_t, T> {
using buffer_type = std::vector<T>;
union {
unit_t filter;
buffer_type buf;
};
path_state() {
new (&buf) buffer_type();
}
path_state(path_state&& other) {
new (&buf) buffer_type(std::move(other.buf));
}
random_topic_scatterer(local_actor* selfptr) : super(selfptr) {
// nop
path_state(const path_state& other) {
new (&buf) buffer_type(other.buf);
}
long credit() const override {
// We receive messages until we have exhausted all downstream credit and
// have filled our buffer to its minimum size.
return this->total_credit();
path_state& operator=(path_state&& other) {
buf = std::move(other.buf);
return *this;
}
void emit_batches() override {
CAF_LOG_TRACE("");
this->fan_out();
for (auto& kvp : this->lanes_) {
auto& l = kvp.second;
super::sort_by_credit(l.paths);
for (auto& x : l.paths) {
auto hint = x->desired_batch_size;
auto next_chunk = [&] {
return super::get_chunk(l.buf, std::min(x->open_credit, hint));
};
for (auto chunk = next_chunk(); !chunk.empty(); chunk = next_chunk()) {
auto csize = static_cast<long>(chunk.size());
x->emit_batch(csize, make_message(std::move(chunk)));
}
}
}
path_state& operator=(const path_state& other) {
buf = other.buf;
return *this;
}
long desired_batch_size() const override {
// TODO: this is an O(n) computation, consider storing the result in a
// member variable for
return super::total_desired_batch_size();
~path_state() {
buf.~buffer_type();
}
};
} // namespace detail
} // namespace caf
#endif // CAF_RANDOM_TOPIC_SCATTERER_HPP
#endif // CAF_DETAIL_PATH_STATE_HPP
......@@ -16,61 +16,20 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_BROADCAST_TOPIC_SCATTERER_HPP
#define CAF_BROADCAST_TOPIC_SCATTERER_HPP
#include <map>
#include <tuple>
#include <deque>
#include <vector>
#include <functional>
#include "caf/topic_scatterer.hpp"
#ifndef CAF_DETAIL_SELECT_ALL_HPP
#define CAF_DETAIL_SELECT_ALL_HPP
namespace caf {
namespace detail {
/// A topic scatterer that delivers data in broadcast fashion to all sinks.
template <class T, class Filter, class Select>
class broadcast_topic_scatterer
: public topic_scatterer<T, Filter, Select> {
public:
/// Base type.
using super = topic_scatterer<T, Filter, Select>;
broadcast_topic_scatterer(local_actor* selfptr) : super(selfptr) {
// nop
}
long credit() const override {
// We receive messages until we have exhausted all downstream credit and
// have filled our buffer to its minimum size.
return this->min_credit();
}
void emit_batches() override {
CAF_LOG_TRACE("");
this->fan_out();
for (auto& kvp : this->lanes_) {
auto& l = kvp.second;
auto hint = super::min_desired_batch_size(l.paths);
auto next_chunk = [&] {
// TODO: this iterates paths_ every time again even though we could
// easily keep track of remaining credit
return super::get_chunk(l.buf,
std::min(super::min_credit(l.paths), hint));
};
for (auto chunk = next_chunk(); !chunk.empty(); chunk = next_chunk()) {
auto csize = static_cast<long>(chunk.size());
auto wrapped_chunk = make_message(std::move(chunk));
for (auto& x : l.paths) {
CAF_ASSERT(x->open_credit >= csize);
x->emit_batch(csize, wrapped_chunk);
}
}
}
struct select_all {
template <class T, class U>
constexpr bool operator()(const T&, const U&) const noexcept {
return true;
}
};
} // namespace detail
} // namespace caf
#endif // CAF_BROADCAST_TOPIC_SCATTERER_HPP
#endif // CAF_DETAIL_SELECT_ALL_HPP
......@@ -27,7 +27,6 @@ namespace caf {
// -- 1 param templates --------------------------------------------------------
template <class> class behavior_type_of;
template <class> class broadcast_scatterer;
template <class> class downstream;
template <class> class expected;
template <class> class intrusive_ptr;
......@@ -48,6 +47,7 @@ template <class, class> class make_sink_result;
template <class, class, int> class actor_cast_access;
template <class, class, class> class broadcast_scatterer;
template <class, class, class> class broadcast_topic_scatterer;
template <class, class, class> class random_topic_scatterer;
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_TOPIC_SCATTERER_HPP
#define CAF_TOPIC_SCATTERER_HPP
#include <map>
#include <tuple>
#include <deque>
#include <vector>
#include <functional>
#include "caf/buffered_scatterer.hpp"
#include "caf/meta/type_name.hpp"
namespace caf {
/// A topic scatterer allows stream nodes to fork into multiple lanes, where
/// each lane carries only a subset of the data. For example, the lane
/// mechanism allows you filter key/value pairs before forwarding them to a set
/// of workers.
template <class T, class Filter, class Select>
class topic_scatterer : public buffered_scatterer<T> {
public:
/// Base type.
using super = buffered_scatterer<T>;
struct lane {
typename super::buffer_type buf;
typename super::path_ptr_vec paths;
template <class Inspector>
friend typename Inspector::result_type inspect(Inspector& f, lane& x) {
return f(meta::type_name("lane"), x.buf, x.paths);
}
};
/// Identifies a lane inside the downstream.
using filter_type = Filter;
using lanes_map = std::map<filter_type, lane>;
topic_scatterer(local_actor* selfptr) : super(selfptr) {
// nop
}
using super::remove_path;
bool remove_path(const stream_id& sid, const actor_addr& x,
error reason, bool silent) override {
CAF_LOG_TRACE(CAF_ARG(sid) << CAF_ARG(x)
<< CAF_ARG(reason) << CAF_ARG(silent));
auto i = this->iter_find(this->paths_, sid, x);
if (i != this->paths_.end()) {
erase_from_lanes(i->get());
return super::remove_path(i, std::move(reason), silent);
}
return false;
}
void add_lane(filter_type f) {
std::sort(f);
lanes_.emplace(std::move(f), typename super::buffer_type{});
}
/// Sets the filter for `x` to `f` and inserts `x` into the appropriate lane.
/// @pre `x` is not registered on *any* lane
template <class Handle>
void set_filter(const stream_id& sid, const Handle& x, filter_type f) {
CAF_LOG_TRACE(CAF_ARG(sid) << CAF_ARG(x) << CAF_ARG(f));
auto ptr = super::find(sid, x);
if (!ptr) {
CAF_LOG_WARNING("unable to set filter for unknown path");
return;
}
erase_from_lanes(ptr);
lanes_[std::move(f)].paths.push_back(ptr);
}
const lanes_map& lanes() const {
return lanes_;
}
Select& selector() {
return select_;
}
const Select& selector() const {
return select_;
}
protected:
void erase_from_lanes(typename super::path_ptr ptr) {
for (auto i = lanes_.begin(); i != lanes_.end(); ++i)
if (erase_from_lane(i->second, ptr)) {
if (i->second.paths.empty())
lanes_.erase(i);
return;
}
}
bool erase_from_lane(lane& l, typename super::path_ptr ptr) {
auto e = l.paths.end();
auto i = std::find(l.paths.begin(), e, ptr);
if (i != e) {
l.paths.erase(i);
return true;
}
return false;
}
/// Spreads the content of `buf_` to `lanes_`.
void fan_out() {
for (auto& kvp : lanes_)
for (auto& x : this->buf_)
if (selected(kvp.first, x))
kvp.second.buf.push_back(x);
this->buf_.clear();
}
/// Returns `true` if `x` is selected by `f`, `false` otherwise.
bool selected(const filter_type& f, const T& x) {
if (select_(f, x))
return true;
return false;
}
lanes_map lanes_;
Select select_;
};
} // namespace caf
#endif // CAF_TOPIC_SCATTERER_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#define CAF_SUITE selective_streaming
#include "caf/test/dsl.hpp"
#include <memory>
#include <numeric>
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/atom.hpp"
#include "caf/broadcast_scatterer.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/stateful_actor.hpp"
using std::string;
using namespace caf;
namespace {
enum class level {
all,
trace,
debug,
warning,
error
};
using value_type = std::pair<level, string>;
struct select {
static bool apply(level x, const value_type& y) noexcept {
return x == level::all || x == y.first;
}
bool operator()(level x, const value_type& y) const noexcept {
return apply(x, y);
}
};
using scatterer = broadcast_scatterer<value_type, level, select>;
using buf = std::vector<value_type>;
buf make_log(level lvl) {
buf result{{level::trace, "trace1"},
{level::trace, "trace2"},
{level::debug, "debug1"},
{level::error, "errro1"},
{level::trace, "trace3"}};
auto predicate = [=](const value_type& x) { return !select::apply(lvl, x); };
auto e = result.end();
auto i = std::remove_if(result.begin(), e, predicate);
if (i != e)
result.erase(i, e);
return result;
}
TESTEE_SETUP();
TESTEE(log_producer) {
using buf = std::vector<value_type>;
return {
[=](level lvl) -> output_stream<value_type> {
auto res = self->make_source(
// initialize state
[=](buf& xs) {
xs = make_log(lvl);
},
// get next element
[](buf& xs, downstream<value_type>& out, size_t num) {
CAF_MESSAGE("push " << num << " messages downstream");
auto n = std::min(num, xs.size());
for (size_t i = 0; i < n; ++i)
out.push(xs[i]);
xs.erase(xs.begin(), xs.begin() + static_cast<ptrdiff_t>(n));
},
// check whether we reached the end
[=](const buf& xs) {
if (xs.empty()) {
CAF_MESSAGE(self->name() << " is done");
return true;
}
return false;
},
policy::arg<scatterer>::value
);
res.ptr()->out().set_filter(res.out(), lvl);
return res;
}
};
}
TESTEE_STATE(log_dispatcher) {
stream_stage_ptr<value_type, message, value_type, scatterer> stage;
};
TESTEE(log_dispatcher) {
self->state.stage = self->make_continuous_stage(
// initialize state
[](unit_t&) {
// nop
},
// processing step
[](unit_t&, downstream<value_type>& out, value_type x) {
out.push(std::move(x));
},
// cleanup
[=](unit_t&) {
CAF_MESSAGE(self->name() << " is done");
},
policy::arg<scatterer>::value
);
return {
[=](join_atom, level lvl) {
auto& stg = self->state.stage;
CAF_MESSAGE("received 'join' request");
auto result = self->add_output_path(stg);
stg->out().set_filter(result.out(), lvl);
return result;
},
[=](const stream<value_type>& in) {
return self->add_input_path(in, self->state.stage);
}
};
}
TESTEE_STATE(log_consumer) {
std::vector<value_type> log;
};
TESTEE(log_consumer) {
return {
[=](stream<value_type>& in) {
return self->make_sink(
// input stream
in,
// initialize state
[=](unit_t&) {
// nop
},
// processing step
[=](unit_t&, value_type x) {
self->state.log.emplace_back(std::move(x));
},
// cleanup and produce result message
[=](unit_t&) {
CAF_MESSAGE(self->name() << " is done");
}
);
}
};
}
struct fixture : test_coordinator_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});
// Make sure the current time isn't invalid.
sched.clock().current_time += cycle;
}
};
} // namespace <anonymous>
// -- unit tests ---------------------------------------------------------------
CAF_TEST_FIXTURE_SCOPE(selective_streaming_tests, fixture)
CAF_TEST(select_all) {
auto src = sys.spawn(log_producer);
auto snk = sys.spawn(log_consumer);
CAF_MESSAGE(CAF_ARG(self) << CAF_ARG(src) << CAF_ARG(snk));
CAF_MESSAGE("initiate stream handshake");
self->send(snk * src, level::all);
sched.run_dispatch_loop(cycle);
CAF_CHECK_EQUAL(deref<log_consumer_actor>(snk).state.log,
make_log(level::all));
}
CAF_TEST(select_trace) {
auto src = sys.spawn(log_producer);
auto snk = sys.spawn(log_consumer);
CAF_MESSAGE(CAF_ARG(self) << CAF_ARG(src) << CAF_ARG(snk));
CAF_MESSAGE("initiate stream handshake");
self->send(snk * src, level::trace);
sched.run_dispatch_loop(cycle);
CAF_CHECK_EQUAL(deref<log_consumer_actor>(snk).state.log,
make_log(level::trace));
}
CAF_TEST(forking) {
auto src = sys.spawn(log_producer);
auto stg = sys.spawn(log_dispatcher);
auto snk1 = sys.spawn(log_consumer);
auto snk2 = sys.spawn(log_consumer);
sched.run();
self->send(stg * src, level::all);
self->send(snk1 * stg, join_atom::value, level::trace);
self->send(snk2 * stg, join_atom::value, level::error);
sched.run();
auto& st = deref<log_dispatcher_actor>(stg).state;
auto predicate = [&] {
return st.stage->inbound_paths().empty() && st.stage->out().clean();
};
sched.run_dispatch_loop(predicate, cycle);
CAF_CHECK_EQUAL(deref<log_consumer_actor>(snk1).state.log,
make_log(level::trace));
CAF_CHECK_EQUAL(deref<log_consumer_actor>(snk2).state.log,
make_log(level::error));
self->send(stg, exit_reason::kill);
}
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