优化处理大型TSV文件的多线程C程序性能方案咨询

核心问题分析 #

你的程序性能瓶颈主要来自三个方面：频繁的内存分配与拷贝、不合理的线程任务划分、低效的磁盘I/O策略，线程的D状态（磁盘睡眠）大概率是内存不足导致的swap交换，或者后期批量写文件时的磁盘阻塞。

具体优化措施 #

1. 减少内存分配与拷贝，优化字符串处理 #

代码中大量不必要的strdup、realloc和内存拷贝是性能损耗的重灾区：

• 替换atoi为直接字符转整数：文件2的每个值是单个数字，无需调用atoi，直接用tokenValue = tsvTokens[columnIndex][0] - '0'（确保每个token为单个字符），速度提升数倍。
• 预分配内存，避免循环内realloc：处理TSV行时，先按每行制表符数量预估列数，一次性分配足够的tsvTokens内存，避免反复扩容。
• 取消冗余的字符串拷贝：
  • 无需strdup(tsvLines[lineIndex])，直接用原字符串进行strtok_r（若原字符串为只读，可提前将TSV文件读入可修改的内存块）。
  • 生成输出行时，直接计算准确长度并一次性分配内存，避免先calloc再strdup的二次拷贝。
• 提前计算字符串长度：比如txtLineContent的长度在循环外计算一次，避免每次循环调用strlen。

修改示例（核心片段）：

// 替换atoi为直接转整数
int tokenValue = tsvTokens[columnIndex][0] - '0';

// 预计算输出行长度
size_t txtLineLen = strlen(txtLineContent);
const char *firstTab = strchr(txtLineContent, '\t');
size_t idLength = firstTab ? (firstTab - txtLineContent) : 0;
size_t outputLineLen = idLength + 1 + tsvTokenCount + (tsvTokenCount - 1); // ID+制表符+列字符+列分隔符
char *outputLine = malloc(outputLineLen + 1); // +1 用于终止符
if (!outputLine) { /* 错误处理 */ }

2. 重构线程任务划分：按国家（Origin）而非行划分 #

当前每个线程处理部分行但要生成所有国家的输出，导致内存占用极高。改为按国家划分线程，每个线程负责一个或多个国家的输出：

• 每个线程仅处理对应国家的逻辑，内存占用大幅降低。
• 线程可直接将结果写入对应输出文件，无需主线程聚合数据，减少内存拷贝和占用。

修改思路：

• 主线程将文件2转换为整数二维数组（而非字符串），文件3读入内存。
• 为每个国家创建线程，参数包含国家键值、文件2的整数数组、文件3的行数据、输出路径。
• 线程遍历所有行，直接生成对应国家的输出并写入文件。

3. 优化磁盘I/O，避免内存溢出 #

线程频繁进入D状态，多因内存不足触发swap，或批量写文件阻塞：

• 边处理边写文件：每个线程负责自己的输出文件，处理一行写一行（或积累若干行批量写入），避免将所有输出数据存于内存。
• 设置大文件缓冲区：用setvbuf为输出文件设置大缓冲区（如128KB），减少系统调用次数：

  FILE *fp = fopen(countryFileName, "w");
  if (fp) {
      char buf[128 * 1024];
      setvbuf(fp, buf, _IOFBF, sizeof(buf));
  }
  

• 分块读取处理：若内存不足，不要一次性读入所有文件，改为每次读1000行，处理完再读下一批，降低常驻内存大小。

4. 优化内存布局，提升缓存命中率 #

当前三维数组localOriginLines内存不连续，CPU缓存命中率低：

• 文件2提前转换为整数二维数组，内存更紧凑，访问时无需反复转整数，缓存命中率更高。
• 按国家连续存储输出数据，或直接写入文件，避免分散的内存存储。

5. 控制线程数，避免上下文切换 #

24线程并非最优，通常线程数等于CPU核心数（或核心数+1）即可，过多线程会导致频繁上下文切换。可通过sysconf(_SC_NPROCESSORS_ONLN)获取核心数，动态设置线程数。

代码重构示例（核心框架） #

以下是按国家划分线程的核心实现：

typedef struct {
    int countryCode;
    const char *countryName;
    const int **tsvData; // 文件2预转换的整数数组
    const char **txtLines; // 文件3的行数据
    int lineCount;
    int columnCount;
} CountryThreadArg;

void *processCountry(void *arg) {
    CountryThreadArg *threadArg = (CountryThreadArg*)arg;
    char fileName[256];
    snprintf(fileName, sizeof(fileName), "%s.txt", threadArg->countryName);
    FILE *fp = fopen(fileName, "w");
    if (!fp) {
        perror("Failed to open output file");
        return NULL;
    }
    // 设置大缓冲区
    char buf[128 * 1024];
    setvbuf(fp, buf, _IOFBF, sizeof(buf));

    for (int lineIndex = 0; lineIndex < threadArg->lineCount; lineIndex++) {
        const char *txtLine = threadArg->txtLines[lineIndex];
        const int *tsvLine = threadArg->tsvData[lineIndex];
        // 写入ID部分
        const char *firstTab = strchr(txtLine, '\t');
        if (firstTab) {
            fwrite(txtLine, 1, firstTab - txtLine, fp);
            fputc('\t', fp);
            txtLine = firstTab + 1;
        }
        // 处理每一列
        for (int col = 0; col < threadArg->columnCount; col++) {
            if (tsvLine[col] == threadArg->countryCode) {
                fputc(txtLine[col * 2], fp); // 适配TSV列间隔格式，需根据实际调整
            } else {
                fputc('9', fp);
            }
            if (col != threadArg->columnCount - 1) {
                fputc('\t', fp);
            }
        }
        fputc('\n', fp);
    }

    fclose(fp);
    free(threadArg);
    return NULL;
}

// 主线程创建线程逻辑
int main() {
    // ... 读取文件1、文件2（转换为整数数组）、文件3 ...
    int cpuCount = sysconf(_SC_NPROCESSORS_ONLN);
    pthread_t *threads = malloc(countryCount * sizeof(pthread_t));
    
    // 控制并发数，避免超过CPU核心数
    for (int i = 0; i < countryCount; i++) {
        CountryThreadArg *arg = malloc(sizeof(CountryThreadArg));
        arg->countryCode = origins[i].code;
        arg->countryName = origins[i].name;
        arg->tsvData = tsvData;
        arg->txtLines = txtLines;
        arg->lineCount = lineCount;
        arg->columnCount = columnCount;
        pthread_create(&threads[i], NULL, processCountry, arg);

        if ((i + 1) % cpuCount == 0) {
            for (int j = i - cpuCount + 1; j <= i; j++) {
                pthread_join(threads[j], NULL);
            }
        }
    }
    // 等待剩余线程完成
    for (int i = countryCount - (countryCount % cpuCount); i < countryCount; i++) {
        pthread_join(threads[i], NULL);
    }
    // ... 清理资源 ...
}

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