communication.h

Go to the documentation of this file.

#ifndef FPMAS_COMMUNICATION_H
#define FPMAS_COMMUNICATION_H
 
#include <cstdint>
#include <string>
#include <unordered_map>
 
#include "fpmas/utils/log.h"
#include "fpmas/api/communication/communication.h"
#include "fpmas/io/json_datapack.h"
 
namespace fpmas {
    void init(int argc, char** argv);
}
 
namespace fpmas { namespace communication {
    using api::communication::DataPack;
    using api::communication::Request;
    using api::communication::Status;
 
    class MpiCommunicatorBase : public virtual api::communication::MpiCommunicator {
        protected:
            int size;
            int rank;
 
            MPI_Group group;
            MPI_Comm comm;
 
        private:
            static void convertStatus(MPI_Status&, Status&, MPI_Datatype datatype);
 
        public:
            MPI_Comm getMpiComm() const;
 
            MPI_Group getMpiGroup() const;
 
            int getRank() const override;
 
            int getSize() const override;
 
            void send(
                    const void* data, int count, MPI_Datatype datatype,
                    int destination, int tag) override;
 
            void send(
                    const DataPack& data, MPI_Datatype datatype,
                    int destination, int tag) override;
 
            void send(int destination, int tag) override;
 
 
            void Isend(
                    const void* data, int count, MPI_Datatype datatype,
                    int destination, int tag, Request& req) override;
 
            void Isend(
                    const DataPack& data, MPI_Datatype datatype,
                    int destination, int tag, Request& req) override;
 
            void Isend(int destination, int tag, Request& req) override;
 
            void Issend(
                    const void* data, int count, MPI_Datatype datatype,
                    int destination, int tag, Request& req) override;
 
            void Issend(
                    const DataPack& data, MPI_Datatype datatype,
                    int destination, int tag, Request& req) override;
            void Issend(int destination, int tag, Request& req) override;
 
            void recv(int source, int tag, Status& status = Status::IGNORE) override;
 
            void recv(
                    void* buffer, int count, MPI_Datatype datatype,
                    int source, int tag, Status& status = Status::IGNORE) override;
 
            void recv(
                    DataPack& data, MPI_Datatype datatype,
                    int source, int tag, Status& status = Status::IGNORE) override;
 
            void probe(MPI_Datatype type, int source, int tag, Status&) override;
 
            bool Iprobe(MPI_Datatype type, int source, int tag, Status&) override;
 
            bool test(Request& req) override;
 
            void wait(Request& req) override;
 
            void waitAll(std::vector<Request>& req) override;
 
            std::unordered_map<int, DataPack> 
                allToAll(std::unordered_map<int, DataPack> export_map, MPI_Datatype datatype) override;
 
            std::vector<DataPack>
                gather(DataPack data, MPI_Datatype datatype, int root) override;
 
            std::vector<DataPack>
                allGather(DataPack data, MPI_Datatype datatype) override;
 
            DataPack bcast(DataPack data, MPI_Datatype datatype, int root) override;
            
            void barrier() override;
    };
 
 
    class MpiCommunicator : public MpiCommunicatorBase {
        public:
            MpiCommunicator();
 
            MpiCommunicator(const MpiCommunicator&) = delete;
            MpiCommunicator& operator=(const MpiCommunicator&) = delete;
 
 
            ~MpiCommunicator();
    };
 
    class MpiCommWorld : public MpiCommunicatorBase {
        public:
        void init() {
            this->comm = MPI_COMM_WORLD;
            MPI_Comm_group(MPI_COMM_WORLD, &this->group);
            MPI_Comm_size(MPI_COMM_WORLD, &this->size);
            MPI_Comm_rank(MPI_COMM_WORLD, &this->rank);
        };
    };
 
    extern MpiCommWorld WORLD;
 
    namespace detail {
        template<typename T, typename PackType>
            class TypedMpi : public api::communication::TypedMpi<T> {
                private:
                    api::communication::MpiCommunicator& comm;
                public:
                    typedef PackType pack_type;
 
                    TypedMpi(api::communication::MpiCommunicator& comm) : comm(comm) {}
 
                    std::unordered_map<int, std::vector<T>>
                        migrate(std::unordered_map<int, std::vector<T>> export_map) override;
 
                    std::unordered_map<int, T>
                        allToAll(std::unordered_map<int, T> export_map) override;
 
                    std::vector<T> gather(const T&, int root) override;
                    std::vector<T> allGather(const T&) override;
                    T bcast(const T&, int root) override;
 
                    void send(const T&, int, int) override;
 
                    void Isend(const T&, int, int, Request&) override;
 
                    void Issend(const T&, int, int, Request&) override;
 
                    void probe(int source, int tag, Status& status) override;
                    bool Iprobe(int source, int tag, Status& status) override;
 
                    T recv(int source, int tag, Status& status = Status::IGNORE) override;
            };
 
        template<typename T, typename PackType> std::unordered_map<int, std::vector<T>>
            TypedMpi<T, PackType>::migrate(std::unordered_map<int, std::vector<T>> export_map) {
                std::unordered_map<int, DataPack> import_data_pack;
 
                {
                    // Pack
                    std::unordered_map<int, DataPack> export_data_pack;
                    for(auto& item : export_map)
                        export_data_pack.emplace(item.first, PackType(item.second).dump());
 
                    // export_data_pack buffers are moved to the temporary allToAll
                    // argument, and automatically freed by the allToAll
                    // implementation
                    import_data_pack = comm.allToAll(
                            std::move(export_data_pack), MPI_CHAR
                            );
                }
 
                std::unordered_map<int, std::vector<T>> import_map;
                for(auto& item : import_data_pack)
                    import_map.emplace(
                            item.first, PackType::parse(std::move(item.second))
                            .template get<std::vector<T>>()
                            );
                
                // Should perform "copy elision"
                return import_map;
            }
 
        template<typename T, typename PackType> std::unordered_map<int, T>
            TypedMpi<T, PackType>::allToAll(std::unordered_map<int, T> export_map) {
                // Pack
                std::unordered_map<int, DataPack> import_data_pack;
 
                {
                    std::unordered_map<int, DataPack> export_data_pack;
                    for(auto& item : export_map)
                        export_data_pack.emplace(item.first, PackType(item.second).dump());
 
                    // export_data_pack buffers are moved to the temporary allToAll
                    // argument, and automatically freed by the allToAll
                    // implementation
                    import_data_pack = comm.allToAll(
                            std::move(export_data_pack), MPI_CHAR
                            );
                }
 
                std::unordered_map<int, T> import_map;
                for(auto& item : import_data_pack)
                    import_map.emplace(
                            item.first, PackType::parse(std::move(item.second))
                            .template get<T>()
                            );
 
                // Should perform "copy elision"
                return import_map;
            }
 
        template<typename T, typename PackType> std::vector<T>
            TypedMpi<T, PackType>::gather(const T& data, int root) {
                DataPack data_pack = PackType(data).dump();
 
                std::vector<DataPack> import_data_pack
                    = comm.gather(data_pack, MPI_CHAR, root);
 
                std::vector<T> import_data;
                for(std::size_t i = 0; i < import_data_pack.size(); i++) {
                    import_data.emplace_back(
                            PackType::parse(
                                std::move(import_data_pack[i])).template get<T>()
                            );
                }
                return import_data;
            }
 
        template<typename T, typename PackType> std::vector<T>
            TypedMpi<T, PackType>::allGather(const T& data) {
                // Pack
                DataPack data_pack = PackType(data).dump();
 
                std::vector<DataPack> import_data_pack
                    = comm.allGather(data_pack, MPI_CHAR);
 
                std::vector<T> import_data;
                for(std::size_t i = 0; i < import_data_pack.size(); i++) {
                    import_data.emplace_back(
                            PackType::parse(
                                std::move(import_data_pack[i])).template get<T>()
                            );
                }
                return import_data;
            }
 
        template<typename T, typename PackType>
            T TypedMpi<T, PackType>::bcast(const T& data, int root) {
                DataPack data_pack = PackType(data).dump();
 
                DataPack recv_data_pack = comm.bcast(data_pack, MPI_CHAR, root);
 
                return PackType::parse(std::move(recv_data_pack)).template get<T>();
            }
 
        template<typename T, typename PackType>
            void TypedMpi<T, PackType>::send(const T& data, int destination, int tag) {
                //FPMAS_LOGD(comm.getRank(), "TYPED_MPI", "Send JSON to process %i : %s", destination, str.c_str());
                DataPack data_pack = PackType(data).dump();
                comm.send(data_pack, MPI_CHAR, destination, tag);
            }
 
        template<typename T, typename PackType>
            void TypedMpi<T, PackType>::Isend(const T& data, int destination, int tag, Request& req) {
                //FPMAS_LOGD(comm.getRank(), "TYPED_MPI", "Issend JSON to process %i : %s", destination, str.c_str());
                DataPack data_pack = PackType(data).dump();
                comm.Isend(data_pack, MPI_CHAR, destination, tag, req);
            }
 
 
        template<typename T, typename PackType>
            void TypedMpi<T, PackType>::Issend(const T& data, int destination, int tag, Request& req) {
                //FPMAS_LOGD(comm.getRank(), "TYPED_MPI", "Issend JSON to process %i : %s", destination, str.c_str());
                DataPack data_pack = PackType(data).dump();
                comm.Issend(data_pack, MPI_CHAR, destination, tag, req);
            }
 
        template<typename T, typename PackType>
            void TypedMpi<T, PackType>::probe(int source, int tag, Status &status) {
                return comm.probe(MPI_CHAR, source, tag, status);
            }
        template<typename T, typename PackType>
            bool TypedMpi<T, PackType>::Iprobe(int source, int tag, Status &status) {
                return comm.Iprobe(MPI_CHAR, source, tag, status);
            }
 
        template<typename T, typename PackType>
            T TypedMpi<T, PackType>::recv(int source, int tag, Status& status) {
                Status message_to_receive_status;
                this->probe(source, tag, message_to_receive_status);
                int count = message_to_receive_status.item_count;
                DataPack data_pack(count, sizeof(char));
                comm.recv(data_pack, MPI_CHAR, source, tag, status);
 
                //FPMAS_LOGD(comm.getRank(), "TYPED_MPI", "Receive JSON from process %i : %s", source, data.c_str());
                return PackType::parse(std::move(data_pack)).template get<T>();
            }
    }
 
    template<typename T>
        struct TypedMpi : public detail::TypedMpi<T, io::datapack::ObjectPack> {
            using detail::TypedMpi<T, io::datapack::ObjectPack>::TypedMpi;
        };
 
 
    template<typename T, typename BinaryOp = std::plus<T>>
        T reduce(
                api::communication::TypedMpi<T>& mpi, int root,
                const T& data, BinaryOp binary_op = BinaryOp()) {
            std::vector<T> data_vec = mpi.gather(data, root);
            if(data_vec.size() > 0)
                return std::accumulate(std::next(data_vec.begin()), data_vec.end(), data_vec[0], binary_op);
            return data;
        }
    template<typename T, typename BinaryOp = std::plus<T>>
        T all_reduce(
                api::communication::TypedMpi<T>& mpi,
                const T& data, BinaryOp binary_op = BinaryOp()) {
            std::vector<T> data_vec = mpi.allGather(data);
            return std::accumulate(std::next(data_vec.begin()), data_vec.end(), data_vec[0], binary_op);
        }
}}
#endif