MPI Parallel Merge Sort

just_read.c

#include "../utils/read_utils.h"
#include <time.h>

int main(int argc, char **argv)
{
    int res;
    int *array;
    int len;
    size_t start, end;
    if (argc < 2) {
        printf("Missing file name argument\n");
        return 1;
    }
    start = time(NULL);
    res = read_integers_from_file(argv[1], &array, &len);
    if (res != 0) {
        printf("Unable to read from file\n");
        return 1;
    }
    end = time(NULL);
    printf("Read from file elapsed time: %ld seconds\n", end - start);
    free(array);
    return 0;
}

mpi_just_read.c

#include <mpi.h>
#include <stdio.h>
#include "../utils/read_utils.h"

int main(int argc, char** argv) {
    // Initialize the MPI environment
    MPI_Init(&argc, &argv);
    // Get the number of processes
    int world_size;
    MPI_Comm_size(MPI_COMM_WORLD, &world_size);

    // Get the rank of the process
    int world_rank;
    MPI_Comm_rank(MPI_COMM_WORLD, &world_rank);

    if (argc < 2) {
        printf("Missing file path argument\n");
        MPI_Finalize();
        return 0;
    }

    if (world_rank == 0) {
        int *array, len, res;
        double start, end;
        start = MPI_Wtime();
        printf("Master: reading from %s\n", argv[1]);
        res = read_integers_from_file(argv[1], &array, &len);
        if (res != 0) {
            printf("Unable to read integers from file: %d\n", res);
            return res;
        }
        end = MPI_Wtime();
        printf("Master: finished reading from %s\n", argv[1]);
        printf("Master: elapsed time: %.2f\n", end - start);
        free(array);
    }

    // Finalize the MPI environment.
    MPI_Finalize();
}

mpi_just_read_barrier.c

#include <mpi.h>
#include <stdio.h>
#include "../utils/read_utils.h"

int main(int argc, char** argv) {
    double start, end;
    // Initialize the MPI environment
    MPI_Init(&argc, &argv);
    // Get the number of processes
    int world_size;
    MPI_Comm_size(MPI_COMM_WORLD, &world_size);

    // Get the rank of the process
    int world_rank;
    MPI_Comm_rank(MPI_COMM_WORLD, &world_rank);

    if (argc < 2) {
        printf("Missing file path argument\n");
        MPI_Finalize();
        return 0;
    }
    start = MPI_Wtime();
    if (world_rank == 0) {
        int *array, len, res;
        printf("Master: reading from %s\n", argv[1]);
        res = read_integers_from_file(argv[1], &array, &len);
        if (res != 0) {
            printf("Unable to read integers from file: %d\n", res);
            return res;
        }
        printf("Master: finished reading from %s\n", argv[1]);
        free(array);
    }

    MPI_Barrier(MPI_COMM_WORLD);
    end = MPI_Wtime();
    if (world_rank == 0) {
        printf("Master: elapsed time: %.2f\n", end - start);
    }
    // Finalize the MPI environment.
    MPI_Finalize();
}

parallel_ms.c

#include <mpi.h>
#include <stdio.h>
#include "../utils/read_utils.h"
#include "../utils/sort_utils.h"
#include "../utils/redtree_utils.h"
#include "../utils/print_utils.h"
#include "../utils/mpi_utils.h"

int main(int argc, char** argv) {
    int res;
    int i;
    int *array;
    int len;
    int pre_len;
    int levels;
    int neighbor;
    int processors;
    int rank;
    double start, end;
    double total = 0;
    
    MPI_Init(&argc, &argv);
    MPI_Comm_size(MPI_COMM_WORLD, &processors);
    MPI_Comm_rank(MPI_COMM_WORLD, &rank);

    if (argc < 2) {
        printf("Missing file name argument.");
        return 1;
    }
    
    res = redtree_levels(processors, &levels);
    if (res != 0) {
        printf("unable to get levels: %d\n", res);
        return res;
    }
    
    if (rank == 0) {
        printf("Master: levels: %d\n", levels);
    }

    if (rank == 0) {
        start = MPI_Wtime();
        res = read_integers_from_file(argv[1], &array, &len);
        if (res != 0) {
            printf("Unable to read from file: %d\n", res);
            return res;
        }
        end = MPI_Wtime();
        
        printf("Master: intial len: %d\n", len);
        printf("Master: read from file elapsed time: %.8f\n", end - start);
        total += end - start;

        start = MPI_Wtime();
        res = mpi_utils_master_distribute_array(processors, array, len);
        if (res != 0) {
            printf("Unable to distribute array: %d\n", res);
            free(array);
            return res;
        }
        array = realloc(array, sizeof(int) * len / processors);
        len = len / processors;
        end = MPI_Wtime();
        printf("Master: distributing array elapsed time: %.8f\n", end - start);
        total += end - start;
    } else {
        res = mpi_utils_receive_array(0, &array, &len);
        if (res != 0) {
            printf("unable to receive slice: %d\n", res);
            return res;
        }
    }

    start = MPI_Wtime();
    // Everybody sort your arrays
    res = merge_sort(array, len);
    if (res != 0) {
        printf("unable to sort my array: %d", res);
        free(array);
        return res;
    }
    
    //printf("Rank %d: my len %d\n", rank, len);

    for (i = 0; i < levels; i++) {
        neighbor = redtree_get_neighbor(i, rank);
        if (!redtree_is_receiver(i, rank)) {
            //printf("Rank %d: sending len %d\n", rank, len);
            res = mpi_utils_send_array(neighbor, array, len);
            if (res != 0) {
                printf("Unable to send array: %d\n", res);
                free(array);
                return res;
            }
            break;
        }
        pre_len = len;
        res = mpi_utils_combine_arrays(neighbor, &array, &len);
        if (res != 0) {
            printf("Unable to combine arrays: %d\n", res);
            free(array);
            return res;
        }
        res = merge_sorted_partitions(array, pre_len, len - pre_len);
        if (res != 0) {
            printf("Unable to merge partitions: %d\n", res);
            free(array);
            return res;
        }

    }

    end = MPI_Wtime();
    printf("Rank %d, computation elapsed time: %.8f\n", rank, end - start);

    if (rank == 0) {
        total += end - start;
        printf("Master: final len: %d\n", len);
        printf("Master: total elapsed time: %.8f\n", total);
        //print_integers(array, len);
    }

    free(array);
    MPI_Finalize();
    return 0;
}

read_utils.c

#include "read_utils.h"

int read_integers_from_file(char *filename, int **r_numbers, int *r_size)
{
    int i;
    int res;
    int count;
    FILE *file;
    int *numbers;
    if (r_numbers == NULL) {
        return UTILS_EBADINPUT;
    }
    if (r_size == NULL) {
        return UTILS_EBADINPUT;
    }
    file = fopen(filename, RU_MODE_READONLY);
    if (file == NULL) {
        return UTILS_EFOPEN;
    }
    res = fscanf(file, "%d", &count);
    if (res != 1) {
        fclose(file);
        return UTILS_EFSCANF;
    }
    numbers = malloc(sizeof(int) * count);
    if (numbers == NULL) {
        fclose(file);
        return UTILS_ECANTALLOC;
    }
    for (i = 0; i < count; i++) {
        res = fscanf(file, "%d", numbers + i);
        if (res != 1) {
            printf("fscanf failure, returned: %d", res);
            fclose(file);
            free(numbers);
            return UTILS_EFSCANF;
        }
    }
    *r_numbers = numbers;
    *r_size = count;
    fclose(file);
    return 0;
}