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

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

save modif for local compilation

  • 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 "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         MPI_Iprobe(MPI_ANY_SOURCE,1,CXios::globalComm, &flag, &status) ;
284         traceOn() ;
285         
286         if (flag==true)
287         {
288           #ifdef _usingMPI
289           rank=status.MPI_SOURCE ;
290           #elif _usingEP
291           rank= status.ep_src ;
292           #endif
293           MPI_Get_count(&status,MPI_CHAR,&count) ;
294           buffer=new char[count] ;
295           MPI_Irecv(buffer,count,MPI_CHAR,rank,1,CXios::globalComm,&request) ;
296           recept=true ;
297         }
298         
299       }
300       else
301       {
302         traceOff() ;
303         MPI_Test(&request,&flag,&status) ;
304         traceOn() ;
305         if (flag==true)
306         {
307           #ifdef _usingMPI
308           rank=status.MPI_SOURCE ;
309           #elif _usingEP
310           rank= status.ep_src ;
311           #endif
312           MPI_Get_count(&status,MPI_CHAR,&count) ;
313           recvContextMessage(buffer,count) ;
314           
315           delete [] buffer ;
316           recept=false ;
317         }
318       }
319     }
320
321     void CServer::recvContextMessage(void* buff,int count)
322     {
323       static map<string,contextMessage> recvContextId ;
324       map<string,contextMessage>::iterator it ;
325
326       CBufferIn buffer(buff,count) ;
327       string id ;
328       int clientLeader ;
329       int nbMessage ;
330
331       buffer>>id>>nbMessage>>clientLeader ;
332
333       it=recvContextId.find(id) ;
334       if (it==recvContextId.end())
335       {
336         contextMessage msg={0,0} ;
337         pair<map<string,contextMessage>::iterator,bool> ret ;
338         ret=recvContextId.insert(pair<string,contextMessage>(id,msg)) ;
339         it=ret.first ;
340       }
341       it->second.nbRecv+=1 ;
342       it->second.leaderRank+=clientLeader ;
343
344       if (it->second.nbRecv==nbMessage)
345       {
346         int size ;
347         MPI_Comm_size(intraComm,&size) ;
348         MPI_Request* requests= new MPI_Request[size-1] ;
349         MPI_Status* status= new MPI_Status[size-1] ;
350
351         for(int i=1;i<size;i++)
352         {
353            MPI_Isend(buff,count,MPI_CHAR,i,2,intraComm,&requests[i-1]) ;
354         }
355         MPI_Waitall(size-1,requests,status) ;
356         registerContext(buff,count,it->second.leaderRank) ;
357
358         recvContextId.erase(it) ;
359         delete [] requests ;
360         delete [] status ;
361
362       }
363     }
364
365     void CServer::listenRootContext(void)
366     {
367
368       MPI_Status status ;
369       int flag ;
370       static void* buffer ;
371       //#pragma omp threadprivate(buffer)
372
373       static MPI_Request request ;
374       //#pragma omp threadprivate(request)
375
376       static bool recept=false ;
377       //#pragma omp threadprivate(recept)
378       
379       int rank ;
380       int count ;
381       const int root=0 ;
382       
383
384       if (recept==false)
385       {
386         traceOff() ;
387         MPI_Iprobe(root,2,intraComm, &flag, &status) ;
388         traceOn() ;
389         if (flag==true)
390         {
391           MPI_Get_count(&status,MPI_CHAR,&count) ;
392           buffer=new char[count] ;
393           MPI_Irecv(buffer,count,MPI_CHAR,root,2,intraComm,&request) ;
394           recept=true ;
395         }
396       }
397       else
398       {
399         MPI_Test(&request,&flag,&status) ;
400         if (flag==true)
401         {
402           MPI_Get_count(&status,MPI_CHAR,&count) ;
403           registerContext(buffer,count) ;
404           delete [] buffer ;
405           recept=false ;
406         }
407       }
408     }
409
410     void CServer::registerContext(void* buff, int count, int leaderRank)
411     {
412       string contextId;
413       CBufferIn buffer(buff, count);
414       buffer >> contextId;
415
416       info(20) << "CServer : Register new Context : " << contextId << endl;
417
418       if (contextList.find(contextId) != contextList.end())
419         ERROR("void CServer::registerContext(void* buff, int count, int leaderRank)",
420               << "Context '" << contextId << "' has already been registred");
421
422       MPI_Comm contextIntercomm;
423       
424       MPI_Intercomm_create(intraComm,0,CXios::globalComm,leaderRank,10+leaderRank,&contextIntercomm);
425
426       MPI_Comm inter;
427       MPI_Intercomm_merge(contextIntercomm,1,&inter);
428       MPI_Barrier(inter);
429
430       info(20) << "CServer : MPI_Intercomm_merge and MPI_Barrier " << contextId << endl;
431       
432
433       CContext* context = CContext::create(contextId);
434       contextList[contextId]=context;
435       context->initServer(intraComm,contextIntercomm);
436       
437       
438
439       contextInterComms.push_back(contextIntercomm);
440       
441       
442       MPI_Comm_free(&inter);
443       
444       info(20) << "CServer : Register new Context OKOK " << contextId << endl;
445     }
446
447     void CServer::contextEventLoop(void)
448     {
449       bool finished ;
450       map<string,CContext*>::iterator it ;
451       for(it=contextList.begin();it!=contextList.end();++it)
452       {
453         finished=it->second->checkBuffersAndListen();
454         if (finished)
455         {
456           contextList.erase(it) ;
457           break ;
458         }
459       }
460
461     }
462
463     //! Get rank of the current process
464     int CServer::getRank()
465     {
466       return rank;
467     }
468
469    /*!
470    * Open a file specified by a suffix and an extension and use it for the given file buffer.
471    * The file name will be suffix+rank+extension.
472    *
473    * \param fileName[in] protype file name
474    * \param ext [in] extension of the file
475    * \param fb [in/out] the file buffer
476    */
477    void CServer::openStream(const StdString& fileName, const StdString& ext, std::filebuf* fb)
478    {
479      StdStringStream fileNameClient;
480      int numDigit = 0;
481      int size = 0;
482      MPI_Comm_size(CXios::globalComm, &size);
483      while (size)
484      {
485        size /= 10;
486        ++numDigit;
487      }
488
489      fileNameClient << fileName << "_" << std::setfill('0') << std::setw(numDigit) << getRank() << ext;
490      fb->open(fileNameClient.str().c_str(), std::ios::out);
491      if (!fb->is_open())
492        ERROR("void CServer::openStream(const StdString& fileName, const StdString& ext, std::filebuf* fb)",
493              << std::endl << "Can not open <" << fileNameClient << "> file to write the server log(s).");
494    }
495
496    /*!
497    * \brief Open a file stream to write the info logs
498    * Open a file stream with a specific file name suffix+rank
499    * to write the info logs.
500    * \param fileName [in] protype file name
501    */
502    void CServer::openInfoStream(const StdString& fileName)
503    {
504      std::filebuf* fb = m_infoStream.rdbuf();
505      openStream(fileName, ".out", fb);
506
507      info.write2File(fb);
508      report.write2File(fb);
509    }
510
511    //! Write the info logs to standard output
512    void CServer::openInfoStream()
513    {
514      info.write2StdOut();
515      report.write2StdOut();
516    }
517
518    //! Close the info logs file if it opens
519    void CServer::closeInfoStream()
520    {
521      if (m_infoStream.is_open()) m_infoStream.close();
522    }
523
524    /*!
525    * \brief Open a file stream to write the error log
526    * Open a file stream with a specific file name suffix+rank
527    * to write the error log.
528    * \param fileName [in] protype file name
529    */
530    void CServer::openErrorStream(const StdString& fileName)
531    {
532      std::filebuf* fb = m_errorStream.rdbuf();
533      openStream(fileName, ".err", fb);
534
535      error.write2File(fb);
536    }
537
538    //! Write the error log to standard error output
539    void CServer::openErrorStream()
540    {
541      error.write2StdErr();
542    }
543
544    //! Close the error log file if it opens
545    void CServer::closeErrorStream()
546    {
547      if (m_errorStream.is_open()) m_errorStream.close();
548    }
549}
Note: See TracBrowser for help on using the repository browser.