main.cc

#include <iostream>
#include <stdio.h>
#include <memory>
#include <stdlib.h>
#include <valarray>
#include <mpi.h>

void printresult(float *arr, int size, float elapsed_time){
    printf("Result array :\n");
    for(int i = 0; i < size; i++){
        printf("%f ", arr[i]);
    }
    printf("\n");
    printf("Elapsed time : %f\n", elapsed_time);
}

void quickSort(float *arr, int low, int high){
    if(low < high){
        float pivot = arr[high];
        int i = low - 1;
        for(int j = low; j <= high - 1; j++){
            if(arr[j] < pivot){
                i++;
                std::swap(arr[i], arr[j]);
            }
        }
        std::swap(arr[i + 1], arr[high]);
        quickSort(arr, low, i);
        quickSort(arr, i + 2, high);
    }
}

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

std::unique_ptr<float[]> merge(float* local, int localsize, float* recv, int recvsize) {
    int mergedSize = localsize + recvsize;
    auto mergedArray = std::make_unique<float[]>(mergedSize);

    int i = 0; 
    int j = 0; 
    int k = 0; 

    while(i < localsize && j < recvsize){
        if (local[i] <= recv[j]) {
            mergedArray[k++] = local[i++];
        }
        else {
           mergedArray[k++] = recv[j++];
        }
    }

    while(i < localsize){
        mergedArray[k++] = local[i++];
    }

    
    while(j < recvsize){
         mergedArray[k++] = recv[j++];
    }
    return mergedArray;
}

int main(int argc, char *argv[]){
    int rank, size;
    int ARRAY_SIZE = atoi(argv[1]);
    char *input_filename = argv[2], *output_filename = argv[3];
    MPI_File input_file, output_file;
    float start_time, end_time;
    
    MPI_Init(&argc, &argv); 
    MPI_Comm_rank(MPI_COMM_WORLD, &rank); 
    MPI_Comm_size(MPI_COMM_WORLD, &size);

    int chunksize = ARRAY_SIZE / size;

    int display = rank * chunksize;

    
    if(size & (size - 1)){
        MPI_Finalize(); 
        return 0; 
    } 

    
    /* TODO : read a chunk of input to each process */ 

    auto chunk = std::make_unique<float[]>(chunksize);
    
    auto ret = MPI_File_open(MPI_COMM_WORLD, input_filename, MPI_MODE_RDONLY, MPI_INFO_NULL, &input_file);
    if(ret != MPI_SUCCESS) {
        std::cerr << rank << ": fails to open file\n";
        MPI_Finalize(); 
        return 0; 
    }
    //             | MPI_FILE  | offset(by rank),       |    chunk   | chunksize |  type  |  status
    auto ret2 = MPI_File_read_at(input_file, sizeof(float) * display, chunk.get(), chunksize, MPI_FLOAT, MPI_STATUS_IGNORE);
    if(ret2 != MPI_SUCCESS) {
        std::cerr << rank << ": fails to read file\n";
        MPI_Finalize(); 
        return 0; 
    }
    MPI_File_close(&input_file);
    
    // first, sort the local array before iteration
    std::sort(chunk.get(), chunk.get()+chunksize);
    // quickSort(chunk.get(),0,chunksize-1);
    

    MPI_Request recv_request, send_request;
    int recv_chunksize;
    float *result;
    
    for(int step = 1; step < size; step *= 2){
        // rank is "what am i"
        if(rank % (2 * step) != 0){
            /* TODO : Send the chunksize and chunk to the pair process */
            int pair_rank = rank - step;
            // what to send, n, type, rank, tag
            MPI_Isend(chunk.get(), chunksize, MPI_FLOAT, pair_rank, 1, MPI_COMM_WORLD, &send_request);

            MPI_Isend(&chunksize, 1, MPI_INT, pair_rank, 0, MPI_COMM_WORLD, &send_request);

            break;
        }
        /* TODO : Recv the chunksize and chunk from the pair process, and allocate space for received chunk */ 
        int pair_rank = rank + step;
        // recieve, n, type, rank, tag, comm, status
        MPI_Irecv(&recv_chunksize, 1, MPI_INT, pair_rank, 0, MPI_COMM_WORLD, &recv_request);

        MPI_Wait(&recv_request, MPI_STATUS_IGNORE);

        auto chunk_received = std::make_unique<float[]>(recv_chunksize);

        MPI_Irecv(chunk_received.get(), recv_chunksize, MPI_FLOAT, pair_rank, 1, MPI_COMM_WORLD, &recv_request);

        MPI_Wait(&recv_request, MPI_STATUS_IGNORE);
        
        // Merge 
        auto merged_chunk = merge(chunk.get(), chunksize, chunk_received.get(), recv_chunksize);
        chunk = std::move(merged_chunk); 
        chunksize += recv_chunksize;
    }
    
    // // TODO : Output the sorted array with MPI-IO
    MPI_File_open(MPI_COMM_WORLD, output_filename,  MPI_MODE_CREATE | MPI_MODE_RDWR, MPI_INFO_NULL, &output_file);
    MPI_File_write_at(output_file, sizeof(float) * display, chunk.get(), chunksize, MPI_FLOAT, MPI_STATUS_IGNORE);
    MPI_File_close(&output_file);

    // if(rank == 0) printresult(chunk.get(), chunksize, end_time - start_time); 
    MPI_Finalize(); 
    return 0;
}