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

Last change on this file since 1063 was 1063, checked in by yushan, 8 years ago

server mode OK for both multiple and one file mode. Tested with test_client

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