MPI 并行桶排序算法实现

#include <stdio.h> #include <stdlib.h> #include <mpi.h> #include <sys/time.h> #include <time.h>

// #define PRINT_RES

// return the difference between end and start in microseconds int timeDiff(struct timeval start, struct timeval end) { return (end.tv_sec-start.tv_sec)*1000000 + (end.tv_usec-start.tv_usec); }

// The compar for qsort int cmpfunc(const void a, const void b){ return ((int)a - (int)b); }

void print_arr(int* arr, int len){ int i; for(i = 0; i < len; ++i){ printf("%d ", arr[i]); } printf("\n"); }

int main(int argc, char *argv[]){ int my_rank, comm_sz; int n, i, local_n; struct timeval start_time, end_time; long local_runtime, total_runtime; int *total_array, *local_array, num_buckets; int *local_small_buckets, *local_large_bucket; int *len_local_small_buckets, len_local_large_bucket; int *sdispls, *recvcounts, *rdispls;

MPI_Init(&argc, &argv);
MPI_Comm_rank(MPI_COMM_WORLD, &my_rank);
MPI_Comm_size(MPI_COMM_WORLD, &comm_sz);

sdispls = malloc(sizeof(int) * comm_sz);
recvcounts = malloc(sizeof(int) * comm_sz);
rdispls = malloc(sizeof(int) * comm_sz);

num_buckets = comm_sz; // The number of buckets is equal to the number of processes
// get the size of array from the command
if (argc < 2) {
    n = 10000;
} else {
    n = atoi(argv[1]);
}
local_n = n / comm_sz;
if(local_n <= 0){
    local_n = 1;
}
n = local_n * comm_sz;
if(my_rank == 0){
    srand(0);
    total_array = (int*) malloc(sizeof(int) * n);
    local_array = (int*) malloc(sizeof(int) * local_n);
    for (i = 0; i < n; ++i){
        total_array[i] = rand();
    }
}else{
    local_array = (int*) malloc(sizeof(int) * local_n);
}

gettimeofday(&start_time, NULL);

// The master needs to partition total_array and send each partition to the correct slave process
MPI_Scatter(total_array, local_n, MPI_INT, local_array, local_n, MPI_INT, 0, MPI_COMM_WORLD);

// Small bucket distribution
local_small_buckets = malloc(sizeof(int*) * local_n * num_buckets);
len_local_small_buckets = calloc(num_buckets, sizeof(int));

int tmp = RAND_MAX / num_buckets + 1;
for (i = 0; i < local_n; ++i){
    int idx = local_array[i] / tmp;
    local_small_buckets[local_n * idx + len_local_small_buckets[idx]++] = local_array[i];
}

// Small buckets then emptied into p final buckets for sorting
local_large_bucket = malloc(sizeof(int) * n);
len_local_large_bucket = 0;
for (i = 0; i < comm_sz; ++i){
    sdispls[i] = local_n * i;
}
MPI_Alltoall(len_local_small_buckets, 1, MPI_INT, recvcounts, 1, MPI_INT, MPI_COMM_WORLD);
rdispls[0] = 0;
len_local_large_bucket = recvcounts[0];
for (i = 1; i < comm_sz; ++i){
    rdispls[i] = rdispls[i-1] + recvcounts[i-1];
    len_local_large_bucket += recvcounts[i];
}
MPI_Alltoallv(local_small_buckets, len_local_small_buckets, sdispls, MPI_INT, local_large_bucket, recvcounts, rdispls, MPI_INT, MPI_COMM_WORLD);

// Sort the local_large_bucket
qsort(local_large_bucket, len_local_large_bucket, sizeof(int), cmpfunc);

gettimeofday(&end_time, NULL);
local_runtime = timeDiff(start_time, end_time);
MPI_Reduce(&local_runtime, &total_runtime, 1, MPI_LONG, MPI_MAX, 0, MPI_COMM_WORLD);

// Gather all the data to the main process
MPI_Gather(&len_local_large_bucket, 1, MPI_INT, recvcounts, 1, MPI_INT, 0, MPI_COMM_WORLD);
if(my_rank == 0){
    rdispls[0] = 0;
    for (i = 1; i < comm_sz; ++i){
        rdispls[i] = rdispls[i-1] + recvcounts[i-1];
    }
}
MPI_Gatherv(local_large_bucket, len_local_large_bucket, MPI_INT, total_array, recvcounts, rdispls, MPI_INT, 0, MPI_COMM_WORLD);

if (my_rank == 0){
    int is_sorted = 1;
    for(i = 1; i < n; ++i){
        if (total_array[i] < total_array[i-1]){
            is_sorted = 0;
            break;
        }
    }
    if(is_sorted){
        printf("The Data has been sorted.\n");
    }else{
        printf("Data Not Sorted in index %d with %d > %d\n", i, total_array[i-1], total_array[i]);
    }

#ifdef PRINT_RES print_arr(total_array, n); #endif printf("Total runtime (in microseconds): %ld\n", total_runtime); free(total_array); } free(local_array); free(len_local_small_buckets); free(local_small_buckets); free(local_large_bucket); free(rdispls); free(recvcounts); free(sdispls); MPI_Finalize(); return 0; } 分析此代码内容：这段代码实现了一个并行的桶排序算法。算法的主要思路是将待排序的数据分成多个小桶，然后将小桶中的数据分发给不同的进程进行排序，最后再将排序后的数据合并到一起。

代码中使用了MPI库来实现进程间的通信。首先，主进程将待排序的数据分发给各个进程，然后每个进程根据数据的值将数据放入不同的小桶中。接着，进程间通过MPI_Alltoallv函数进行通信，将各个进程的小桶中的数据合并到一个大桶中。然后，每个进程对自己负责的部分数据进行排序，最后通过MPI_Gatherv函数将排序后的数据收集到主进程中。

代码中还使用了qsort函数来对每个进程负责的部分数据进行排序，最后在主进程中检查排序结果是否正确，并打印出总运行时间。

总体来说，这段代码实现了一个基于桶排序的并行排序算法，并使用MPI库实现了进程间的通信。