source: XIOS/dev/branch_yushan_merged/src/server.cpp @ 1185

Last change on this file since 1185 was 1185, checked in by yushan, 4 years ago

save dev. recv_test OK

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