Commit: 4e91aae83ea0476ef7112e00d6fe2b9a5031574c - MolDS (git)

Commit MetaInfo

• Revision: 4e91aae83ea0476ef7112e00d6fe2b9a5031574c (tree)
• Time: 2013-07-12 15:59:42
• Author: Mikiya Fujii <mikiya.fujii@gmai...>
• Commiter: Mikiya Fujii

Log Message

Too big messages are split to be send by MPI. #31588

git-svn-id: https://svn.sourceforge.jp/svnroot/molds/branches/mpi-cis@1388 1136aad2-a195-0410-b898-f5ea1d11b9d8

Change Summary

• modified: src/mpi/MpiProcess.cpp
• modified: src/mpi/MpiProcess.h

Incremental Difference

--- a/src/mpi/MpiProcess.cpp
+++ b/src/mpi/MpiProcess.cpp
@@ -37,6 +37,7 @@ MpiProcess::MpiProcess(){
 MpiProcess::MpiProcess(int argc, char *argv[]){
    this->environment  = new boost::mpi::environment(argc, argv);
    this->communicator = new boost::mpi::communicator();
+   this->messageLimit = INT_MAX;
 }
 
 MpiProcess::~MpiProcess(){

--- a/src/mpi/MpiProcess.h
+++ b/src/mpi/MpiProcess.h
@@ -18,9 +18,9 @@
 //************************************************************************//
 #ifndef INCLUDED_MPIPROCESS
 #define INCLUDED_MPIPROCESS
+#include<limits.h>
 #include<boost/mpi.hpp>
 namespace MolDS_mpi{
-
 // MpiProcess is singleton
 class MpiProcess: private MolDS_base::Uncopyable{
 public:
@@ -29,9 +29,28 @@ public:
    static MpiProcess* GetInstance();
    int GetRank() const{return this->communicator->rank();}
    int GetSize() const{return this->communicator->size();}
-   template<typename T> void Send(int dest, int tag, const T* values, int n) const{this->communicator->send(dest, tag, values, n);}
-   template<typename T> void Recv(int source, int tag, T* values, int n) const{this->communicator->recv(source, tag, values, n);}
-   template<typename T> void Broadcast(T* values, int n, int root) const{broadcast(*this->communicator, values, n, root);}
+   template<typename T> void Send(int dest, int tag, const T* values, intptr_t num) const{
+      std::vector<Chunk> chunks;
+      this->SplitMessage2Chunks(chunks, tag, values, num);
+      for(intptr_t i=0; i<chunks.size(); i++){
+         this->communicator->send(dest, chunks[i].tag, &values[chunks[i].first], chunks[i].num);
+      }
+   }
+   template<typename T> void Recv(int source, int tag, T* values, intptr_t num) const{
+      std::vector<Chunk> chunks;
+      this->SplitMessage2Chunks(chunks, tag, values, num);
+      for(intptr_t i=0; i<chunks.size(); i++){
+         this->communicator->recv(source, chunks[i].tag, &values[chunks[i].first], chunks[i].num);
+      }
+   }
+   template<typename T> void Broadcast(T* values, intptr_t num, int root) const{
+      std::vector<Chunk> chunks;
+      intptr_t tag=0;
+      this->SplitMessage2Chunks(chunks, tag, values, num);
+      for(intptr_t i=0; i<chunks.size(); i++){
+         broadcast(*this->communicator, &values[chunks[i].first], chunks[i].num, root);
+      }
+   }
 private:
    static MpiProcess* mpiProcess;
    MpiProcess();
@@ -39,6 +58,31 @@ private:
    ~MpiProcess();
    boost::mpi::environment*  environment;
    boost::mpi::communicator* communicator;
+   struct Chunk{int tag; intptr_t first; int num;};
+   double messageLimit;
+   template<typename T> void SplitMessage2Chunks(std::vector<Chunk>& chunks, const int origianlTag, T* values, intptr_t num) const{
+      if(this->messageLimit < static_cast<double>(sizeof(T))*static_cast<double>(num) ){
+         int  elementsLimit = static_cast<intptr_t>(messageLimit/sizeof(T));
+         intptr_t numChunks = num/elementsLimit;
+         int      remaining = num%elementsLimit;
+         if(0 < remaining){
+            numChunks++;
+         }
+         int tagBase = origianlTag*numChunks;
+         for(intptr_t i=0; i<numChunks; i++){
+            int tag = tagBase+i;
+            Chunk chunk = {tag, i*elementsLimit, elementsLimit};
+            chunks.push_back(chunk);
+         }
+         if(0 < remaining){
+            chunks[numChunks-1].num = remaining;
+         }
+      }
+      else{
+         Chunk chunk = {origianlTag, 0, num};
+         chunks.push_back(chunk);
+      }
+   }
 };
 
 }