MPIControl.h

Go to the documentation of this file.

/***********************************************************************************/
/*  Copyright 2009-2015 WSL Institute for Snow and Avalanche Research    SLF-DAVOS      */
/***********************************************************************************/
/* This file is part of Alpine3D.
    Alpine3D is free software: you can redistribute it and/or modify
    it under the terms of the GNU Lesser General Public License as published by
    the Free Software Foundation, either version 3 of the License, or
    (at your option) any later version.

    Alpine3D is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU Lesser General Public License for more details.

    You should have received a copy of the GNU Lesser General Public License
    along with Alpine3D.  If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef MPIWRAPPER_H
#define MPIWRAPPER_H

#ifdef ENABLE_MPI
        #include <mpi.h>
#endif

#ifdef _OPENMP
        #include <omp.h>
#endif

#include <meteoio/MeteoIO.h>
#include <cstdio>

class MPIControl
{
        public:
                static MPIControl& instance();

                // The following methods refer to OpenMP
                bool openmp() const;

                size_t thread() const;

                size_t max_threads() const;

                // The following methods refer to MPI
                bool master() const;

                size_t master_rank() const;

                size_t rank() const;

                size_t size() const;

                std::string name() const;

                void barrier() const;

                void gather(const int& send_value, std::vector<int>& receive_vector, const size_t& root = 0);


                void allreduce_max(double& value);
                void allreduce_min(double& value);
                void allreduce_sum(double& value);
                void allreduce_sum(int& value);

                template<class T> static void deserialize(const void* in, const size_t& len, T& obj)
                {
                        std::stringstream obj_stream;
                        obj_stream.write(reinterpret_cast<const char*>(in), len);
                        obj_stream >> obj;
                }

                template<class T> static size_t serialize(void** out, const T& obj, const bool alloc=false)
                {
                        std::stringstream obj_stream;
                        obj_stream << obj;
                        const size_t len = obj_stream.str().size();
                        if (alloc) {
                                *out = new char[len];
                        }
                        obj_stream.read(reinterpret_cast<char*>(*out), len);
                        return len;
                }

                #ifdef ENABLE_MPI

                template <class T> static void op_sum_func(void* in, void* out, int* /*len*/, MPI_Datatype* datatype)
                {
                        int tmp_size;
                        MPI_Type_size(*datatype, &tmp_size);

                        T in_obj, out_obj;

                        deserialize(in, static_cast<size_t>(tmp_size), in_obj);
                        deserialize(out, static_cast<size_t>(tmp_size), out_obj);

                        out_obj += in_obj;

                        serialize(&out, out_obj);
                }
                #endif

                #ifdef ENABLE_MPI

                template <class T> void allreduce_sum(T& obj)
                {
                        if (size_ <= 1) return;

                        MPI_Op op;
                        MPI_Op_create(op_sum_func<T>, true, &op);

                        void* in_obj_char=NULL;
                        void* out_obj_char=NULL;

                        const size_t len = serialize(&in_obj_char, obj, true);
                        out_obj_char = new char[len];

                        MPI_Datatype chars_datatype;
                        MPI_Type_contiguous(static_cast<int>(len), MPI_CHAR, &chars_datatype);
                        MPI_Type_commit(&chars_datatype);

                        MPI_Allreduce(in_obj_char, out_obj_char, 1, chars_datatype, op, MPI_COMM_WORLD);
                        MPI_Op_free(&op);

                        deserialize(out_obj_char, len, obj);

                        delete[] (char*)out_obj_char;
                        delete[] (char*)in_obj_char;
                }
                #else
                template <class T> void allreduce_sum(T& /*obj*/) {}
                #endif

                #ifdef ENABLE_MPI

                template <class T> void broadcast(T& obj, const size_t& root = 0)
                {
                        if (size_ <= 1) return;

                        std::string obj_string;

                        if (rank_ == root) { // Build the string that represents the object
                                std::stringstream obj_stream;
                                obj_stream << obj;
                                obj_string.insert(0, obj_stream.str());
                        }

                        broadcast(obj_string, root);

                        if (rank_ != root) { // Deserialize the object broadcasted
                                std::stringstream obj_stream;
                                obj_stream << obj_string;
                                obj_stream >> obj;
                        }
                }
                #else
                template <class T> void broadcast(T& /*obj*/, const size_t& root=0) {(void)root;}
                #endif

                #ifdef ENABLE_MPI

                template <class T> void broadcast(std::vector<T>& vec_obj, const size_t& root = 0)
                {
                        if (size_ <= 1) return;

                        std::string obj_string;
                        size_t vec_size;

                        if (rank_ == root) { // Build the string that represents the objects
                                std::stringstream objs_stream;
                                for (size_t ii=0; ii<vec_obj.size(); ii++)
                                        objs_stream << vec_obj[ii];
                                obj_string.insert(0, objs_stream.str());
                                vec_size = vec_obj.size();
                        }

                        broadcast(vec_size, root);
                        broadcast(obj_string, root);

                        if (rank_ != root) { // Deserialize the objects broadcasted
                                vec_obj.clear();

                                std::stringstream objs_stream;
                                objs_stream << obj_string;

                                T tmp;
                                for (size_t ii=0; ii<vec_size; ii++) {
                                        objs_stream >> tmp;
                                        vec_obj.push_back(tmp);
                                }
                        }
                }
                #else
                template <class T> void broadcast(std::vector<T>& /*vec_obj*/, const size_t& root=0) {(void)root;}
                #endif

                #ifdef ENABLE_MPI

                template <class T> void scatter(std::vector<T*>& vec_local, const size_t& root = 0)
                {
                        if (size_ <= 1) return;

                        if (rank_ == root) {
                                for (size_t ii=1; ii<size_; ii++) { // HACK: Assuming root is master
                                        size_t startx, deltax;
                                        getArraySliceParams(vec_local.size(), ii, startx, deltax);

                                        send((int)deltax, ii);
                                        for (size_t jj=startx; jj<(startx+deltax); jj++) {
                                                std::string obj_string;
                                                std::stringstream objs_stream;

                                                objs_stream << *(vec_local[jj]);
                                                obj_string.insert(0, objs_stream.str());
                                                send(obj_string, ii);
                                                delete vec_local[jj];
                                        }
                                }

                                size_t startx, deltax;
                                getArraySliceParams(vec_local.size(), startx, deltax);
                                vec_local.resize(deltax);
                        } else {
                                std::string obj_string;
                                int deltax;
                                receive(deltax, root);
                                vec_local.resize((size_t)deltax);

                                T obj; // temporary object
                                for (size_t ii=0; ii<vec_local.size(); ii++) {
                                        std::stringstream objs_stream;
                                        receive(obj_string, root);

                                        objs_stream << obj_string;

                                        objs_stream >> obj;
                                        vec_local[ii] = new T(obj);
                                }
                        }
                }
                #else
                template <class T> void scatter(std::vector<T*>& /*vec_local*/, const size_t& root=0) {(void)root;}
                #endif

                #ifdef ENABLE_MPI

                template <class T> void send(const std::vector<T*>& vec_local, const size_t& destination, const int& tag=0);
                template <class T> void send(const std::vector<T>& vec_local, const size_t& destination, const int& tag=0);
                #else
                template <class T> void send(const std::vector<T*>& /*vec_local*/, const size_t& /*destination*/, const int& tag=0) {(void)tag;}
                template <class T> void send(const std::vector<T>& /*vec_local*/, const size_t& /*destination*/, const int& tag=0) {(void)tag;}
                #endif

                #ifdef ENABLE_MPI

                template <class T> void receive(std::vector<T*>& vec_local, const size_t& source, const int& tag=0);
                template <class T> void receive(std::vector<T>& vec_local, const size_t& source, const int& tag=0);
                #else
                template <class T> void receive(std::vector<T*>& /*vec_local*/, const size_t& /*source*/, const int& tag=0) {(void)tag;}
                template <class T> void receive(std::vector<T>& /*vec_local*/, const size_t& /*source*/, const int& tag=0) {(void)tag;}
                #endif

                #ifdef ENABLE_MPI

                template <class T> void gather(std::vector<T*>& vec_local, const size_t& root = 0)
                {
                        if (size_ <= 1) return;

                        std::vector<int> vec_sizes;
                        const size_t sum = vec_local.size();

                        gather(vec_local.size(), vec_sizes); //global offsets
                        allreduce_sum(sum); //global amount of T*

                        if (rank_ == root) {
                                std::string obj_string;
                                size_t local_sum = vec_local.size();

                                vec_local.resize(sum);

                                T obj; //temporary object
                                for (size_t ii=1; ii<vec_sizes.size(); ii++) {//HACK: Assuming master is always rank 0
                                        for (size_t jj=0; jj<vec_sizes[ii]; jj++) {
                                                std::stringstream objs_stream;
                                                receive(obj_string, ii);

                                                objs_stream << obj_string;
                                                objs_stream >> obj;
                                                vec_local[local_sum+jj] = new T(obj);
                                        }
                                        local_sum += vec_sizes[ii];
                                }

                        } else {
                                for (size_t ii=0; ii<vec_local.size(); ii++) {
                                        std::string obj_string;
                                        std::stringstream objs_stream;

                                        objs_stream << *(vec_local[ii]);
                                        obj_string.insert(0, objs_stream.str());
                                        send(obj_string, root);
                                }
                        }
                }
                #else
                template <class T> void gather(std::vector<T*>& /*vec_local*/, const size_t& root=0) {(void)root;}
                #endif

                void getArraySliceParams(const size_t& dimx, size_t& startx_sub, size_t& nx_sub) const {getArraySliceParams(dimx, size_, rank_, startx_sub, nx_sub);}
                void getArraySliceParams(const size_t& dimx, const size_t& idx_wk, size_t& startx_sub, size_t& nx_sub) const {getArraySliceParams(dimx, size_, idx_wk, startx_sub, nx_sub);}

                void getArraySliceParamsOptim(const size_t& dimx, size_t& startx_sub, size_t& nx_sub,const mio::DEMObject& dem, const mio::Grid2DObject& landuse)
                     {getArraySliceParamsOptim(dimx, rank_, startx_sub, nx_sub, dem, landuse);}

                void getArraySliceParamsOptim(const size_t& dimx, const size_t& idx_wk, size_t& startx_sub, size_t& nx_sub,const mio::DEMObject& dem, const mio::Grid2DObject& landuse);

                static void getArraySliceParams(const size_t& dimx, const size_t& nbworkers, const size_t& idx_wk, size_t& startx_sub, size_t& nx_sub);

                static void getArraySliceParams(const size_t& dimx, const size_t& nbworkers, std::vector<size_t>& offset, std::vector<size_t>& nx);

        private:
                MPIControl();
                ~MPIControl();

                MPIControl(MPIControl const&);     //No implementation allowed - singleton class
                void operator=(MPIControl const&); //No implementation allowed - singleton class
                void broadcast(std::string& message, const size_t& root);

                void send(std::string& message, const size_t& recipient, const int& tag=0);
                void receive(std::string& message, const size_t& source, const int& tag=0);
                void send(const int& value, const size_t& recipient, const int& tag=0);
                void receive(int& value, const size_t& source, const int& tag=0);

                static void checkSuccess(const int& ierr);

                size_t rank_;          // the rank of this process
                size_t size_;          // the number of all processes
                std::string name_;  // the name of the node this process is running on
};

#endif