source: XIOS/dev/dev_ym/XIOS_SERVICES/src/client.cpp @ 1761

Last change on this file since 1761 was 1761, checked in by ymipsl, 5 years ago

implementing first guess for service functionnalities.

YM

  • 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: 29.3 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
19
20namespace xios
21{
22
23    const double serverPublishDefaultTimeout=10;
24
25    MPI_Comm CClient::intraComm ;
26    MPI_Comm CClient::interComm ;
27    MPI_Comm CClient::clientsComm_ ;
28
29    std::list<MPI_Comm> CClient::contextInterComms;
30    int CClient::serverLeader ;
31    bool CClient::is_MPI_Initialized ;
32    int CClient::rank_ = INVALID_RANK;
33    StdOFStream CClient::m_infoStream;
34    StdOFStream CClient::m_errorStream;
35    CPoolRessource* CClient::poolRessource_=nullptr ;
36
37    MPI_Comm& CClient::getInterComm(void)   { return (interComm); }
38     
39///---------------------------------------------------------------
40/*!
41 * \fn void CClient::initialize(const string& codeId, MPI_Comm& localComm, MPI_Comm& returnComm)
42 * Function creates intraComm (CClient::intraComm) for client group with id=codeId and interComm (CClient::interComm) between client and server groups.
43 * \param [in] codeId identity of context.
44 * \param [in/out] localComm local communicator.
45 * \param [in/out] returnComm (intra)communicator of client group.
46 */
47
48    void CClient::initRessources(void)
49    {
50
51 /*     
52      int commRank;
53      MPI_Comm_rank(CXios::globalComm,&commRank) ;
54      if (commRank==0)
55      {
56        ressources.createPool("ioserver1",ressources.getRessourcesSize()/2) ;
57      }
58      else if (commRank==1)
59      {
60        ressources.createPool("ioserver2",ressources.getRessourcesSize()/2) ;
61      }
62  */
63    }
64
65    void CClient::initialize(const string& codeId, MPI_Comm& localComm, MPI_Comm& returnComm)
66    {
67   
68       MPI_Comm clientComm ;
69      // initialize MPI if not initialized
70      int initialized ;
71      MPI_Initialized(&initialized) ;
72      if (initialized) is_MPI_Initialized=true ;
73      else is_MPI_Initialized=false ;
74     
75      MPI_Comm globalComm=CXios::getGlobalComm() ;
76
77      /////////////////////////////////////////
78      ///////////// PART 1 ////////////////////
79      /////////////////////////////////////////
80     
81
82      // localComm isn't given
83      if (localComm == MPI_COMM_NULL)
84      {
85         
86        // don't use OASIS
87        if (!CXios::usingOasis)
88        {
89
90          if (!is_MPI_Initialized)
91          {
92            MPI_Init(NULL, NULL);
93          }
94          CTimer::get("XIOS").resume() ;
95          CTimer::get("XIOS init/finalize",false).resume() ;
96         
97          // split the global communicator
98          // get hash from all model to attribute a unique color (int) and then split to get client communicator
99          // every mpi process of globalComm (MPI_COMM_WORLD) must participate
100
101          int commRank, commSize ;
102          MPI_Comm_rank(globalComm,&commRank) ;
103          MPI_Comm_size(globalComm,&commSize) ;
104
105          std::hash<string> hashString ;
106          size_t hashClient=hashString(codeId) ;
107         
108          size_t* hashAll = new size_t[commSize] ;
109          MPI_Allgather(&hashClient,1,MPI_UNSIGNED_LONG,hashAll,1,MPI_LONG,globalComm) ;
110         
111          int color=0 ;
112          set<size_t> listHash ;
113          for(int i=0 ; i<=commRank ; i++) 
114            if (listHash.count(hashAll[i])==0)
115            {
116              listHash.insert(hashAll[i]) ;
117              color=color+1 ;
118            }
119          delete[] hashAll ;
120
121          MPI_Comm_split(globalComm, color, commRank, &clientComm) ;
122        }
123        else // using oasis to split communicator
124        {
125          if (!is_MPI_Initialized) oasis_init(codeId) ;
126          oasis_get_localcomm(clientComm) ;
127        }
128      }
129      else // localComm is given
130      {
131        MPI_Comm_dup(localComm,&clientComm) ;
132      }
133     
134     
135      /////////////////////////////////////////
136      ///////////// PART 2 ////////////////////
137      /////////////////////////////////////////
138     
139
140      // Create the XIOS communicator for every process which is related
141      // to XIOS, as well on client side as on server side
142     
143      MPI_Comm xiosGlobalComm ;
144      string strIds=CXios::getin<string>("clients_code_id","") ;
145      vector<string> clientsCodeId=splitRegex(strIds,"\\s*,\\s*") ;
146      if (strIds.empty())
147      {
148         // no code Ids given, suppose XIOS initialisation is global           
149         int commRank, commGlobalRank, serverLeader, clientLeader,serverRemoteLeader,clientRemoteLeader ;
150         MPI_Comm splitComm,interComm ;
151         MPI_Comm_rank(globalComm,&commGlobalRank) ;
152         MPI_Comm_split(globalComm, 0, commGlobalRank, &splitComm) ;
153         int splitCommSize, globalCommSize ;
154       
155         MPI_Comm_size(splitComm,&splitCommSize) ;
156         MPI_Comm_size(globalComm,&globalCommSize) ;
157         if (splitCommSize==globalCommSize) // no server
158         {
159           MPI_Comm_dup(globalComm,&xiosGlobalComm) ;
160           CXios::setXiosComm(xiosGlobalComm) ;
161         }
162         else
163         {
164           MPI_Comm_rank(splitComm,&commRank) ;
165           if (commRank==0) clientLeader=commGlobalRank ;
166           else clientLeader=0 ;
167           serverLeader=0 ;
168           MPI_Allreduce(&clientLeader,&clientRemoteLeader,1,MPI_INT,MPI_SUM,globalComm) ;
169           MPI_Allreduce(&serverLeader,&serverRemoteLeader,1,MPI_INT,MPI_SUM,globalComm) ;
170           MPI_Intercomm_create(splitComm, 0, globalComm, serverRemoteLeader,1341,&interComm) ;
171           MPI_Intercomm_merge(interComm,true,&xiosGlobalComm) ;
172           CXios::setXiosComm(xiosGlobalComm) ;
173         }
174      }
175      else
176      {
177
178        xiosGlobalCommByFileExchange(clientComm, codeId) ;
179     
180      }
181
182      int commRank ;
183      MPI_Comm_rank(CXios::getXiosComm(), &commRank) ;
184      MPI_Comm_split(CXios::getXiosComm(),false,commRank, &clientsComm_) ;
185     
186      // is using server or not ?
187      int xiosCommSize, clientsCommSize ; 
188      MPI_Comm_size(CXios::getXiosComm(), &xiosCommSize) ;
189      MPI_Comm_size(clientsComm_, &clientsCommSize) ;
190      if (xiosCommSize==clientsCommSize) CXios::setUsingServer() ;
191      else CXios::setNotUsingServer() ;
192
193
194      CXios::setGlobalRegistry(new CRegistry(clientsComm_)) ;
195      /////////////////////////////////////////
196      ///////////// PART 3 ////////////////////
197      /////////////////////////////////////////
198     
199      CXios::launchDaemonsManager(false) ;
200      poolRessource_ = new CPoolRessource(clientComm, codeId) ;
201
202      /////////////////////////////////////////
203      ///////////// PART 4 ////////////////////
204      /////////////////////////////////////////     
205     
206      // create the services
207/*
208      int commRank ;
209      MPI_Comm_rank(clientComm,&commRank) ;
210      auto contextsManager=CXios::getContextsManager() ;
211   
212      if (commRank==0)
213      {
214        contextsManager->createServerContext(CXios::defaultPoolId, CXios::defaultGathererId, 0, codeId) ;
215      }
216     
217      MPI_Comm interComm ;
218
219      contextsManager->createServerContextIntercomm(CXios::defaultPoolId, CXios::defaultGathererId, 0, codeId, clientComm, interComm) ;
220*/ 
221/*      while (true)
222      {
223
224      }
225*/     
226      returnComm = clientComm ;
227    }
228
229
230    void CClient::xiosGlobalCommByFileExchange(MPI_Comm clientComm, const string& codeId)
231    {
232 
233      MPI_Comm globalComm=CXios::getGlobalComm() ;
234      MPI_Comm xiosGlobalComm ;
235
236      string strIds=CXios::getin<string>("clients_code_id","") ;
237      vector<string> clientsCodeId=splitRegex(strIds,"\\s*,\\s*") ;
238
239      int commRank, globalRank, clientRank, serverRank ;
240      MPI_Comm_rank(clientComm, &commRank) ;
241      MPI_Comm_rank(globalComm, &globalRank) ;
242      string clientFileName("__xios_publisher::"+codeId+"__to_remove__") ;
243           
244      int error ;
245
246      if (commRank==0) // if root process publish name
247      { 
248        std::ofstream ofs (clientFileName, std::ofstream::out);
249        ofs<<globalRank ;
250        ofs.close();
251       
252  // get server root rank
253
254        std::ifstream ifs ;
255        string fileName=("__xios_publisher::"+CXios::xiosCodeId+"__to_remove__") ;
256     
257        double timeout = CXios::getin<double>("server_puplish_timeout",serverPublishDefaultTimeout) ;
258        double time ;
259         
260        do
261        {
262          CTimer::get("server_publish_timeout").resume() ; 
263          ifs.clear() ;
264          ifs.open(fileName, std::ifstream::in) ;
265          CTimer::get("server_publish_timeout").suspend() ;
266        } while (ifs.fail() && CTimer::get("server_publish_timeout").getCumulatedTime()<timeout) ;
267       
268        if (CTimer::get("server_publish_timeout").getCumulatedTime()>=timeout || ifs.fail())
269        {
270          ifs.clear() ;
271          ifs.close() ;
272          ifs.clear() ;
273          error=true ;           
274        }
275        else 
276        {
277          ifs>>serverRank ;
278          ifs.close() ;
279          error=false ;
280        } 
281
282      } 
283     
284      MPI_Bcast(&error,1,MPI_INT,0,clientComm) ;
285     
286      if (error==false)  // you have a server
287      {
288        MPI_Comm intraComm ;
289        MPI_Comm_dup(clientComm,&intraComm) ;
290        MPI_Comm interComm ;
291       
292        int pos=0 ;
293        for(int i=0 ; codeId!=clientsCodeId[i]; i++) pos=pos+1 ;
294
295        bool high=true ;
296        for(int i=pos ; i<clientsCodeId.size(); i++)
297        { 
298          MPI_Intercomm_create(intraComm, 0, globalComm, serverRank, 3141, &interComm);
299          MPI_Comm_free(&intraComm) ;
300          MPI_Intercomm_merge(interComm,high, &intraComm ) ;
301          high=false ;
302        }
303        xiosGlobalComm=intraComm ;
304      }
305      else  // no server detected
306      {
307        vector<int> clientsRank(clientsCodeId.size()) ;
308       
309        if (commRank==0)
310        { 
311          for(int i=0;i<clientsRank.size();i++)
312          {
313            std::ifstream ifs ;
314            string fileName=("__xios_publisher::"+clientsCodeId[i]+"__to_remove__") ;
315            do
316            {
317              ifs.clear() ;
318              ifs.open(fileName, std::ifstream::in) ;
319            } while (ifs.fail()) ;
320            ifs>>clientsRank[i] ;
321            ifs.close() ;
322          }
323        }
324         
325        int client ;
326        MPI_Comm intraComm ;
327        MPI_Comm_dup(clientComm,&intraComm) ;
328        MPI_Comm interComm ;
329       
330        int pos=0 ;
331        for(int i=0 ; codeId!=clientsCodeId[i]; i++) pos=pos+1 ;
332       
333        bool high=true ;
334        for(int i=pos+1 ; i<clientsCodeId.size(); i++)
335        { 
336          if (codeId==clientsCodeId[0])   // first model play the server rule
337          {         
338            MPI_Intercomm_create(intraComm, 0, globalComm, clientsRank[i], 3141, &interComm);
339            MPI_Intercomm_merge(interComm,false, &intraComm ) ;
340          }
341          else
342          {         
343            MPI_Intercomm_create(intraComm, 0, globalComm, clientsRank[0], 3141, &interComm);
344            MPI_Intercomm_merge(interComm,high, &intraComm ) ;
345            high=false ;
346          }
347        }
348        xiosGlobalComm=intraComm ;
349      }
350
351      MPI_Barrier(xiosGlobalComm);
352      if (commRank==0) std::remove(clientFileName.c_str()) ;         
353      MPI_Barrier(xiosGlobalComm);
354 
355      CXios::setXiosComm(xiosGlobalComm) ;
356
357    }
358
359    void CClient::xiosGlobalCommByPublishing(MPI_Comm clientComm, const string& codeId)
360    {
361
362      // untested. need to be developped an a true MPI compliant library
363
364/*
365        // try to discover other client/server
366        // do you have a xios server ?
367        char portName[MPI_MAX_PORT_NAME];
368        int ierr ;
369        int commRank ;
370        MPI_Comm_rank(clientComm,&commRank) ;
371
372        MPI_Barrier(globalComm) ;
373        if (commRank==0)
374        {
375             
376          MPI_Comm_set_errhandler(MPI_COMM_WORLD, MPI_ERRORS_RETURN );
377          const char* serviceName=CXios::xiosCodeId.c_str() ;
378          ierr=MPI_Lookup_name(CXios::xiosCodeId.c_str(), MPI_INFO_NULL, portName);
379          MPI_Comm_set_errhandler(MPI_COMM_WORLD, MPI_ERRORS_ARE_FATAL );
380        }
381        ierr=MPI_SUCCESS ;
382        MPI_Bcast(&ierr,1,MPI_INT,0,clientComm) ;
383
384        if (ierr==MPI_SUCCESS) // you have a server
385        { 
386          MPI_Comm intraComm=clientComm ;
387          MPI_Comm interComm ;
388          for(int i=0 ; i<clientsCodeId.size(); i++)
389          { 
390            MPI_Comm_connect(portName, MPI_INFO_NULL, 0, intraComm, &interComm);
391            MPI_Intercomm_merge(interComm, true, &intraComm ) ;
392          }
393          xiosGlobalComm=intraComm ;
394        }
395        else  // you don't have any server
396        {
397          if (codeId==clientsCodeId[0]) // first code will publish his name
398          {
399
400            if (commRank==0) // if root process publish name
401            { 
402              MPI_Open_port(MPI_INFO_NULL, portName);
403              MPI_Publish_name(CXios::xiosCodeId.c_str(), MPI_INFO_NULL, portName);
404            }
405
406            MPI_Comm intraComm=clientComm ;
407            MPI_Comm interComm ;
408            for(int i=0 ; i<clientsCodeId.size()-1; i++)
409            { 
410              MPI_Comm_accept(portName, MPI_INFO_NULL, 0, intraComm, &interComm);
411              MPI_Intercomm_merge(interComm,false, &intraComm ) ;
412            }
413          }
414          else  // other clients are connecting to the first one
415          {
416            if (commRank==0)
417            {
418
419              MPI_Comm_set_errhandler(MPI_COMM_WORLD, MPI_ERRORS_RETURN );
420              ierr=MPI_Lookup_name(CXios::xiosCodeId.c_str(), MPI_INFO_NULL, portName);
421              MPI_Comm_set_errhandler(MPI_COMM_WORLD, MPI_ERRORS_ARE_FATAL );
422             }
423
424            MPI_Bcast(&ierr,1,MPI_INT,0,clientComm) ;
425
426            if (ierr==MPI_SUCCESS) // you can connect
427            { 
428              MPI_Comm intraComm=clientComm ;
429              MPI_Comm interComm ;
430              for(int i=0 ; i<clientsCodeId.size()-1; i++)
431              { 
432                MPI_Comm_connect(portName, MPI_INFO_NULL, 0, intraComm, &interComm);
433                MPI_Intercomm_merge(interComm, true, &intraComm ) ;
434              }
435              xiosGlobalComm=intraComm ;
436            }
437          }
438        } 
439      */
440    }
441
442    void CClient::initialize_old(const string& codeId, MPI_Comm& localComm, MPI_Comm& returnComm)
443    {
444      int initialized ;
445      MPI_Initialized(&initialized) ;
446      if (initialized) is_MPI_Initialized=true ;
447      else is_MPI_Initialized=false ;
448      int rank ;
449     
450      CXios::launchRessourcesManager(false) ;
451      CXios::launchServicesManager( false) ;
452      CXios::launchContextsManager(false) ;
453
454      initRessources() ;
455// don't use OASIS
456      if (!CXios::usingOasis)
457      {
458// localComm isn't given
459        if (localComm == MPI_COMM_NULL)
460        {
461          if (!is_MPI_Initialized)
462          {
463            MPI_Init(NULL, NULL);
464          }
465          CTimer::get("XIOS").resume() ;
466          CTimer::get("XIOS init/finalize",false).resume() ;
467          boost::hash<string> hashString ;
468
469          unsigned long hashClient=hashString(codeId) ;
470          unsigned long hashServer=hashString(CXios::xiosCodeId) ;
471          unsigned long* hashAll ;
472          int size ;
473          int myColor ;
474          int i,c ;
475          MPI_Comm newComm ;
476
477          MPI_Comm_size(CXios::globalComm,&size) ;
478          MPI_Comm_rank(CXios::globalComm,&rank_);
479
480          hashAll=new unsigned long[size] ;
481
482          MPI_Allgather(&hashClient,1,MPI_LONG,hashAll,1,MPI_LONG,CXios::globalComm) ;
483
484          map<unsigned long, int> colors ;
485          map<unsigned long, int> leaders ;
486
487          for(i=0,c=0;i<size;i++)
488          {
489            if (colors.find(hashAll[i])==colors.end())
490            {
491              colors[hashAll[i]] =c ;
492              leaders[hashAll[i]]=i ;
493              c++ ;
494            }
495          }
496
497          // Verify whether we are on server mode or not
498          CXios::setNotUsingServer();
499          for (i=0; i < size; ++i)
500          {
501            if (hashServer == hashAll[i])
502            {
503              CXios::setUsingServer();
504              break;
505            }
506          }
507
508          myColor=colors[hashClient];
509          MPI_Comm_split(CXios::globalComm,myColor,rank_,&intraComm) ;
510
511          if (CXios::usingServer)
512          {
513            int clientLeader=leaders[hashClient] ;
514            serverLeader=leaders[hashServer] ;
515            int intraCommSize, intraCommRank ;
516            MPI_Comm_size(intraComm,&intraCommSize) ;
517            MPI_Comm_rank(intraComm,&intraCommRank) ;
518            info(50)<<"intercommCreate::client "<<rank_<<" intraCommSize : "<<intraCommSize
519                   <<" intraCommRank :"<<intraCommRank<<"  clientLeader "<< serverLeader<<endl ;
520             MPI_Intercomm_create(intraComm, 0, CXios::globalComm, serverLeader, 0, &interComm) ;
521             //rank_ = intraCommRank;
522          }
523          else
524          {
525            MPI_Comm_dup(intraComm,&interComm) ;
526          }
527          delete [] hashAll ;
528        }
529        // localComm argument is given
530        else
531        {
532          if (CXios::usingServer)
533          {
534            //ERROR("void CClient::initialize(const string& codeId,MPI_Comm& localComm,MPI_Comm& returnComm)", << " giving a local communictor is not compatible with using server mode") ;
535          }
536          else
537          {
538            MPI_Comm_dup(localComm,&intraComm) ;
539            MPI_Comm_dup(intraComm,&interComm) ;
540          }
541        }
542      }
543      // using OASIS
544      else
545      {
546        // localComm isn't given
547        if (localComm == MPI_COMM_NULL)
548        {
549          if (!is_MPI_Initialized) oasis_init(codeId) ;
550          oasis_get_localcomm(localComm) ;
551        }
552        MPI_Comm_dup(localComm,&intraComm) ;
553
554        CTimer::get("XIOS").resume() ;
555        CTimer::get("XIOS init/finalize",false).resume() ;
556
557        if (CXios::usingServer)
558        {
559          MPI_Status status ;
560          MPI_Comm_rank(intraComm,&rank_) ;
561
562          oasis_get_intercomm(interComm,CXios::xiosCodeId) ;
563          if (rank_==0) MPI_Recv(&serverLeader,1, MPI_INT, 0, 0, interComm, &status) ;
564          MPI_Bcast(&serverLeader,1,MPI_INT,0,intraComm) ;
565        }
566        else MPI_Comm_dup(intraComm,&interComm) ;
567      }
568
569      MPI_Comm_dup(intraComm,&returnComm) ;
570    }
571
572
573
574    void CClient::registerContext(const string& id, MPI_Comm contextComm)
575    {
576      int commRank, commSize ;
577      MPI_Comm_rank(contextComm,&commRank) ;
578      MPI_Comm_size(contextComm,&commSize) ;
579
580      getPoolRessource()->createService(contextComm, id, 0, CServicesManager::CLIENT, 1) ;
581      getPoolRessource()->createService(contextComm, CXios::defaultServerId, 0, CServicesManager::IO_SERVER, 1) ;
582
583      if (commRank==0) while (!CXios::getServicesManager()->hasService(getPoolRessource()->getId(), id, 0)) { CXios::getDaemonsManager()->eventLoop();}
584
585      if (commRank==0) CXios::getContextsManager()->createServerContext(getPoolRessource()->getId(), id, 0, id) ;
586      int type=CServicesManager::CLIENT ;
587      string name = CXios::getContextsManager()->getServerContextName(getPoolRessource()->getId(), id, 0, type, id) ;
588      while (!CXios::getContextsManager()->hasContext(name, contextComm) )
589      {
590        CXios::getDaemonsManager()->eventLoop() ;
591      }
592
593/*     
594
595      CContext::setCurrent(id) ;
596      CContext* context=CContext::create(id);
597     
598      // register the new client side context to the contexts manager
599      if (commRank==0)
600      {
601        MPI_Comm_rank(CXios::getXiosComm(),&commRank) ;
602        SRegisterContextInfo contextInfo ;
603        contextInfo.serviceType=CServicesManager::CLIENT ;
604        contextInfo.partitionId=0 ;
605        contextInfo.leader=commRank ;
606        contextInfo.size=commSize ;
607        CXios::getContextsManager()->registerContext(id, contextInfo) ;
608      }
609      context->initClient(contextComm) ;
610*/ 
611    }
612
613
614///---------------------------------------------------------------
615/*!
616 * \fn void CClient::registerContext(const string& id, MPI_Comm contextComm)
617 * \brief Sends a request to create a context to server. Creates client/server contexts.
618 * \param [in] id id of context.
619 * \param [in] contextComm.
620 * Function is only called by client.
621 */
622    void CClient::registerContext_old(const string& id, MPI_Comm contextComm)
623    {
624      CContext::setCurrent(id) ;
625      CContext* context=CContext::create(id);
626      StdString idServer(id);
627      idServer += "_server";
628
629      if (CXios::isServer && !context->hasServer)
630      // Attached mode
631      {
632        MPI_Comm contextInterComm ;
633        MPI_Comm_dup(contextComm,&contextInterComm) ;
634        CContext* contextServer = CContext::create(idServer);
635
636        // Firstly, initialize context on client side
637        context->initClient(contextComm,contextInterComm, contextServer);
638
639        // Secondly, initialize context on server side
640        contextServer->initServer(contextComm,contextInterComm, context);
641
642        // Finally, we should return current context to context client
643        CContext::setCurrent(id);
644
645        contextInterComms.push_back(contextInterComm);
646      }
647      else
648      {
649        int size,rank,globalRank ;
650        size_t message_size ;
651        int leaderRank ;
652        MPI_Comm contextInterComm ;
653
654        MPI_Comm_size(contextComm,&size) ;
655        MPI_Comm_rank(contextComm,&rank) ;
656        MPI_Comm_rank(CXios::globalComm,&globalRank) ;
657        if (rank!=0) globalRank=0 ;
658
659        CMessage msg ;
660        msg<<idServer<<size<<globalRank ;
661//        msg<<id<<size<<globalRank ;
662
663        int messageSize=msg.size() ;
664        char * buff = new char[messageSize] ;
665        CBufferOut buffer((void*)buff,messageSize) ;
666        buffer<<msg ;
667
668        MPI_Send((void*)buff,buffer.count(),MPI_CHAR,serverLeader,1,CXios::globalComm) ;
669
670        MPI_Intercomm_create(contextComm,0,CXios::globalComm,serverLeader,10+globalRank,&contextInterComm) ;
671        info(10)<<"Register new Context : "<<id<<endl ;
672        MPI_Comm inter ;
673        MPI_Intercomm_merge(contextInterComm,0,&inter) ;
674        MPI_Barrier(inter) ;
675
676        context->initClient(contextComm,contextInterComm) ;
677
678        contextInterComms.push_back(contextInterComm);
679        MPI_Comm_free(&inter);
680        delete [] buff ;
681
682      }
683    }
684
685/*!
686 * \fn void CClient::callOasisEnddef(void)
687 * \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
688 * Function is only called by client.
689 */
690    void CClient::callOasisEnddef(void)
691    {
692      bool oasisEnddef=CXios::getin<bool>("call_oasis_enddef",true) ;
693      if (!oasisEnddef) ERROR("void CClient::callOasisEnddef(void)", <<"Function xios_oasis_enddef called but variable <call_oasis_enddef> is set to false."<<endl
694                                                                     <<"Variable <call_oasis_enddef> must be set to true"<<endl) ;
695      if (CXios::isServer)
696      // Attached mode
697      {
698        // nothing to do   
699      }
700      else
701      {
702        int rank ;
703        int msg=0 ;
704
705        MPI_Comm_rank(intraComm,&rank) ;
706        if (rank==0) 
707        {
708          MPI_Send(&msg,1,MPI_INT,0,5,interComm) ; // tags oasis_endded = 5
709        }
710
711      }
712    }
713
714    void CClient::finalize(void)
715    {
716     
717      MPI_Barrier(clientsComm_) ;
718      int commRank ;
719      MPI_Comm_rank(clientsComm_, &commRank) ;
720      if (commRank==0) CXios::getRessourcesManager()->finalize() ;
721     
722      auto globalRegistry=CXios::getGlobalRegistry() ;
723      globalRegistry->hierarchicalGatherRegistry() ;
724
725      if (commRank==0)
726      {
727        info(80)<<"Write data base Registry"<<endl<<globalRegistry->toString()<<endl ;
728        globalRegistry->toFile("xios_registry.bin") ;
729      }
730      delete globalRegistry ;
731
732      CTimer::get("XIOS init/finalize",false).suspend() ;
733      CTimer::get("XIOS").suspend() ;
734      if (!is_MPI_Initialized)
735      {
736        if (CXios::usingOasis) oasis_finalize();
737        else MPI_Finalize() ;
738      }
739     
740      info(20) << "Client side context is finalized"<<endl ;
741      report(0) <<" Performance report : Whole time from XIOS init and finalize: "<< CTimer::get("XIOS init/finalize").getCumulatedTime()<<" s"<<endl ;
742      report(0) <<" Performance report : total time spent for XIOS : "<< CTimer::get("XIOS").getCumulatedTime()<<" s"<<endl ;
743      report(0)<< " Performance report : time spent for waiting free buffer : "<< CTimer::get("Blocking time").getCumulatedTime()<<" s"<<endl ;
744      report(0)<< " Performance report : Ratio : "<< CTimer::get("Blocking time").getCumulatedTime()/CTimer::get("XIOS init/finalize").getCumulatedTime()*100.<<" %"<<endl ;
745      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 ;
746//      report(0)<< " Memory report : Current buffer_size : "<<CXios::bufferSize<<endl ;
747      report(0)<< " Memory report : Minimum buffer size required : " << CClientBuffer::maxRequestSize << " bytes" << endl ;
748      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 ;
749      report(100)<<CTimer::getAllCumulatedTime()<<endl ;
750    }
751   
752
753    void CClient::finalize_old(void)
754    {
755      int rank ;
756      int msg=0 ;
757
758      MPI_Comm_rank(intraComm,&rank) ;
759 
760      if (!CXios::isServer)
761      {
762        MPI_Comm_rank(intraComm,&rank) ;
763        if (rank==0)
764        {
765          MPI_Send(&msg,1,MPI_INT,0,0,interComm) ;
766        }
767      }
768
769      for (std::list<MPI_Comm>::iterator it = contextInterComms.begin(); it != contextInterComms.end(); it++)
770        MPI_Comm_free(&(*it));
771      MPI_Comm_free(&interComm);
772      MPI_Comm_free(&intraComm);
773
774      CTimer::get("XIOS init/finalize",false).suspend() ;
775      CTimer::get("XIOS").suspend() ;
776
777      if (!is_MPI_Initialized)
778      {
779        if (CXios::usingOasis) oasis_finalize();
780        else MPI_Finalize() ;
781      }
782     
783      info(20) << "Client side context is finalized"<<endl ;
784      report(0) <<" Performance report : Whole time from XIOS init and finalize: "<< CTimer::get("XIOS init/finalize").getCumulatedTime()<<" s"<<endl ;
785      report(0) <<" Performance report : total time spent for XIOS : "<< CTimer::get("XIOS").getCumulatedTime()<<" s"<<endl ;
786      report(0)<< " Performance report : time spent for waiting free buffer : "<< CTimer::get("Blocking time").getCumulatedTime()<<" s"<<endl ;
787      report(0)<< " Performance report : Ratio : "<< CTimer::get("Blocking time").getCumulatedTime()/CTimer::get("XIOS init/finalize").getCumulatedTime()*100.<<" %"<<endl ;
788      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 ;
789//      report(0)<< " Memory report : Current buffer_size : "<<CXios::bufferSize<<endl ;
790      report(0)<< " Memory report : Minimum buffer size required : " << CClientBuffer::maxRequestSize << " bytes" << endl ;
791      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 ;
792      report(100)<<CTimer::getAllCumulatedTime()<<endl ;
793   }
794
795    /*!
796    * Return global rank without oasis and current rank in model intraComm in case of oasis
797    */
798   int CClient::getRank()
799   {
800     return rank_;
801   }
802
803    /*!
804    * Open a file specified by a suffix and an extension and use it for the given file buffer.
805    * The file name will be suffix+rank+extension.
806    *
807    * \param fileName[in] protype file name
808    * \param ext [in] extension of the file
809    * \param fb [in/out] the file buffer
810    */
811    void CClient::openStream(const StdString& fileName, const StdString& ext, std::filebuf* fb)
812    {
813      StdStringStream fileNameClient;
814      int numDigit = 0;
815      int size = 0;
816      int rank;
817      MPI_Comm_size(CXios::getGlobalComm(), &size);
818      MPI_Comm_rank(CXios::getGlobalComm(),&rank);
819      while (size)
820      {
821        size /= 10;
822        ++numDigit;
823      }
824
825      fileNameClient << fileName << "_" << std::setfill('0') << std::setw(numDigit) << rank << ext;
826
827      fb->open(fileNameClient.str().c_str(), std::ios::out);
828      if (!fb->is_open())
829        ERROR("void CClient::openStream(const StdString& fileName, const StdString& ext, std::filebuf* fb)",
830              << std::endl << "Can not open <" << fileNameClient.str() << "> file to write the client log(s).");
831    }
832
833    /*!
834    * \brief Open a file stream to write the info logs
835    * Open a file stream with a specific file name suffix+rank
836    * to write the info logs.
837    * \param fileName [in] protype file name
838    */
839    void CClient::openInfoStream(const StdString& fileName)
840    {
841      std::filebuf* fb = m_infoStream.rdbuf();
842      openStream(fileName, ".out", fb);
843
844      info.write2File(fb);
845      report.write2File(fb);
846    }
847
848    //! Write the info logs to standard output
849    void CClient::openInfoStream()
850    {
851      info.write2StdOut();
852      report.write2StdOut();
853    }
854
855    //! Close the info logs file if it opens
856    void CClient::closeInfoStream()
857    {
858      if (m_infoStream.is_open()) m_infoStream.close();
859    }
860
861    /*!
862    * \brief Open a file stream to write the error log
863    * Open a file stream with a specific file name suffix+rank
864    * to write the error log.
865    * \param fileName [in] protype file name
866    */
867    void CClient::openErrorStream(const StdString& fileName)
868    {
869      std::filebuf* fb = m_errorStream.rdbuf();
870      openStream(fileName, ".err", fb);
871
872      error.write2File(fb);
873    }
874
875    //! Write the error log to standard error output
876    void CClient::openErrorStream()
877    {
878      error.write2StdErr();
879    }
880
881    //! Close the error log file if it opens
882    void CClient::closeErrorStream()
883    {
884      if (m_errorStream.is_open()) m_errorStream.close();
885    }
886}
Note: See TracBrowser for help on using the repository browser.