source: XIOS/dev/dev_ym/XIOS_COUPLING/src/client.cpp @ 2332

Last change on this file since 2332 was 2332, checked in by jderouillat, 2 years ago

Suffix intraComm and interComm members of CClient and CServer with _ (some occurrences of these varaiables exist locally).

  • 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: 20.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#include "string_tools.hpp"
14#include "ressources_manager.hpp"
15#include "services_manager.hpp"
16#include <functional>
17#include <cstdio>
18#include "workflow_graph.hpp"
19#include "release_static_allocation.hpp"
20
21namespace xios
22{
23
24    const double serverPublishDefaultTimeout=10;
25
26    MPI_Comm CClient::intraComm_ ;
27    MPI_Comm CClient::interComm_ ;
28    MPI_Comm CClient::clientsComm_ ;
29
30    std::list<MPI_Comm> CClient::contextInterComms;
31    int CClient::serverLeader ;
32    bool CClient::is_MPI_Initialized ;
33    int CClient::rank_ = INVALID_RANK;
34    StdOFStream CClient::m_infoStream;
35    StdOFStream CClient::m_errorStream;
36    CPoolRessource* CClient::poolRessource_=nullptr ;
37
38    MPI_Comm& CClient::getInterComm(void)   { return (interComm_); }
39     
40///---------------------------------------------------------------
41/*!
42 * \fn void CClient::initialize(const string& codeId, MPI_Comm& localComm, MPI_Comm& returnComm)
43 * Function creates intraComm (CClient::intraComm) for client group with id=codeId and interComm (CClient::interComm) between client and server groups.
44 * \param [in] codeId identity of context.
45 * \param [in/out] localComm local communicator.
46 * \param [in/out] returnComm (intra)communicator of client group.
47 */
48
49    void CClient::initialize(const string& codeId, MPI_Comm& localComm, MPI_Comm& returnComm)
50    {
51   
52       MPI_Comm clientComm ;
53      // initialize MPI if not initialized
54      int initialized ;
55      MPI_Initialized(&initialized) ;
56      if (initialized) is_MPI_Initialized=true ;
57      else is_MPI_Initialized=false ;
58     
59      MPI_Comm globalComm=CXios::getGlobalComm() ;
60
61      /////////////////////////////////////////
62      ///////////// PART 1 ////////////////////
63      /////////////////////////////////////////
64     
65
66      // localComm isn't given
67      if (localComm == MPI_COMM_NULL)
68      {
69         
70        // don't use OASIS
71        if (!CXios::usingOasis)
72        {
73
74          if (!is_MPI_Initialized)
75          {
76            MPI_Init(NULL, NULL);
77          }
78          CTimer::get("XIOS").resume() ;
79          CTimer::get("XIOS init/finalize",false).resume() ;
80         
81          // split the global communicator
82          // get hash from all model to attribute a unique color (int) and then split to get client communicator
83          // every mpi process of globalComm (MPI_COMM_WORLD) must participate
84
85          int commRank, commSize ;
86          MPI_Comm_rank(globalComm,&commRank) ;
87          MPI_Comm_size(globalComm,&commSize) ;
88
89          std::hash<string> hashString ;
90          size_t hashClient=hashString(codeId) ;
91         
92          size_t* hashAll = new size_t[commSize] ;
93          MPI_Allgather(&hashClient,1,MPI_SIZE_T,hashAll,1,MPI_SIZE_T,globalComm) ;
94         
95          int color=0 ;
96          map<size_t,int> listHash ;
97          for(int i=0 ; i<=commSize ; i++) 
98            if (listHash.count(hashAll[i])==0) 
99            {
100              listHash[hashAll[i]]=color ;
101              color=color+1 ;
102            }
103            color=listHash[hashClient] ;
104          delete[] hashAll ;
105
106          MPI_Comm_split(globalComm, color, commRank, &clientComm) ;
107        }
108        else // using oasis to split communicator
109        {
110          if (!is_MPI_Initialized) oasis_init(codeId) ;
111          oasis_get_localcomm(clientComm) ;
112        }
113      }
114      else // localComm is given
115      {
116        MPI_Comm_dup(localComm,&clientComm) ;
117      }
118     
119     
120      /////////////////////////////////////////
121      ///////////// PART 2 ////////////////////
122      /////////////////////////////////////////
123     
124
125      // Create the XIOS communicator for every process which is related
126      // to XIOS, as well on client side as on server side
127     
128      MPI_Comm xiosGlobalComm ;
129      string strIds=CXios::getin<string>("clients_code_id","") ;
130      vector<string> clientsCodeId=splitRegex(strIds,"\\s*,\\s*") ;
131      if (strIds.empty())
132      {
133         // no code Ids given, suppose XIOS initialisation is global           
134         int commRank, commGlobalRank, serverLeader, clientLeader,serverRemoteLeader,clientRemoteLeader ;
135         MPI_Comm splitComm,interComm ;
136         MPI_Comm_rank(globalComm,&commGlobalRank) ;
137         MPI_Comm_split(globalComm, 0, commGlobalRank, &splitComm) ;
138         int splitCommSize, globalCommSize ;
139       
140         MPI_Comm_size(splitComm,&splitCommSize) ;
141         MPI_Comm_size(globalComm,&globalCommSize) ;
142         if (splitCommSize==globalCommSize) // no server
143         {
144           MPI_Comm_dup(globalComm,&xiosGlobalComm) ;
145           CXios::setXiosComm(xiosGlobalComm) ;
146         }
147         else
148         {
149           MPI_Comm_rank(splitComm,&commRank) ;
150           if (commRank==0) clientLeader=commGlobalRank ;
151           else clientLeader=0 ;
152           serverLeader=0 ;
153           MPI_Allreduce(&clientLeader,&clientRemoteLeader,1,MPI_INT,MPI_SUM,globalComm) ;
154           MPI_Allreduce(&serverLeader,&serverRemoteLeader,1,MPI_INT,MPI_SUM,globalComm) ;
155           MPI_Intercomm_create(splitComm, 0, globalComm, serverRemoteLeader,1341,&interComm) ;
156           MPI_Intercomm_merge(interComm,true,&xiosGlobalComm) ;
157           CXios::setXiosComm(xiosGlobalComm) ;
158         }
159      }
160      else
161      {
162
163        xiosGlobalCommByFileExchange(clientComm, codeId) ;
164     
165      }
166
167      int commRank ;
168      MPI_Comm_rank(CXios::getXiosComm(), &commRank) ;
169      MPI_Comm_split(CXios::getXiosComm(),false,commRank, &clientsComm_) ;
170     
171      // is using server or not ?
172      int xiosCommSize, clientsCommSize ; 
173      MPI_Comm_size(CXios::getXiosComm(), &xiosCommSize) ;
174      MPI_Comm_size(clientsComm_, &clientsCommSize) ;
175      if (xiosCommSize==clientsCommSize) CXios::setUsingServer() ;
176      else CXios::setNotUsingServer() ;
177
178      /////////////////////////////////////////
179      ///////////// PART 3 ////////////////////
180      /////////////////////////////////////////
181     
182      CXios::launchDaemonsManager(false) ;
183      poolRessource_ = new CPoolRessource(clientComm, codeId) ;
184
185      /////////////////////////////////////////
186      ///////////// PART 4 ////////////////////
187      /////////////////////////////////////////     
188     
189      returnComm = clientComm ;
190    }
191
192
193    void CClient::xiosGlobalCommByFileExchange(MPI_Comm clientComm, const string& codeId)
194    {
195 
196      MPI_Comm globalComm=CXios::getGlobalComm() ;
197      MPI_Comm xiosGlobalComm ;
198
199      string strIds=CXios::getin<string>("clients_code_id","") ;
200      vector<string> clientsCodeId=splitRegex(strIds,"\\s*,\\s*") ;
201
202      int commRank, globalRank, clientRank, serverRank ;
203      MPI_Comm_rank(clientComm, &commRank) ;
204      MPI_Comm_rank(globalComm, &globalRank) ;
205      string clientFileName("__xios_publisher::"+codeId+"__to_remove__") ;
206           
207      int error ;
208
209      if (commRank==0) // if root process publish name
210      { 
211        std::ofstream ofs (clientFileName, std::ofstream::out);
212        ofs<<globalRank ;
213        ofs.close();
214       
215  // get server root rank
216
217        std::ifstream ifs ;
218        string fileName=("__xios_publisher::"+CXios::xiosCodeId+"__to_remove__") ;
219     
220        double timeout = CXios::getin<double>("server_puplish_timeout",serverPublishDefaultTimeout) ;
221        double time ;
222         
223        do
224        {
225          CTimer::get("server_publish_timeout").resume() ; 
226          ifs.clear() ;
227          ifs.open(fileName, std::ifstream::in) ;
228          CTimer::get("server_publish_timeout").suspend() ;
229        } while (ifs.fail() && CTimer::get("server_publish_timeout").getCumulatedTime()<timeout) ;
230       
231        if (CTimer::get("server_publish_timeout").getCumulatedTime()>=timeout || ifs.fail())
232        {
233          ifs.clear() ;
234          ifs.close() ;
235          ifs.clear() ;
236          error=true ;           
237        }
238        else 
239        {
240          ifs>>serverRank ;
241          ifs.close() ;
242          error=false ;
243        } 
244
245      } 
246     
247      MPI_Bcast(&error,1,MPI_INT,0,clientComm) ;
248     
249      if (error==false)  // you have a server
250      {
251        MPI_Comm intraComm ;
252        MPI_Comm_dup(clientComm,&intraComm) ;
253        MPI_Comm interComm ;
254       
255        int pos=0 ;
256        for(int i=0 ; codeId!=clientsCodeId[i]; i++) pos=pos+1 ;
257
258        bool high=true ;
259        for(int i=pos ; i<clientsCodeId.size(); i++)
260        { 
261          MPI_Intercomm_create(intraComm, 0, globalComm, serverRank, 3141, &interComm);
262          MPI_Comm_free(&intraComm) ;
263          MPI_Intercomm_merge(interComm,high, &intraComm ) ;
264          high=false ;
265        }
266        xiosGlobalComm=intraComm ;
267      }
268      else  // no server detected
269      {
270        vector<int> clientsRank(clientsCodeId.size()) ;
271       
272        if (commRank==0)
273        { 
274          for(int i=0;i<clientsRank.size();i++)
275          {
276            std::ifstream ifs ;
277            string fileName=("__xios_publisher::"+clientsCodeId[i]+"__to_remove__") ;
278            do
279            {
280              ifs.clear() ;
281              ifs.open(fileName, std::ifstream::in) ;
282            } while (ifs.fail()) ;
283            ifs>>clientsRank[i] ;
284            ifs.close() ;
285          }
286        }
287         
288        int client ;
289        MPI_Comm intraComm ;
290        MPI_Comm_dup(clientComm,&intraComm) ;
291        MPI_Comm interComm ;
292       
293        int pos=0 ;
294        for(int i=0 ; codeId!=clientsCodeId[i]; i++) pos=pos+1 ;
295       
296        bool high=true ;
297        for(int i=pos+1 ; i<clientsCodeId.size(); i++)
298        { 
299          if (codeId==clientsCodeId[0])   // first model play the server rule
300          {         
301            MPI_Intercomm_create(intraComm, 0, globalComm, clientsRank[i], 3141, &interComm);
302            MPI_Intercomm_merge(interComm,false, &intraComm ) ;
303          }
304          else
305          {         
306            MPI_Intercomm_create(intraComm, 0, globalComm, clientsRank[0], 3141, &interComm);
307            MPI_Intercomm_merge(interComm,high, &intraComm ) ;
308            high=false ;
309          }
310        }
311        xiosGlobalComm=intraComm ;
312      }
313
314      MPI_Barrier(xiosGlobalComm);
315      if (commRank==0) std::remove(clientFileName.c_str()) ;         
316      MPI_Barrier(xiosGlobalComm);
317 
318      CXios::setXiosComm(xiosGlobalComm) ;
319
320      MPI_Comm commUnfree ;
321      MPI_Comm_dup(clientComm, &commUnfree ) ;
322 
323    }
324
325// to check on other architecture
326    void CClient::xiosGlobalCommByPublishing(MPI_Comm clientComm, const string& codeId)
327    {
328
329      // untested. need to be developped an a true MPI compliant library
330
331/*
332        // try to discover other client/server
333        // do you have a xios server ?
334        char portName[MPI_MAX_PORT_NAME];
335        int ierr ;
336        int commRank ;
337        MPI_Comm_rank(clientComm,&commRank) ;
338
339        MPI_Barrier(globalComm) ;
340        if (commRank==0)
341        {
342             
343          MPI_Comm_set_errhandler(MPI_COMM_WORLD, MPI_ERRORS_RETURN );
344          const char* serviceName=CXios::xiosCodeId.c_str() ;
345          ierr=MPI_Lookup_name(CXios::xiosCodeId.c_str(), MPI_INFO_NULL, portName);
346          MPI_Comm_set_errhandler(MPI_COMM_WORLD, MPI_ERRORS_ARE_FATAL );
347        }
348        ierr=MPI_SUCCESS ;
349        MPI_Bcast(&ierr,1,MPI_INT,0,clientComm) ;
350
351        if (ierr==MPI_SUCCESS) // you have a server
352        { 
353          MPI_Comm intraComm=clientComm ;
354          MPI_Comm interComm ;
355          for(int i=0 ; i<clientsCodeId.size(); i++)
356          { 
357            MPI_Comm_connect(portName, MPI_INFO_NULL, 0, intraComm, &interComm);
358            MPI_Intercomm_merge(interComm, true, &intraComm ) ;
359          }
360          xiosGlobalComm=intraComm ;
361        }
362        else  // you don't have any server
363        {
364          if (codeId==clientsCodeId[0]) // first code will publish his name
365          {
366
367            if (commRank==0) // if root process publish name
368            { 
369              MPI_Open_port(MPI_INFO_NULL, portName);
370              MPI_Publish_name(CXios::xiosCodeId.c_str(), MPI_INFO_NULL, portName);
371            }
372
373            MPI_Comm intraComm=clientComm ;
374            MPI_Comm interComm ;
375            for(int i=0 ; i<clientsCodeId.size()-1; i++)
376            { 
377              MPI_Comm_accept(portName, MPI_INFO_NULL, 0, intraComm, &interComm);
378              MPI_Intercomm_merge(interComm,false, &intraComm ) ;
379            }
380          }
381          else  // other clients are connecting to the first one
382          {
383            if (commRank==0)
384            {
385
386              MPI_Comm_set_errhandler(MPI_COMM_WORLD, MPI_ERRORS_RETURN );
387              ierr=MPI_Lookup_name(CXios::xiosCodeId.c_str(), MPI_INFO_NULL, portName);
388              MPI_Comm_set_errhandler(MPI_COMM_WORLD, MPI_ERRORS_ARE_FATAL );
389             }
390
391            MPI_Bcast(&ierr,1,MPI_INT,0,clientComm) ;
392
393            if (ierr==MPI_SUCCESS) // you can connect
394            { 
395              MPI_Comm intraComm=clientComm ;
396              MPI_Comm interComm ;
397              for(int i=0 ; i<clientsCodeId.size()-1; i++)
398              { 
399                MPI_Comm_connect(portName, MPI_INFO_NULL, 0, intraComm, &interComm);
400                MPI_Intercomm_merge(interComm, true, &intraComm ) ;
401              }
402              xiosGlobalComm=intraComm ;
403            }
404          }
405        } 
406      */
407    }
408
409
410///---------------------------------------------------------------
411/*!
412 * \fn void CClient::registerContext(const string& id, MPI_Comm contextComm)
413 * \brief Sends a request to create a context to server. Creates client/server contexts.
414 * \param [in] id id of context.
415 * \param [in] contextComm.
416 * Function is only called by client.
417 */
418    void CClient::registerContext(const string& id, MPI_Comm contextComm)
419    {
420      int commRank, commSize ;
421      MPI_Comm_rank(contextComm,&commRank) ;
422      MPI_Comm_size(contextComm,&commSize) ;
423
424      getPoolRessource()->createService(contextComm, id, 0, CServicesManager::CLIENT, 1) ;
425      getPoolRessource()->createService(contextComm, id+"_"+CXios::defaultServerId, 0, CServicesManager::IO_SERVER, 1) ;
426
427      if (commRank==0) while (!CXios::getServicesManager()->hasService(getPoolRessource()->getId(), id, 0)) { CXios::getDaemonsManager()->eventLoop();}
428
429      if (commRank==0) CXios::getContextsManager()->createServerContext(getPoolRessource()->getId(), id, 0, id) ;
430      int type=CServicesManager::CLIENT ;
431      string name = CXios::getContextsManager()->getServerContextName(getPoolRessource()->getId(), id, 0, type, id) ;
432      double time ;
433      double lastTime=0 ;
434      double latency=0 ;
435      bool out=false ;
436      while (!out)
437      {
438        time=MPI_Wtime() ;
439        if (time-lastTime > latency) 
440        {
441          out=CXios::getContextsManager()->hasContext(name, contextComm);
442          lastTime=time ;
443        }
444        if (!out) CXios::getDaemonsManager()->eventLoop() ;
445      }
446
447    }
448
449
450
451/*!
452 * \fn void CClient::callOasisEnddef(void)
453 * \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
454 * Function is only called by client.
455 */
456    void CClient::callOasisEnddef(void)
457    {
458      bool oasisEnddef=CXios::getin<bool>("call_oasis_enddef",true) ;
459      if (!oasisEnddef) ERROR("void CClient::callOasisEnddef(void)", <<"Function xios_oasis_enddef called but variable <call_oasis_enddef> is set to false."<<endl
460                                                                     <<"Variable <call_oasis_enddef> must be set to true"<<endl) ;
461      if (CXios::isServer)
462      // Attached mode
463      {
464        // nothing to do   
465      }
466      else
467      {
468        int rank ;
469        int msg=0 ;
470
471        MPI_Comm_rank(intraComm_,&rank) ;
472        if (rank==0) 
473        {
474          MPI_Send(&msg,1,MPI_INT,0,5,interComm_) ; // tags oasis_endded = 5
475        }
476
477      }
478    }
479
480    void CClient::finalize(void)
481    {
482     
483      MPI_Barrier(clientsComm_) ;
484      int commRank ;
485      MPI_Comm_rank(clientsComm_, &commRank) ;
486      if (commRank==0) CXios::getRessourcesManager()->finalize() ;
487     
488      CTimer::get("XIOS init/finalize",false).suspend() ;
489      CTimer::get("XIOS").suspend() ;
490      CXios::finalizeDaemonsManager() ;
491      finalizePoolRessource() ;
492      CContext::removeAllContexts() ; // free memory for related context
493
494      CXios::getMpiGarbageCollector().release() ; // release unfree MPI ressources
495      if (!is_MPI_Initialized)
496      {
497        if (CXios::usingOasis) oasis_finalize();
498        else MPI_Finalize() ;
499      }
500     
501      info(20) << "Client side context is finalized"<<endl ;
502      report(0) <<" Performance report : Whole time from XIOS init and finalize: "<< CTimer::get("XIOS init/finalize").getCumulatedTime()<<" s"<<endl ;
503      report(0) <<" Performance report : total time spent for XIOS : "<< CTimer::get("XIOS").getCumulatedTime()<<" s"<<endl ;
504      report(0)<< " Performance report : time spent for waiting free buffer : "<< CTimer::get("Blocking time").getCumulatedTime()<<" s"<<endl ;
505      report(0)<< " Performance report : Ratio : "<< CTimer::get("Blocking time").getCumulatedTime()/CTimer::get("XIOS init/finalize").getCumulatedTime()*100.<<" %"<<endl ;
506      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 ;
507//      report(0)<< " Memory report : Current buffer_size : "<<CXios::bufferSize<<endl ;
508      report(0)<< " Memory report : Minimum buffer size required : " << CClientBuffer::maxRequestSize << " bytes" << endl ;
509      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 ;
510      report(100)<<CTimer::getAllCumulatedTime()<<endl ;
511      CWorkflowGraph::drawWorkFlowGraph_client();
512
513      xios::releaseStaticAllocation() ;
514
515    }
516   
517    void CClient::finalizePoolRessource() 
518    { 
519      delete poolRessource_ ; poolRessource_=nullptr ;
520    }
521
522    /*!
523    * Return global rank without oasis and current rank in model intraComm in case of oasis
524    */
525   int CClient::getRank()
526   {
527     return rank_;
528   }
529
530    /*!
531    * Open a file specified by a suffix and an extension and use it for the given file buffer.
532    * The file name will be suffix+rank+extension.
533    *
534    * \param fileName[in] protype file name
535    * \param ext [in] extension of the file
536    * \param fb [in/out] the file buffer
537    */
538    void CClient::openStream(const StdString& fileName, const StdString& ext, std::filebuf* fb)
539    {
540      StdStringStream fileNameClient;
541      int numDigit = 0;
542      int size = 0;
543      int rank;
544      MPI_Comm_size(CXios::getGlobalComm(), &size);
545      MPI_Comm_rank(CXios::getGlobalComm(),&rank);
546      while (size)
547      {
548        size /= 10;
549        ++numDigit;
550      }
551
552      fileNameClient << fileName << "_" << std::setfill('0') << std::setw(numDigit) << rank << ext;
553
554      fb->open(fileNameClient.str().c_str(), std::ios::out);
555      if (!fb->is_open())
556        ERROR("void CClient::openStream(const StdString& fileName, const StdString& ext, std::filebuf* fb)",
557              << std::endl << "Can not open <" << fileNameClient.str() << "> file to write the client log(s).");
558    }
559
560    /*!
561    * \brief Open a file stream to write the info logs
562    * Open a file stream with a specific file name suffix+rank
563    * to write the info logs.
564    * \param fileName [in] protype file name
565    */
566    void CClient::openInfoStream(const StdString& fileName)
567    {
568      std::filebuf* fb = m_infoStream.rdbuf();
569      openStream(fileName, ".out", fb);
570
571      info.write2File(fb);
572      report.write2File(fb);
573    }
574
575    //! Write the info logs to standard output
576    void CClient::openInfoStream()
577    {
578      info.write2StdOut();
579      report.write2StdOut();
580    }
581
582    //! Close the info logs file if it opens
583    void CClient::closeInfoStream()
584    {
585      if (m_infoStream.is_open()) m_infoStream.close();
586    }
587
588    /*!
589    * \brief Open a file stream to write the error log
590    * Open a file stream with a specific file name suffix+rank
591    * to write the error log.
592    * \param fileName [in] protype file name
593    */
594    void CClient::openErrorStream(const StdString& fileName)
595    {
596      std::filebuf* fb = m_errorStream.rdbuf();
597      openStream(fileName, ".err", fb);
598
599      error.write2File(fb);
600    }
601
602    //! Write the error log to standard error output
603    void CClient::openErrorStream()
604    {
605      error.write2StdErr();
606    }
607
608    //! Close the error log file if it opens
609    void CClient::closeErrorStream()
610    {
611      if (m_errorStream.is_open()) m_errorStream.close();
612    }
613}
Note: See TracBrowser for help on using the repository browser.