source: XIOS/dev/branch_openmp/src/server.cpp @ 1469

Last change on this file since 1469 was 1469, checked in by yushan, 6 years ago

LMDZ, XIOS_Tests works fine with 2-level server mode. Tested on Curie. (8*8+2)

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