c - 在 C 中合并排序时间

Question

这是我的虚拟机:
CPU:4核
内存:4096 MB
操作系统:Ubuntu 18.04(64位)
问题1:为什么会有阈值4193790？
我用 C 写了一个归并排序:

#include<stdio.h>
#include<stdlib.h>
#include<omp.h>
#include<sys/time.h>
#include"helper.c"
#define N 4193789

void merge(double* li,int left1,int right1,int left2,int right2,int size){
    double *li_tmp;
    li_tmp = (double *)malloc(sizeof(double)*size);
    int i = left1;
    int j = left2;
    int k = left1;

    while(i<=right1 && j<=right2){ 
        if(li[i] < li[j]){
            li_tmp[k] = li[i++]; 
        }     
        else{
            li_tmp[k] = li[j++];
        } 
        k++; 
    }
    if(i>right1){
        while(j<=right2){
            li_tmp[k++] = li[j++];
        }
    }
    else if(j>right2){
        while(i<=right1){
            li_tmp[k++] = li[i++];
        }
    }

    for(i=left1; i<right2+1; i++){
        li[i] = li_tmp[i];
    }
    free(li_tmp); 
}

void merge_sort(double* li,int left,int right,int size){
    if (left<right){        
        int mid = (left + right)/2;
        merge_sort(li,left,mid,size);
        merge_sort(li,mid+1,right,size);
        merge(li,left,mid,mid+1,right,size);
    }
}

int main(int argc, char *argv[])
{
    // Just call merge_sort(), check the correctness and print the time consumption.
    double *data;
    data = (double *)malloc(sizeof(double)*N);
    srand(1234567);
    gen_rand(data,N);

    struct timeval start, end;
    gettimeofday(&start, NULL);

    merge_sort(data, 0, N-1, N);

    gettimeofday(&end, NULL);
    double delta = ((end.tv_sec  - start.tv_sec) * 1000000u + end.tv_usec - start.tv_usec) / 1.e6;

    bool correct = true;
    for(int i=0;i<N-1;i++){
        if (data[i]>data[i+1]){
            correct = false;
        }
    }
    if (correct){
        printf("Correct!\n");
    }else{
        printf("Not correct!\n");
    }

    printf("time spent=%12.10f\n",delta);
}

这是我的helper.c，只是在double *a中生成一个随机的double数组。

#include<stdio.h>
#include<stdlib.h>
#include<stdbool.h>

void gen_rand(double *a, int num){
    for (int i=0;i<num;i++){
        a[i] = 1.0 * rand() / RAND_MAX * num;
    }
}

我发现了一个非常奇怪的场景:
我知道归并排序是O(nlogn)，所以当我尝试不同的数组长度时，我发现一个区域的时间消耗变化很大，并不适合O(nlogn)。经过多次尝试，我发现一个阈值。
当我把N定义为4193789时，时间是1s，但是当我把N改成4193790时，时间会增加到34s!
我想知道为什么会有这样的阈值。

vagrant@hang2:~/data/$ gcc -fopenmp merge_sort_main.c 
vagrant@hang2:~/data/$ ./a.out 
Correct!
time spent=1.1103340000
vagrant@hang2:~/data/$ gcc -fopenmp merge_sort_main.c 
vagrant@hang2:~/data/$ ./a.out 
Correct!
time spent=34.5053590000

问题2:为什么在大数组(超过4193790)时omp方法会变慢？
omp的另一个问题:
这是我的主要内容:

#pragma omp parallel num_threads(4)
    {
    #pragma omp single 
        merge_sort_omp(data, 0, N-1, N);
    }

和merge_sort_omp():

void merge_sort_omp(double* li,int left,int right,int size){
    if (left<right){
        if (right-left>10000){        
            int mid = (left + right)/2;
            #pragma omp task firstprivate (li, left, mid)
            merge_sort_omp(li,left,mid,size);
            #pragma omp task firstprivate (li, mid, right)
            merge_sort_omp(li,mid+1,right,size);
            #pragma omp taskwait
            merge(li,left,mid,mid+1,right,size);
        }else{
            int mid = (left + right)/2;
            merge_sort_omp(li,left,mid,size);
            merge_sort_omp(li,mid+1,right,size);
            merge(li,left,mid,mid+1,right,size);
        }
    }
}

我尝试了 N=4000000 和 N=4193790 如下:

vagrant@hang2:~/data$ gcc -fopenmp merge_sort_main.c 
vagrant@hang2:~/data$ ./a.out 
Correct!
time spent=1.1358180000
vagrant@hang2:~/data$ gcc -fopenmp merge_sort_omp_main.c 
vagrant@hang2:~/data$ ./a.out 
Correct!
time spent=0.4998150000
vagrant@hang2:~/data$ gcc -fopenmp merge_sort_main.c 
vagrant@hang2:~/data$ ./a.out 
Correct!
time spent=34.3504340000
vagrant@hang2:~/data$ gcc -fopenmp merge_sort_omp_main.c 
vagrant@hang2:~/data$ ./a.out 
Correct!
time spent=111.9368700000

我想知道为什么并行代码在 N=4000000 处是串行代码的两倍，但串行代码在 N=4193790 处较慢。几乎慢了三倍。我想知道为什么 omp 变慢了？

Answer 1

除了调用 -O2 构建或 -O3在评论中，您的代码所做的最昂贵的事情是调用 malloc和 free在每次调用 merge_sort 时构建一个临时数组.在高性能循环的每次迭代中进行内存分配会大大减慢速度。
简单的解决方法是只对该临时缓冲区进行一次单一分配 - 并使其足够大以适应所有场景。
而不是这个:

void merge(double* li,int left1,int right1,int left2,int right2,int size){
    double *li_tmp;
    li_tmp = (double *)malloc(sizeof(double)*size);

这个:

double* li_tmp = NULL;  // li_tmp is now global

void merge(double* li,int left1,int right1,int left2,int right2,int size){

    if (li_tmp == NULL) {
        li_tmp = (double *)malloc(sizeof(double)*N); // allocate for size N just once
    }

然后删除 free合并函数底部的语句。
而不是这个:

    for(i=left1; i<right2+1; i++){
        li[i] = li_tmp[i];
    }
    free(li_tmp); 
}

只是这个:

    for(i=left1; i<right2+1; i++){
        li[i] = li_tmp[i];
    }
    
}

然后在合并排序运行完成后在其他地方释放 li_tmp。
至于为什么 N 的大小不同导致不同的性能变化，我认为如果没有应用这些优化和编译器开关，这是不值得的。最可能的假设是，由不同大小的 N 引起的数组大小，会在高速缓存和主 RAM 之间触发更多的“分页”。或者这些大块分配以不同的方式给内存管理器带来压力。