source: XIOS/trunk/src/server.cpp @ 925

Last change on this file since 925 was 925, checked in by rlacroix, 5 years ago

Fix MPI_Init calls to be more compliant with the standard.

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