Commit 24808de1 authored by Dominik Charousset's avatar Dominik Charousset

Propagate capacity to policy via hint parameter

parent 1ec54972
...@@ -72,13 +72,13 @@ public: ...@@ -72,13 +72,13 @@ public:
/// Returns the minimal credit of all sinks in O(n). /// Returns the minimal credit of all sinks in O(n).
size_t min_credit() const; size_t min_credit() const;
/// Broadcasts the first `min_credit()` elements of the buffer on all paths /// Broadcasts the first `*hint` elements of the buffer on all paths. If
/// for all topics. /// `hint == nullptr` then `min_credit()` is used instead.
virtual void broadcast() = 0; virtual void broadcast(size_t* hint = nullptr) = 0;
/// Sends `total_credit()` elements of the buffer to any available path /// Sends `*hint` elements of the buffer to available paths. If
/// on any topic. /// `hint == nullptr` then `total_credit()` is used instead.
virtual void anycast() = 0; virtual void anycast(size_t* hint = nullptr) = 0;
/// Returns the size of the output buffer. /// Returns the size of the output buffer.
virtual size_t buf_size() const = 0; virtual size_t buf_size() const = 0;
......
...@@ -54,9 +54,8 @@ public: ...@@ -54,9 +54,8 @@ public:
return buf_; return buf_;
} }
void broadcast() override { void broadcast(size_t* hint) override {
auto chunk = get_chunk(hint ? *hint : min_credit());
auto chunk = get_chunk(min_credit());
auto csize = chunk.size(); auto csize = chunk.size();
if (csize == 0) if (csize == 0)
return; return;
...@@ -67,7 +66,7 @@ public: ...@@ -67,7 +66,7 @@ public:
} }
} }
void anycast() override { void anycast(size_t*) override {
sort_by_credit(); 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);
......
...@@ -38,8 +38,9 @@ public: ...@@ -38,8 +38,9 @@ public:
/// Queries the optimal amount of data for the next `push` operation to `x`. /// Queries the optimal amount of data for the next `push` operation to `x`.
virtual size_t desired_buffer_size(const abstract_downstream& x) = 0; virtual size_t desired_buffer_size(const abstract_downstream& x) = 0;
/// Pushes data to the downstream paths of `x`. /// Pushes data to the downstream paths of `x`, optionally passing the last
virtual void push(abstract_downstream& x) = 0; /// result of `desired_buffer_size` for `x` as second argument.
virtual void push(abstract_downstream& x, size_t* hint = nullptr) = 0;
// TODO: callbacks für new and closed downstream pahts // TODO: callbacks für new and closed downstream pahts
......
...@@ -43,9 +43,9 @@ public: ...@@ -43,9 +43,9 @@ public:
return sec::downstream_already_exists; return sec::downstream_already_exists;
} }
error push() final { error push(size_t* hint = nullptr) final {
if (out().buf_size() > 0) if (out().buf_size() > 0)
out().policy().push(out()); out().policy().push(out(), hint);
return none; return none;
} }
......
...@@ -27,7 +27,7 @@ namespace policy { ...@@ -27,7 +27,7 @@ namespace policy {
class anycast final : public downstream_policy { class anycast final : public downstream_policy {
public: public:
void push(abstract_downstream& out) override; void push(abstract_downstream& out, size_t* hint) override;
size_t desired_buffer_size(const abstract_downstream& out) override; size_t desired_buffer_size(const abstract_downstream& out) override;
}; };
......
...@@ -27,7 +27,7 @@ namespace policy { ...@@ -27,7 +27,7 @@ namespace policy {
class broadcast final : public downstream_policy { class broadcast final : public downstream_policy {
public: public:
void push(abstract_downstream& out) override; void push(abstract_downstream& out, size_t* hint) override;
size_t desired_buffer_size(const abstract_downstream& out) override; size_t desired_buffer_size(const abstract_downstream& out) override;
}; };
......
...@@ -54,9 +54,10 @@ public: ...@@ -54,9 +54,10 @@ public:
/// @pre `new_demand > 0` /// @pre `new_demand > 0`
virtual error downstream_demand(strong_actor_ptr& hdl, size_t new_demand); virtual error downstream_demand(strong_actor_ptr& hdl, size_t new_demand);
/// Push new data to downstream by sending batches. The amount of pushed data /// Push new data to downstream actors by sending batches. The amount of
/// is limited by the available credit. /// pushed data is limited by `hint` or the available credit if
virtual error push(); /// `hint == nullptr`.
virtual error push(size_t* hint = nullptr);
// -- handler for upstream events -------------------------------------------- // -- handler for upstream events --------------------------------------------
...@@ -83,19 +84,21 @@ public: ...@@ -83,19 +84,21 @@ public:
virtual bool done() const = 0; virtual bool done() const = 0;
/// Queries whether this handler is a sink, i.e., /// Returns the downstream if this handler is a sink or stage.
/// does not accept downstream actors. virtual optional<abstract_downstream&> get_downstream();
virtual bool is_sink() const;
/// Queries whether this handler is a hdl, i.e., /// Returns the upstream if this handler is a source or stage.
/// does not accepts upstream actors. virtual optional<abstract_upstream&> get_upstream();
virtual bool is_source() const;
/// Returns a type-erased `stream<T>` as handshake token for downstream /// Returns a type-erased `stream<T>` as handshake token for downstream
/// actors. Returns an empty message for sinks. /// actors. Returns an empty message for sinks.
virtual message make_output_token(const stream_id&) const; virtual message make_output_token(const stream_id&) const;
}; };
/// A reference counting pointer to a `stream_handler`.
/// @relates stream_handler
using stream_handler_ptr = intrusive_ptr<stream_handler>;
} // namespace caf } // namespace caf
#endif // CAF_STREAM_HANDLER_HPP #endif // CAF_STREAM_HANDLER_HPP
...@@ -63,6 +63,10 @@ public: ...@@ -63,6 +63,10 @@ public:
return trait::make_result(state_, fin_); return trait::make_result(state_, fin_);
} }
optional<abstract_upstream&> get_upstream() final {
return in_;
}
state_type& state() { state_type& state() {
return state_; return state_;
} }
......
...@@ -58,6 +58,10 @@ public: ...@@ -58,6 +58,10 @@ public:
return pred_(state_); return pred_(state_);
} }
optional<abstract_downstream&> get_downstream() final {
return out_;
}
state_type& state() { state_type& state() {
return state_; return state_;
} }
......
...@@ -66,6 +66,14 @@ public: ...@@ -66,6 +66,14 @@ public:
return make_message(stream<output_type>{x}); return make_message(stream<output_type>{x});
} }
optional<abstract_downstream&> get_downstream() final {
return out_;
}
optional<abstract_upstream&> get_upstream() final {
return in_;
}
state_type& state() { state_type& state() {
return state_; return state_;
} }
......
...@@ -24,8 +24,8 @@ ...@@ -24,8 +24,8 @@
namespace caf { namespace caf {
namespace policy { namespace policy {
void anycast::push(abstract_downstream& out) { void anycast::push(abstract_downstream& out, size_t* hint) {
out.anycast(); out.anycast(hint);
} }
size_t anycast::desired_buffer_size(const abstract_downstream& out) { size_t anycast::desired_buffer_size(const abstract_downstream& out) {
......
...@@ -24,8 +24,8 @@ ...@@ -24,8 +24,8 @@
namespace caf { namespace caf {
namespace policy { namespace policy {
void broadcast::push(abstract_downstream& out) { void broadcast::push(abstract_downstream& out, size_t* hint) {
out.broadcast(); out.broadcast(hint);
} }
size_t broadcast::desired_buffer_size(const abstract_downstream& out) { size_t broadcast::desired_buffer_size(const abstract_downstream& out) {
......
...@@ -48,7 +48,7 @@ error stream_handler::downstream_demand(strong_actor_ptr&, size_t) { ...@@ -48,7 +48,7 @@ error stream_handler::downstream_demand(strong_actor_ptr&, size_t) {
return sec::invalid_downstream; return sec::invalid_downstream;
} }
error stream_handler::push() { error stream_handler::push(size_t*) {
CAF_LOG_ERROR("Cannot push to a stream marked as no-downstreams"); CAF_LOG_ERROR("Cannot push to a stream marked as no-downstreams");
return sec::invalid_downstream; return sec::invalid_downstream;
} }
...@@ -77,12 +77,12 @@ error stream_handler::pull(size_t) { ...@@ -77,12 +77,12 @@ error stream_handler::pull(size_t) {
return sec::invalid_upstream; return sec::invalid_upstream;
} }
bool stream_handler::is_sink() const { optional<abstract_downstream&> stream_handler::get_downstream() {
return false; return none;
} }
bool stream_handler::is_source() const { optional<abstract_upstream&> stream_handler::get_upstream() {
return false; return none;
} }
message stream_handler::make_output_token(const stream_id&) const { message stream_handler::make_output_token(const stream_id&) const {
......
...@@ -51,7 +51,7 @@ error stream_source::downstream_demand(strong_actor_ptr& hdl, size_t value) { ...@@ -51,7 +51,7 @@ error stream_source::downstream_demand(strong_actor_ptr& hdl, size_t value) {
auto size_hint = out_ptr_->policy().desired_buffer_size(*out_ptr_); auto size_hint = out_ptr_->policy().desired_buffer_size(*out_ptr_);
if (current_size < size_hint) if (current_size < size_hint)
generate(size_hint - current_size); generate(size_hint - current_size);
return push(); return push(&size_hint);
} }
// transmit cached elements before closing paths // transmit cached elements before closing paths
if (buf_size() > 0) if (buf_size() > 0)
......
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