feat: calculate the io throughput in background in ReadLimiter

dbms/src/Encryption/RateLimiter.cpp

@@ -289,47 +289,31 @@ void WriteLimiter::updateMaxBytesPerSec(Int64 max_bytes_per_sec)
 }
 
 ReadLimiter::ReadLimiter(
-    std::function<Int64()> getIOStatistic_,
+    std::function<Int64()> get_read_bytes_,
     Int64 rate_limit_per_sec_,
     LimiterType type_,
-    Int64 get_io_stat_period_us,
     UInt64 refill_period_ms_)
     : WriteLimiter(rate_limit_per_sec_, type_, refill_period_ms_)
-    , getIOStatistic(std::move(getIOStatistic_))
-    , last_stat_bytes(getIOStatistic())
-    , last_stat_time(now())
+    , get_read_bytes(std::move(get_read_bytes_))
+    , last_stat_bytes(get_read_bytes())
     , log(Logger::get("ReadLimiter"))
-    , get_io_statistic_period_us(get_io_stat_period_us)
 {}
 
 Int64 ReadLimiter::getAvailableBalance()
 {
-    TimePoint us = now();
-    // Not call getIOStatisctics() every time for performance.
-    // If the clock back, elapsed_us could be negative.
-    Int64 elapsed_us = std::chrono::duration_cast<std::chrono::microseconds>(us - last_stat_time).count();
-    if (get_io_statistic_period_us != 0 && elapsed_us < get_io_statistic_period_us)
-    {
-        return available_balance;
-    }
-
-    return refreshAvailableBalance();
-}
-
-Int64 ReadLimiter::refreshAvailableBalance()
-{
-    TimePoint us = now();
-    Int64 bytes = getIOStatistic();
-    if (bytes < last_stat_bytes)
+    Int64 bytes = get_read_bytes();
+    if (unlikely(bytes < last_stat_bytes))
     {
         LOG_FMT_WARNING(
             log,
-            "last_stat {}:{} current_stat {}:{}",
-            last_stat_time.time_since_epoch().count(),
+            "last_stat: {} current_stat: {}",
             last_stat_bytes,
-            us.time_since_epoch().count(),
             bytes);
     }
+    else if (likely(bytes == last_stat_bytes))
+    {
+        return available_balance;
+    }
     else
     {
         Int64 real_alloc_bytes = bytes - last_stat_bytes;
@@ -338,7 +322,6 @@ Int64 ReadLimiter::refreshAvailableBalance()
         alloc_bytes += real_alloc_bytes;
     }
     last_stat_bytes = bytes;
-    last_stat_time = us;
     return available_balance;
 }
 
@@ -381,17 +364,18 @@ void ReadLimiter::refillAndAlloc()
     }
 }
 
-IORateLimiter::IORateLimiter()
+IORateLimiter::IORateLimiter(UInt64 update_read_info_period_ms_)
     : log(Logger::get("IORateLimiter"))
     , stop(false)
+    , update_read_info_period_ms(update_read_info_period_ms_)
 {}
 
 IORateLimiter::~IORateLimiter()
 {
     stop.store(true, std::memory_order_relaxed);
-    if (auto_tune_thread.joinable())
+    if (auto_tune_and_get_read_info_thread.joinable())
     {
-        auto_tune_thread.join();
+        auto_tune_and_get_read_info_thread.join();
     }
 }
 
@@ -409,13 +393,13 @@ extern thread_local bool is_background_thread;
 
 WriteLimiterPtr IORateLimiter::getWriteLimiter()
 {
-    std::lock_guard lock(mtx_);
+    std::lock_guard lock(mtx);
     return is_background_thread ? bg_write_limiter : fg_write_limiter;
 }
 
 ReadLimiterPtr IORateLimiter::getReadLimiter()
 {
-    std::lock_guard lock(mtx_);
+    std::lock_guard lock(mtx);
     return is_background_thread ? bg_read_limiter : fg_read_limiter;
 }
 
@@ -426,7 +410,7 @@ void IORateLimiter::updateConfig(Poco::Util::AbstractConfiguration & config_)
     {
         return;
     }
-    std::lock_guard lock(mtx_);
+    std::lock_guard lock(mtx);
     updateReadLimiter(io_config.getBgReadMaxBytesPerSec(), io_config.getFgReadMaxBytesPerSec());
     updateWriteLimiter(io_config.getBgWriteMaxBytesPerSec(), io_config.getFgWriteMaxBytesPerSec());
 }
@@ -455,11 +439,10 @@ void IORateLimiter::updateReadLimiter(Int64 bg_bytes, Int64 fg_bytes)
 {
     LOG_FMT_INFO(log, "updateReadLimiter: bg_bytes {} fg_bytes {}", bg_bytes, fg_bytes);
     auto get_bg_read_io_statistic = [&]() {
-        return getCurrentIOInfo().bg_read_bytes;
+        return read_info.bg_read_bytes.load(std::memory_order_relaxed);
     };
     auto get_fg_read_io_statistic = [&]() {
-        auto io_info = getCurrentIOInfo();
-        return std::max(0, io_info.total_read_bytes - io_info.bg_read_bytes);
+        return read_info.fg_read_bytes.load(std::memory_order_relaxed);
     };
 
     if (bg_bytes == 0)
@@ -526,7 +509,7 @@ void IORateLimiter::setBackgroundThreadIds(std::vector<pid_t> thread_ids)
     LOG_FMT_INFO(log, "bg_thread_ids {} => {}", bg_thread_ids.size(), bg_thread_ids);
 }
 
-std::pair<Int64, Int64> IORateLimiter::getReadWriteBytes(const std::string & fname [[maybe_unused]])
+Int64 IORateLimiter::getReadBytes(const std::string & fname [[maybe_unused]])
 {
 #if __linux__
     std::ifstream ifs(fname);
@@ -538,7 +521,6 @@ std::pair<Int64, Int64> IORateLimiter::getReadWriteBytes(const std::string & fna
     }
     std::string s;
     Int64 read_bytes = -1;
-    Int64 write_bytes = -1;
     while (std::getline(ifs, s))
     {
         if (s.empty())
@@ -557,49 +539,43 @@ std::pair<Int64, Int64> IORateLimiter::getReadWriteBytes(const std::string & fna
             boost::algorithm::trim(values[1]);
             read_bytes = std::stoll(values[1]);
         }
-        else if (values[0] == "write_bytes")
-        {
-            boost::algorithm::trim(values[1]);
-            write_bytes = std::stoll(values[1]);
-        }
     }
-    if (read_bytes == -1 || write_bytes == -1)
+    if (read_bytes == -1)
     {
-        auto msg = fmt::format("read_bytes: {} write_bytes: {} Invalid result.", read_bytes, write_bytes);
+        auto msg = fmt::format("read_bytes: {}. Invalid result.", read_bytes);
         LOG_ERROR(log, msg);
         throw Exception(msg, ErrorCodes::UNKNOWN_EXCEPTION);
     }
-    return {read_bytes, write_bytes};
+    return read_bytes;
 #else
-    return {0, 0};
+    return 0;
 #endif
 }
 
-IORateLimiter::IOInfo IORateLimiter::getCurrentIOInfo()
+void IORateLimiter::getCurrentIOInfo()
 {
     static const pid_t pid = getpid();
-    IOInfo io_info;
 
-    // Read I/O info of each background threads.
+    // Read read info of each background threads.
+    Int64 bg_read_bytes_tmp{0};
     for (pid_t tid : bg_thread_ids)
     {
         const std::string thread_io_fname = fmt::format("/proc/{}/task/{}/io", pid, tid);
-        Int64 read_bytes, write_bytes;
-        std::tie(read_bytes, write_bytes) = getReadWriteBytes(thread_io_fname);
-        io_info.bg_read_bytes += read_bytes;
-        io_info.bg_write_bytes += write_bytes;
+        Int64 read_bytes;
+        read_bytes = getReadBytes(thread_io_fname);
+        bg_read_bytes_tmp += read_bytes;
     }
+    read_info.bg_read_bytes.store(bg_read_bytes_tmp, std::memory_order_relaxed);
 
-    // Read I/O info of this process.
+    // Read read info of this process.
     static const std::string proc_io_fname = fmt::format("/proc/{}/io", pid);
-    std::tie(io_info.total_read_bytes, io_info.total_write_bytes) = getReadWriteBytes(proc_io_fname);
-    io_info.update_time = std::chrono::system_clock::now();
-    return io_info;
+    Int64 fg_read_bytes_tmp{getReadBytes(proc_io_fname) - bg_read_bytes_tmp};
+    read_info.fg_read_bytes.store(std::max(0, fg_read_bytes_tmp), std::memory_order_relaxed);
 }
 
 void IORateLimiter::setStop()
 {
-    std::lock_guard lock(mtx_);
+    std::lock_guard lock(mtx);
     if (bg_write_limiter != nullptr)
     {
         auto sz = bg_write_limiter->setStop();
@@ -624,17 +600,28 @@ void IORateLimiter::setStop()
 
 void IORateLimiter::runAutoTune()
 {
-    auto auto_tune_worker = [&]() {
+    auto auto_tune_and_get_read_info_worker = [&]() {
+        using time_point = std::chrono::time_point<std::chrono::system_clock>;
+        using clock = std::chrono::system_clock;
+        time_point auto_tune_time = clock::now();
+        time_point update_read_info_time = auto_tune_time;
         while (!stop.load(std::memory_order_relaxed))
         {
-            ::sleep(io_config.auto_tune_sec > 0 ? io_config.auto_tune_sec : 1);
-            if (io_config.auto_tune_sec > 0)
+            std::this_thread::sleep_for(std::chrono::milliseconds(update_read_info_period_ms));
+            auto now_time_point = clock::now();
+            if ((io_config.auto_tune_sec > 0) && (now_time_point - auto_tune_time >= std::chrono::seconds(io_config.auto_tune_sec)))
             {
                 autoTune();
+                auto_tune_time = now_time_point;
+            }
+            if ((bg_read_limiter || fg_read_limiter) && likely(now_time_point - update_read_info_time >= std::chrono::milliseconds(update_read_info_period_ms)))
+            {
+                getCurrentIOInfo();
+                update_read_info_time = now_time_point;
             }
         }
     };
-    auto_tune_thread = std::thread(auto_tune_worker);
+    auto_tune_and_get_read_info_thread = std::thread(auto_tune_and_get_read_info_worker);
 }
 
 std::unique_ptr<IOLimitTuner> IORateLimiter::createIOLimitTuner()
@@ -643,7 +630,7 @@ std::unique_ptr<IOLimitTuner> IORateLimiter::createIOLimitTuner()
     ReadLimiterPtr bg_read, fg_read;
     StorageIORateLimitConfig t_io_config;
     {
-        std::lock_guard lock(mtx_);
+        std::lock_guard lock(mtx);
         bg_write = bg_write_limiter;
         fg_write = fg_write_limiter;
         bg_read = bg_read_limiter;
@@ -666,12 +653,12 @@ void IORateLimiter::autoTune()
         auto tune_result = tuner->tune();
         if (tune_result.read_tuned)
         {
-            std::lock_guard lock(mtx_);
+            std::lock_guard lock(mtx);
             updateReadLimiter(tune_result.max_bg_read_bytes_per_sec, tune_result.max_fg_read_bytes_per_sec);
         }
         if (tune_result.write_tuned)
         {
-            std::lock_guard lock(mtx_);
+            std::lock_guard lock(mtx);
             updateWriteLimiter(tune_result.max_bg_write_bytes_per_sec, tune_result.max_fg_write_bytes_per_sec);
         }
     }