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

Last change on this file since 925 was 925, checked in by rlacroix, 5 years ago

Fix MPI_Init calls to be more compliant with the standard.

  • Property copyright set to
    Software name : XIOS (Xml I/O Server)
    http://forge.ipsl.jussieu.fr/ioserver
    Creation date : January 2009
    Licence : CeCCIL version2
    see license file in root directory : Licence_CeCILL_V2-en.txt
    or http://www.cecill.info/licences/Licence_CeCILL_V2-en.html
    Holder : CEA/LSCE (Laboratoire des Sciences du CLimat et de l'Environnement)
    CNRS/IPSL (Institut Pierre Simon Laplace)
    Project Manager : Yann Meurdesoif
    yann.meurdesoif@cea.fr
  • Property svn:eol-style set to native
File size: 10.4 KB
Line 
1#include "globalScopeData.hpp"
2#include "xios_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    std::list<MPI_Comm> CClient::contextInterComms;
20    int CClient::serverLeader ;
21    bool CClient::is_MPI_Initialized ;
22    int CClient::rank = INVALID_RANK;
23    StdOFStream CClient::m_infoStream;
24    StdOFStream CClient::m_errorStream;
25
26    void CClient::initialize(const string& codeId,MPI_Comm& localComm,MPI_Comm& returnComm)
27    {
28      int initialized ;
29      MPI_Initialized(&initialized) ;
30      if (initialized) is_MPI_Initialized=true ;
31      else is_MPI_Initialized=false ;
32
33// don't use OASIS
34      if (!CXios::usingOasis)
35      {
36// localComm doesn't given
37        if (localComm == MPI_COMM_NULL)
38        {
39          if (!is_MPI_Initialized)
40          {
41            MPI_Init(NULL, NULL);
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
95            int intraCommSize, intraCommRank ;
96            MPI_Comm_size(intraComm,&intraCommSize) ;
97            MPI_Comm_rank(intraComm,&intraCommRank) ;
98            info(50)<<"intercommCreate::client "<<rank<<" intraCommSize : "<<intraCommSize
99                 <<" intraCommRank :"<<intraCommRank<<"  clientLeader "<< serverLeader<<endl ;
100            MPI_Intercomm_create(intraComm,0,CXios::globalComm,serverLeader,0,&interComm) ;
101          }
102          else
103          {
104            MPI_Comm_dup(intraComm,&interComm) ;
105          }
106          delete [] hashAll ;
107        }
108        // localComm argument is given
109        else
110        {
111          if (CXios::usingServer)
112          {
113            //ERROR("void CClient::initialize(const string& codeId,MPI_Comm& localComm,MPI_Comm& returnComm)", << " giving a local communictor is not compatible with using server mode") ;
114          }
115          else
116          {
117            MPI_Comm_dup(localComm,&intraComm) ;
118            MPI_Comm_dup(intraComm,&interComm) ;
119          }
120        }
121      }
122      // using OASIS
123      else
124      {
125        // localComm doesn't given
126        if (localComm == MPI_COMM_NULL)
127        {
128          if (!is_MPI_Initialized) oasis_init(codeId) ;
129          oasis_get_localcomm(localComm) ;
130        }
131        MPI_Comm_dup(localComm,&intraComm) ;
132
133        CTimer::get("XIOS").resume() ;
134        CTimer::get("XIOS init").resume() ;
135
136        if (CXios::usingServer)
137        {
138          MPI_Status status ;
139          MPI_Comm_rank(intraComm,&rank) ;
140
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      StdString idServer(id);
158      idServer += "_server";
159
160      if (!CXios::isServer)
161      {
162        int size,rank,globalRank ;
163        size_t message_size ;
164        int leaderRank ;
165        MPI_Comm contextInterComm ;
166
167        MPI_Comm_size(contextComm,&size) ;
168        MPI_Comm_rank(contextComm,&rank) ;
169        MPI_Comm_rank(CXios::globalComm,&globalRank) ;
170        if (rank!=0) globalRank=0 ;
171
172
173        CMessage msg ;
174        msg<<idServer<<size<<globalRank ;
175//        msg<<id<<size<<globalRank ;
176
177        int messageSize=msg.size() ;
178        void * buff = new char[messageSize] ;
179        CBufferOut buffer(buff,messageSize) ;
180        buffer<<msg ;
181
182        MPI_Send(buff,buffer.count(),MPI_CHAR,serverLeader,1,CXios::globalComm) ;
183        delete [] buff ;
184
185        MPI_Intercomm_create(contextComm,0,CXios::globalComm,serverLeader,10+globalRank,&contextInterComm) ;
186        info(10)<<"Register new Context : "<<id<<endl ;
187
188        MPI_Comm inter ;
189        MPI_Intercomm_merge(contextInterComm,0,&inter) ;
190        MPI_Barrier(inter) ;
191
192        context->initClient(contextComm,contextInterComm) ;
193
194        contextInterComms.push_back(contextInterComm);
195        MPI_Comm_free(&inter);
196      }
197      else
198      {
199        MPI_Comm contextInterComm ;
200        MPI_Comm_dup(contextComm,&contextInterComm) ;
201        CContext* contextServer = CContext::create(idServer);
202
203        // Firstly, initialize context on client side
204        context->initClient(contextComm,contextInterComm, contextServer);
205
206        // Secondly, initialize context on server side
207        contextServer->initServer(contextComm,contextInterComm, context);
208
209        // Finally, we should return current context to context client
210        CContext::setCurrent(id);
211
212        contextInterComms.push_back(contextInterComm);
213      }
214    }
215
216    void CClient::finalize(void)
217    {
218      int rank ;
219      int msg=0 ;
220
221      MPI_Comm_rank(intraComm,&rank) ;
222 
223      if (!CXios::isServer)
224      {
225        MPI_Comm_rank(intraComm,&rank) ;
226        if (rank==0)
227        {
228          MPI_Send(&msg,1,MPI_INT,0,0,interComm) ;
229        }
230      }
231
232      for (std::list<MPI_Comm>::iterator it = contextInterComms.begin(); it != contextInterComms.end(); it++)
233        MPI_Comm_free(&(*it));
234      MPI_Comm_free(&interComm);
235      MPI_Comm_free(&intraComm);
236
237      CTimer::get("XIOS finalize").suspend() ;
238      CTimer::get("XIOS").suspend() ;
239
240      if (!is_MPI_Initialized)
241      {
242        if (CXios::usingOasis) oasis_finalize();
243        else MPI_Finalize() ;
244      }
245     
246      info(20) << "Client side context is finalized"<<endl ;
247      report(0) <<" Performance report : total time spent for XIOS : "<< CTimer::get("XIOS").getCumulatedTime()<<" s"<<endl ;
248      report(0)<< " Performance report : time spent for waiting free buffer : "<< CTimer::get("Blocking time").getCumulatedTime()<<" s"<<endl ;
249      report(0)<< " Performance report : Ratio : "<< CTimer::get("Blocking time").getCumulatedTime()/CTimer::get("XIOS").getCumulatedTime()*100.<<" %"<<endl ;
250      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 ;
251//      report(0)<< " Memory report : Current buffer_size : "<<CXios::bufferSize<<endl ;
252      report(0)<< " Memory report : Minimum buffer size required : " << CClientBuffer::maxRequestSize << " bytes" << endl ;
253      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 ;
254   }
255
256   int CClient::getRank()
257   {
258     return rank;
259   }
260
261    /*!
262    * Open a file specified by a suffix and an extension and use it for the given file buffer.
263    * The file name will be suffix+rank+extension.
264    *
265    * \param fileName[in] protype file name
266    * \param ext [in] extension of the file
267    * \param fb [in/out] the file buffer
268    */
269    void CClient::openStream(const StdString& fileName, const StdString& ext, std::filebuf* fb)
270    {
271      StdStringStream fileNameClient;
272      int numDigit = 0;
273      int size = 0;
274      MPI_Comm_size(CXios::globalComm, &size);
275      while (size)
276      {
277        size /= 10;
278        ++numDigit;
279      }
280
281      fileNameClient << fileName << "_" << std::setfill('0') << std::setw(numDigit) << getRank() << ext;
282      fb->open(fileNameClient.str().c_str(), std::ios::out);
283      if (!fb->is_open())
284        ERROR("void CClient::openStream(const StdString& fileName, const StdString& ext, std::filebuf* fb)",
285              << std::endl << "Can not open <" << fileNameClient << "> file to write the client log(s).");
286    }
287
288    /*!
289    * \brief Open a file stream to write the info logs
290    * Open a file stream with a specific file name suffix+rank
291    * to write the info logs.
292    * \param fileName [in] protype file name
293    */
294    void CClient::openInfoStream(const StdString& fileName)
295    {
296      std::filebuf* fb = m_infoStream.rdbuf();
297      openStream(fileName, ".out", fb);
298
299      info.write2File(fb);
300      report.write2File(fb);
301    }
302
303    //! Write the info logs to standard output
304    void CClient::openInfoStream()
305    {
306      info.write2StdOut();
307      report.write2StdOut();
308    }
309
310    //! Close the info logs file if it opens
311    void CClient::closeInfoStream()
312    {
313      if (m_infoStream.is_open()) m_infoStream.close();
314    }
315
316    /*!
317    * \brief Open a file stream to write the error log
318    * Open a file stream with a specific file name suffix+rank
319    * to write the error log.
320    * \param fileName [in] protype file name
321    */
322    void CClient::openErrorStream(const StdString& fileName)
323    {
324      std::filebuf* fb = m_errorStream.rdbuf();
325      openStream(fileName, ".err", fb);
326
327      error.write2File(fb);
328    }
329
330    //! Write the error log to standard error output
331    void CClient::openErrorStream()
332    {
333      error.write2StdErr();
334    }
335
336    //! Close the error log file if it opens
337    void CClient::closeErrorStream()
338    {
339      if (m_errorStream.is_open()) m_errorStream.close();
340    }
341}
Note: See TracBrowser for help on using the repository browser.