Commit ea760a5f authored by Dominik Charousset's avatar Dominik Charousset

Allow stage drivers to signal congestions

parent 72e58355
...@@ -49,8 +49,10 @@ public: ...@@ -49,8 +49,10 @@ public:
using state_type = typename trait::state; using state_type = typename trait::state;
template <class Init> template <class Init>
stream_stage_driver_impl(Init init, Process f, Finalize fin) stream_stage_driver_impl(DownstreamManager& out, Init init, Process f,
: process_(std::move(f)), Finalize fin)
: super(out),
process_(std::move(f)),
fin_(std::move(fin)) { fin_(std::move(fin)) {
init(state_); init(state_);
} }
...@@ -68,7 +70,6 @@ private: ...@@ -68,7 +70,6 @@ private:
state_type state_; state_type state_;
Process process_; Process process_;
Finalize fin_; Finalize fin_;
message result_;
}; };
} // namespace detail } // namespace detail
......
...@@ -51,7 +51,7 @@ public: ...@@ -51,7 +51,7 @@ public:
stream_stage_impl(scheduled_actor* self, Ts&&... xs) stream_stage_impl(scheduled_actor* self, Ts&&... xs)
: stream_manager(self), : stream_manager(self),
super(self), super(self),
driver_(std::forward<Ts>(xs)...) { driver_(this->out_, std::forward<Ts>(xs)...) {
// nop // nop
} }
...@@ -74,7 +74,7 @@ public: ...@@ -74,7 +74,7 @@ public:
} }
bool congested() const noexcept override { bool congested() const noexcept override {
return this->out_.capacity() == 0; return driver_.congested();
} }
int32_t acquire_credit(inbound_path* path, int32_t desired) override { int32_t acquire_credit(inbound_path* path, int32_t desired) override {
......
...@@ -53,6 +53,10 @@ public: ...@@ -53,6 +53,10 @@ public:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
stream_stage_driver(DownstreamManager& out) : out_(out) {
// nop
}
virtual ~stream_stage_driver() { virtual ~stream_stage_driver() {
// nop // nop
} }
...@@ -68,6 +72,12 @@ public: ...@@ -68,6 +72,12 @@ public:
// nop // nop
} }
/// Can mark the stage as congested. The default implementation signals a
/// congestion if the downstream manager has no capacity left in its buffer.
virtual bool congested() const noexcept {
return out_.capacity() == 0;
}
/// Acquires credit on an inbound path. The calculated credit to fill our /// Acquires credit on an inbound path. The calculated credit to fill our
/// queue fro two cycles is `desired`, but the driver is allowed to return /// queue fro two cycles is `desired`, but the driver is allowed to return
/// any non-negative value. /// any non-negative value.
...@@ -75,6 +85,9 @@ public: ...@@ -75,6 +85,9 @@ public:
CAF_IGNORE_UNUSED(path); CAF_IGNORE_UNUSED(path);
return desired; return desired;
} }
protected:
DownstreamManager& out_;
}; };
} // namespace caf } // namespace caf
......
...@@ -267,7 +267,11 @@ public: ...@@ -267,7 +267,11 @@ public:
using downstream_manager = broadcast_downstream_manager<int>; using downstream_manager = broadcast_downstream_manager<int>;
struct driver final : public stream_stage_driver<int, downstream_manager> { struct driver final : public stream_stage_driver<int, downstream_manager> {
public: public:
driver(vector<int>* log) : log_(log) { using super = stream_stage_driver<int, downstream_manager>;
driver(downstream_manager& out, vector<int>* log)
: super(out),
log_(log) {
// nop // nop
} }
......
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