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

Last change on this file since 1077 was 1077, checked in by oabramkina, 4 years ago

dev: Intermediate commit

Done:
Registries: ok with two server levels

To do (on a single server level):
test_complete
reading

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