source: XIOS/branchs/xios-2.5/src/server.cpp @ 1867

Last change on this file since 1867 was 1867, checked in by ymipsl, 5 years ago

Remove freeing MPI communicator at finalize that can raise deadlock on some recent MPI library

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: 29.3 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    std::list<MPI_Comm> CServer::interCommLeft ;
21    std::list<MPI_Comm> CServer::interCommRight ;
22    std::list<MPI_Comm> CServer::contextInterComms;
23    std::list<MPI_Comm> CServer::contextIntraComms;
24    int CServer::serverLevel = 0 ;
25    int CServer::nbContexts = 0;
26    bool CServer::isRoot = false ;
27    int CServer::rank_ = INVALID_RANK;
28    StdOFStream CServer::m_infoStream;
29    StdOFStream CServer::m_errorStream;
30    map<string,CContext*> CServer::contextList ;
31    vector<int> CServer::sndServerGlobalRanks;
32    bool CServer::finished=false ;
33    bool CServer::is_MPI_Initialized ;
34    CEventScheduler* CServer::eventScheduler = 0;
35
36//---------------------------------------------------------------
37/*!
38 * \fn void CServer::initialize(void)
39 * Creates intraComm for each possible type of servers (classical, primary or secondary).
40 * Creates interComm and stores them into the following lists:
41 *   classical server -- interCommLeft
42 *   primary server -- interCommLeft and interCommRight
43 *   secondary server -- interCommLeft for each pool.
44 *   IMPORTANT: CXios::usingServer2 should NOT be used beyond this function. Use CServer::serverLevel instead.
45 */
46    void CServer::initialize(void)
47    {
48      int initialized ;
49      MPI_Initialized(&initialized) ;
50      if (initialized) is_MPI_Initialized=true ;
51      else is_MPI_Initialized=false ;
52      int rank ;
53
54      // Not using OASIS
55      if (!CXios::usingOasis)
56      {
57
58        if (!is_MPI_Initialized)
59        {
60          MPI_Init(NULL, NULL);
61        }
62        CTimer::get("XIOS").resume() ;
63
64        boost::hash<string> hashString ;
65        unsigned long hashServer = hashString(CXios::xiosCodeId);
66
67        unsigned long* hashAll ;
68        unsigned long* srvLevelAll ;
69
70        int size ;
71        int myColor ;
72        int i,c ;
73        MPI_Comm newComm;
74
75        MPI_Comm_size(CXios::globalComm, &size) ;
76        MPI_Comm_rank(CXios::globalComm, &rank_);
77
78        hashAll=new unsigned long[size] ;
79        MPI_Allgather(&hashServer, 1, MPI_LONG, hashAll, 1, MPI_LONG, CXios::globalComm) ;
80
81        map<unsigned long, int> colors ;
82        map<unsigned long, int> leaders ;
83        map<unsigned long, int>::iterator it ;
84
85        // (1) Establish client leaders, distribute processes between two server levels
86        std::vector<int> srvRanks;
87        for(i=0,c=0;i<size;i++)
88        {
89          if (colors.find(hashAll[i])==colors.end())
90          {
91            colors[hashAll[i]]=c ;
92            leaders[hashAll[i]]=i ;
93            c++ ;
94          }
95          if (CXios::usingServer2)
96            if (hashAll[i] == hashServer)
97              srvRanks.push_back(i);
98        }
99
100        if (CXios::usingServer2)
101        {
102          int reqNbProc = srvRanks.size()*CXios::ratioServer2/100.;
103          if (reqNbProc<1 || reqNbProc==srvRanks.size())
104          {
105            error(0)<<"WARNING: void CServer::initialize(void)"<<endl
106                << "It is impossible to dedicate the requested number of processes = "<<reqNbProc
107                <<" to secondary server. XIOS will run in the classical server mode."<<endl;
108          }
109          else
110          {
111            if (CXios::nbPoolsServer2 == 0) CXios::nbPoolsServer2 = reqNbProc;
112            int firstSndSrvRank = srvRanks.size()*(100.-CXios::ratioServer2)/100. ;
113            int poolLeader = firstSndSrvRank;
114//*********** (1) Comment out the line below to set one process per pool
115            sndServerGlobalRanks.push_back(srvRanks[poolLeader]);
116            int nbPools = CXios::nbPoolsServer2;
117            if ( nbPools > reqNbProc || nbPools < 1)
118            {
119              error(0)<<"WARNING: void CServer::initialize(void)"<<endl
120                  << "It is impossible to allocate the requested number of pools = "<<nbPools
121                  <<" on the secondary server. It will be set so that there is one process per pool."<<endl;
122              nbPools = reqNbProc;
123            }
124            int remainder = ((int) (srvRanks.size()*CXios::ratioServer2/100.)) % nbPools;
125            int procsPerPool = ((int) (srvRanks.size()*CXios::ratioServer2/100.)) / nbPools;
126            for (i=0; i<srvRanks.size(); i++)
127            {
128              if (i >= firstSndSrvRank)
129              {
130                if (rank_ == srvRanks[i])
131                {
132                  serverLevel=2;
133                }
134                poolLeader += procsPerPool;
135                if (remainder != 0)
136                {
137                  ++poolLeader;
138                  --remainder;
139                }
140//*********** (2) Comment out the two lines below to set one process per pool
141                if (poolLeader < srvRanks.size())
142                  sndServerGlobalRanks.push_back(srvRanks[poolLeader]);
143//*********** (3) Uncomment the line below to set one process per pool
144//                sndServerGlobalRanks.push_back(srvRanks[i]);
145              }
146              else
147              {
148                if (rank_ == srvRanks[i]) serverLevel=1;
149              }
150            }
151            if (serverLevel==2)
152            {
153              info(50)<<"The number of secondary server pools is "<< sndServerGlobalRanks.size() <<endl ;
154              for (i=0; i<sndServerGlobalRanks.size(); i++)
155              {
156                if (rank_>= sndServerGlobalRanks[i])
157                {
158                  if ( i == sndServerGlobalRanks.size()-1)
159                  {
160                    myColor = colors.size() + sndServerGlobalRanks[i];
161                  }
162                  else if (rank_< sndServerGlobalRanks[i+1])
163                  {
164                    myColor = colors.size() + sndServerGlobalRanks[i];
165                    break;
166                  }
167                }
168              }
169            }
170          }
171        }
172
173        // (2) Create intraComm
174        if (serverLevel != 2) myColor=colors[hashServer];
175        MPI_Comm_split(CXios::globalComm, myColor, rank_, &intraComm) ;
176
177        // (3) Create interComm
178        if (serverLevel == 0)
179        {
180          int clientLeader;
181          for(it=leaders.begin();it!=leaders.end();it++)
182          {
183            if (it->first!=hashServer)
184            {
185              clientLeader=it->second ;
186              int intraCommSize, intraCommRank ;
187              MPI_Comm_size(intraComm,&intraCommSize) ;
188              MPI_Comm_rank(intraComm,&intraCommRank) ;
189              info(50)<<"intercommCreate::server (classical mode) "<<rank_<<" intraCommSize : "<<intraCommSize
190                       <<" intraCommRank :"<<intraCommRank<<"  clientLeader "<< clientLeader<<endl ;
191
192              MPI_Intercomm_create(intraComm, 0, CXios::globalComm, clientLeader, 0, &newComm) ;
193              interCommLeft.push_back(newComm) ;
194            }
195          }
196        }
197        else if (serverLevel == 1)
198        {
199          int clientLeader, srvSndLeader;
200          int srvPrmLeader ;
201
202          for (it=leaders.begin();it!=leaders.end();it++)
203          {
204            if (it->first != hashServer)
205            {
206              clientLeader=it->second ;
207              int intraCommSize, intraCommRank ;
208              MPI_Comm_size(intraComm, &intraCommSize) ;
209              MPI_Comm_rank(intraComm, &intraCommRank) ;
210              info(50)<<"intercommCreate::server (server level 1) "<<rank_<<" intraCommSize : "<<intraCommSize
211                       <<" intraCommRank :"<<intraCommRank<<"  clientLeader "<< clientLeader<<endl ;
212              MPI_Intercomm_create(intraComm, 0, CXios::globalComm, clientLeader, 0, &newComm) ;
213              interCommLeft.push_back(newComm) ;
214            }
215          }
216
217          for (int i = 0; i < sndServerGlobalRanks.size(); ++i)
218          {
219            int intraCommSize, intraCommRank ;
220            MPI_Comm_size(intraComm, &intraCommSize) ;
221            MPI_Comm_rank(intraComm, &intraCommRank) ;
222            info(50)<<"intercommCreate::client (server level 1) "<<rank_<<" intraCommSize : "<<intraCommSize
223                <<" intraCommRank :"<<intraCommRank<<"  clientLeader "<< sndServerGlobalRanks[i]<<endl ;
224            MPI_Intercomm_create(intraComm, 0, CXios::globalComm, sndServerGlobalRanks[i], 1, &newComm) ;
225            interCommRight.push_back(newComm) ;
226          }
227        }
228        else
229        {
230          int clientLeader;
231          clientLeader = leaders[hashString(CXios::xiosCodeId)];
232          int intraCommSize, intraCommRank ;
233          MPI_Comm_size(intraComm, &intraCommSize) ;
234          MPI_Comm_rank(intraComm, &intraCommRank) ;
235          info(50)<<"intercommCreate::server (server level 2) "<<rank_<<" intraCommSize : "<<intraCommSize
236                   <<" intraCommRank :"<<intraCommRank<<"  clientLeader "<< clientLeader<<endl ;
237
238          MPI_Intercomm_create(intraComm, 0, CXios::globalComm, clientLeader, 1, &newComm) ;
239          interCommLeft.push_back(newComm) ;
240        }
241
242        delete [] hashAll ;
243
244      }
245      // using OASIS
246      else
247      {
248        int size;
249        int myColor;
250        int* srvGlobalRanks;
251        if (!is_MPI_Initialized) oasis_init(CXios::xiosCodeId);
252
253        CTimer::get("XIOS").resume() ;
254        MPI_Comm localComm;
255        oasis_get_localcomm(localComm);
256        MPI_Comm_rank(localComm,&rank_) ;
257
258//      (1) Create server intraComm
259        if (!CXios::usingServer2)
260        {
261          MPI_Comm_dup(localComm, &intraComm);
262        }
263        else
264        {
265          int globalRank;
266          MPI_Comm_size(localComm,&size) ;
267          MPI_Comm_rank(CXios::globalComm,&globalRank) ;
268          srvGlobalRanks = new int[size] ;
269          MPI_Allgather(&globalRank, 1, MPI_INT, srvGlobalRanks, 1, MPI_INT, localComm) ;
270
271          int reqNbProc = size*CXios::ratioServer2/100.;
272          if (reqNbProc < 1 || reqNbProc == size)
273          {
274            error(0)<<"WARNING: void CServer::initialize(void)"<<endl
275                << "It is impossible to dedicate the requested number of processes = "<<reqNbProc
276                <<" to secondary server. XIOS will run in the classical server mode."<<endl;
277            MPI_Comm_dup(localComm, &intraComm);
278          }
279          else
280          {
281            int firstSndSrvRank = size*(100.-CXios::ratioServer2)/100. ;
282            int poolLeader = firstSndSrvRank;
283//*********** (1) Comment out the line below to set one process per pool
284//            sndServerGlobalRanks.push_back(srvGlobalRanks[poolLeader]);
285            int nbPools = CXios::nbPoolsServer2;
286            if ( nbPools > reqNbProc || nbPools < 1)
287            {
288              error(0)<<"WARNING: void CServer::initialize(void)"<<endl
289                  << "It is impossible to allocate the requested number of pools = "<<nbPools
290                  <<" on the secondary server. It will be set so that there is one process per pool."<<endl;
291              nbPools = reqNbProc;
292            }
293            int remainder = ((int) (size*CXios::ratioServer2/100.)) % nbPools;
294            int procsPerPool = ((int) (size*CXios::ratioServer2/100.)) / nbPools;
295            for (int i=0; i<size; i++)
296            {
297              if (i >= firstSndSrvRank)
298              {
299                if (globalRank == srvGlobalRanks[i])
300                {
301                  serverLevel=2;
302                }
303                poolLeader += procsPerPool;
304                if (remainder != 0)
305                {
306                  ++poolLeader;
307                  --remainder;
308                }
309//*********** (2) Comment out the two lines below to set one process per pool
310//                if (poolLeader < size)
311//                  sndServerGlobalRanks.push_back(srvGlobalRanks[poolLeader]);
312//*********** (3) Uncomment the line below to set one process per pool
313                sndServerGlobalRanks.push_back(srvGlobalRanks[i]);
314              }
315              else
316              {
317                if (globalRank == srvGlobalRanks[i]) serverLevel=1;
318              }
319            }
320            if (serverLevel==2)
321            {
322              info(50)<<"The number of secondary server pools is "<< sndServerGlobalRanks.size() <<endl ;
323              for (int i=0; i<sndServerGlobalRanks.size(); i++)
324              {
325                if (globalRank>= sndServerGlobalRanks[i])
326                {
327                  if (i == sndServerGlobalRanks.size()-1)
328                  {
329                    myColor = sndServerGlobalRanks[i];
330                  }
331                  else if (globalRank< sndServerGlobalRanks[i+1])
332                  {
333                    myColor = sndServerGlobalRanks[i];
334                    break;
335                  }
336                }
337              }
338            }
339            if (serverLevel != 2) myColor=0;
340            MPI_Comm_split(localComm, myColor, rank_, &intraComm) ;
341          }
342        }
343
344        string codesId=CXios::getin<string>("oasis_codes_id") ;
345        vector<string> splitted ;
346        boost::split( splitted, codesId, boost::is_any_of(","), boost::token_compress_on ) ;
347        vector<string>::iterator it ;
348
349        MPI_Comm newComm ;
350        int globalRank ;
351        MPI_Comm_rank(CXios::globalComm,&globalRank);
352
353//      (2) Create interComms with models
354        for(it=splitted.begin();it!=splitted.end();it++)
355        {
356          oasis_get_intercomm(newComm,*it) ;
357          if ( serverLevel == 0 || serverLevel == 1)
358          {
359            interCommLeft.push_back(newComm) ;
360            if (rank_==0) MPI_Send(&globalRank,1,MPI_INT,0,0,newComm) ;
361          }
362        }
363
364//      (3) Create interComms between primary and secondary servers
365        int intraCommSize, intraCommRank ;
366        MPI_Comm_size(intraComm,&intraCommSize) ;
367        MPI_Comm_rank(intraComm, &intraCommRank) ;
368
369        if (serverLevel == 1)
370        {
371          for (int i = 0; i < sndServerGlobalRanks.size(); ++i)
372          {
373            int srvSndLeader = sndServerGlobalRanks[i];
374            info(50)<<"intercommCreate::client (server level 1) "<<globalRank<<" intraCommSize : "<<intraCommSize
375                <<" intraCommRank :"<<intraCommRank<<"  clientLeader "<< srvSndLeader<<endl ;
376            MPI_Intercomm_create(intraComm, 0, CXios::globalComm, srvSndLeader, 0, &newComm) ;
377            interCommRight.push_back(newComm) ;
378          }
379        }
380        else if (serverLevel == 2)
381        {
382          info(50)<<"intercommCreate::server (server level 2)"<<globalRank<<" intraCommSize : "<<intraCommSize
383                   <<" intraCommRank :"<<intraCommRank<<"  clientLeader "<< srvGlobalRanks[0] <<endl ;
384          MPI_Intercomm_create(intraComm, 0, CXios::globalComm, srvGlobalRanks[0], 0, &newComm) ;
385          interCommLeft.push_back(newComm) ;
386        }
387        if (CXios::usingServer2) delete [] srvGlobalRanks ;
388        oasis_enddef() ;
389      }
390
391
392      MPI_Comm_rank(intraComm, &rank) ;
393      if (rank==0) isRoot=true;
394      else isRoot=false;
395     
396      eventScheduler = new CEventScheduler(intraComm) ;
397    }
398
399    void CServer::finalize(void)
400    {
401      CTimer::get("XIOS").suspend() ;
402     
403      delete eventScheduler ;
404
405      for (std::list<MPI_Comm>::iterator it = contextInterComms.begin(); it != contextInterComms.end(); it++)
406        /* MPI_Comm_free(&(*it)) */; // WARNING remove freeing communicator !! --> deadlock raised, to be checked
407
408      for (std::list<MPI_Comm>::iterator it = contextIntraComms.begin(); it != contextIntraComms.end(); it++)
409        MPI_Comm_free(&(*it));
410
411//      for (std::list<MPI_Comm>::iterator it = interComm.begin(); it != interComm.end(); it++)
412//        MPI_Comm_free(&(*it));
413
414//        for (std::list<MPI_Comm>::iterator it = interCommLeft.begin(); it != interCommLeft.end(); it++)
415//          MPI_Comm_free(&(*it));
416
417        for (std::list<MPI_Comm>::iterator it = interCommRight.begin(); it != interCommRight.end(); it++)
418          /* MPI_Comm_free(&(*it)) */ ; // WARNING remove freeing communicator !! --> deadlock raised, to be checked
419
420      MPI_Comm_free(&intraComm);
421
422      if (!is_MPI_Initialized)
423      {
424        if (CXios::usingOasis) oasis_finalize();
425        else MPI_Finalize() ;
426      }
427      report(0)<<"Performance report : Time spent for XIOS : "<<CTimer::get("XIOS server").getCumulatedTime()<<endl  ;
428      report(0)<<"Performance report : Time spent in processing events : "<<CTimer::get("Process events").getCumulatedTime()<<endl  ;
429      report(0)<<"Performance report : Ratio : "<<CTimer::get("Process events").getCumulatedTime()/CTimer::get("XIOS server").getCumulatedTime()*100.<<"%"<<endl  ;
430      report(100)<<CTimer::getAllCumulatedTime()<<endl ;
431    }
432
433     void CServer::eventLoop(void)
434     {
435       bool stop=false ;
436
437       CTimer::get("XIOS server").resume() ;
438       while(!stop)
439       {
440         if (isRoot)
441         {
442           listenContext();
443           listenRootContext();
444           if (!finished) listenFinalize() ;
445         }
446         else
447         {
448           listenRootContext();
449           if (!finished) listenRootFinalize() ;
450         }
451
452         contextEventLoop() ;
453         if (finished && contextList.empty()) stop=true ;
454         eventScheduler->checkEvent() ;
455       }
456       CTimer::get("XIOS server").suspend() ;
457     }
458
459     void CServer::listenFinalize(void)
460     {
461        list<MPI_Comm>::iterator it, itr;
462        int msg ;
463        int flag ;
464
465        for(it=interCommLeft.begin();it!=interCommLeft.end();it++)
466        {
467           MPI_Status status ;
468           traceOff() ;
469           MPI_Iprobe(0,0,*it,&flag,&status) ;
470           traceOn() ;
471           if (flag==true)
472           {
473              MPI_Recv(&msg,1,MPI_INT,0,0,*it,&status) ;
474              info(20)<<" CServer : Receive client finalize"<<endl ;
475              // Sending server finalize message to secondary servers (if any)
476              for(itr=interCommRight.begin();itr!=interCommRight.end();itr++)
477              {
478                MPI_Send(&msg,1,MPI_INT,0,0,*itr) ;
479              }
480              /* MPI_Comm_free(&(*it)); */ // WARNING remove freeing communicator !! --> deadlock raised, to be checked
481              interCommLeft.erase(it) ;
482              break ;
483            }
484         }
485
486         if (interCommLeft.empty())
487         {
488           int i,size ;
489           MPI_Comm_size(intraComm,&size) ;
490           MPI_Request* requests= new MPI_Request[size-1] ;
491           MPI_Status* status= new MPI_Status[size-1] ;
492
493           for(int i=1;i<size;i++) MPI_Isend(&msg,1,MPI_INT,i,4,intraComm,&requests[i-1]) ;
494           MPI_Waitall(size-1,requests,status) ;
495
496           finished=true ;
497           delete [] requests ;
498           delete [] status ;
499         }
500     }
501
502
503     void CServer::listenRootFinalize()
504     {
505        int flag ;
506        MPI_Status status ;
507        int msg ;
508
509        traceOff() ;
510        MPI_Iprobe(0,4,intraComm, &flag, &status) ;
511        traceOn() ;
512        if (flag==true)
513        {
514           MPI_Recv(&msg,1,MPI_INT,0,4,intraComm,&status) ;
515           finished=true ;
516        }
517      }
518
519     void CServer::listenContext(void)
520     {
521
522       MPI_Status status ;
523       int flag ;
524       static char* buffer ;
525       static MPI_Request request ;
526       static bool recept=false ;
527       int rank ;
528       int count ;
529
530       if (recept==false)
531       {
532         traceOff() ;
533         MPI_Iprobe(MPI_ANY_SOURCE,1,CXios::globalComm, &flag, &status) ;
534         traceOn() ;
535         if (flag==true)
536         {
537           rank=status.MPI_SOURCE ;
538           MPI_Get_count(&status,MPI_CHAR,&count) ;
539           buffer=new char[count] ;
540           MPI_Irecv((void*)buffer,count,MPI_CHAR,rank,1,CXios::globalComm,&request) ;
541           recept=true ;
542         }
543       }
544       else
545       {
546         traceOff() ;
547         MPI_Test(&request,&flag,&status) ;
548         traceOn() ;
549         if (flag==true)
550         {
551           rank=status.MPI_SOURCE ;
552           MPI_Get_count(&status,MPI_CHAR,&count) ;
553           recvContextMessage((void*)buffer,count) ;
554           delete [] buffer ;
555           recept=false ;
556         }
557       }
558     }
559
560     void CServer::recvContextMessage(void* buff,int count)
561     {
562       static map<string,contextMessage> recvContextId;
563       map<string,contextMessage>::iterator it ;
564       CBufferIn buffer(buff,count) ;
565       string id ;
566       int clientLeader ;
567       int nbMessage ;
568
569       buffer>>id>>nbMessage>>clientLeader ;
570
571       it=recvContextId.find(id) ;
572       if (it==recvContextId.end())
573       {
574         contextMessage msg={0,0} ;
575         pair<map<string,contextMessage>::iterator,bool> ret ;
576         ret=recvContextId.insert(pair<string,contextMessage>(id,msg)) ;
577         it=ret.first ;
578       }
579       it->second.nbRecv+=1 ;
580       it->second.leaderRank+=clientLeader ;
581
582       if (it->second.nbRecv==nbMessage)
583       {
584         int size ;
585         MPI_Comm_size(intraComm,&size) ;
586//         MPI_Request* requests= new MPI_Request[size-1] ;
587//         MPI_Status* status= new MPI_Status[size-1] ;
588         MPI_Request* requests= new MPI_Request[size] ;
589         MPI_Status* status= new MPI_Status[size] ;
590
591         CMessage msg ;
592         msg<<id<<it->second.leaderRank;
593         int messageSize=msg.size() ;
594         void * sendBuff = new char[messageSize] ;
595         CBufferOut sendBuffer(sendBuff,messageSize) ;
596         sendBuffer<<msg ;
597
598         // Include root itself in order not to have a divergence
599         for(int i=0; i<size; i++)
600         {
601           MPI_Isend(sendBuff,sendBuffer.count(),MPI_CHAR,i,2,intraComm,&requests[i]) ;
602         }
603
604         recvContextId.erase(it) ;
605         delete [] requests ;
606         delete [] status ;
607
608       }
609     }
610
611     void CServer::listenRootContext(void)
612     {
613       MPI_Status status ;
614       int flag ;
615       static std::vector<void*> buffers;
616       static std::vector<MPI_Request> requests ;
617       static std::vector<int> counts ;
618       static std::vector<bool> isEventRegistered ;
619       static std::vector<bool> isEventQueued ;
620       MPI_Request request;
621
622       int rank ;
623       const int root=0 ;
624       boost::hash<string> hashString;
625       size_t hashId = hashString("RegisterContext");
626
627       // (1) Receive context id from the root, save it into a buffer
628       traceOff() ;
629       MPI_Iprobe(root,2,intraComm, &flag, &status) ;
630       traceOn() ;
631       if (flag==true)
632       {
633         counts.push_back(0);
634         MPI_Get_count(&status,MPI_CHAR,&(counts.back())) ;
635         buffers.push_back(new char[counts.back()]) ;
636         requests.push_back(request);
637         MPI_Irecv((void*)(buffers.back()),counts.back(),MPI_CHAR,root,2,intraComm,&(requests.back())) ;
638         isEventRegistered.push_back(false);
639         isEventQueued.push_back(false);
640         nbContexts++;
641       }
642
643       for (int ctxNb = 0; ctxNb < nbContexts; ctxNb++ )
644       {
645         // (2) If context id is received, register an event
646         MPI_Test(&requests[ctxNb],&flag,&status) ;
647         if (flag==true && !isEventRegistered[ctxNb])
648         {
649           eventScheduler->registerEvent(ctxNb,hashId);
650           isEventRegistered[ctxNb] = true;
651         }
652         // (3) If event has been scheduled, call register context
653         if (eventScheduler->queryEvent(ctxNb,hashId) && !isEventQueued[ctxNb])
654         {
655           registerContext(buffers[ctxNb],counts[ctxNb]) ;
656           isEventQueued[ctxNb] = true;
657           delete [] buffers[ctxNb] ;
658         }
659       }
660
661     }
662
663     void CServer::registerContext(void* buff, int count, int leaderRank)
664     {
665       string contextId;
666       CBufferIn buffer(buff, count);
667//       buffer >> contextId;
668       buffer >> contextId>>leaderRank;
669       CContext* context;
670
671       info(20) << "CServer : Register new Context : " << contextId << endl;
672
673       if (contextList.find(contextId) != contextList.end())
674         ERROR("void CServer::registerContext(void* buff, int count, int leaderRank)",
675               << "Context '" << contextId << "' has already been registred");
676
677       context=CContext::create(contextId);
678       contextList[contextId]=context;
679
680       // Primary or classical server: create communication channel with a client
681       // (1) create interComm (with a client)
682       // (2) initialize client and server (contextClient and contextServer)
683       MPI_Comm inter;
684       if (serverLevel < 2)
685       {
686         MPI_Comm contextInterComm;
687         MPI_Intercomm_create(intraComm, 0, CXios::globalComm, leaderRank, 10+leaderRank, &contextInterComm);
688         MPI_Intercomm_merge(contextInterComm,1,&inter);
689         MPI_Barrier(inter);
690         MPI_Comm_free(&inter);
691         context->initServer(intraComm,contextInterComm);
692         contextInterComms.push_back(contextInterComm);
693
694       }
695       // Secondary server: create communication channel with a primary server
696       // (1) duplicate interComm with a primary server
697       // (2) initialize client and server (contextClient and contextServer)
698       // Remark: in the case of the secondary server there is no need to create an interComm calling MPI_Intercomm_create,
699       //         because interComm of CContext is defined on the same processes as the interComm of CServer.
700       //         So just duplicate it.
701       else if (serverLevel == 2)
702       {
703         MPI_Comm_dup(interCommLeft.front(), &inter);
704         contextInterComms.push_back(inter);
705         context->initServer(intraComm, contextInterComms.back());
706       }
707
708       // Primary server:
709       // (1) send create context message to secondary servers
710       // (2) initialize communication channels with secondary servers (create contextClient and contextServer)
711       if (serverLevel == 1)
712       {
713         int i = 0, size;
714         MPI_Comm_size(intraComm, &size) ;
715         for (std::list<MPI_Comm>::iterator it = interCommRight.begin(); it != interCommRight.end(); it++, ++i)
716         {
717           StdString str = contextId +"_server_" + boost::lexical_cast<string>(i);
718           CMessage msg;
719           int messageSize;
720           msg<<str<<size<<rank_ ;
721           messageSize = msg.size() ;
722           buff = new char[messageSize] ;
723           CBufferOut buffer(buff,messageSize) ;
724           buffer<<msg ;
725           MPI_Send(buff, buffer.count(), MPI_CHAR, sndServerGlobalRanks[i], 1, CXios::globalComm) ;
726           MPI_Comm_dup(*it, &inter);
727           contextInterComms.push_back(inter);
728           MPI_Comm_dup(intraComm, &inter);
729           contextIntraComms.push_back(inter);
730           context->initClient(contextIntraComms.back(), contextInterComms.back()) ;
731           delete [] buff ;
732         }
733       }
734     }
735
736     void CServer::contextEventLoop(bool enableEventsProcessing /*= true*/)
737     {
738       bool isFinalized ;
739       map<string,CContext*>::iterator it ;
740
741       for(it=contextList.begin();it!=contextList.end();it++)
742       {
743         isFinalized=it->second->isFinalized();
744         if (isFinalized)
745         {
746           contextList.erase(it) ;
747           break ;
748         }
749         else
750           it->second->checkBuffersAndListen(enableEventsProcessing);
751       }
752     }
753
754     //! Get rank of the current process in the intraComm
755     int CServer::getRank()
756     {
757       int rank;
758       MPI_Comm_rank(intraComm,&rank);
759       return rank;
760     }
761
762     vector<int>& CServer::getSecondaryServerGlobalRanks()
763     {
764       return sndServerGlobalRanks;
765     }
766
767    /*!
768    * Open a file specified by a suffix and an extension and use it for the given file buffer.
769    * The file name will be suffix+rank+extension.
770    *
771    * \param fileName[in] protype file name
772    * \param ext [in] extension of the file
773    * \param fb [in/out] the file buffer
774    */
775    void CServer::openStream(const StdString& fileName, const StdString& ext, std::filebuf* fb)
776    {
777      StdStringStream fileNameClient;
778      int numDigit = 0;
779      int size = 0;
780      int id;
781      MPI_Comm_size(CXios::globalComm, &size);
782      while (size)
783      {
784        size /= 10;
785        ++numDigit;
786      }
787      id = rank_; //getRank();
788
789      fileNameClient << fileName << "_" << std::setfill('0') << std::setw(numDigit) << id << ext;
790      fb->open(fileNameClient.str().c_str(), std::ios::out);
791      if (!fb->is_open())
792        ERROR("void CServer::openStream(const StdString& fileName, const StdString& ext, std::filebuf* fb)",
793              << std::endl << "Can not open <" << fileNameClient << "> file to write the server log(s).");
794    }
795
796    /*!
797    * \brief Open a file stream to write the info logs
798    * Open a file stream with a specific file name suffix+rank
799    * to write the info logs.
800    * \param fileName [in] protype file name
801    */
802    void CServer::openInfoStream(const StdString& fileName)
803    {
804      std::filebuf* fb = m_infoStream.rdbuf();
805      openStream(fileName, ".out", fb);
806
807      info.write2File(fb);
808      report.write2File(fb);
809    }
810
811    //! Write the info logs to standard output
812    void CServer::openInfoStream()
813    {
814      info.write2StdOut();
815      report.write2StdOut();
816    }
817
818    //! Close the info logs file if it opens
819    void CServer::closeInfoStream()
820    {
821      if (m_infoStream.is_open()) m_infoStream.close();
822    }
823
824    /*!
825    * \brief Open a file stream to write the error log
826    * Open a file stream with a specific file name suffix+rank
827    * to write the error log.
828    * \param fileName [in] protype file name
829    */
830    void CServer::openErrorStream(const StdString& fileName)
831    {
832      std::filebuf* fb = m_errorStream.rdbuf();
833      openStream(fileName, ".err", fb);
834
835      error.write2File(fb);
836    }
837
838    //! Write the error log to standard error output
839    void CServer::openErrorStream()
840    {
841      error.write2StdErr();
842    }
843
844    //! Close the error log file if it opens
845    void CServer::closeErrorStream()
846    {
847      if (m_errorStream.is_open()) m_errorStream.close();
848    }
849}
Note: See TracBrowser for help on using the repository browser.