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

Last change on this file since 697 was 697, checked in by ymipsl, 6 years ago

Implement registryIn and registryOut functionnalities.
YM

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