MPI Parallel Bucket Sort: Efficient Sorting with Multiple Processes

#include <stdio.h>\n#include <stdlib.h>\n#include <mpi.h>\n#include <sys/time.h>\n#include <time.h>\n\n/* #define PRINT_RES */\n\n/* return the difference between end and start in microseconds */\nint timeDiff(struct timeval start, struct timeval end) {\n return (end.tv_sec-start.tv_sec)*1000000 + (end.tv_usec-start.tv_usec);\n}\n\n/* The compar for qsort /\nint cmpfunc(const void a, const void b){\n return ((int)a - (int)b);\n}\n\nvoid print_arr(int arr, int len){\n int i;\n for(i = 0; i < len; ++i){\n printf("%d ", arr[i]);\n }\n printf("\n");\n}\n\nint main(int argc, char *argv[]){\n int my_rank, comm_sz;\n int n, i, local_n;\n struct timeval start_time, end_time;\n long local_runtime, total_runtime;\n int *total_array, *local_array, num_buckets;\n int *local_small_buckets, *local_large_bucket;\n int *len_local_small_buckets, len_local_large_bucket;\n int *sdispls, *recvcounts, rdispls;\n\n MPI_Init(&argc, &argv);\n MPI_Comm_rank(MPI_COMM_WORLD, &my_rank);\n MPI_Comm_size(MPI_COMM_WORLD, &comm_sz);\n\n sdispls = malloc(sizeof(int) * comm_sz);\n recvcounts = malloc(sizeof(int) * comm_sz);\n rdispls = malloc(sizeof(int) * comm_sz);\n\n num_buckets = comm_sz; /* The number of buckets is equal to the number of processes /\n /* get the size of array from the command /\n if (argc < 2) {\n n = 10000;\n } else {\n n = atoi(argv[1]);\n }\n local_n = n / comm_sz;\n if(local_n <= 0){\n local_n = 1;\n }\n n = local_n * comm_sz;\n if(my_rank == 0){\n srand(0);\n total_array = (int) malloc(sizeof(int) * n);\n local_array = (int) malloc(sizeof(int) * local_n);\n for (i = 0; i < n; ++i){\n total_array[i] = rand();\n }\n }else{\n local_array = (int) malloc(sizeof(int) * local_n);\n }\n\n gettimeofday(&start_time, NULL);\n\n /* The master needs to partition total_array and send each partition to the correct slave process */\n MPI_Scatter(total_array, local_n, MPI_INT, local_array, local_n, MPI_INT, 0, MPI_COMM_WORLD);\n\n /* Small bucket distribution /\n local_small_buckets = malloc(sizeof(int) * local_n * num_buckets);\n len_local_small_buckets = calloc(num_buckets, sizeof(int));\n \n int tmp = RAND_MAX / num_buckets + 1;\n for (i = 0; i < local_n; ++i){\n int idx = local_array[i] / tmp;\n local_small_buckets[local_n * idx + len_local_small_buckets[idx]++] = local_array[i];\n }\n\n /* Small buckets then emptied into p final buckets for sorting */\n local_large_bucket = malloc(sizeof(int) * n);\n len_local_large_bucket = 0;\n for (i = 0; i < comm_sz; ++i){\n sdispls[i] = local_n * i;\n }\n MPI_Alltoall(len_local_small_buckets, 1, MPI_INT, recvcounts, 1, MPI_INT, MPI_COMM_WORLD);\n rdispls[0] = 0;\n len_local_large_bucket = recvcounts[0];\n for (i = 1; i < comm_sz; ++i){\n rdispls[i] = rdispls[i-1] + recvcounts[i-1];\n len_local_large_bucket += recvcounts[i];\n }\n MPI_Alltoallv(local_small_buckets, len_local_small_buckets, sdispls, MPI_INT, local_large_bucket, recvcounts, rdispls, MPI_INT, MPI_COMM_WORLD);\n\n /* Sort the local_large_bucket */\n qsort(local_large_bucket, len_local_large_bucket, sizeof(int), cmpfunc);\n\n gettimeofday(&end_time, NULL);\n local_runtime = timeDiff(start_time, end_time);\n MPI_Reduce(&local_runtime, &total_runtime, 1, MPI_LONG, MPI_MAX, 0, MPI_COMM_WORLD);\n\n /* Gather all the data to the main process */\n MPI_Gather(&len_local_large_bucket, 1, MPI_INT, recvcounts, 1, MPI_INT, 0, MPI_COMM_WORLD);\n if(my_rank == 0){\n rdispls[0] = 0;\n for (i = 1; i < comm_sz; ++i){\n rdispls[i] = rdispls[i-1] + recvcounts[i-1];\n }\n }\n MPI_Gatherv(local_large_bucket, len_local_large_bucket, MPI_INT, total_array, recvcounts, rdispls, MPI_INT, 0, MPI_COMM_WORLD);\n\n if (my_rank == 0){\n int is_sorted = 1;\n for(i = 1; i < n; ++i){\n if (total_array[i] < total_array[i-1]){\n is_sorted = 0;\n break;\n }\n }\n if(is_sorted){\n printf("The Data has been sorted.\n");\n }else{\n printf("Data Not Sorted in index %d with %d > %d\n", i, total_array[i-1], total_array[i]);\n }\n#ifdef PRINT_RES\n print_arr(total_array, n);\n#endif\n printf("Total runtime (in microseconds): %ld\n", total_runtime);\n free(total_array);\n }\n free(local_array);\n free(len_local_small_buckets);\n free(local_small_buckets);\n free(local_large_bucket);\n free(rdispls);\n free(recvcounts);\n free(sdispls);\n MPI_Finalize();\n return 0;\n