Commit 49333dc6 authored by Dominik Charousset's avatar Dominik Charousset

Allow users to trigger new credit

parent ea760a5f
...@@ -67,6 +67,10 @@ public: ...@@ -67,6 +67,10 @@ public:
/// client. /// client.
virtual void stop(); virtual void stop();
/// Tries to advance the stream by generating more credit or by sending
/// batches.
void advance();
/// Aborts a stream after any stream message handler returned a non-default /// Aborts a stream after any stream message handler returned a non-default
/// constructed error `reason` or the parent actor terminates with a /// constructed error `reason` or the parent actor terminates with a
/// non-default error. /// non-default error.
......
...@@ -21,6 +21,8 @@ ...@@ -21,6 +21,8 @@
#include "caf/actor_addr.hpp" #include "caf/actor_addr.hpp"
#include "caf/actor_cast.hpp" #include "caf/actor_cast.hpp"
#include "caf/actor_control_block.hpp" #include "caf/actor_control_block.hpp"
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/error.hpp" #include "caf/error.hpp"
#include "caf/expected.hpp" #include "caf/expected.hpp"
#include "caf/inbound_path.hpp" #include "caf/inbound_path.hpp"
...@@ -112,6 +114,30 @@ void stream_manager::stop() { ...@@ -112,6 +114,30 @@ void stream_manager::stop() {
self_->erase_inbound_paths_later(this); self_->erase_inbound_paths_later(this);
} }
void stream_manager::advance() {
CAF_LOG_TRACE("");
// Try to emit more credit.
if (!inbound_paths_.empty()) {
using std::chrono::microseconds;
auto& cfg = self_->system().config();
microseconds bc{cfg.streaming_desired_batch_complexity_us};
microseconds interval{cfg.streaming_tick_duration_us()};
auto& mbox = self_->mailbox();
auto& qs = get<2>(mbox.queue().queues()).queues();
// Iterate all queues for inbound traffic.
for (auto& kvp : qs) {
auto inptr = kvp.second.policy().handler.get();
// Ignore inbound paths of other managers.
if (inptr->mgr.get() == this) {
auto bs = static_cast<int32_t>(kvp.second.total_task_size());
inptr->emit_ack_batch(self_, bs, interval, bc);
}
}
}
// Try to generate more batches.
push();
}
void stream_manager::abort(error reason) { void stream_manager::abort(error reason) {
CAF_LOG_TRACE(CAF_ARG(reason)); CAF_LOG_TRACE(CAF_ARG(reason));
out().abort(reason); out().abort(reason);
......
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