Refine storage read thread count. (#6885) (#8577)

close #8580
pingcap · Dec 25, 2023 · 13da6ff · 13da6ff
1 parent d3aafae
commit 13da6ff
Show file tree

Hide file tree

Showing 6 changed files with 16 additions and 10 deletions.
diff --git a/dbms/src/Interpreters/Settings.h b/dbms/src/Interpreters/Settings.h
@@ -319,7 +319,8 @@ struct Settings
     M(SettingUInt64, dt_checksum_frame_size, DBMS_DEFAULT_BUFFER_SIZE, "Frame size for delta tree stable storage")                                                                                                                      \
                                                                                                                                                                                                                                         \
     M(SettingDouble, dt_page_gc_threshold, 0.5, "Max valid rate of deciding to do a GC in PageStorage")                                                                                                                                 \
-    M(SettingBool, dt_enable_read_thread, true, "Enable storage read thread or not")                                                                                                                                                   \
+    M(SettingBool, dt_enable_read_thread, true, "Enable storage read thread or not")                                                                                                                                                    \
+    M(SettingDouble, dt_read_thread_count_scale, 1.0, "Number of read thread = number of logical cpu cores * dt_read_thread_count_scale.  Only has meaning at server startup.")                                                         \
     M(SettingUInt64, dt_max_sharing_column_bytes_for_all, 2048 * Constant::MB, "Memory limitation for data sharing of all requests. 0 means disable data sharing")                                                                      \
     M(SettingUInt64, dt_max_sharing_column_count, 5, "ColumnPtr object limitation for data sharing of each DMFileReader::Stream. 0 means disable data sharing")                                                                         \
                                                                                                                                                                                                                                         \

diff --git a/dbms/src/Server/Server.cpp b/dbms/src/Server/Server.cpp
@@ -1169,7 +1169,7 @@ int Server::main(const std::vector<std::string> & /*args*/)
     LOG_INFO(log, "dt_enable_read_thread {}", global_context->getSettingsRef().dt_enable_read_thread);
     // `DMFileReaderPool` should be constructed before and destructed after `SegmentReaderPoolManager`.
     DM::DMFileReaderPool::instance();
-    DM::SegmentReaderPoolManager::instance().init(server_info);
+    DM::SegmentReaderPoolManager::instance().init(server_info.cpu_info.logical_cores, settings.dt_read_thread_count_scale);
     DM::SegmentReadTaskScheduler::instance();
 
     {

diff --git a/dbms/src/Storages/DeltaMerge/ReadThread/SegmentReader.cpp b/dbms/src/Storages/DeltaMerge/ReadThread/SegmentReader.cpp
@@ -208,18 +208,19 @@ SegmentReaderPoolManager::SegmentReaderPoolManager()
 
 SegmentReaderPoolManager::~SegmentReaderPoolManager() = default;
 
-void SegmentReaderPoolManager::init(const ServerInfo & server_info)
+void SegmentReaderPoolManager::init(UInt32 logical_cpu_cores, double read_thread_count_scale)
 {
+    double total_thread_count = logical_cpu_cores * read_thread_count_scale;
     auto numa_nodes = getNumaNodes(log);
-    LOG_INFO(log, "numa_nodes {} => {}", numa_nodes.size(), numa_nodes);
+    RUNTIME_CHECK(!numa_nodes.empty());
+    UInt32 thread_count_per_node = std::ceil(total_thread_count / numa_nodes.size());
     for (const auto & node : numa_nodes)
     {
-        int thread_count = node.empty() ? server_info.cpu_info.logical_cores : node.size();
-        reader_pools.push_back(std::make_unique<SegmentReaderPool>(thread_count, node));
+        reader_pools.push_back(std::make_unique<SegmentReaderPool>(thread_count_per_node, node));
         auto ids = reader_pools.back()->getReaderIds();
         reader_ids.insert(ids.begin(), ids.end());
     }
-    LOG_INFO(log, "num_readers={}", reader_ids.size());
+    LOG_INFO(log, "numa_nodes={} number_of_readers={}", numa_nodes, reader_ids.size());
 }
 
 void SegmentReaderPoolManager::addTask(MergedTaskPtr && task)

diff --git a/dbms/src/Storages/DeltaMerge/ReadThread/SegmentReader.h b/dbms/src/Storages/DeltaMerge/ReadThread/SegmentReader.h
@@ -59,7 +59,7 @@ class SegmentReaderPoolManager
         static SegmentReaderPoolManager pool_manager;
         return pool_manager;
     }
-    void init(const ServerInfo & server_info);
+    void init(UInt32 logical_cpu_cores, double read_thread_count_scale);
     ~SegmentReaderPoolManager();
     DISALLOW_COPY_AND_MOVE(SegmentReaderPoolManager);
 

diff --git a/dbms/src/Storages/DeltaMerge/workload/MainEntry.cpp b/dbms/src/Storages/DeltaMerge/workload/MainEntry.cpp
@@ -256,7 +256,9 @@ void dailyRandomTest(WorkloadOptions & opts)
 void initReadThread()
 {
     DB::ServerInfo server_info;
-    DB::DM::SegmentReaderPoolManager::instance().init(server_info);
+    DB::DM::SegmentReaderPoolManager::instance().init(
+        server_info.cpu_info.logical_cores,
+        TiFlashTestEnv::getGlobalContext().getSettingsRef().dt_read_thread_count_scale);
     DB::DM::SegmentReadTaskScheduler::instance();
     DB::DM::DMFileReaderPool::instance();
 }

diff --git a/dbms/src/TestUtils/gtests_dbms_main.cpp b/dbms/src/TestUtils/gtests_dbms_main.cpp
@@ -63,7 +63,9 @@ int main(int argc, char ** argv)
     DB::ServerInfo server_info;
     // `DMFileReaderPool` should be constructed before and destructed after `SegmentReaderPoolManager`.
     DB::DM::DMFileReaderPool::instance();
-    DB::DM::SegmentReaderPoolManager::instance().init(server_info);
+    DB::DM::SegmentReaderPoolManager::instance().init(
+        server_info.cpu_info.logical_cores,
+        DB::tests::TiFlashTestEnv::getGlobalContext().getSettingsRef().dt_read_thread_count_scale);
     DB::DM::SegmentReadTaskScheduler::instance();
 
 #ifdef FIU_ENABLE