C++多线程读取非重叠文件段的性能优化问题

问题现象 #

我开发了一个多线程处理超大文件（数百GB）的C++程序，逻辑上为每个线程分配非重叠的文件区间，线程顺序读取对应段并处理，理论上完全可并行。但实际运行中线程越多性能越差：处理18GB文件时，单线程CPU占用约71%；4线程时总CPU占用仅112%（单线程平均28%），多线程收益极低。

已知瓶颈在IO，但需要明确具体原因并找到解决方法。

当前实现细节 #

• 每个线程拥有独立文件描述符
• 调用posix_fadvise告知内核将顺序访问对应段
• 每次读取新块时，调用posix_fadvise(POSIX_FADV_DONTNEED)移除页缓存中的旧块，避免文件远大于页缓存时出现页缓存颠簸
• 原本期望每个文件描述符的预读（readahead）机制能提升性能，但未达预期

复现代码 #

#include "file_iterator.h"
#include <thread>
#include <mutex>
#include <sys/stat.h>
#include <vector>


int main(int argc, char** argv){

    std::string file = std::string(argv[1]); 
    size_t freq[256]={0};
    std::mutex mutex;

    //匿名函数传递给线程
    // [start, end) 是分配给线程的文件区间
    auto worker =[&](size_t start, size_t end){

        //本示例中线程的工作为统计区间内的符号频率
        size_t t_freq[256]={0};
        file_iterator<uint8_t> it(file, start, end);
        while(!it.consumed()){
            t_freq[*it]++;
            ++it;
        }

        //线程完成工作后，将计算结果存入全局变量
        {
            const std::lock_guard<std::mutex> lock(mutex);
            for(size_t i=0;i<256;i++){
                freq[i]+=t_freq[i];
            }

        }
    };

    //argv[2]为线程数量
    size_t n_threads=size_t(atoi(argv[2]));

    //获取输入文件大小
    struct stat st{};
        stat(file.c_str(), &st);
        off_t fsz = st.st_size;

    std::cout<<"文件包含 "<<fsz<<" 个符号 "<<std::endl;
    std::cout<<"线程数量: "<<n_threads<<std::endl;

    off_t syms_per_thread = 1 + ((fsz - 1) / n_threads);  //向上取整

    std::cout<<"每个线程处理符号数 "<<syms_per_thread<<std::endl;

    //将工作负载平均分配给线程
    std::vector<std::pair<size_t, size_t>> ranges;
    for(size_t i=0;i<n_threads;i++){
        size_t start = syms_per_thread*i;
        size_t end = std::min<off_t>(fsz, syms_per_thread*(i+1));
        std::cout<<"线程 "<<i<<" : "<<start<<" "<<end<<std::endl; //end为开区间
        ranges.emplace_back(start, end);
    }

    //创建并等待线程完成
    std::vector<std::thread> threads;
    for(size_t i=0;i<n_threads;i++){
        threads.emplace_back(worker, std::ref(ranges[i].first), std::ref(ranges[i].second));
    }

    for(size_t i=0;i<n_threads;i++){
        threads[i].join();
    }

    //输出结果
    std::cout<<"符号频率 : "<<std::endl;
    for(size_t i=0;i<256;i++){
        if(freq[i]!=0){
            std::cout<<(uint8_t)i<<" : "<<freq[i]<<std::endl;
        }
    }
}

file_iterator.h实现 #

#include <unistd.h>
#include <sys/stat.h>
#include <fcntl.h>

#include <cassert>
#include <string>
#include <iostream>

template<typename sym_t,
     off_t buff_size=8*1024*1024>
class file_iterator{

    int fd=-1;
    static constexpr uint8_t w_bytes = sizeof(sym_t);
    sym_t *buffer=nullptr;
    std::string file;
    size_t b_pos=0;
    size_t curr_pos;
    size_t last_pos;
    size_t f_offset;

public:
    explicit file_iterator(std::string &_file, off_t start=0, off_t end=0) : file(_file),
                                                                         curr_pos(start){
        struct stat st{};
        stat(file.c_str(), &st);
        off_t fsz = st.st_size;
        off_t tot_syms = fsz / w_bytes;
        assert(tot_syms>0);

        if(start<tot_syms){

            if(end==0) end = tot_syms;
            assert(start<end && end<=tot_syms);

            fd = open(file.c_str(), O_RDONLY);
            assert(fd>0);

#ifdef __linux__
            ssize_t tot_bytes = (end-start)*w_bytes;
            posix_fadvise(fd, start*w_bytes, tot_bytes, POSIX_FADV_SEQUENTIAL);
#endif
            buffer = (sym_t *)malloc(buff_size);

            f_offset = start*w_bytes;
            lseek(fd, f_offset, SEEK_SET);
            size_t read_elm = read(fd, buffer, buff_size);
            assert(read_elm>0 && read_elm<=buff_size);
            last_pos = end-1;
        }else{
            curr_pos = tot_syms;
            last_pos = tot_syms;
        }
    }

    inline sym_t operator *() const {
        return buffer[b_pos];
    }

    inline ~file_iterator(){
        free(buffer);
    }

    inline void operator++(){
        b_pos++;
        curr_pos++;
        if(b_pos==buff_size && curr_pos<last_pos){
            ssize_t read_elm = read(fd, buffer, buff_size);
            assert(read_elm>0 && read_elm<=buff_size);
#ifdef __linux__
            posix_fadvise(fd, f_offset, buff_size, POSIX_FADV_DONTNEED);
#endif
            f_offset +=read_elm; 
            b_pos=0;
        }
    }

    [[nodiscard]] inline bool consumed() const {
        return curr_pos>last_pos;
    }
};

性能优化建议 #

1. 修复预读机制失效问题 #

• 所有线程共享同一个文件描述符，改用pread指定偏移量读取。独立打开文件会让内核无法感知整体顺序访问模式，导致磁头频繁切换、预读失效，共享FD能让内核统一优化预读策略。
• 验证posix_fadvise参数准确性：确认start*w_bytes字节偏移、(end-start)*w_bytes区间长度完全覆盖线程负责的文件段。

2. 调整页缓存回收策略 #

• 延迟POSIX_FADV_DONTNEED调用时机：不要读取新块后立即释放旧块，可在当前块处理过半时标记旧块，或改用POSIX_FADV_NOREUSE（告知内核数据不会复用，可优先回收但不强制立即释放），避免打断内核预读流程。
• 若系统内存充足，可暂时关闭DONTNEED调用，观察性能变化，确认是否是缓存回收导致的额外IO开销。

3. 优化线程负载与IO对齐 #

• 线程区间划分对齐磁盘物理块大小（通常4KB或更大），避免跨块读取导致的额外寻道。当前按符号数均分可能导致区间起始/结束位置落在块中间，增加磁盘IO次数。
• 控制线程数量匹配磁盘并行能力：机械硬盘通常仅支持1-2个并发顺序流，SSD可支持更多但也有限，超过上限只会增加调度开销和磁头切换成本。

4. 调整读取缓冲区与IO方式 #

• 测试不同缓冲区大小：当前8MB缓冲区可尝试调整为4MB、16MB，且确保大小是磁盘块的整数倍，平衡系统调用次数与IO带宽占用。
• 尝试用mmap替代read：将文件区间映射到内存，减少用户态与内核态拷贝开销，让内核更高效管理预读。

5. 极小化同步开销 #

• 当前锁竞争极小，但可进一步用原子变量数组替代mutex做结果合并，彻底消除同步开销（对整体性能影响有限，但能避免极端情况的锁等待）。

内容的提问来源于stack exchange，提问作者AAA