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

Last change on this file since 1079 was 1079, checked in by yushan, 5 years ago

backup modification

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