Commit 396c79f3 authored by Dominik Charousset's avatar Dominik Charousset

Make sure advance_streams is called at least once

parent 85067563
...@@ -357,8 +357,11 @@ scheduled_actor::resume(execution_unit* ctx, size_t max_throughput) { ...@@ -357,8 +357,11 @@ scheduled_actor::resume(execution_unit* ctx, size_t max_throughput) {
// Set a new receive timeout if we called our behavior at least once. // Set a new receive timeout if we called our behavior at least once.
if (handled_msgs > 0) if (handled_msgs > 0)
set_receive_timeout(); set_receive_timeout();
// Set a new stream timeout if we called `advance_streams` at least once. // Set a new stream timeout.
if (!stream_managers_.empty() && tout.time_since_epoch().count() != 0) { if (!stream_managers_.empty()) {
// Make sure we call `advance_streams` at least once.
if (tout.time_since_epoch().count() != 0)
tout = advance_streams(clock().now());
set_stream_timeout(tout); set_stream_timeout(tout);
} }
}; };
...@@ -445,8 +448,10 @@ bool scheduled_actor::is_active_receive_timeout(uint64_t tid) const { ...@@ -445,8 +448,10 @@ bool scheduled_actor::is_active_receive_timeout(uint64_t tid) const {
uint64_t scheduled_actor::set_stream_timeout(actor_clock::time_point x) { uint64_t scheduled_actor::set_stream_timeout(actor_clock::time_point x) {
CAF_LOG_TRACE(x); CAF_LOG_TRACE(x);
// Do not request 'infinite' timeouts. // Do not request 'infinite' timeouts.
if (x == actor_clock::time_point::max()) if (x == actor_clock::time_point::max()) {
CAF_LOG_DEBUG("drop infinite timeout");
return 0; return 0;
}
// Do not request a timeout if all streams are idle. // Do not request a timeout if all streams are idle.
std::vector<stream_manager_ptr> mgrs; std::vector<stream_manager_ptr> mgrs;
for (auto& kvp : stream_managers_) for (auto& kvp : stream_managers_)
...@@ -1062,8 +1067,10 @@ scheduled_actor::handle_open_stream_msg(mailbox_element& x) { ...@@ -1062,8 +1067,10 @@ scheduled_actor::handle_open_stream_msg(mailbox_element& x) {
actor_clock::time_point actor_clock::time_point
scheduled_actor::advance_streams(actor_clock::time_point now) { scheduled_actor::advance_streams(actor_clock::time_point now) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
if (!stream_ticks_.started()) if (!stream_ticks_.started()) {
CAF_LOG_DEBUG("tick emitter not started yet");
return actor_clock::time_point::max(); return actor_clock::time_point::max();
}
/// Advance time for driving forced batches and credit. /// Advance time for driving forced batches and credit.
auto bitmask = stream_ticks_.timeouts(now, {max_batch_delay_ticks_, auto bitmask = stream_ticks_.timeouts(now, {max_batch_delay_ticks_,
credit_round_ticks_}); credit_round_ticks_});
......
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