source: XIOS/dev/branch_yushan/src/server.cpp @ 1110

Last change on this file since 1110 was 1110, checked in by yushan, 7 years ago

redefinition of mpi_any_source and mpi_any_tag

  • 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: 14.8 KB
Line 
1#include "globalScopeData.hpp"
2#include "xios_spl.hpp"
3#include "cxios.hpp"
4#include "server.hpp"
5#include "type.hpp"
6#include "context.hpp"
7#include "object_template.hpp"
8#include "oasis_cinterface.hpp"
9#include <boost/functional/hash.hpp>
10#include <boost/algorithm/string.hpp>
11#include "tracer.hpp"
12#include "timer.hpp"
13#include "event_scheduler.hpp"
14
15namespace xios
16{
17    MPI_Comm CServer::intraComm ;
18    list<MPI_Comm> CServer::interComm ;
19    std::list<MPI_Comm> CServer::contextInterComms;
20    bool CServer::isRoot ;
21    int CServer::rank = INVALID_RANK;
22    StdOFStream CServer::m_infoStream;
23    StdOFStream CServer::m_errorStream;
24    map<string,CContext*> CServer::contextList ;
25    bool CServer::finished=false ;
26    bool CServer::is_MPI_Initialized ;
27
28   
29    CEventScheduler* CServer::eventScheduler = 0;
30   
31    void CServer::initialize(void)
32    {
33      // int initialized ;
34      // MPI_Initialized(&initialized) ;
35      // if (initialized) is_MPI_Initialized=true ;
36      // else is_MPI_Initialized=false ;
37
38      // Not using OASIS
39      if (!CXios::usingOasis)
40      {
41        // if (!is_MPI_Initialized)
42        // {
43        //   MPI_Init(NULL, NULL);
44        // }
45
46        CTimer::get("XIOS").resume() ;
47
48        boost::hash<string> hashString ;
49
50        unsigned long hashServer=hashString(CXios::xiosCodeId) ;
51        unsigned long* hashAll ;
52
53
54        int size ;
55        int myColor ;
56        int i,c ;
57        MPI_Comm newComm ;
58
59        MPI_Comm_size(CXios::globalComm,&size) ;
60        MPI_Comm_rank(CXios::globalComm,&rank);
61        hashAll=new unsigned long[size] ;
62
63        MPI_Allgather(&hashServer,1,MPI_LONG,hashAll,1,MPI_LONG,CXios::globalComm) ;
64
65        map<unsigned long, int> colors ;
66        map<unsigned long, int> leaders ;
67        map<unsigned long, int>::iterator it ;
68
69        for(i=0,c=0;i<size;i++)
70        {
71          if (colors.find(hashAll[i])==colors.end())
72          {
73            colors[hashAll[i]]=c ;
74            leaders[hashAll[i]]=i ;
75            c++ ;
76          }
77        }
78
79        myColor=colors[hashServer] ;
80
81
82        MPI_Comm_split(CXios::globalComm,myColor,rank,&intraComm) ;
83
84       
85        int serverLeader=leaders[hashServer] ;
86        int clientLeader;
87
88         serverLeader=leaders[hashServer] ;
89         for(it=leaders.begin();it!=leaders.end();++it)
90         {
91           if (it->first!=hashServer)
92           {
93             clientLeader=it->second ;
94             int intraCommSize, intraCommRank ;
95             MPI_Comm_size(intraComm,&intraCommSize) ;
96             MPI_Comm_rank(intraComm,&intraCommRank) ;
97             info(50)<<"intercommCreate::server "<<rank<<" intraCommSize : "<<intraCommSize
98                     <<" intraCommRank :"<<intraCommRank<<"  clientLeader "<< clientLeader<<endl ;
99
100             MPI_Intercomm_create(intraComm,0,CXios::globalComm,clientLeader,0,&newComm) ;
101             interComm.push_back(newComm) ;
102           }
103         }
104
105         delete [] hashAll ;
106      }
107      // using OASIS
108      else
109      {
110        int size;
111        if (!is_MPI_Initialized) oasis_init(CXios::xiosCodeId);
112
113        CTimer::get("XIOS").resume() ;
114        MPI_Comm localComm;
115        oasis_get_localcomm(localComm);
116        MPI_Comm_dup(localComm, &intraComm);
117
118        MPI_Comm_rank(intraComm,&rank) ;
119        MPI_Comm_size(intraComm,&size) ;
120        string codesId=CXios::getin<string>("oasis_codes_id") ;
121
122        vector<string> splitted ;
123        boost::split( splitted, codesId, boost::is_any_of(","), boost::token_compress_on ) ;
124        vector<string>::iterator it ;
125
126        MPI_Comm newComm ;
127        int globalRank ;
128        MPI_Comm_rank(CXios::globalComm,&globalRank);
129
130        for(it=splitted.begin();it!=splitted.end();it++)
131        {
132          oasis_get_intercomm(newComm,*it) ;
133          if (rank==0) MPI_Send(&globalRank,1,MPI_INT,0,0,newComm) ;
134          MPI_Comm_remote_size(newComm,&size);
135          interComm.push_back(newComm) ;
136        }
137              oasis_enddef() ;
138      }
139
140      MPI_Comm_rank(intraComm,&rank) ;
141      if (rank==0) isRoot=true;
142      else isRoot=false;
143     
144      eventScheduler = new CEventScheduler(intraComm) ;
145    }
146
147    void CServer::finalize(void)
148    {
149      CTimer::get("XIOS").suspend() ;
150     
151      delete eventScheduler ;
152     
153     
154
155      for (std::list<MPI_Comm>::iterator it = contextInterComms.begin(); it != contextInterComms.end(); ++it)
156        MPI_Comm_free(&(*it));
157
158      for (std::list<MPI_Comm>::iterator it = interComm.begin(); it != interComm.end(); ++it)
159        MPI_Comm_free(&(*it));
160
161      MPI_Comm_free(&intraComm);
162
163      if (!is_MPI_Initialized)
164      {
165        if (CXios::usingOasis) oasis_finalize();
166        //else  {MPI_Finalize() ;}
167      }
168
169     
170      report(0)<<"Performance report : Time spent for XIOS : "<<CTimer::get("XIOS server").getCumulatedTime()<<endl  ;
171      report(0)<<"Performance report : Time spent in processing events : "<<CTimer::get("Process events").getCumulatedTime()<<endl  ;
172      report(0)<<"Performance report : Ratio : "<<CTimer::get("Process events").getCumulatedTime()/CTimer::get("XIOS server").getCumulatedTime()*100.<<"%"<<endl  ;
173    }
174
175     void CServer::eventLoop(void)
176     {
177       bool stop=false ;
178
179       CTimer::get("XIOS server").resume() ;
180       while(!stop)
181       {
182         if (isRoot)
183         {
184           listenContext(); 
185           if (!finished) listenFinalize() ; 
186         }
187         else
188         {
189           listenRootContext(); 
190           if (!finished) 
191           {
192             listenRootFinalize() ; 
193           }
194         }
195         
196         contextEventLoop() ;
197         if (finished && contextList.empty()) stop=true ;
198         
199         eventScheduler->checkEvent() ;
200       }
201       
202       
203       CTimer::get("XIOS server").suspend() ;
204     }
205
206     void CServer::listenFinalize(void)
207     {
208        list<MPI_Comm>::iterator it;
209        int msg ;
210        int flag ;
211       
212
213        for(it=interComm.begin();it!=interComm.end();++it)
214        {
215           MPI_Status status ;
216           traceOff() ;
217           MPI_Iprobe(0,0,*it,&flag,&status) ;
218           traceOn() ;
219           if (flag==true)
220           {
221              MPI_Recv(&msg,1,MPI_INT,0,0,*it,&status) ;
222              info(20)<<" CServer : Receive client finalize"<<endl ;
223
224              MPI_Comm_free(&(*it));
225              interComm.erase(it) ;
226              break ;
227            }
228         }
229
230         if (interComm.empty())
231         {
232           int i,size ;
233           MPI_Comm_size(intraComm,&size) ;
234           MPI_Request* requests= new MPI_Request[size-1] ;
235           MPI_Status* status= new MPI_Status[size-1] ;
236
237           for(int i=1;i<size;i++) MPI_Isend(&msg,1,MPI_INT,i,4,intraComm,&requests[i-1]) ;
238           MPI_Waitall(size-1,requests,status) ;
239
240           finished=true ;
241           delete [] requests ;
242           delete [] status ;
243         }
244     }
245
246
247     void CServer::listenRootFinalize()
248     {
249        int flag ;
250        MPI_Status status ;
251        int msg ;
252       
253        traceOff() ;
254        MPI_Iprobe(0,4,intraComm, &flag, &status) ;
255        traceOn() ;
256        if (flag==true)
257        {
258           MPI_Recv(&msg,1,MPI_INT,0,4,intraComm,&status) ;
259           finished=true ;
260        }
261      }
262
263     void CServer::listenContext(void)
264     {
265
266       MPI_Status status ;
267       int flag = false ;
268       static void* buffer ;
269       //#pragma omp threadprivate(buffer)
270
271       static MPI_Request request ;
272       //#pragma omp threadprivate(request)
273
274       static bool recept=false ;
275       //#pragma omp threadprivate(recept)
276
277       int rank ;
278       int count ; 
279
280       if (recept==false)
281       {     
282         traceOff() ;
283         #ifdef _usingEP
284         MPI_Iprobe(-1,1,CXios::globalComm, &flag, &status) ;
285         #else
286         MPI_Iprobe(MPI_ANY_SOURCE,1,CXios::globalComm, &flag, &status) ;
287         #endif
288         traceOn() ;
289         
290         if (flag==true)
291         {
292           #ifdef _usingMPI
293           rank=status.MPI_SOURCE ;
294           #elif _usingEP
295           rank= status.ep_src ;
296           #endif
297           MPI_Get_count(&status,MPI_CHAR,&count) ;
298           buffer=new char[count] ;
299           MPI_Irecv(buffer,count,MPI_CHAR,rank,1,CXios::globalComm,&request) ;
300           recept=true ;
301         }
302         
303       }
304       else
305       {
306         traceOff() ;
307         MPI_Test(&request,&flag,&status) ;
308         traceOn() ;
309         if (flag==true)
310         {
311           #ifdef _usingMPI
312           rank=status.MPI_SOURCE ;
313           #elif _usingEP
314           rank= status.ep_src ;
315           #endif
316           MPI_Get_count(&status,MPI_CHAR,&count) ;
317           recvContextMessage(buffer,count) ;
318           
319           delete [] buffer ;
320           recept=false ;
321         }
322       }
323     }
324
325     void CServer::recvContextMessage(void* buff,int count)
326     {
327       static map<string,contextMessage> recvContextId ;
328       map<string,contextMessage>::iterator it ;
329
330       CBufferIn buffer(buff,count) ;
331       string id ;
332       int clientLeader ;
333       int nbMessage ;
334
335       buffer>>id>>nbMessage>>clientLeader ;
336
337       it=recvContextId.find(id) ;
338       if (it==recvContextId.end())
339       {
340         contextMessage msg={0,0} ;
341         pair<map<string,contextMessage>::iterator,bool> ret ;
342         ret=recvContextId.insert(pair<string,contextMessage>(id,msg)) ;
343         it=ret.first ;
344       }
345       it->second.nbRecv+=1 ;
346       it->second.leaderRank+=clientLeader ;
347
348       if (it->second.nbRecv==nbMessage)
349       {
350         int size ;
351         MPI_Comm_size(intraComm,&size) ;
352         MPI_Request* requests= new MPI_Request[size-1] ;
353         MPI_Status* status= new MPI_Status[size-1] ;
354
355         for(int i=1;i<size;i++)
356         {
357            MPI_Isend(buff,count,MPI_CHAR,i,2,intraComm,&requests[i-1]) ;
358         }
359         MPI_Waitall(size-1,requests,status) ;
360         registerContext(buff,count,it->second.leaderRank) ;
361
362         recvContextId.erase(it) ;
363         delete [] requests ;
364         delete [] status ;
365
366       }
367     }
368
369     void CServer::listenRootContext(void)
370     {
371
372       MPI_Status status ;
373       int flag ;
374       static void* buffer ;
375       //#pragma omp threadprivate(buffer)
376
377       static MPI_Request request ;
378       //#pragma omp threadprivate(request)
379
380       static bool recept=false ;
381       //#pragma omp threadprivate(recept)
382       
383       int rank ;
384       int count ;
385       const int root=0 ;
386       
387
388       if (recept==false)
389       {
390         traceOff() ;
391         MPI_Iprobe(root,2,intraComm, &flag, &status) ;
392         traceOn() ;
393         if (flag==true)
394         {
395           MPI_Get_count(&status,MPI_CHAR,&count) ;
396           buffer=new char[count] ;
397           MPI_Irecv(buffer,count,MPI_CHAR,root,2,intraComm,&request) ;
398           recept=true ;
399         }
400       }
401       else
402       {
403         MPI_Test(&request,&flag,&status) ;
404         if (flag==true)
405         {
406           MPI_Get_count(&status,MPI_CHAR,&count) ;
407           registerContext(buffer,count) ;
408           delete [] buffer ;
409           recept=false ;
410         }
411       }
412     }
413
414     void CServer::registerContext(void* buff, int count, int leaderRank)
415     {
416       string contextId;
417       CBufferIn buffer(buff, count);
418       buffer >> contextId;
419
420       info(20) << "CServer : Register new Context : " << contextId << endl;
421
422       if (contextList.find(contextId) != contextList.end())
423         ERROR("void CServer::registerContext(void* buff, int count, int leaderRank)",
424               << "Context '" << contextId << "' has already been registred");
425
426       MPI_Comm contextIntercomm;
427       
428       MPI_Intercomm_create(intraComm,0,CXios::globalComm,leaderRank,10+leaderRank,&contextIntercomm);
429
430       MPI_Comm inter;
431       MPI_Intercomm_merge(contextIntercomm,1,&inter);
432       MPI_Barrier(inter);
433
434       info(20) << "CServer : MPI_Intercomm_merge and MPI_Barrier " << contextId << endl;
435       
436
437       CContext* context = CContext::create(contextId);
438       contextList[contextId]=context;
439       context->initServer(intraComm,contextIntercomm);
440       
441       
442
443       contextInterComms.push_back(contextIntercomm);
444       
445       
446       MPI_Comm_free(&inter);
447       
448       info(20) << "CServer : Register new Context OKOK " << contextId << endl;
449     }
450
451     void CServer::contextEventLoop(void)
452     {
453       bool finished ;
454       map<string,CContext*>::iterator it ;
455       for(it=contextList.begin();it!=contextList.end();++it)
456       {
457         finished=it->second->checkBuffersAndListen();
458         if (finished)
459         {
460           contextList.erase(it) ;
461           break ;
462         }
463       }
464
465     }
466
467     //! Get rank of the current process
468     int CServer::getRank()
469     {
470       return rank;
471     }
472
473    /*!
474    * Open a file specified by a suffix and an extension and use it for the given file buffer.
475    * The file name will be suffix+rank+extension.
476    *
477    * \param fileName[in] protype file name
478    * \param ext [in] extension of the file
479    * \param fb [in/out] the file buffer
480    */
481    void CServer::openStream(const StdString& fileName, const StdString& ext, std::filebuf* fb)
482    {
483      StdStringStream fileNameClient;
484      int numDigit = 0;
485      int size = 0;
486      MPI_Comm_size(CXios::globalComm, &size);
487      while (size)
488      {
489        size /= 10;
490        ++numDigit;
491      }
492
493      fileNameClient << fileName << "_" << std::setfill('0') << std::setw(numDigit) << getRank() << ext;
494      fb->open(fileNameClient.str().c_str(), std::ios::out);
495      if (!fb->is_open())
496        ERROR("void CServer::openStream(const StdString& fileName, const StdString& ext, std::filebuf* fb)",
497              << std::endl << "Can not open <" << fileNameClient << "> file to write the server log(s).");
498    }
499
500    /*!
501    * \brief Open a file stream to write the info logs
502    * Open a file stream with a specific file name suffix+rank
503    * to write the info logs.
504    * \param fileName [in] protype file name
505    */
506    void CServer::openInfoStream(const StdString& fileName)
507    {
508      std::filebuf* fb = m_infoStream.rdbuf();
509      openStream(fileName, ".out", fb);
510
511      info.write2File(fb);
512      report.write2File(fb);
513    }
514
515    //! Write the info logs to standard output
516    void CServer::openInfoStream()
517    {
518      info.write2StdOut();
519      report.write2StdOut();
520    }
521
522    //! Close the info logs file if it opens
523    void CServer::closeInfoStream()
524    {
525      if (m_infoStream.is_open()) m_infoStream.close();
526    }
527
528    /*!
529    * \brief Open a file stream to write the error log
530    * Open a file stream with a specific file name suffix+rank
531    * to write the error log.
532    * \param fileName [in] protype file name
533    */
534    void CServer::openErrorStream(const StdString& fileName)
535    {
536      std::filebuf* fb = m_errorStream.rdbuf();
537      openStream(fileName, ".err", fb);
538
539      error.write2File(fb);
540    }
541
542    //! Write the error log to standard error output
543    void CServer::openErrorStream()
544    {
545      error.write2StdErr();
546    }
547
548    //! Close the error log file if it opens
549    void CServer::closeErrorStream()
550    {
551      if (m_errorStream.is_open()) m_errorStream.close();
552    }
553}
Note: See TracBrowser for help on using the repository browser.