大文件的并行处理问题

问题

论坛上碰见一个问题：

1. 有 N 个 1-2G 大的文件，其中第六列是点分表示的 IPV4 地址，需要把它转为整数表示。

   #!/bin/awk -f
   {
   　　len = split($6, a, ".")
   　　if (len == 4) {
   　　　　ip = lshift(a[1], 24) + lshift(a[2], 16) + lshift(a[3], 8) + a[4]
   　　　　printf $1"\t"$2"\t"$3"\t"$4"\t"$5"\t"ip"\n" > o
   　　}
   }
   

   CPU 是 6 核的，就起了 6 个进程去转，每个 awk 进程基本都是 CPU 100 了
   但转换一个文件还是需要 15 分钟左右，还是太慢
   求教有什么方法可以让 AWK 跑的更快些吗？

思考

1. 6 个 awk 进程应该是同时转 6 个文件，这个可以用 parallel 命令（一般发行版都带）实现

   parallel awk -f ip.awk ::: {1..6}.log
   

   但这样无法加速单个文件的处理。

2. awk 进程 CPU 100，说明 IO 不是瓶颈了
   标准 IO（Buffered I/O）的数据流程： IO <—> 内核页缓存 <——> 用户应用缓冲 <—> 数据处理

3. 如果是这种情况：把文件 A、B、C 合并成文件 D，则并行处理 A、B、C 和单个处理 D 相比，哪个更快？
   在 4 核环境下，用一个 700m 文件试了下，并行处理省掉约一半的时间

   time awk -f ip.awk D.log
   time parallel awk -f ip.awk ::: {A..C}.log
   

尝试

已知把大文件拆成小文件，做并行处理，可以加速，那么：

1. 把大文件均匀分割，保存为若干小文件，然后再并行处理。但这样会增加不少额外 IO，或许可以用 Direct I/O 减少分割与合并开销。 —- 待验证
2. 把大文件均匀分割，通过管道传给 awk 并行处理
   man 了下 parallel，发现有个 —pipepart 选项，估计就是用来干这事的，不过标注的还是 Debug 特性。

   751m /opt/sa.txt

   time parallel --pipepart awk \‘{print \$6}\‘ :::: /opt/sa.txt > par.txt
   5.34s user 1.88s system 23% cpu 30.240 total
   

   time awk ‘{print $6}‘ /opt/sa.txt > awk.txt
   4.21s user 0.35s system 35% cpu 12.901 total
   

   看上去磁盘真不给力！

硬盘 IO 是个问题，不过 pipepart 的 CPU 这么低，不像是并行的节奏啊！再验证下：

1. 划分文件成 M 个小块（一般和核数一致），但确保以换行作为分割点
2. 使用 mmap 映射文件（吞吐量比 read 好？）
3. 开 M 个线程，每个线程建立一个 FIFO，然后写入文件块
4. M 个 awk 并行读取这些 FIFO

#include <unistd.h>
#include <stdio.h>
#include <stdlib.h>
#include <assert.h>
#include <string.h>
#include <fcntl.h>
#include <sys/mman.h>
#include <sys/stat.h>
#include <signal.h>
#include <pthread.h>

#define  EOL      ‘\n‘
#define  PATHLEN  128

// [addr-range <-- addr] 搜索第一个换行，返回偏移
int eol_offset(const char* addr, int range)
{
    assert(addr);
    assert(range > 0);
    int i = 0;    
    while (*(addr - i) != EOL)
    {
        i++;
        if (i >= range)
            return 0;
    }
    assert(0 <= i);
    return (i - 1);
}

typedef struct {
    char* addr;
    long len;
    char path[PATHLEN];
} TaskData, *LTaskData;

// 线程任务：写文件块到 fifo
void* output(void* args)
{
    assert(args);
    LTaskData td = (LTaskData)args;
    printf("%ld: %ldB --> %s\n", pthread_self(), td->len, td->path);

    int fd = open(td->path, O_WRONLY, 0);
    int wrum = write(fd, td->addr, td->len);
    printf("%s: %d bytes done!\n", td->path, wrum);

    close(fd); fd = -1;
    assert(unlink(td->path) == 0);
    free(td); td = NULL; args = NULL;

    return NULL;
}

//~ 信号处理
void setup_signal()
{
    sigset_t bset;
    sigemptyset(&bset);
    sigaddset(&bset, SIGPIPE); //建立一个信号集，将SIGPIPE添加进去
    assert(pthread_sigmask(SIG_BLOCK, &bset, NULL) == 0);
}

int main(int argc, char* argv[])
{
    assert(argc == 4);
    setup_signal();

    // 命令行参数
    int part_num = atoi(argv[1]);
    assert(0 < part_num);
    const char* pipename = argv[2];
    assert(strlen(pipename) < PATHLEN);
    const char* filename = argv[3];

    // mmap 映射文件
    int fdin = open(filename, O_RDONLY);
    assert(fdin >= 0);
    struct stat statbuf = {0};
    assert(fstat(fdin, &statbuf) == 0);
    size_t srclen = statbuf.st_size;
    void* src = mmap(0, srclen, PROT_READ, MAP_PRIVATE, fdin, 0);;
    assert(src != MAP_FAILED);
    close(fdin); fdin = -1;

    // 划分文件
    int part_len = srclen / part_num + 1;   
    char** joint_addr = (char**)malloc((part_num + 1) * sizeof(char*));
    assert(joint_addr);
    joint_addr[0] = src;
    for (int i = 1; i < part_num; i++)
    {
        char* addr = (char*)src + part_len * i;
        joint_addr[i] = addr - eol_offset(addr, part_len);
    }
    joint_addr[part_num] = (char*)src + srclen;

    // 建立fifo，多线程处理
    pthread_t* tid = (pthread_t*)malloc(part_num * sizeof(pthread_t));
    assert(tid);
    for (int i = 0; i < part_num; i++)
    {
        LTaskData td = (LTaskData)malloc(sizeof(TaskData));
        assert(td);
        td->addr = joint_addr[i];
        td->len = joint_addr[i + 1] - joint_addr[i];
        snprintf(td->path, PATHLEN, "%d%s", i+1, pipename);
        // 数据预读，可能失败
        madvise(td->addr, td->len, MADV_WILLNEED | MADV_SEQUENTIAL);
        if (access(td->path, F_OK) == 0)
            assert(unlink(td->path) == 0);
        assert(mkfifo(td->path, (S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH)) == 0);

        assert(pthread_create(&tid[i], NULL, output, td) == 0);
    }

    // 等待处理结束
    for (int i = 0; i < part_num; i++)
        pthread_join(tid[i], NULL);

    free(joint_addr); joint_addr = NULL;
    free(tid); tid = NULL;
    munmap(src, srclen); src = NULL;

    return 0;
}

编译参数： gcc -Wall -pedantic -std=gnu99 map.c -o map -lpthread
运行参数：

./map 4 pip /opt/sa.txt ==> time parallel awk \‘{print \$6}\‘ ::: {1..4}pip > pip.txt
5.20s user 0.70s system 24% cpu 24.495 total

总结

效果不理想，瓶颈在哪，IO 读写？FIFO？不得而知
该如何测试程序的并行性能（确定性能瓶颈）？ 涉及 IPC

以上的尝试，均是数据预处理 + 并行 awk 处理，所以本质仍是 IPC 数据交换，不管是以文件还是管道形式。那么，通信的开销与并行加速的好处相比，哪家强呢？ How to Measure it？

参考

1. Linux 中的零拷贝技术 (1), (2)
2. 深入剖析命名管道 FIFO 对程序性能的影响
3. Speeding up file I/O: mmap() vs. read()
4. awk 源码读取输入部分分析