MHO_MPIInterface.hh

Go to the documentation of this file.

#ifndef MHO_MPIInterface_HH__
#define MHO_MPIInterface_HH__
 
#include "mpi.h"
 
#include "MHO_Message.hh"
#include <string>
#include <vector>
 
#define LOCAL_RANK_MPI
 
namespace hops
{
 
//helper template for mapping basic types to MPI type codes
template< typename T > MPI_Datatype mpi_type_for();
 
template<> inline MPI_Datatype mpi_type_for< int >()
{
    return MPI_INT;
}
 
template<> inline MPI_Datatype mpi_type_for< double >()
{
    return MPI_DOUBLE;
}
 
template<> inline MPI_Datatype mpi_type_for< float >()
{
    return MPI_FLOAT;
}
 
class MHO_MPIInterface
{
    public:
        //singleton interface
        static MHO_MPIInterface* GetInstance();
 
        void Initialize(int* argc, char*** argv, bool split_mode = true);
        void Finalize();
 
        bool Check() const { return (fGlobalProcessID >= 0) && (fNProcesses > 0); }
 
        int GetGlobalProcessID() const { return fGlobalProcessID; }
 
        int GetNProcesses() const { return fNProcesses; }
 
        int GetLocalProcessID() const { return fLocalProcessID; }
 
        std::string GetHostName() const { return fHostName; };
 
        //use to isolate a section of code, so each process completes it
        //one at a time
        void BeginSequentialProcess();
        void EndSequentialProcess();
 
        void GlobalBarrier() const { MPI_Barrier(MPI_COMM_WORLD); }
 
        void PrintMessage(std::string msg);
 
        //broadcast a string message to all processes
        void BroadcastString(std::string& msg);
 
        //routines to be used by programs which split the processes into two
        //groups bases on even/odd local process rank
        bool SplitMode() { return fSplitMode; };
 
        bool IsSplitValid() { return fValidSplit; };
 
        bool IsEvenGroupMember() { return fIsEvenGroupMember; };
 
        int GetNSubGroupProcesses() { return fNSubGroupProcesses; }
 
        int GetSubGroupRank() { return fSubGroupRank; };
 
        int GetPartnerProcessID() { return fPartnerProcessID; };
 
        MPI_Group* GetSubGroup()
        {
            if(fIsEvenGroupMember)
            {
                return &fEvenGroup;
            }
            else
            {
                return &fOddGroup;
            };
        }
 
        MPI_Comm* GetSubGroupCommunicator()
        {
            if(fIsEvenGroupMember)
            {
                return &fEvenCommunicator;
            }
            else
            {
                return &fOddCommunicator;
            };
        }
 
        MPI_Group* GetEvenGroup() { return &fEvenGroup; };
 
        MPI_Group* GetOddGroup() { return &fOddGroup; };
 
        MPI_Comm* GetEvenCommunicator() { return &fEvenCommunicator; };
 
        MPI_Comm* GetOddCommunicator() { return &fOddCommunicator; };
 
        template< typename T > std::map< std::string, T > MergeMap(const std::map< std::string, T >& local_map)
        {
            //serialize the local map into primitive types
            std::vector< int > key_lengths;
            std::string concatenated_keys;
            std::vector< T > values;
 
            for(const auto& kv : local_map)
            {
                key_lengths.push_back(static_cast< int >(kv.first.size()));
                concatenated_keys += kv.first;
                values.push_back(kv.second);
            }
 
            int local_entry_count = static_cast< int >(local_map.size());
            int local_char_count = static_cast< int >(concatenated_keys.size());
            // gather map entry counts
            std::vector< int > entry_counts(fNProcesses);
            MPI_Gather(&local_entry_count, 1, MPI_INT, entry_counts.data(), 1, MPI_INT, 0, MPI_COMM_WORLD);
 
            //gather key character counts
            std::vector< int > char_counts(fNProcesses);
            MPI_Gather(&local_char_count, 1, MPI_INT, char_counts.data(), 1, MPI_INT, 0, MPI_COMM_WORLD);
 
            std::map< std::string, T > merged;
 
            //only rank 0 reconstructs ---
            if(fGlobalProcessID == 0)
            {
                // Compute displacements
                std::vector< int > entry_displs(fNProcesses, 0);
                std::vector< int > char_displs(fNProcesses, 0);
 
                for(int i = 1; i < fNProcesses; ++i)
                {
                    entry_displs[i] = entry_displs[i - 1] + entry_counts[i - 1];
                    char_displs[i] = char_displs[i - 1] + char_counts[i - 1];
                }
 
                int total_entries = entry_displs[fNProcesses - 1] + entry_counts[fNProcesses - 1];
                int total_chars = char_displs[fNProcesses - 1] + char_counts[fNProcesses - 1];
 
                // allocate receive buffers
                std::vector< int > all_key_lengths(total_entries);
                std::vector< char > all_chars(total_chars);
                std::vector< T > all_values(total_entries);
 
                // gather key lengths
                MPI_Gatherv(key_lengths.data(), local_entry_count, MPI_INT, all_key_lengths.data(), entry_counts.data(),
                            entry_displs.data(), MPI_INT, 0, MPI_COMM_WORLD);
 
                // gather keys
                MPI_Gatherv(concatenated_keys.data(), local_char_count, MPI_CHAR, all_chars.data(), char_counts.data(),
                            char_displs.data(), MPI_CHAR, 0, MPI_COMM_WORLD);
 
                // gather values
                MPI_Gatherv(values.data(), local_entry_count, mpi_type_for< T >(), all_values.data(), entry_counts.data(),
                            entry_displs.data(), mpi_type_for< T >(), 0, MPI_COMM_WORLD);
 
                // Reconstruct merged map
                size_t pos = 0;
                for(int i = 0; i < total_entries; ++i)
                {
                    int len = all_key_lengths[i];
                    std::string key(all_chars.begin() + pos, all_chars.begin() + pos + len);
                    pos += len;
                    merged[key] = all_values[i];
                }
            }
            else
            {
                //non-root ranks just need to participate in gathers
                MPI_Gatherv(key_lengths.data(), local_entry_count, MPI_INT, nullptr, nullptr, nullptr, MPI_INT, 0,
                            MPI_COMM_WORLD);
 
                MPI_Gatherv(concatenated_keys.data(), local_char_count, MPI_CHAR, nullptr, nullptr, nullptr, MPI_CHAR, 0,
                            MPI_COMM_WORLD);
 
                MPI_Gatherv(values.data(), local_entry_count, mpi_type_for< T >(), nullptr, nullptr, nullptr,
                            mpi_type_for< T >(), 0, MPI_COMM_WORLD);
            }
 
            return merged;
        }
 
        std::set< std::string > MergeStringSet(const std::set< std::string >& local_set);
 
    protected:
        MHO_MPIInterface();
        virtual ~MHO_MPIInterface();
 
        int fGlobalProcessID;
        int fNProcesses;
        int fLocalProcessID;
        std::string fHostName;
        std::vector< int > fCoHostedProcessIDs;
 
        //groups and communicators for splitting processes into
        //two sets, based on whether they have even/odd (local) ranks
        bool fSplitMode;
        MPI_Group fEvenGroup;       //even process subgroup
        MPI_Group fOddGroup;        //odd process subgroup
        MPI_Comm fEvenCommunicator; //comm for even group
        MPI_Comm fOddCommunicator;  //comm for odd group
        bool fValidSplit;           //true if the size of the subgroups is equal
        bool fIsEvenGroupMember;    //true if this process is a member of the even subgroup
        int fSubGroupRank;          //rank of this process in its subgroup
        int fNSubGroupProcesses;    //number of processes in the subgroup this process belongs to
        int fPartnerProcessID;      //global rank of partner process in other subgroup
 
        void DetermineLocalRank();
        void SetupSubGroups();
 
        MPI_Status fStatus;
};
 
} //end of namespace hops
 
#endif