[COMMIT seastar master] io_queue: Introduce "flow monitor"

[Commit Bot]

From: Pavel Emelyanov <xe...@scylladb.com>
Committer: Pavel Emelyanov <xe...@scylladb.com>
Branch: master

io_queue: Introduce "flow monitor"

The monitor evaluates EMA value for the dispatch speed to complete speed
ratio. The ratio is expected to be 1.0, but according to the the patch
series description, sometimes it can grow larger.

The ratio of dispatch and completion rates is fed into EMA filter every
100ms with the smoothing factor of 0.95, but both values are placed in
io_queue::config and can later be added to io_properties of options.

The calculated value is exported as a io_queue_flow_control metrics
labeled with mountpoint, shard and IO-group.

Signed-off-by: Pavel Emelyanov <xe...@scylladb.com>

---
diff --git a/include/seastar/core/io_queue.hh b/include/seastar/core/io_queue.hh
--- a/include/seastar/core/io_queue.hh
+++ b/include/seastar/core/io_queue.hh
@@ -33,6 +33,7 @@
#include <seastar/core/metrics_registration.hh>
#include <seastar/core/future.hh>
#include <seastar/core/internal/io_request.hh>
+#include <seastar/core/lowres_clock.hh>
#include <seastar/util/spinlock.hh>
#include <seastar/util/modules.hh>

@@ -116,6 +117,15 @@ private:
size_t _requests_executing = 0;
uint64_t _requests_dispatched = 0;
uint64_t _requests_completed = 0;
+
+ // Flow monitor
+ uint64_t _prev_dispatched = 0;
+ uint64_t _prev_completed = 0;
+ double _flow_ratio = 1.0;
+ timer<lowres_clock> _flow_ratio_update;
+
+ void update_flow_ratio() noexcept;
+
public:

using clock_type = std::chrono::steady_clock;
@@ -144,6 +154,8 @@ public:
float rate_factor = 1.0;
std::chrono::duration<double> rate_limit_duration = std::chrono::milliseconds(1);
size_t block_count_limit_min = 1;
+ unsigned flow_ratio_ticks = 100;
+ double flow_ratio_ema_factor = 0.95;
};

io_queue(io_group_ptr group, internal::io_sink& sink);
diff --git a/src/core/io_queue.cc b/src/core/io_queue.cc
--- a/src/core/io_queue.cc
+++ b/src/core/io_queue.cc
@@ -536,6 +536,20 @@ const fair_group& get_fair_group(const io_queue& ioq, unsigned stream) {

} // internal namespace

+template <typename T>
+void update_moving_average(T& result, T value, double factor) noexcept {
+ result = result * factor + value * (1.0 - factor);
+}
+
+void io_queue::update_flow_ratio() noexcept {
+ if (_requests_completed > _prev_completed) {
+ auto instant = double(_requests_dispatched - _prev_dispatched) / double(_requests_completed - _prev_completed);
+ update_moving_average(_flow_ratio, instant, get_config().flow_ratio_ema_factor);
+ _prev_dispatched = _requests_dispatched;
+ _prev_completed = _requests_completed;
+ }
+}
+
void
io_queue::complete_request(io_desc_read_write& desc) noexcept {
_requests_executing--;
@@ -553,6 +567,7 @@ io_queue::io_queue(io_group_ptr group, internal::io_sink& sink)
: _priority_classes()
, _group(std::move(group))
, _sink(sink)
+ , _flow_ratio_update([this] { update_flow_ratio(); })
{
auto& cfg = get_config();
if (cfg.duplex) {
@@ -563,6 +578,7 @@ io_queue::io_queue(io_group_ptr group, internal::io_sink& sink)
} else {
_streams.emplace_back(*_group->_fgs[0], make_fair_queue_config(cfg, "rw"));
}
+ _flow_ratio_update.arm_periodic(std::chrono::duration_cast<std::chrono::milliseconds>(group->io_latency_goal() * cfg.flow_ratio_ticks));
}

fair_group::config io_group::make_fair_group_config(const io_queue::config& qcfg) noexcept {