source: CONFIG_DEVT/IPSLCM6.5_work_ENSEMBLES/modeles/XIOS/src/client.cpp @ 5861

Last change on this file since 5861 was 5861, checked in by ymipsl, 3 years ago
  • Adapt XIOS sources to manage ensemble runs, using new version of oasis

YM
File size: 16.8 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#include "string_tools.hpp"
14
15namespace xios
16{
17
18    MPI_Comm CClient::intraComm ;
19    MPI_Comm CClient::interComm ;
20    std::list<MPI_Comm> CClient::contextInterComms;
21    int CClient::serverLeader ;
22    bool CClient::is_MPI_Initialized ;
23    int CClient::rank_ = INVALID_RANK;
24    StdOFStream CClient::m_infoStream;
25    StdOFStream CClient::m_errorStream;
26    MPI_Comm& CClient::getInterComm(void)   { return (interComm); }
27     
28///---------------------------------------------------------------
29/*!
30 * \fn void CClient::initialize(const string& codeId, MPI_Comm& localComm, MPI_Comm& returnComm)
31 * Function creates intraComm (CClient::intraComm) for client group with id=codeId and interComm (CClient::interComm) between client and server groups.
32 * \param [in] codeId identity of context.
33 * \param [in/out] localComm local communicator.
34 * \param [in/out] returnComm (intra)communicator of client group.
35 */
36
37    void CClient::initialize(const string& codeId, MPI_Comm& localComm, MPI_Comm& returnComm)
38    {
39      int initialized ;
40      MPI_Initialized(&initialized) ;
41      if (initialized) is_MPI_Initialized=true ;
42      else is_MPI_Initialized=false ;
43      int rank ;
44      bool ensembleManagement ;
45      int ensembleSize ;
46
47// don't use OASIS
48      if (!CXios::usingOasis)
49      {
50// localComm isn't given
51        if (localComm == MPI_COMM_NULL)
52        {
53          if (!is_MPI_Initialized)
54          {
55            MPI_Init(NULL, NULL);
56          }
57          CTimer::get("XIOS").resume() ;
58          CTimer::get("XIOS init/finalize").resume() ;
59          boost::hash<string> hashString ;
60
61          unsigned long hashClient=hashString(codeId) ;
62          unsigned long hashServer=hashString(CXios::xiosCodeId) ;
63          unsigned long* hashAll ;
64          int size ;
65          int myColor ;
66          int i,c ;
67          MPI_Comm newComm ;
68
69          MPI_Comm_size(CXios::globalComm,&size) ;
70          MPI_Comm_rank(CXios::globalComm,&rank_);
71
72          hashAll=new unsigned long[size] ;
73
74          MPI_Allgather(&hashClient,1,MPI_LONG,hashAll,1,MPI_LONG,CXios::globalComm) ;
75
76          map<unsigned long, int> colors ;
77          map<unsigned long, int> leaders ;
78
79          for(i=0,c=0;i<size;i++)
80          {
81            if (colors.find(hashAll[i])==colors.end())
82            {
83              colors[hashAll[i]] =c ;
84              leaders[hashAll[i]]=i ;
85              c++ ;
86            }
87          }
88
89          // Verify whether we are on server mode or not
90          CXios::setNotUsingServer();
91          for (i=0; i < size; ++i)
92          {
93            if (hashServer == hashAll[i])
94            {
95              CXios::setUsingServer();
96              break;
97            }
98          }
99
100          myColor=colors[hashClient];
101          MPI_Comm_split(CXios::globalComm,myColor,rank_,&intraComm) ;
102
103          if (CXios::usingServer)
104          {
105            int clientLeader=leaders[hashClient] ;
106            serverLeader=leaders[hashServer] ;
107            int intraCommSize, intraCommRank ;
108            MPI_Comm_size(intraComm,&intraCommSize) ;
109            MPI_Comm_rank(intraComm,&intraCommRank) ;
110            info(50)<<"intercommCreate::client "<<rank_<<" intraCommSize : "<<intraCommSize
111                   <<" intraCommRank :"<<intraCommRank<<"  clientLeader "<< serverLeader<<endl ;
112             MPI_Intercomm_create(intraComm, 0, CXios::globalComm, serverLeader, 0, &interComm) ;
113             //rank_ = intraCommRank;
114          }
115          else
116          {
117            MPI_Comm_dup(intraComm,&interComm) ;
118          }
119          delete [] hashAll ;
120        }
121        // localComm argument is given
122        else
123        {
124          if (CXios::usingServer)
125          {
126            //ERROR("void CClient::initialize(const string& codeId,MPI_Comm& localComm,MPI_Comm& returnComm)", << " giving a local communictor is not compatible with using server mode") ;
127          }
128          else
129          {
130            MPI_Comm_dup(localComm,&intraComm) ;
131            MPI_Comm_dup(intraComm,&interComm) ;
132          }
133        }
134      }
135      // using OASIS
136      else
137      {
138
139        // manage ensemble with oasis
140        // code id is splitted follow the base code Id and the num number using the ":" separator
141        // codeId == baseCodeId:member
142        // member must integer convertible
143        vector<string> splittedCodeId=splitRegex(codeId,"\\s*:\\s*") ;
144        if (splittedCodeId.size()>2) ERROR("void CClient::initialize(const string& codeId, MPI_Comm& localComm, MPI_Comm& returnComm)",
145                                            <<"CodeId is badly formed "<<codeId) ;
146        string baseCodeId=splittedCodeId[0] ;
147        int member ;
148        if (splittedCodeId.size()==2)
149        {
150          try
151          {
152            member = stoi(splittedCodeId[1]) ;
153          } 
154          catch(...)
155          {
156            ERROR("void CClient::initialize(const string& codeId, MPI_Comm& localComm, MPI_Comm& returnComm)",
157                                            <<"CodeId is badly formed "<<codeId) ;
158          }
159          ensembleManagement=true ;
160        }
161        else ensembleManagement=false ;
162
163        string codesId=CXios::getin<string>("oasis_codes_id") ;
164        vector<string> oasisCodesId=splitRegex(codesId,"\\s*,\\s*") ;
165        for(auto& oasisCodeId : oasisCodesId)
166        {
167          info(10)<<"oasisCodeId "<<oasisCodeId<<endl ;
168          vector<string> splittedOasisCodeId = splitRegex(oasisCodeId,"\\s*:\\s*") ;
169          string baseOasisCodeId=splittedOasisCodeId[0] ;
170          info(10)<<"baseOasisCodeId "<<baseOasisCodeId<<"  baseCodeId "<<baseCodeId<<endl ;
171          if (baseOasisCodeId==baseCodeId)
172          {
173            if (splittedOasisCodeId.size()==2)
174            {
175              try
176              {
177                ensembleSize = stoi(splittedOasisCodeId[1]) ;
178                info(10)<<"ensembleSize "<<ensembleSize<<endl ;
179              } 
180              catch(...)
181              {
182                ERROR("void CClient::initialize(const string& codeId, MPI_Comm& localComm, MPI_Comm& returnComm)",
183                                                <<"CodeId is badly formed "<<codeId) ;
184              }
185              if (ensembleManagement==false)  ERROR("void CClient::initialize(const string& codeId, MPI_Comm& localComm, MPI_Comm& returnComm)",
186                                                <<"inconsistancy for ensemble management between local codeId and OASIS codes Id  : "
187                                                <<codeId<<"  "<< codesId); 
188            }
189            else if (ensembleManagement==true)  ERROR("void CClient::initialize(const string& codeId, MPI_Comm& localComm, MPI_Comm& returnComm)",
190                                                <<"inconsistancy for ensemble management between local codeId and OASIS codes Id "
191                                                <<codeId<<"  "<< codesId); 
192          }
193        }
194       
195       
196        // localComm isn't given
197        if (localComm == MPI_COMM_NULL)
198        {
199          if (!is_MPI_Initialized) oasis_init(codeId) ;
200          if(ensembleManagement) 
201          {
202            vector<string> intraCommStr(ensembleSize) ;
203            vector<int>    rootRanks ;
204            for(int i=0;i<ensembleSize;i++) intraCommStr[i]=baseCodeId+":"+std::to_string(i) ;
205            oasis_get_multi_intracomm(localComm, intraCommStr, rootRanks) ;
206          }
207          else oasis_get_localcomm(localComm) ;
208        }
209        MPI_Comm_dup(localComm,&intraComm) ;
210
211        CTimer::get("XIOS").resume() ;
212        CTimer::get("XIOS init/finalize").resume() ;
213
214        if (CXios::usingServer)
215        {
216          MPI_Status status ;
217          MPI_Comm_rank(intraComm,&rank_) ;
218         
219          if(ensembleManagement) 
220          {
221            vector<string> intraCommStr(ensembleSize+1) ;
222            vector<int>    rootRanks ;
223            for(int i=0;i<ensembleSize;i++) intraCommStr[i]=baseCodeId+":"+std::to_string(i) ;
224            intraCommStr[ensembleSize] = CXios::xiosCodeId ;
225           
226            oasis_get_multi_intracomm(localComm, intraCommStr, rootRanks) ;
227   
228            int myColor= 0 ; // 0:client 1:server
229            int myrank ;
230            MPI_Comm splittedComm ;
231            MPI_Comm_rank(localComm,&myrank) ;
232            MPI_Comm_split(localComm,myColor,myrank, &splittedComm) ;
233            MPI_Intercomm_create(splittedComm,0,localComm,rootRanks[ensembleSize],10,&interComm) ;
234         
235          }
236          else oasis_get_intercomm(interComm,CXios::xiosCodeId) ;
237          if (rank_==0) MPI_Recv(&serverLeader,1, MPI_INT, 0, 0, interComm, &status) ;
238          MPI_Bcast(&serverLeader,1,MPI_INT,0,intraComm) ;
239        }
240        else MPI_Comm_dup(intraComm,&interComm) ;
241      }
242
243      MPI_Comm_dup(intraComm,&returnComm) ;
244    }
245
246///---------------------------------------------------------------
247/*!
248 * \fn void CClient::registerContext(const string& id, MPI_Comm contextComm)
249 * \brief Sends a request to create a context to server. Creates client/server contexts.
250 * \param [in] id id of context.
251 * \param [in] contextComm.
252 * Function is only called by client.
253 */
254    void CClient::registerContext(const string& id, MPI_Comm contextComm)
255    {
256      CContext::setCurrent(id) ;
257      CContext* context=CContext::create(id);
258      StdString idServer(id);
259      idServer += "_server";
260
261      if (CXios::isServer && !context->hasServer)
262      // Attached mode
263      {
264        MPI_Comm contextInterComm ;
265        MPI_Comm_dup(contextComm,&contextInterComm) ;
266        CContext* contextServer = CContext::create(idServer);
267
268        // Firstly, initialize context on client side
269        context->initClient(contextComm,contextInterComm, contextServer);
270
271        // Secondly, initialize context on server side
272        contextServer->initServer(contextComm,contextInterComm, context);
273
274        // Finally, we should return current context to context client
275        CContext::setCurrent(id);
276
277        contextInterComms.push_back(contextInterComm);
278      }
279      else
280      {
281        int size,rank,globalRank ;
282        size_t message_size ;
283        int leaderRank ;
284        MPI_Comm contextInterComm ;
285
286        MPI_Comm_size(contextComm,&size) ;
287        MPI_Comm_rank(contextComm,&rank) ;
288        MPI_Comm_rank(CXios::globalComm,&globalRank) ;
289        if (rank!=0) globalRank=0 ;
290
291        CMessage msg ;
292        msg<<idServer<<size<<globalRank ;
293//        msg<<id<<size<<globalRank ;
294
295        int messageSize=msg.size() ;
296        char * buff = new char[messageSize] ;
297        CBufferOut buffer((void*)buff,messageSize) ;
298        buffer<<msg ;
299
300        MPI_Send((void*)buff,buffer.count(),MPI_CHAR,serverLeader,1,CXios::globalComm) ;
301
302        MPI_Intercomm_create(contextComm,0,CXios::globalComm,serverLeader,10+globalRank,&contextInterComm) ;
303        info(10)<<"Register new Context : "<<id<<endl ;
304        MPI_Comm inter ;
305        MPI_Intercomm_merge(contextInterComm,0,&inter) ;
306        MPI_Barrier(inter) ;
307        MPI_Comm_free(&inter);
308
309        context->initClient(contextComm,contextInterComm) ;
310
311        contextInterComms.push_back(contextInterComm);
312        delete [] buff ;
313
314      }
315    }
316
317/*!
318 * \fn void CClient::callOasisEnddef(void)
319 * \brief Send the order to the servers to call "oasis_enddef". It must be done by each compound of models before calling oasis_enddef on client side
320 * Function is only called by client.
321 */
322    void CClient::callOasisEnddef(void)
323    {
324      bool oasisEnddef=CXios::getin<bool>("call_oasis_enddef",true) ;
325      if (!oasisEnddef) ERROR("void CClient::callOasisEnddef(void)", <<"Function xios_oasis_enddef called but variable <call_oasis_enddef> is set to false."<<endl
326                                                                     <<"Variable <call_oasis_enddef> must be set to true"<<endl) ;
327      if (CXios::isServer)
328      // Attached mode
329      {
330        // nothing to do   
331      }
332      else
333      {
334        int rank ;
335        int msg=0 ;
336
337        MPI_Comm_rank(intraComm,&rank) ;
338        if (rank==0) 
339        {
340          MPI_Send(&msg,1,MPI_INT,0,5,interComm) ; // tags oasis_endded = 5
341        }
342
343      }
344    }
345
346
347    void CClient::finalize(void)
348    {
349      int rank ;
350      int msg=0 ;
351
352      MPI_Comm_rank(intraComm,&rank) ;
353 
354      if (!CXios::isServer)
355      {
356        MPI_Comm_rank(intraComm,&rank) ;
357        if (rank==0)
358        {
359          MPI_Send(&msg,1,MPI_INT,0,0,interComm) ;
360        }
361      }
362
363
364      MPI_Comm_free(&interComm);
365      for (std::list<MPI_Comm>::iterator it = contextInterComms.begin(); it != contextInterComms.end(); it++)
366        MPI_Comm_free(&(*it));
367      MPI_Comm_free(&intraComm);
368
369      CTimer::get("XIOS init/finalize").suspend() ;
370      CTimer::get("XIOS").suspend() ;
371
372      if (!is_MPI_Initialized)
373      {
374        if (CXios::usingOasis) oasis_finalize();
375        else MPI_Finalize() ;
376      }
377     
378      info(20) << "Client side context is finalized"<<endl ;
379      report(0) <<" Performance report : Whole time from XIOS init and finalize: "<< CTimer::get("XIOS init/finalize").getCumulatedTime()<<" s"<<endl ;
380      report(0) <<" Performance report : total time spent for XIOS : "<< CTimer::get("XIOS").getCumulatedTime()<<" s"<<endl ;
381      report(0)<< " Performance report : time spent for waiting free buffer : "<< CTimer::get("Blocking time").getCumulatedTime()<<" s"<<endl ;
382      report(0)<< " Performance report : Ratio : "<< CTimer::get("Blocking time").getCumulatedTime()/CTimer::get("XIOS init/finalize").getCumulatedTime()*100.<<" %"<<endl ;
383      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 ;
384//      report(0)<< " Memory report : Current buffer_size : "<<CXios::bufferSize<<endl ;
385      report(0)<< " Memory report : Minimum buffer size required : " << CClientBuffer::maxRequestSize << " bytes" << endl ;
386      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 ;
387      report(100)<<CTimer::getAllCumulatedTime()<<endl ;
388   }
389
390    /*!
391    * Return global rank without oasis and current rank in model intraComm in case of oasis
392    */
393   int CClient::getRank()
394   {
395     return rank_;
396   }
397
398    /*!
399    * Open a file specified by a suffix and an extension and use it for the given file buffer.
400    * The file name will be suffix+rank+extension.
401    *
402    * \param fileName[in] protype file name
403    * \param ext [in] extension of the file
404    * \param fb [in/out] the file buffer
405    */
406    void CClient::openStream(const StdString& fileName, const StdString& ext, std::filebuf* fb)
407    {
408      StdStringStream fileNameClient;
409      int numDigit = 0;
410      int size = 0;
411      int rank;
412      MPI_Comm_size(CXios::globalComm, &size);
413      while (size)
414      {
415        size /= 10;
416        ++numDigit;
417      }
418
419      if (CXios::usingOasis)
420      {
421        MPI_Comm_rank(CXios::globalComm,&rank);
422        fileNameClient << fileName << "_" << std::setfill('0') << std::setw(numDigit) << rank << ext;
423      }
424      else
425        fileNameClient << fileName << "_" << std::setfill('0') << std::setw(numDigit) << getRank() << ext;
426
427
428      fb->open(fileNameClient.str().c_str(), std::ios::out);
429      if (!fb->is_open())
430        ERROR("void CClient::openStream(const StdString& fileName, const StdString& ext, std::filebuf* fb)",
431              << std::endl << "Can not open <" << fileNameClient.str() << "> file to write the client log(s).");
432    }
433
434    /*!
435    * \brief Open a file stream to write the info logs
436    * Open a file stream with a specific file name suffix+rank
437    * to write the info logs.
438    * \param fileName [in] protype file name
439    */
440    void CClient::openInfoStream(const StdString& fileName)
441    {
442      std::filebuf* fb = m_infoStream.rdbuf();
443      openStream(fileName, ".out", fb);
444
445      info.write2File(fb);
446      report.write2File(fb);
447    }
448
449    //! Write the info logs to standard output
450    void CClient::openInfoStream()
451    {
452      info.write2StdOut();
453      report.write2StdOut();
454    }
455
456    //! Close the info logs file if it opens
457    void CClient::closeInfoStream()
458    {
459      if (m_infoStream.is_open()) m_infoStream.close();
460    }
461
462    /*!
463    * \brief Open a file stream to write the error log
464    * Open a file stream with a specific file name suffix+rank
465    * to write the error log.
466    * \param fileName [in] protype file name
467    */
468    void CClient::openErrorStream(const StdString& fileName)
469    {
470      std::filebuf* fb = m_errorStream.rdbuf();
471      openStream(fileName, ".err", fb);
472
473      error.write2File(fb);
474    }
475
476    //! Write the error log to standard error output
477    void CClient::openErrorStream()
478    {
479      error.write2StdErr();
480    }
481
482    //! Close the error log file if it opens
483    void CClient::closeErrorStream()
484    {
485      if (m_errorStream.is_open()) m_errorStream.close();
486    }
487}
Note: See TracBrowser for help on using the repository browser.