message_passing.hpp

Go to the documentation of this file.

/*
 * Copyright 2008,2009,2010,2011,2012 Stephan Ossowski, Korbinian Schneeberger,
 * Felix Ott, Joerg Hagmann, Alf Scotland, Sebastian Bender
 * 
 * This file is part of SHORE.
 * 
 * SHORE is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 * 
 * SHORE 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 General Public License for more details.
 * 
 * You should have received a copy of the GNU General Public License
 * along with SHORE.  If not, see <http://www.gnu.org/licenses/>.
 */


#ifndef SHORE_PROCESSING_MESSAGE_PASSING_HPP__
#define SHORE_PROCESSING_MESSAGE_PASSING_HPP__

#include <algorithm>
#include <sstream>
#include <stdexcept>
#include <vector>

#ifdef WITH_MPI
  #define MPICH_IGNORE_CXX_SEEK 1
  #include <mpi.h>
  #include <boost/serialization/string.hpp>
  #include <boost/archive/text_oarchive.hpp>
  #include <boost/archive/text_iarchive.hpp>
  #define SMPI_INFO(x) std::string(x)
#else
  #define SMPI_INFO(x) std::string()
#endif

namespace shore {

class mpi
{
 private:

        static mpi m_instance;

        class exception
        :public std::runtime_error
        {
         public:

                exception(const std::string& label,const int)
                :std::runtime_error("MPI error: "+label)
                {}
        };

        bool m_initialized;

        size_t m_comm_bufsize;
        char* m_comm_buffer;

        mpi();
        ~mpi();

        // no copy construction or assignment allowed
        mpi(const mpi&);
        mpi& operator=(const mpi&);


 #ifdef WITH_MPI

        void ensure_bufsize(const size_t size);

        template<typename T>
        size_t serialize(const T& what)
        {
                std::ostringstream str;
                {
                        boost::archive::text_oarchive arch(str);
                        arch<<what;
                }

                const std::string serial=str.str();
                //if(serial.size()>=m_instance.m_comm_bufsize) throw std::runtime_error(
                //  "mpi::serialize: message size limit exceeded");
                m_instance.ensure_bufsize(serial.size()+1);

                std::copy(serial.begin(),serial.end(),m_instance.m_comm_buffer);

                // send around zero terminated strings, since
                // MPI_Status::count does not seem to be standard
                m_instance.m_comm_buffer[serial.size()]=0;

                return serial.size()+1;
        }

        template<typename T>
        size_t deserialize(T& what,const size_t offset=0)
        {
                const std::string serial(m_comm_buffer+offset);

                std::istringstream str(serial);
                {
                        boost::archive::text_iarchive arch(str);
                        arch>>what;
                }

                return serial.size()+1;
        }
 #endif // WITH_MPI

 public:

        class status
        {
         private:

                friend class mpi;

                status();

         public:

                int source;
                int tag;
        };

        static void no_mpi(const std::string& name);

        static void start(int ac,char**av);
        static void stop();


        static int env_maxtag();

        static size_t comm_size();
        static size_t comm_rank();
        static bool comm_isroot();

        static void comm_barrier();

 #ifdef WITH_MPI
        static status comm_probe();

        static void comm_send(const int dest,const int tag);

        static status comm_recv(const int src,const int tag);

        static void comm_send(const int dest,const int tag,const int &what);

        static status comm_recv(const int src,const int tag,int &what);

        template<typename T>
        static void comm_send(const int dest,const int tag,const T& what)
        {
                const int msgsize=m_instance.serialize(what);
                
                comm_send(dest,tag,msgsize);

                const int err=MPI_Ssend(m_instance.m_comm_buffer,
                  msgsize,MPI_CHAR,dest,tag,MPI_COMM_WORLD);

                if(err!=MPI_SUCCESS)
                        throw exception("MPI_Ssend (3)",err);
        }

        template<typename T>
        static status comm_recv(const int src,const int tag,T& what)
        {
                status ret;

                int msgsize=0;
                ret=comm_recv(src,tag,msgsize);

                m_instance.ensure_bufsize(msgsize);

                MPI_Status stat;
                const int err=MPI_Recv(m_instance.m_comm_buffer,
                                       msgsize,MPI_CHAR,ret.source,ret.tag,
                                       MPI_COMM_WORLD,&stat);

                if(err!=MPI_SUCCESS)
                        throw exception("MPI_Recv (3)",err);

                ret.source=stat.MPI_SOURCE;
                ret.tag=stat.MPI_TAG;

                m_instance.deserialize(what);

                return ret;
        }

        template<typename T>
        static status comm_recv(const int src,T& what)
        {
                return comm_recv(src,MPI_ANY_TAG,what);
        }

        template<typename T>
        static status comm_recv(T& what)
        {
                return comm_recv(MPI_ANY_SOURCE,MPI_ANY_TAG,what);
        }

 #endif // WITH_MPI

        static void comm_broadcast(int &what,int root=0);

        template<typename T>
        static void comm_broadcast(T& what,int root=0)
        {
 #ifdef WITH_MPI

                int msgsize=0;
                const size_t rnk=comm_rank();

                if(rnk==size_t(root))
                        msgsize=m_instance.serialize(what);

                // message size must be known to all processes
                comm_broadcast(msgsize,root);

                if(rnk!=size_t(root)) m_instance.ensure_bufsize(msgsize);

                const int err=MPI_Bcast(m_instance.m_comm_buffer,
                                        msgsize,MPI_CHAR,root,MPI_COMM_WORLD);

                if(err!=MPI_SUCCESS)
                        throw exception("MPI_Bcast (2)",err);

                if(rnk!=size_t(root))
                        m_instance.deserialize(what);
 #endif // WITH_MPI
        }

        static std::vector<int> comm_gather(const int &what,int root=0);
};

} // namespace

#endif // SHORE_PROCESSING_MESSAGE_PASSING_HPP__