communication.h

Go to the documentation of this file.

#ifndef FPMAS_COMMUNICATION_API_H
#define FPMAS_COMMUNICATION_API_H
 
#include "fpmas/api/graph/distributed_id.h"
 
namespace fpmas { namespace api { namespace communication {
 
 
    class DataPack {
        public:
            std::size_t size;
            std::size_t count;
            std::size_t data_size;
 
            char* buffer;
 
            DataPack(std::size_t count, std::size_t data_size)
                : size(data_size*count), count(count), data_size(data_size) {
                buffer = (char*) std::malloc(size);
            }
 
            DataPack() : DataPack(0, 0) {
            }
 
            DataPack(const DataPack& other) :
                size(other.size), count(other.count), data_size(other.data_size),
                buffer((char*) std::malloc(size)) {
                    std::memcpy(buffer, other.buffer, size);
                }
 
            DataPack(DataPack&& other)
                : size(other.size), count(other.count), data_size(other.data_size) {
                    buffer = other.buffer;
                    other.buffer = (char*) std::malloc(0);
                    other.size = 0;
                    other.count = 0;
                }
 
            DataPack& operator=(const DataPack& other) {
                size = other.size;
                count = other.count;
                data_size = other.data_size;
                buffer = (char*) std::realloc(buffer, size);
                std::memcpy(buffer, other.buffer, size);
                return *this;
            }
 
            DataPack& operator=(DataPack&& other) {
                std::free(buffer);
                size = other.size;
                count = other.count;
                data_size = other.data_size;
                buffer = other.buffer;
                other.buffer = (char*) std::malloc(0);
                other.size = 0;
                other.count = 0;
                return *this;
            }
 
            void resize(std::size_t count) {
                this->count = count;
                this->size = count * data_size;
                buffer = (char*) std::realloc(buffer, this->size);
            }
 
            void free() {
                std::free(buffer);
                // A buffer of size 0 is assigned to prevent issues when
                // std::free() is called again in the DataPack destructor.
                buffer = (char*) std::malloc(0);
                size = 0;
                count = 0;
            }
 
            ~DataPack() {
                std::free(buffer);
            }
    };
 
    struct Request {
        MPI_Request __mpi_request;
        DataPack* __data;
 
        Request() : __mpi_request(), __data(nullptr) {}
        Request(const Request&) = delete;
        Request(Request&& request) {
            this->__mpi_request = request.__mpi_request;
            this->__data = request.__data;
            request.__data = nullptr;
        }
 
        Request& operator=(const Request&) = delete;
        Request& operator=(Request&& request) {
            this->__mpi_request = request.__mpi_request;
            this->__data = request.__data;
            request.__data = nullptr;
            return *this;
        }
 
        void free() {
            if(__data!=nullptr) {
                delete __data;
                __data = nullptr;
            }
        }
    };
 
    struct Status {
        static Status IGNORE;
        int size;
        int item_count;
        int source;
        int tag;
 
        Status() : size(0), item_count(0), source(0), tag(0) {}
    };
 
    class MpiCommunicator {
        public:
            static MPI_Datatype IGNORE_TYPE;
 
            virtual int getRank() const = 0;
            virtual int getSize() const = 0;
 
            virtual void send(
                    const void* data, int count, MPI_Datatype datatype,
                    int destination, int tag) = 0;
 
            virtual void send(
                    const DataPack& data, MPI_Datatype datatype,
                    int destination, int tag) = 0;
 
            virtual void send(int destination, int tag) = 0;
 
            virtual void Isend(
                    const void* data, int count, MPI_Datatype datatype,
                    int destination, int tag, Request& request) = 0;
 
            virtual void Isend(
                    const DataPack& data, MPI_Datatype datatype,
                    int destination, int tag, Request& request) = 0;
 
            virtual void Isend(
                    int destination, int tag, Request& request) = 0;
 
            virtual void Issend(
                    const void* data, int count, MPI_Datatype datatype,
                    int destination, int tag, Request& request) = 0;
 
            virtual void Issend(
                    const DataPack& data, MPI_Datatype datatype,
                    int destination, int tag, Request& request) = 0;
 
            virtual void Issend(int destination, int tag, Request& request) = 0;
 
            virtual void probe(MPI_Datatype type, int source, int tag, Status& status) = 0;
 
            virtual bool Iprobe(MPI_Datatype type, int source, int tag, Status& status) = 0;
 
 
            virtual void recv(int source, int tag, Status& status = Status::IGNORE) = 0;
 
 
            virtual void recv(
                    void* buffer, int count, MPI_Datatype datatype,
                    int source, int tag, Status& status) = 0;
 
            virtual void recv(
                    DataPack& data, MPI_Datatype datatype,
                    int source, int tag, Status& status) = 0;
 
            virtual bool test(Request& request) = 0;
 
            virtual void wait(Request& request) = 0;
 
            virtual void waitAll(std::vector<Request>& requests) = 0;
 
            // TODO: should std::vector be used instead of map? pb: what if we
            // want to send "no data" to process?
            virtual std::unordered_map<int, DataPack> allToAll(
                    std::unordered_map<int, DataPack> export_map, MPI_Datatype datatype
                    ) = 0;
 
            virtual std::vector<DataPack>
                gather(DataPack data, MPI_Datatype datatype, int root) = 0;
 
            virtual std::vector<DataPack>
                allGather(DataPack data, MPI_Datatype datatype) = 0;
 
            virtual DataPack bcast(DataPack data, MPI_Datatype datatype, int root) = 0;
 
            virtual void barrier() = 0;
 
            virtual ~MpiCommunicator() {};
    };
 
    template<typename T>
        class TypedMpi {
            public:
                // TODO: Remove this in 2.0
                virtual std::unordered_map<int, std::vector<T>>
                    migrate(std::unordered_map<int, std::vector<T>> export_map) = 0;
 
                virtual std::unordered_map<int, T>
                    allToAll(std::unordered_map<int, T> export_map) = 0;
 
                virtual std::vector<T>
                    gather(const T& data, int root) = 0;
 
                virtual std::vector<T>
                    allGather(const T& data) = 0;
 
                virtual T bcast(const T& data, int root) = 0;
 
                virtual void send(const T& data, int destination, int tag) = 0;
 
                virtual void Isend(const T& data, int destination, int tag, Request& req) = 0;
 
                virtual void Issend(const T& data, int destination, int tag, Request& req) = 0;
 
                virtual void probe(int source, int tag, Status& status) = 0;
 
                virtual bool Iprobe(int source, int tag, Status& status) = 0;
 
                virtual T recv(int source, int tag, Status& status = Status::IGNORE) = 0;
 
                virtual ~TypedMpi() {};
        };
 
    bool operator==(const DataPack& d1, const DataPack& d2);
}
}}
#endif