controllers: allow memtable I/O controller to have shares statically set

This is so it looks more like the CPU controller. The end goal is to integrate them.

Signed-off-by: Glauber Costa <[email protected]>

backlog_controller.hh

@@ -112,6 +112,12 @@ class backlog_io_controller : public backlog_controller {
     future<> _inflight_update;
 
 public:
+    backlog_io_controller(const ::io_priority_class& iop, float static_shares)
+        : _io_priority(iop)
+        , _inflight_update(make_ready_future<>())
+    {
+        update_controller(static_shares);
+    }
     backlog_io_controller(const ::io_priority_class& iop, std::chrono::milliseconds interval, std::vector<backlog_controller::control_point> control_points, std::function<float()> backlog)
         : backlog_controller(interval, std::move(control_points), backlog)
         , _io_priority(iop)
@@ -156,6 +162,7 @@ public:
 class flush_io_controller : public backlog_io_controller {
     static constexpr float hard_dirty_limit = 1.0f;
 public:
+    flush_io_controller(const ::io_priority_class& iop, float static_shares) : backlog_io_controller(iop, static_shares) {}
     flush_io_controller(const ::io_priority_class& iop, std::chrono::milliseconds interval, float soft_limit, std::function<float()> current_backlog)
         : backlog_io_controller(iop, std::move(interval),
           std::vector<backlog_controller::control_point>({{soft_limit, 10}, {soft_limit + (hard_dirty_limit - soft_limit) / 2, 100}, {hard_dirty_limit, 1000}}),

database.cc

@@ -2056,13 +2056,22 @@ future<> distributed_loader::populate_column_family(distributed<database>& db, s
 
 inline
 flush_cpu_controller
-make_flush_cpu_controller(db::config& cfg, seastar::scheduling_group sg, std::function<double()> fn) {
+make_flush_controller(db::config& cfg, seastar::scheduling_group sg, std::function<double()> fn) {
     if (cfg.memtable_flush_static_shares() > 0) {
         return flush_cpu_controller(sg, cfg.memtable_flush_static_shares());
     }
     return flush_cpu_controller(sg, 250ms, cfg.virtual_dirty_soft_limit(), std::move(fn));
 }
 
+inline
+flush_io_controller
+make_flush_controller(db::config& cfg, const ::io_priority_class& iop, std::function<double()> fn) {
+    if (cfg.memtable_flush_static_shares() > 0) {
+        return flush_io_controller(iop, cfg.memtable_flush_static_shares());
+    }
+    return flush_io_controller(iop, 250ms, cfg.virtual_dirty_soft_limit(), std::move(fn));
+}
+
 utils::UUID database::empty_version = utils::UUID_gen::get_name_UUID(bytes{});
 
 database::database() : database(db::config(), database_config())
@@ -2077,16 +2086,16 @@ database::database(const db::config& cfg, database_config dbcfg)
     , _dirty_memory_manager(*this, memory::stats().total_memory() * 0.45, cfg.virtual_dirty_soft_limit())
     , _streaming_dirty_memory_manager(*this, memory::stats().total_memory() * 0.10, cfg.virtual_dirty_soft_limit())
     , _dbcfg(dbcfg)
-    , _memtable_cpu_controller(make_flush_cpu_controller(*_cfg, dbcfg.memtable_scheduling_group, [this, limit = float(_dirty_memory_manager.throttle_threshold())] {
+    , _memtable_cpu_controller(make_flush_controller(*_cfg, dbcfg.memtable_scheduling_group, [this, limit = float(_dirty_memory_manager.throttle_threshold())] {
         return (_dirty_memory_manager.virtual_dirty_memory()) / limit;
     }))
     , _data_query_stage("data_query", _dbcfg.query_scheduling_group, &column_family::query)
     , _version(empty_version)
     , _compaction_manager(std::make_unique<compaction_manager>(dbcfg.compaction_scheduling_group))
     , _enable_incremental_backups(cfg.incremental_backups())
-    , _flush_io_controller(service::get_local_memtable_flush_priority(), 250ms, cfg.virtual_dirty_soft_limit(), [this, limit = float(_dirty_memory_manager.throttle_threshold())] {
+    , _flush_io_controller(make_flush_controller(*_cfg, service::get_local_memtable_flush_priority(), [this, limit = float(_dirty_memory_manager.throttle_threshold())] {
         return _dirty_memory_manager.virtual_dirty_memory() / limit;
-    })
+    }))
     , _compaction_io_controller(service::get_local_compaction_priority(), 250ms, [this] () -> float {
         auto backlog = _compaction_manager->backlog();
         // This means we are using an unimplemented strategy