Commit 384c78a6 authored by Dominik Charousset's avatar Dominik Charousset

Allow subtypes of `downstream<T>`

parent f22bf810
...@@ -63,13 +63,40 @@ public: ...@@ -63,13 +63,40 @@ public:
virtual ~abstract_downstream(); virtual ~abstract_downstream();
/// Returns the total available credit for all sinks in O(n). /// Returns the total available credit for all sinks in `xs` in O(n).
template <class PathContainer>
static size_t total_credit(const PathContainer& xs) {
return fold(xs, 0u,
[](size_t x, const typename PathContainer::value_type& y) {
return x + y->open_credit;
});
}
/// Returns the total available credit for all sinks in `paths_` in O(n).
size_t total_credit() const; size_t total_credit() const;
/// Returns the maximum credit of all sinks in O(n). /// Returns the maximum credit of all sinks in `paths_` in O(n).
template <class PathContainer>
static size_t max_credit(const PathContainer& xs) {
return fold(xs, 0,
[](size_t x, const typename PathContainer::value_type& y) {
return std::max(x, y->open_credit);
});
}
/// Returns the maximum credit of all sinks in `paths_` in O(n).
size_t max_credit() const; size_t max_credit() const;
/// Returns the minimal credit of all sinks in O(n). /// Returns the minimal credit of all sinks in `xs` in O(n).
template <class PathContainer>
static size_t min_credit(const PathContainer& xs) {
return fold(xs, std::numeric_limits<size_t>::max(),
[](size_t x, const typename PathContainer::value_type& y) {
return std::min(x, y->open_credit);
});
}
/// Returns the minimal credit of all sinks in `paths_` in O(n).
size_t min_credit() const; size_t min_credit() const;
/// Broadcasts the first `*hint` elements of the buffer on all paths. If /// Broadcasts the first `*hint` elements of the buffer on all paths. If
...@@ -99,7 +126,19 @@ public: ...@@ -99,7 +126,19 @@ public:
/// Sends an abort message to all paths and closes the stream. /// Sends an abort message to all paths and closes the stream.
void abort(strong_actor_ptr& cause, const error& reason); void abort(strong_actor_ptr& cause, const error& reason);
optional<path&> find(const strong_actor_ptr& ptr) const; template <class PathContainer>
static path* find(PathContainer& xs, const strong_actor_ptr& ptr) {
auto predicate = [&](const typename PathContainer::value_type& y) {
return y->hdl == ptr;
};
auto e = xs.end();
auto i = std::find_if(xs.begin(), e, predicate);
if (i != e)
return &(*(*i)); // Ugly, but works for both raw and smart pointers.
return nullptr;
}
path* find(const strong_actor_ptr& ptr) const;
size_t available_credit() const; size_t available_credit() const;
...@@ -122,15 +161,25 @@ public: ...@@ -122,15 +161,25 @@ public:
protected: protected:
void send_batch(downstream_path& dest, size_t chunk_size, message chunk); void send_batch(downstream_path& dest, size_t chunk_size, message chunk);
/// Sorts `xs` in descending order by available credit.
template <class PathContainer>
static void sort_by_credit(PathContainer& xs) {
using value_type = typename PathContainer::value_type;
auto cmp = [](const value_type& x, const value_type& y) {
return x->open_credit > y->open_credit;
};
std::sort(xs.begin(), xs.end(), cmp);
}
/// Sorts `paths_` in descending order by available credit. /// Sorts `paths_` in descending order by available credit.
void sort_by_credit(); void sort_by_credit();
template <class F> template <class PathContainer, class F>
size_t fold_paths(size_t init, F f) const { static size_t fold(PathContainer& xs, size_t init, F f) {
auto b = paths_.begin(); auto b = xs.begin();
auto e = paths_.end(); auto e = xs.end();
auto g = [&](size_t x, const path_uptr& y) { auto g = [&](size_t x, const typename PathContainer::value_type& y) {
return f(x, *y); return f(x, y);
}; };
return b != e ? std::accumulate(b, e, init, g) : 0; return b != e ? std::accumulate(b, e, init, g) : 0;
} }
......
...@@ -30,10 +30,13 @@ ...@@ -30,10 +30,13 @@
namespace caf { namespace caf {
template <class T> template <class T>
class downstream final : public abstract_downstream { class downstream : public abstract_downstream {
public: public:
/// A chunk of data for sending batches downstream. /// A chunk of data for sending batches downstream.
using chunk = std::vector<T>; using chunk_type = std::vector<T>;
/// A queue of items for temporary storage before moving them into chunks.
using queue_type = std::deque<T>;
downstream(local_actor* ptr, const stream_id& sid, downstream(local_actor* ptr, const stream_id& sid,
typename abstract_downstream::policy_ptr pptr) typename abstract_downstream::policy_ptr pptr)
...@@ -46,11 +49,11 @@ public: ...@@ -46,11 +49,11 @@ public:
buf_.emplace_back(std::forward<Ts>(xs)...); buf_.emplace_back(std::forward<Ts>(xs)...);
} }
std::deque<T>& buf() { queue_type& buf() {
return buf_; return buf_;
} }
const std::deque<T>& buf() const { const queue_type& buf() const {
return buf_; return buf_;
} }
...@@ -67,15 +70,14 @@ public: ...@@ -67,15 +70,14 @@ public:
} }
void anycast(size_t*) override { void anycast(size_t*) override {
sort_by_credit(); this->sort_by_credit();
for (auto& x : paths_) { for (auto& x : paths_) {
auto chunk = get_chunk(x->open_credit); auto chunk = get_chunk(x->open_credit);
auto csize = chunk.size(); auto csize = chunk.size();
if (csize == 0) if (csize == 0)
return; return;
x->open_credit -= csize; x->open_credit -= csize;
auto wrapped_chunk = make_message(std::move(chunk)); send_batch(*x, csize, std::move(make_message(std::move(chunk))));
send_batch(*x, csize, std::move(wrapped_chunk));
} }
} }
...@@ -83,54 +85,31 @@ public: ...@@ -83,54 +85,31 @@ public:
return buf_.size(); return buf_.size();
} }
private: protected:
/// Returns the minimum of `desired` and `buf_.size()`.
size_t clamp_chunk_size(size_t desired) const {
return std::min(desired, buf_.size());
}
using scratch_space_value = std::tuple<T, bool, atom_value>;
using scratch_space = std::vector<scratch_space_value>;
/// Iterator over a `scratch_space` that behaves like a move
/// iterator over `vector<T>`.
struct scratch_space_move_iter {
typename scratch_space::iterator i;
scratch_space_move_iter& operator++() {
++i;
return *this;
}
T&& operator*() {
return std::move(std::get<0>(*i));
}
bool operator!=(const scratch_space_move_iter& x) {
return i != x.i;
}
};
/// @pre `n <= buf_.size()` /// @pre `n <= buf_.size()`
chunk get_chunk(size_t n) { static chunk_type get_chunk(queue_type& buf, size_t n) {
chunk xs; chunk_type xs;
if (n > 0) { if (n > 0) {
xs.reserve(n); xs.reserve(n);
if (n < buf_.size()) { if (n < buf.size()) {
auto first = buf_.begin(); auto first = buf.begin();
auto last = first + static_cast<ptrdiff_t>(n); auto last = first + static_cast<ptrdiff_t>(n);
std::move(first, last, std::back_inserter(xs)); std::move(first, last, std::back_inserter(xs));
buf_.erase(first, last); buf.erase(first, last);
} else { } else {
std::move(buf_.begin(), buf_.end(), std::back_inserter(xs)); std::move(buf.begin(), buf.end(), std::back_inserter(xs));
buf_.clear(); buf.clear();
} }
} }
return xs; return xs;
} }
std::deque<T> buf_; /// @pre `n <= buf_.size()`
chunk_type get_chunk(size_t n) {
return get_chunk(buf_, n);
}
queue_type buf_;
}; };
} // namespace caf } // namespace caf
......
...@@ -27,7 +27,15 @@ ...@@ -27,7 +27,15 @@
namespace caf { namespace caf {
template <class Fun, class Cleanup> struct default_stream_stage_policy {
template <class InputType>
using upstream_type = upstream<InputType>;
template <class OutputType>
using downstream_type = downstream<OutputType>;
};
template <class Fun, class Cleanup, class Policy = default_stream_stage_policy>
class stream_stage_impl : public stream_stage { class stream_stage_impl : public stream_stage {
public: public:
using trait = stream_stage_trait_t<Fun>; using trait = stream_stage_trait_t<Fun>;
...@@ -38,6 +46,10 @@ public: ...@@ -38,6 +46,10 @@ public:
using output_type = typename trait::output; using output_type = typename trait::output;
using upstream_type = typename Policy::template upstream_type<output_type>;
using downstream_type = typename Policy::template downstream_type<output_type>;
stream_stage_impl(local_actor* self, stream_stage_impl(local_actor* self,
const stream_id& sid, const stream_id& sid,
typename abstract_upstream::policy_ptr in_policy, typename abstract_upstream::policy_ptr in_policy,
...@@ -86,12 +98,20 @@ public: ...@@ -86,12 +98,20 @@ public:
return state_; return state_;
} }
upstream_type& in() {
return in_;
}
downstream_type& out() {
return out_;
}
private: private:
state_type state_; state_type state_;
Fun fun_; Fun fun_;
Cleanup cleanup_; Cleanup cleanup_;
upstream<output_type> in_; upstream_type in_;
downstream<input_type> out_; downstream_type out_;
}; };
} // namespace caf } // namespace caf
......
...@@ -41,20 +41,15 @@ abstract_downstream::~abstract_downstream() { ...@@ -41,20 +41,15 @@ abstract_downstream::~abstract_downstream() {
} }
size_t abstract_downstream::total_credit() const { size_t abstract_downstream::total_credit() const {
auto f = [](size_t x, path_cref y) { return total_credit(paths_);
return x + y.open_credit;
};
return fold_paths(0, f);
} }
size_t abstract_downstream::max_credit() const { size_t abstract_downstream::max_credit() const {
auto f = [](size_t x, path_cref y) { return std::max(x, y.open_credit); }; return max_credit(paths_);
return fold_paths(0, f);
} }
size_t abstract_downstream::min_credit() const { size_t abstract_downstream::min_credit() const {
auto f = [](size_t x, path_cref y) { return std::min(x, y.open_credit); }; return min_credit(paths_);
return fold_paths(std::numeric_limits<size_t>::max(), f);
} }
bool abstract_downstream::add_path(strong_actor_ptr ptr) { bool abstract_downstream::add_path(strong_actor_ptr ptr) {
...@@ -116,16 +111,9 @@ void abstract_downstream::abort(strong_actor_ptr& cause, const error& reason) { ...@@ -116,16 +111,9 @@ void abstract_downstream::abort(strong_actor_ptr& cause, const error& reason) {
make<stream_msg::abort>(this->sid_, reason)); make<stream_msg::abort>(this->sid_, reason));
} }
auto abstract_downstream::find(const strong_actor_ptr& ptr) const abstract_downstream::path*
-> optional<path&> { abstract_downstream::find(const strong_actor_ptr& ptr) const {
auto predicate = [&](const path_uptr& y) { return find(paths_, ptr);
return y->hdl == ptr;
};
auto e = paths_.end();
auto i = std::find_if(paths_.begin(), e, predicate);
if (i != e)
return *(*i);
return none;
} }
size_t abstract_downstream::available_credit() const { size_t abstract_downstream::available_credit() const {
...@@ -143,10 +131,7 @@ void abstract_downstream::send_batch(downstream_path& dest, size_t chunk_size, ...@@ -143,10 +131,7 @@ void abstract_downstream::send_batch(downstream_path& dest, size_t chunk_size,
} }
void abstract_downstream::sort_by_credit() { void abstract_downstream::sort_by_credit() {
auto cmp = [](const path_uptr& x, const path_uptr& y) { sort_by_credit(paths_);
return x->open_credit > y->open_credit;
};
std::sort(paths_.begin(), paths_.end(), cmp);
} }
} // namespace caf } // namespace caf
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