source: XIOS/dev/dev_olga/src/server.cpp @ 983

Last change on this file since 983 was 983, checked in by oabramkina, 5 years ago

My branch

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