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

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

branch_openmp merged with trunk r1544

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