DistDataCollector.h

#include <mpi.h>
#include <set>
#include <vector>
#include <limits>
 
#ifndef DIST_DATA_COLLECTOR
#define DIST_DATA_COLLECTOR
 

class DistDataCollector {
 
    private:
 
        // MPI communicator to use for communication
        MPI_Comm comm;
 
        // number of chunks
        std::size_t numChunks;
 
        // local size of the data
        std::size_t numSize;
 
        // the array that collects the data of size numChunks * numSize on rootRank
        double* collectedData;
 
        // MPI window for remote memory access
        MPI_Win win;
 
        // MPI rank that holds the collected data
        int rootRank;
 
        public:
 
        // initial values
        const double BAD_VALUE = std::numeric_limits<double>::quiet_NaN();

    DistDataCollector(MPI_Comm comm, int numChunks, int numSize, int rootRank = 0);

    ~DistDataCollector();

    MPI_Win &getWin(){
        return this->win;
    }

    void fence();

    void inline startEpoch() {
        // Start a passive target shared local access epoch for all processes in the communicator
        MPI_Win_lock_all(MPI_MODE_NOCHECK, this->win);
    }

    void inline flush() {
        MPI_Win_flush(this->rootRank, this->win);
    }

    void inline endEpoch() {
        MPI_Win_unlock_all(this->win);
    }   

    void put(int chunkId, const double* data);

    void inline putAsync(int chunkId, const double* data) {
        MPI_Put(data, this->numSize, MPI_DOUBLE, this->rootRank, chunkId * this->numSize, this->numSize, MPI_DOUBLE, this->win);
    }

    void accumulate(int chunkId, const double* data);

    std::vector<double> get(int chunkId);

    void get(int chunkId, double* buffer);

    void inline getAsync(int chunkId, double* buffer) {
        MPI_Get(buffer, this->numSize, MPI_DOUBLE, this->rootRank, chunkId * this->numSize, this->numSize, MPI_DOUBLE, this->win);
    }

    double* getCollectedDataPtr() {
        return this->collectedData;
    }

    int getNumChunks() const {
        return this->numChunks;
    }

    int getNumSize() const {
        return this->numSize;
    }

    void free() { // Should this be removed and just implemented in the destructor?
        if (this->win != MPI_WIN_NULL) {
            MPI_Win_free(&this->win);
        }
        //No need to free the data, MPI_Win_free will free the pointer
        //MPI_Free_mem(this->collectedData);
    }
 
    // Disable copy and assignment operations
    // to prevent accidental copying of the DistDataCollector instance
    // This is important because the class manages an MPI window and data pointer
    // which should not be copied or assigned.
    DistDataCollector(const DistDataCollector&) = delete; // Disable copy constructor
    DistDataCollector& operator=(const DistDataCollector&) = delete; // Disable assignment operator
    DistDataCollector(DistDataCollector&& other) noexcept; // Move constructor
    DistDataCollector& operator=(DistDataCollector&& other) noexcept; // Move assignment operator
};
 
#endif // DIST_DATA_COLLECTOR