source: XIOS/trunk/src/client.cpp @ 491

Last change on this file since 491 was 491, checked in by mhnguyen, 8 years ago

Removing using_server from iodef.xml file. From now on, XIOS is capable of dectectin automatically server mode

+) Check whether server is in the MPI global group
+) Add some more methods to set/unset using_server
+) Do some cleaning stuff

Test
+) On Curie
+) All test passed correctly without using_server id

  • Property svn:eol-style set to native
File size: 8.1 KB
Line 
1#include "globalScopeData.hpp"
2#include "xmlioserver_spl.hpp"
3#include "cxios.hpp"
4#include "client.hpp"
5#include <boost/functional/hash.hpp>
6#include "type.hpp"
7#include "context.hpp"
8#include "context_client.hpp"
9#include "oasis_cinterface.hpp"
10#include "mpi.hpp"
11#include "timer.hpp"
12#include "buffer_client.hpp"
13
14namespace xios
15{
16
17    MPI_Comm CClient::intraComm ;
18    MPI_Comm CClient::interComm ;
19    int CClient::serverLeader ;
20    bool CClient::is_MPI_Initialized ;
21    int CClient::rank = INVALID_RANK;
22    StdOFStream CClient::m_infoStream;
23
24    void CClient::initialize(const string& codeId,MPI_Comm& localComm,MPI_Comm& returnComm)
25    {
26      int initialized ;
27      MPI_Initialized(&initialized) ;
28      if (initialized) is_MPI_Initialized=true ;
29      else is_MPI_Initialized=false ;
30
31// don't use OASIS
32      if (!CXios::usingOasis)
33      {
34// localComm doesn't given
35        if (localComm == MPI_COMM_NULL)
36        {
37          if (!is_MPI_Initialized)
38          {
39            int argc=0;
40            char** argv=NULL;
41            MPI_Init(&argc,&argv) ;
42          }
43          CTimer::get("XIOS").resume() ;
44          CTimer::get("XIOS init").resume() ;
45          boost::hash<string> hashString ;
46
47          unsigned long hashClient=hashString(codeId) ;
48          unsigned long hashServer=hashString(CXios::xiosCodeId) ;
49          unsigned long* hashAll ;
50          int size ;
51          int myColor ;
52          int i,c ;
53          MPI_Comm newComm ;
54
55          MPI_Comm_size(CXios::globalComm,&size) ;
56          MPI_Comm_rank(CXios::globalComm,&rank);
57
58          hashAll=new unsigned long[size] ;
59
60          MPI_Allgather(&hashClient,1,MPI_LONG,hashAll,1,MPI_LONG,CXios::globalComm) ;
61
62          map<unsigned long, int> colors ;
63          map<unsigned long, int> leaders ;
64
65          for(i=0,c=0;i<size;i++)
66          {
67            if (colors.find(hashAll[i])==colors.end())
68            {
69              colors[hashAll[i]] =c ;
70              leaders[hashAll[i]]=i ;
71              c++ ;
72            }
73          }
74
75          // Verify whether we are on server mode or not
76          CXios::setNotUsingServer();
77          for (i=0; i < size; ++i)
78          {
79            if (hashServer == hashAll[i])
80            {
81              CXios::setUsingServer();
82              break;
83            }
84          }
85
86          myColor=colors[hashClient] ;
87
88          MPI_Comm_split(CXios::globalComm,myColor,rank,&intraComm) ;
89
90          if (CXios::usingServer)
91          {
92            int clientLeader=leaders[hashClient] ;
93            serverLeader=leaders[hashServer] ;
94            MPI_Intercomm_create(intraComm,0,CXios::globalComm,serverLeader,0,&interComm) ;
95          }
96          else
97          {
98            MPI_Comm_dup(intraComm,&interComm) ;
99          }
100          delete [] hashAll ;
101        }
102        // localComm argument is given
103        else
104        {
105          if (CXios::usingServer)
106          {
107            //ERROR("void CClient::initialize(const string& codeId,MPI_Comm& localComm,MPI_Comm& returnComm)", << " giving a local communictor is not compatible with using server mode") ;
108          }
109          else
110          {
111            MPI_Comm_dup(localComm,&intraComm) ;
112            MPI_Comm_dup(intraComm,&interComm) ;
113          }
114        }
115      }
116      // using OASIS
117      else
118      {
119        // localComm doesn't given
120        if (localComm == MPI_COMM_NULL)
121        {
122          if (!is_MPI_Initialized) oasis_init(codeId) ;
123          oasis_get_localcomm(intraComm) ;
124        }
125        else MPI_Comm_dup(localComm,&intraComm) ;
126        CTimer::get("XIOS").resume() ;
127        CTimer::get("XIOS init").resume() ;
128
129          // Verify whether we are on server mode or not
130        CXios::setNotUsingServer();
131        int interCommSize = 0, intraCommSize = 0;
132        oasis_get_intercomm(interComm,CXios::xiosCodeId);
133        MPI_Comm_size(interComm, &interCommSize);
134        MPI_Comm_size(intraComm, &intraCommSize);
135        if (interCommSize == intraCommSize) CXios::setUsingServer();
136
137        if (CXios::usingServer)
138        {
139          MPI_Status status ;
140          MPI_Comm_rank(intraComm,&rank) ;
141          oasis_get_intercomm(interComm,CXios::xiosCodeId) ;
142          if (rank==0) MPI_Recv(&serverLeader,1, MPI_INT, 0, 0, interComm, &status) ;
143          MPI_Bcast(&serverLeader,1,MPI_INT,0,intraComm) ;
144
145        }
146        else MPI_Comm_dup(intraComm,&interComm) ;
147      }
148
149      MPI_Comm_dup(intraComm,&returnComm) ;
150    }
151
152
153    void CClient::registerContext(const string& id,MPI_Comm contextComm)
154    {
155      CContext::setCurrent(id) ;
156      CContext* context=CContext::create(id) ;
157
158      if (!CXios::isServer)
159      {
160        int size,rank,globalRank ;
161        size_t message_size ;
162        int leaderRank ;
163        MPI_Comm contextInterComm ;
164
165        MPI_Comm_size(contextComm,&size) ;
166        MPI_Comm_rank(contextComm,&rank) ;
167        MPI_Comm_rank(CXios::globalComm,&globalRank) ;
168        if (rank!=0) globalRank=0 ;
169
170
171        CMessage msg ;
172        msg<<id<<size<<globalRank ;
173
174        int messageSize=msg.size() ;
175        void * buff = new char[messageSize] ;
176        CBufferOut buffer(buff,messageSize) ;
177        buffer<<msg ;
178
179        MPI_Send(buff,buffer.count(),MPI_CHAR,serverLeader,1,CXios::globalComm) ;
180        delete [] buff ;
181
182        MPI_Intercomm_create(contextComm,0,CXios::globalComm,serverLeader,10+globalRank,&contextInterComm) ;
183        info(10)<<"Register new Context : "<<id<<endl ;
184
185        MPI_Comm inter ;
186        MPI_Intercomm_merge(contextInterComm,0,&inter) ;
187        MPI_Barrier(inter) ;
188
189        context->initClient(contextComm,contextInterComm) ;
190      }
191      else
192      {
193        MPI_Comm contextInterComm ;
194        MPI_Comm_dup(contextComm,&contextInterComm) ;
195        context->initClient(contextComm,contextInterComm) ;
196        context->initServer(contextComm,contextInterComm) ;
197      }
198    }
199
200    void CClient::finalize(void)
201    {
202      int rank ;
203      int msg=0 ;
204      if (!CXios::isServer)
205      {
206        MPI_Comm_rank(intraComm,&rank) ;
207        if (rank==0)
208        {
209          MPI_Send(&msg,1,MPI_INT,0,0,interComm) ;
210        }
211      }
212
213     CTimer::get("XIOS finalize").suspend() ;
214     CTimer::get("XIOS").suspend() ;
215
216      if (!is_MPI_Initialized)
217      {
218        if (CXios::usingOasis) oasis_finalize();
219        else MPI_Finalize() ;
220      }
221      info(20) << "Client side context is finalized"<<endl ;
222      report(0) <<" Performance report : total time spent for XIOS : "<< CTimer::get("XIOS").getCumulatedTime()<<" s"<<endl ;
223      report(0)<< " Performance report : time spent for waiting free buffer : "<< CTimer::get("Blocking time").getCumulatedTime()<<" s"<<endl ;
224      report(0)<< " Performance report : Ratio : "<< CTimer::get("Blocking time").getCumulatedTime()/CTimer::get("XIOS").getCumulatedTime()*100.<<" %"<<endl ;
225      report(0)<< " Performance report : This ratio must be close to zero. Otherwise it may be usefull to increase buffer size or numbers of server"<<endl ;
226      report(0)<< " Memory report : Current buffer_size : "<<CXios::bufferSize<<endl ;
227      report(0)<< " Memory report : Minimum buffer size required : "<<maxRequestSize*2<<endl ;
228      report(0)<< " Memory report : increasing it by a factor will increase performance, depending of the volume of data wrote in file at each time step of the file"<<endl ;
229   }
230
231   int CClient::getRank()
232   {
233     return rank;
234   }
235
236     /*!
237      * \brief Open file stream to write in
238      *   Opening a file stream with a specific file name suffix-client+rank
239      * \param [in] protype file name
240     */
241     void CClient::openInfoStream(const StdString& fileName)
242     {
243       std::filebuf* fb = m_infoStream.rdbuf();
244       StdStringStream fileNameClient;
245       fileNameClient << fileName <<"_client_" << getRank() << ".out";
246       fb->open(fileNameClient.str().c_str(), std::ios::out);
247       if (!fb->is_open())
248       ERROR("void CClient::openInfoStream(const StdString& fileName)",
249            <<endl<< "Can not open <"<<fileNameClient<<"> file to write" );
250
251       info.write2File(fb);
252     }
253
254     //! Write out to standard output
255     void CClient::openInfoStream()
256     {
257       info.write2StdOut();
258     }
259
260     //! Close file if it opens
261     void CClient::closeInfoStream()
262     {
263       if (m_infoStream.is_open()) m_infoStream.close();
264     }
265
266
267}
Note: See TracBrowser for help on using the repository browser.