disagg: Limit the compute node cache download IOPS and bandwidth (#10558)

close #10557

* Prior to this PR, both the number of threads handling foreground reads and the number of background threads downloading files for FileCache were controlled by the same parameter: profiles.default.io_thread_count_scale, making it difficult to fine-tune these settings independently. This PR introduces two separate parameters to control the concurrency and queue length of FileCache downloads performed by Compute Node background threads, allowing adjustments to background thread settings without affecting the foreground read thread pool configuration:
  - Added profiles.default.dt_filecache_downloading_count_scale, with a default value of 2.0. This means the concurrency for FileCache background downloads is set to logical_cores profiles.default.dt_filecache_downloading_count_scale.
  - Changed the default value of profiles.default.dt_filecache_max_downloading_count_scale from 1.0 to 10.0. This parameter now defines the maximum queue length for FileCache background downloads as logical_cores profiles.default.dt_filecache_max_downloading_count_scale. Before this PR, the queue length is logical_cores * profiles.default.io_thread_count_scale * 2, so changing the default value is just making it not related to io_thread_count_scale any more.
* FileCache downloads files from S3 to local storage, and this process is rate-limited by the rate_limiter parameter.
* After persisting data received from Write Nodes into the LocalPageCache, Compute Nodes no longer invoke fsync, reducing their IOPS requirements.
* Optimized the lock contention scope in `FileCache::get`.

Signed-off-by: JaySon-Huang <tshent@qq.com>

Author: JaySon-Huang

dbms/src/Common/TiFlashMetrics.h

@@ -845,6 +845,8 @@ static_assert(RAFT_REGION_BIG_WRITE_THRES * 4 < RAFT_REGION_BIG_WRITE_MAX, "Inva
       Counter,                                                                                                                      \
       F(type_dtfile_hit, {"type", "dtfile_hit"}),                                                                                   \
       F(type_dtfile_miss, {"type", "dtfile_miss"}),                                                                                 \
+      F(type_dtfile_not_cache_type, {"type", "dtfile_not_cache_type"}),                                                             \
+      F(type_dtfile_too_many_download, {"type", "dtfile_too_many_download"}),                                                       \
       F(type_dtfile_evict, {"type", "dtfile_evict"}),                                                                               \
       F(type_dtfile_full, {"type", "dtfile_full"}),                                                                                 \
       F(type_dtfile_download, {"type", "dtfile_download"}),                                                                         \

dbms/src/IO/BaseFile/RateLimiter.cpp

@@ -470,6 +470,12 @@ WriteLimiterPtr IORateLimiter::getWriteLimiter()
     return is_background_thread ? bg_write_limiter : fg_write_limiter;
 }
 
+WriteLimiterPtr IORateLimiter::getBgWriteLimiter()
+{
+    std::lock_guard lock(limiter_mtx);
+    return bg_write_limiter;
+}
+
 ReadLimiterPtr IORateLimiter::getReadLimiter()
 {
     std::lock_guard lock(limiter_mtx);

dbms/src/IO/BaseFile/RateLimiter.h

@@ -213,6 +213,8 @@ class IORateLimiter
     ~IORateLimiter();
 
     WriteLimiterPtr getWriteLimiter();
+    WriteLimiterPtr getBgWriteLimiter();
+
     ReadLimiterPtr getReadLimiter();
     void init(Poco::Util::AbstractConfiguration & config_);
     void updateConfig(Poco::Util::AbstractConfiguration & config_);

dbms/src/Interpreters/Settings.h

@@ -244,7 +244,8 @@ struct Settings
     M(SettingInt64, remote_gc_small_size, 128 * 1024, "The files with total size less than this threshold will be compacted")                                                                                                           \
     /* Disagg arch reading settings */ \
     M(SettingUInt64, dt_write_page_cache_limit_size, 2 * 1024 * 1024, "Limit size per write batch when compute node writing to PageStorage cache")                                                                                      \
-    M(SettingDouble, dt_filecache_max_downloading_count_scale, 1.0, "Max downloading task count of FileCache = io thread count * dt_filecache_max_downloading_count_scale.")                                                            \
+    M(SettingDouble, dt_filecache_downloading_count_scale, 2.0, "Max concurrency of download task count of FileCache = number of logical cpu cores * dt_filecache_downloading_count_scale.")                                                              \
+    M(SettingDouble, dt_filecache_max_downloading_count_scale, 10.0, "Max queue size of download task count of FileCache = number of logical cpu cores * dt_filecache_max_downloading_count_scale.")                                    \
     M(SettingUInt64, dt_filecache_min_age_seconds, 1800, "Files of the same priority can only be evicted from files that were not accessed within `dt_filecache_min_age_seconds` seconds.")                                             \
     M(SettingBool, dt_enable_fetch_memtableset, true, "Whether fetching delta cache in FetchDisaggPages")                                                                                                                               \
     M(SettingUInt64, dt_fetch_pages_packet_limit_size, 512 * 1024, "Response packet bytes limit of FetchDisaggPages, 0 means one page per packet")                                                                                      \

dbms/src/Interpreters/SharedContexts/Disagg.cpp

@@ -37,6 +37,27 @@ void SharedContextDisagg::initReadNodePageCache(
         PSDiskDelegatorPtr delegator;
         if (!cache_dir.empty())
         {
+            {
+                // Compute node disable fsync to improve performance, so we need to
+                // clear the cache dir before use to avoid corrupted data.
+                auto dir = Poco::File(cache_dir);
+                if (dir.exists())
+                {
+                    LOG_INFO(
+                        Logger::get(),
+                        "Local cache directory will be cleared before setting up LocalPageCache",
+                        cache_dir);
+                    dir.remove(true);
+                }
+                else
+                {
+                    LOG_INFO(
+                        Logger::get(),
+                        "Local cache directory not exist when setting up LocalPageCache",
+                        cache_dir);
+                }
+            }
+
             delegator = path_pool.getPSDiskDelegatorFixedDirectory(cache_dir);
             LOG_INFO(
                 Logger::get(),

dbms/src/Server/Server.cpp

@@ -328,9 +328,11 @@ void adjustThreadPoolSize(const Settings & settings, size_t logical_cores)
     }
     if (S3FileCachePool::instance)
     {
-        S3FileCachePool::instance->setMaxThreads(max_io_thread_count);
-        S3FileCachePool::instance->setMaxFreeThreads(max_io_thread_count / 2);
-        S3FileCachePool::instance->setQueueSize(max_io_thread_count * 2);
+        auto concurrency = logical_cores * settings.dt_filecache_downloading_count_scale;
+        auto queue_size = logical_cores * settings.dt_filecache_max_downloading_count_scale;
+        S3FileCachePool::instance->setMaxThreads(concurrency);
+        S3FileCachePool::instance->setMaxFreeThreads(concurrency / 2);
+        S3FileCachePool::instance->setQueueSize(queue_size);
     }
     if (RNWritePageCachePool::instance)
     {
@@ -691,7 +693,11 @@ try
     if (const auto & config = storage_config.remote_cache_config; config.isCacheEnabled() && is_disagg_compute_mode)
     {
         config.initCacheDir();
-        FileCache::initialize(global_context->getPathCapacity(), config);
+        FileCache::initialize(
+            global_context->getPathCapacity(),
+            config,
+            server_info.cpu_info.logical_cores,
+            global_context->getIORateLimiter());
     }
 
     /// Determining PageStorage run mode based on current files on disk and storage config.

dbms/src/Storages/DeltaMerge/File/tests/gtest_dm_meta_version.cpp

@@ -473,7 +473,12 @@ try
         .dir = fmt::format("{}/fs_cache", getTemporaryPath()),
         .capacity = 1 * 1000 * 1000 * 1000,
     };
-    FileCache::initialize(db_context->getGlobalContext().getPathCapacity(), file_cache_config);
+    UInt16 vcores = 8;
+    FileCache::initialize(
+        db_context->getGlobalContext().getPathCapacity(),
+        file_cache_config,
+        vcores,
+        db_context->getGlobalContext().getIORateLimiter());
 
     auto dm_file = prepareDMFileRemote(/* file_id= */ 1);
     ASSERT_TRUE(dm_file->path().starts_with("s3://"));

dbms/src/Storages/DeltaMerge/Index/VectorIndex/tests/gtest_dm_vector_index.cpp

@@ -1926,7 +1926,13 @@ class VectorIndexSegmentOnS3Test
             .dir = fmt::format("{}/fs_cache", getTemporaryPath()),
             .capacity = 1 * 1000 * 1000 * 1000,
         };
-        FileCache::initialize(global_context.getPathCapacity(), file_cache_config);
+
+        UInt16 vcores = 8;
+        FileCache::initialize(
+            global_context.getPathCapacity(),
+            file_cache_config,
+            vcores,
+            global_context.getIORateLimiter());
 
         auto cols = DMTestEnv::getDefaultColumns();
         cols->emplace_back(cdVec());

dbms/src/Storages/DeltaMerge/Remote/RNLocalPageCache.cpp

@@ -83,6 +83,7 @@ void RNLocalPageCache::write(
     }
 
     UniversalWriteBatch cache_wb;
+    cache_wb.disableFSync(); // RNLocalPageCache doesn't need fsync. Disable it to reduce IOPS requirements.
     cache_wb.putPage(key, 0, read_buffer, size, field_sizes);
     GET_METRIC(tiflash_storage_remote_cache, type_page_download).Increment();
     GET_METRIC(tiflash_storage_remote_cache_bytes, type_page_download_bytes).Increment(size);
@@ -113,6 +114,7 @@ void RNLocalPageCache::write(UniversalWriteBatch && wb)
                 itr->second.size);
         }
     }
+    wb.disableFSync(); // RNLocalPageCache doesn't need fsync. Disable it to reduce IOPS requirements.
     GET_METRIC(tiflash_storage_remote_cache, type_page_download).Increment(wb.getWrites().size());
     GET_METRIC(tiflash_storage_remote_cache_bytes, type_page_download_bytes).Increment(wb.getTotalDataSize());
     storage->write(std::move(wb));

dbms/src/Storages/DeltaMerge/tests/gtest_segment_replace_stable_data.cpp

@@ -426,7 +426,12 @@ class SegmentReplaceStableDataDisaggregated
                 .dir = fmt::format("{}/fs_cache", getTemporaryPath()),
                 .capacity = 1 * 1000 * 1000 * 1000,
             };
-            FileCache::initialize(global_context.getPathCapacity(), file_cache_config);
+            UInt16 vcores = 8;
+            FileCache::initialize(
+                global_context.getPathCapacity(),
+                file_cache_config,
+                vcores,
+                global_context.getIORateLimiter());
         }
 
         table_columns = DMTestEnv::getDefaultColumns();