source: XIOS/dev/XIOS_DEV_CMIP6/src/server.cpp @ 1301

Last change on this file since 1301 was 1271, checked in by oabramkina, 7 years ago

Corrections in addition to r1270. The algorithm for registering context is improved to avoid possible interference among contexts.

  • 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: 28.8 KB
RevLine 
[490]1#include "globalScopeData.hpp"
[591]2#include "xios_spl.hpp"
[300]3#include "cxios.hpp"
[342]4#include "server.hpp"
[983]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>
[382]12#include "mpi.hpp"
[347]13#include "tracer.hpp"
14#include "timer.hpp"
[492]15#include "event_scheduler.hpp"
[300]16
[335]17namespace xios
[490]18{
[300]19    MPI_Comm CServer::intraComm ;
[1180]20    std::list<MPI_Comm> CServer::interCommLeft ;
21    std::list<MPI_Comm> CServer::interCommRight ;
[655]22    std::list<MPI_Comm> CServer::contextInterComms;
[1071]23    std::list<MPI_Comm> CServer::contextIntraComms;
[1021]24    int CServer::serverLevel = 0 ;
[1148]25    int CServer::nbContexts = 0;
[983]26    bool CServer::isRoot = false ;
[1077]27    int CServer::rank_ = INVALID_RANK;
[490]28    StdOFStream CServer::m_infoStream;
[523]29    StdOFStream CServer::m_errorStream;
[490]30    map<string,CContext*> CServer::contextList ;
[1152]31    vector<int> CServer::sndServerGlobalRanks;
[300]32    bool CServer::finished=false ;
33    bool CServer::is_MPI_Initialized ;
[597]34    CEventScheduler* CServer::eventScheduler = 0;
[983]35
36//---------------------------------------------------------------
37/*!
38 * \fn void CServer::initialize(void)
[1054]39 * Creates intraComm for each possible type of servers (classical, primary or secondary).
[1234]40 * (For now the assumption is that there is one proc per secondary server pool.)
[1148]41 * Creates interComm and stores them into the following lists:
[1054]42 *   classical server -- interCommLeft
43 *   primary server -- interCommLeft and interCommRight
[1148]44 *   secondary server -- interCommLeft for each pool.
[1234]45 *   IMPORTANT: CXios::usingServer2 should NOT be used beyond this function. Use CServer::serverLevel instead.
[983]46 */
[300]47    void CServer::initialize(void)
48    {
49      int initialized ;
50      MPI_Initialized(&initialized) ;
51      if (initialized) is_MPI_Initialized=true ;
52      else is_MPI_Initialized=false ;
[1152]53      int rank ;
[490]54
[300]55      // Not using OASIS
56      if (!CXios::usingOasis)
57      {
[490]58
59        if (!is_MPI_Initialized)
[300]60        {
[925]61          MPI_Init(NULL, NULL);
[300]62        }
[359]63        CTimer::get("XIOS").resume() ;
[490]64
65        boost::hash<string> hashString ;
[1021]66        unsigned long hashServer = hashString(CXios::xiosCodeId);
[490]67
[300]68        unsigned long* hashAll ;
[1152]69        unsigned long* srvLevelAll ;
[490]70
[300]71        int size ;
72        int myColor ;
73        int i,c ;
[1152]74        MPI_Comm newComm;
[490]75
[983]76        MPI_Comm_size(CXios::globalComm, &size) ;
[1077]77        MPI_Comm_rank(CXios::globalComm, &rank_);
[1009]78
[300]79        hashAll=new unsigned long[size] ;
[983]80        MPI_Allgather(&hashServer, 1, MPI_LONG, hashAll, 1, MPI_LONG, CXios::globalComm) ;
[490]81
[1021]82        map<unsigned long, int> colors ;
[300]83        map<unsigned long, int> leaders ;
84        map<unsigned long, int>::iterator it ;
[490]85
[1152]86        // (1) Establish client leaders, distribute processes between two server levels
[1234]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          }
[1152]96          if (CXios::usingServer2)
[1180]97            if (hashAll[i] == hashServer)
98              srvRanks.push_back(i);
99        }
[1234]100
101        if (CXios::usingServer2)
[1180]102        {
[1234]103          int reqNbProc = srvRanks.size()*CXios::ratioServer2/100.;
104          if (reqNbProc<1 || reqNbProc==srvRanks.size())
[1152]105          {
[1234]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;
[1152]109          }
[1180]110          else
111          {
[1243]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;
[1234]126            for (i=0; i<srvRanks.size(); i++)
127            {
[1243]128              if (i >= firstSndSrvRank)
[1234]129              {
[1243]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
[1234]144                sndServerGlobalRanks.push_back(srvRanks[i]);
145              }
146              else
147              {
148                if (rank_ == srvRanks[i]) serverLevel=1;
149              }
150            }
[1243]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            }
[1180]170          }
[300]171        }
[1243]172
[1152]173        // (2) Create intraComm
[1243]174        if (serverLevel != 2) myColor=colors[hashServer];
[1152]175        MPI_Comm_split(CXios::globalComm, myColor, rank_, &intraComm) ;
176
177        // (3) Create interComm
[1021]178        if (serverLevel == 0)
[983]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) ;
[1142]189              info(50)<<"intercommCreate::server (classical mode) "<<rank_<<" intraCommSize : "<<intraCommSize
[983]190                       <<" intraCommRank :"<<intraCommRank<<"  clientLeader "<< clientLeader<<endl ;
[490]191
[1009]192              MPI_Intercomm_create(intraComm, 0, CXios::globalComm, clientLeader, 0, &newComm) ;
[1180]193              interCommLeft.push_back(newComm) ;
[983]194            }
195          }
196        }
[1021]197        else if (serverLevel == 1)
[983]198        {
[1054]199          int clientLeader, srvSndLeader;
200          int srvPrmLeader ;
[1152]201
[1021]202          for (it=leaders.begin();it!=leaders.end();it++)
[983]203          {
[1021]204            if (it->first != hashServer)
[983]205            {
[1021]206              clientLeader=it->second ;
207              int intraCommSize, intraCommRank ;
208              MPI_Comm_size(intraComm, &intraCommSize) ;
209              MPI_Comm_rank(intraComm, &intraCommRank) ;
[1142]210              info(50)<<"intercommCreate::server (server level 1) "<<rank_<<" intraCommSize : "<<intraCommSize
[1021]211                       <<" intraCommRank :"<<intraCommRank<<"  clientLeader "<< clientLeader<<endl ;
212              MPI_Intercomm_create(intraComm, 0, CXios::globalComm, clientLeader, 0, &newComm) ;
213              interCommLeft.push_back(newComm) ;
[983]214            }
[1021]215          }
[1009]216
[1152]217          for (int i = 0; i < sndServerGlobalRanks.size(); ++i)
[983]218          {
[1054]219            int intraCommSize, intraCommRank ;
220            MPI_Comm_size(intraComm, &intraCommSize) ;
221            MPI_Comm_rank(intraComm, &intraCommRank) ;
[1142]222            info(50)<<"intercommCreate::client (server level 1) "<<rank_<<" intraCommSize : "<<intraCommSize
[1152]223                <<" intraCommRank :"<<intraCommRank<<"  clientLeader "<< sndServerGlobalRanks[i]<<endl ;
224            MPI_Intercomm_create(intraComm, 0, CXios::globalComm, sndServerGlobalRanks[i], 1, &newComm) ;
[1054]225            interCommRight.push_back(newComm) ;
[1021]226          }
[1168]227        }
[1021]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) ;
[1142]235          info(50)<<"intercommCreate::server (server level 2) "<<rank_<<" intraCommSize : "<<intraCommSize
[1021]236                   <<" intraCommRank :"<<intraCommRank<<"  clientLeader "<< clientLeader<<endl ;
[983]237
[1142]238          MPI_Intercomm_create(intraComm, 0, CXios::globalComm, clientLeader, 1, &newComm) ;
[1021]239          interCommLeft.push_back(newComm) ;
[1168]240        }
[983]241
242        delete [] hashAll ;
243
[300]244      }
245      // using OASIS
246      else
247      {
[1152]248        int size;
[1130]249        int myColor;
[1234]250        int* srvGlobalRanks;
[490]251        if (!is_MPI_Initialized) oasis_init(CXios::xiosCodeId);
252
[359]253        CTimer::get("XIOS").resume() ;
[655]254        MPI_Comm localComm;
255        oasis_get_localcomm(localComm);
[1234]256        MPI_Comm_rank(localComm,&rank_) ;
[655]257
[1234]258//      (1) Create server intraComm
[1130]259        if (!CXios::usingServer2)
[1167]260        {
[1130]261          MPI_Comm_dup(localComm, &intraComm);
[1167]262        }
[1130]263        else
264        {
[1180]265          int globalRank;
[1152]266          MPI_Comm_size(localComm,&size) ;
[1180]267          MPI_Comm_rank(CXios::globalComm,&globalRank) ;
[1234]268          srvGlobalRanks = new int[size] ;
269          MPI_Allgather(&globalRank, 1, MPI_INT, srvGlobalRanks, 1, MPI_INT, localComm) ;
[1152]270
[1234]271          int reqNbProc = size*CXios::ratioServer2/100.;
272          if (reqNbProc < 1 || reqNbProc == size)
[1130]273          {
[1234]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);
[1130]278          }
279          else
280          {
[1243]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;
[1234]295            for (int i=0; i<size; i++)
296            {
[1243]297              if (i >= firstSndSrvRank)
[1234]298              {
[1243]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
[1234]313                sndServerGlobalRanks.push_back(srvGlobalRanks[i]);
314              }
315              else
316              {
317                if (globalRank == srvGlobalRanks[i]) serverLevel=1;
318              }
319            }
[1243]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;
[1234]340            MPI_Comm_split(localComm, myColor, rank_, &intraComm) ;
[1130]341          }
[1234]342        }
[1130]343
[300]344        string codesId=CXios::getin<string>("oasis_codes_id") ;
345        vector<string> splitted ;
[483]346        boost::split( splitted, codesId, boost::is_any_of(","), boost::token_compress_on ) ;
[300]347        vector<string>::iterator it ;
348
349        MPI_Comm newComm ;
350        int globalRank ;
351        MPI_Comm_rank(CXios::globalComm,&globalRank);
[490]352
[1234]353//      (2) Create interComms with models
[300]354        for(it=splitted.begin();it!=splitted.end();it++)
355        {
356          oasis_get_intercomm(newComm,*it) ;
[1234]357          if ( serverLevel == 0 || serverLevel == 1)
[1150]358          {
[1130]359            interCommLeft.push_back(newComm) ;
[1150]360            if (rank_==0) MPI_Send(&globalRank,1,MPI_INT,0,0,newComm) ;
361          }
[1180]362        }
363
[1234]364//      (3) Create interComms between primary and secondary servers
[1243]365        int intraCommSize, intraCommRank ;
366        MPI_Comm_size(intraComm,&intraCommSize) ;
367        MPI_Comm_rank(intraComm, &intraCommRank) ;
368
[1180]369        if (serverLevel == 1)
370        {
371          for (int i = 0; i < sndServerGlobalRanks.size(); ++i)
[1130]372          {
[1180]373            int srvSndLeader = sndServerGlobalRanks[i];
[1243]374            info(50)<<"intercommCreate::client (server level 1) "<<globalRank<<" intraCommSize : "<<intraCommSize
375                <<" intraCommRank :"<<intraCommRank<<"  clientLeader "<< srvSndLeader<<endl ;
[1180]376            MPI_Intercomm_create(intraComm, 0, CXios::globalComm, srvSndLeader, 0, &newComm) ;
377            interCommRight.push_back(newComm) ;
[1130]378          }
[300]379        }
[1180]380        else if (serverLevel == 2)
381        {
[1244]382          info(50)<<"intercommCreate::server (server level 2)"<<globalRank<<" intraCommSize : "<<intraCommSize
[1243]383                   <<" intraCommRank :"<<intraCommRank<<"  clientLeader "<< srvGlobalRanks[0] <<endl ;
[1180]384          MPI_Intercomm_create(intraComm, 0, CXios::globalComm, srvGlobalRanks[0], 0, &newComm) ;
385          interCommLeft.push_back(newComm) ;
386        }
[1184]387        if (CXios::usingServer2) delete [] srvGlobalRanks ;
388        oasis_enddef() ;
[300]389      }
[490]390
[1184]391
[1152]392      MPI_Comm_rank(intraComm, &rank) ;
393      if (rank==0) isRoot=true;
[490]394      else isRoot=false;
[492]395     
396      eventScheduler = new CEventScheduler(intraComm) ;
[300]397    }
[490]398
[300]399    void CServer::finalize(void)
400    {
[361]401      CTimer::get("XIOS").suspend() ;
[697]402     
[492]403      delete eventScheduler ;
[655]404
405      for (std::list<MPI_Comm>::iterator it = contextInterComms.begin(); it != contextInterComms.end(); it++)
406        MPI_Comm_free(&(*it));
[983]407
[1071]408      for (std::list<MPI_Comm>::iterator it = contextIntraComms.begin(); it != contextIntraComms.end(); it++)
409        MPI_Comm_free(&(*it));
410
[992]411//      for (std::list<MPI_Comm>::iterator it = interComm.begin(); it != interComm.end(); it++)
412//        MPI_Comm_free(&(*it));
413
[1077]414//        for (std::list<MPI_Comm>::iterator it = interCommLeft.begin(); it != interCommLeft.end(); it++)
415//          MPI_Comm_free(&(*it));
[983]416
[1054]417        for (std::list<MPI_Comm>::iterator it = interCommRight.begin(); it != interCommRight.end(); it++)
418          MPI_Comm_free(&(*it));
[992]419
[655]420      MPI_Comm_free(&intraComm);
421
[300]422      if (!is_MPI_Initialized)
[490]423      {
[300]424        if (CXios::usingOasis) oasis_finalize();
425        else MPI_Finalize() ;
426      }
[347]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  ;
[1158]430      report(100)<<CTimer::getAllCumulatedTime()<<endl ;
[300]431    }
[490]432
[300]433     void CServer::eventLoop(void)
434     {
435       bool stop=false ;
[490]436
[347]437       CTimer::get("XIOS server").resume() ;
[300]438       while(!stop)
439       {
440         if (isRoot)
441         {
442           listenContext();
[1148]443           listenRootContext();
[300]444           if (!finished) listenFinalize() ;
445         }
446         else
447         {
448           listenRootContext();
449           if (!finished) listenRootFinalize() ;
450         }
[490]451
[300]452         contextEventLoop() ;
453         if (finished && contextList.empty()) stop=true ;
[956]454         eventScheduler->checkEvent() ;
[300]455       }
[347]456       CTimer::get("XIOS server").suspend() ;
[300]457     }
[490]458
[300]459     void CServer::listenFinalize(void)
460     {
[992]461        list<MPI_Comm>::iterator it, itr;
[300]462        int msg ;
463        int flag ;
[490]464
[992]465        for(it=interCommLeft.begin();it!=interCommLeft.end();it++)
[300]466        {
467           MPI_Status status ;
[347]468           traceOff() ;
[300]469           MPI_Iprobe(0,0,*it,&flag,&status) ;
[347]470           traceOn() ;
[300]471           if (flag==true)
472           {
[1054]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              }
[655]480              MPI_Comm_free(&(*it));
[992]481              interCommLeft.erase(it) ;
[300]482              break ;
483            }
484         }
[490]485
[1054]486         if (interCommLeft.empty())
[300]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] ;
[490]492
[300]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     }
[490]501
502
[300]503     void CServer::listenRootFinalize()
504     {
505        int flag ;
506        MPI_Status status ;
507        int msg ;
[490]508
[347]509        traceOff() ;
[300]510        MPI_Iprobe(0,4,intraComm, &flag, &status) ;
[347]511        traceOn() ;
[300]512        if (flag==true)
513        {
514           MPI_Recv(&msg,1,MPI_INT,0,4,intraComm,&status) ;
515           finished=true ;
516        }
517      }
[490]518
[300]519     void CServer::listenContext(void)
520     {
[490]521
[300]522       MPI_Status status ;
523       int flag ;
[1158]524       static char* buffer ;
[300]525       static MPI_Request request ;
526       static bool recept=false ;
527       int rank ;
528       int count ;
[490]529
[300]530       if (recept==false)
531       {
[347]532         traceOff() ;
[300]533         MPI_Iprobe(MPI_ANY_SOURCE,1,CXios::globalComm, &flag, &status) ;
[347]534         traceOn() ;
[490]535         if (flag==true)
[300]536         {
537           rank=status.MPI_SOURCE ;
538           MPI_Get_count(&status,MPI_CHAR,&count) ;
539           buffer=new char[count] ;
[1158]540           MPI_Irecv((void*)buffer,count,MPI_CHAR,rank,1,CXios::globalComm,&request) ;
[490]541           recept=true ;
[300]542         }
543       }
544       else
545       {
[347]546         traceOff() ;
[300]547         MPI_Test(&request,&flag,&status) ;
[347]548         traceOn() ;
[300]549         if (flag==true)
550         {
551           rank=status.MPI_SOURCE ;
552           MPI_Get_count(&status,MPI_CHAR,&count) ;
[1158]553           recvContextMessage((void*)buffer,count) ;
554           delete [] buffer ;
[490]555           recept=false ;
[300]556         }
557       }
558     }
[490]559
[300]560     void CServer::recvContextMessage(void* buff,int count)
561     {
[983]562       static map<string,contextMessage> recvContextId;
[300]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 ;
[490]570
[300]571       it=recvContextId.find(id) ;
572       if (it==recvContextId.end())
[490]573       {
[300]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 ;
[490]578       }
[300]579       it->second.nbRecv+=1 ;
580       it->second.leaderRank+=clientLeader ;
[490]581
[300]582       if (it->second.nbRecv==nbMessage)
[490]583       {
[300]584         int size ;
585         MPI_Comm_size(intraComm,&size) ;
[1148]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] ;
[490]590
[1148]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++)
[300]600         {
[1148]601           MPI_Isend(sendBuff,count,MPI_CHAR,i,2,intraComm,&requests[i]) ;
[300]602         }
603
604         recvContextId.erase(it) ;
605         delete [] requests ;
606         delete [] status ;
607
608       }
[490]609     }
610
[300]611     void CServer::listenRootContext(void)
612     {
613       MPI_Status status ;
614       int flag ;
[1270]615       static std::vector<void*> buffers;
616       static std::vector<MPI_Request> requests ;
617       static std::vector<int> counts ;
[1271]618       static std::vector<bool> isEventRegistered ;
[1270]619       MPI_Request request;
620
[300]621       int rank ;
622       const int root=0 ;
[1148]623       boost::hash<string> hashString;
624       size_t hashId = hashString("RegisterContext");
[490]625
[1270]626       // (1) Receive context id from the root, save it into a buffer
[1271]627       traceOff() ;
628       MPI_Iprobe(root,2,intraComm, &flag, &status) ;
629       traceOn() ;
630       if (flag==true)
[300]631       {
[1271]632         counts.push_back(0);
633         MPI_Get_count(&status,MPI_CHAR,&(counts.back())) ;
634         buffers.push_back(new char[counts.back()]) ;
635         requests.push_back(request);
636         MPI_Irecv((void*)(buffers.back()),counts.back(),MPI_CHAR,root,2,intraComm,&(requests.back())) ;
637         isEventRegistered.push_back(false);
638         nbContexts++;
[300]639       }
[1271]640
641       for (int ctxNb = 0; ctxNb < nbContexts; ctxNb++ )
[300]642       {
[1271]643         // (2) If context id is received, register an event
644         MPI_Test(&requests[ctxNb],&flag,&status) ;
645         if (flag==true && !isEventRegistered[ctxNb])
[300]646         {
[1271]647           eventScheduler->registerEvent(ctxNb,hashId);
648           isEventRegistered[ctxNb] = true;
[300]649         }
[1271]650         // (3) If event has been scheduled, call register context
651         if (eventScheduler->queryEvent(ctxNb,hashId))
652         {
653           registerContext(buffers[ctxNb],counts[ctxNb]) ;
654           delete [] buffers[ctxNb] ;
655         }
[300]656       }
[1271]657
[490]658     }
659
[655]660     void CServer::registerContext(void* buff, int count, int leaderRank)
[300]661     {
662       string contextId;
[655]663       CBufferIn buffer(buff, count);
[1148]664//       buffer >> contextId;
665       buffer >> contextId>>leaderRank;
[983]666       CContext* context;
[300]667
[680]668       info(20) << "CServer : Register new Context : " << contextId << endl;
[490]669
[680]670       if (contextList.find(contextId) != contextList.end())
671         ERROR("void CServer::registerContext(void* buff, int count, int leaderRank)",
672               << "Context '" << contextId << "' has already been registred");
[490]673
[983]674       context=CContext::create(contextId);
[655]675       contextList[contextId]=context;
676
[1148]677       // Primary or classical server: create communication channel with a client
678       // (1) create interComm (with a client)
679       // (2) initialize client and server (contextClient and contextServer)
[1071]680       MPI_Comm inter;
[1054]681       if (serverLevel < 2)
682       {
683         MPI_Comm contextInterComm;
684         MPI_Intercomm_create(intraComm, 0, CXios::globalComm, leaderRank, 10+leaderRank, &contextInterComm);
685         MPI_Intercomm_merge(contextInterComm,1,&inter);
686         MPI_Barrier(inter);
687         MPI_Comm_free(&inter);
688         context->initServer(intraComm,contextInterComm);
689         contextInterComms.push_back(contextInterComm);
[1071]690
[1054]691       }
[1148]692       // Secondary server: create communication channel with a primary server
693       // (1) duplicate interComm with a primary server
694       // (2) initialize client and server (contextClient and contextServer)
695       // Remark: in the case of the secondary server there is no need to create an interComm calling MPI_Intercomm_create,
696       //         because interComm of CContext is defined on the same processes as the interComm of CServer.
697       //         So just duplicate it.
[1054]698       else if (serverLevel == 2)
699       {
[1071]700         MPI_Comm_dup(interCommLeft.front(), &inter);
701         contextInterComms.push_back(inter);
702         context->initServer(intraComm, contextInterComms.back());
[1054]703       }
704
[1148]705       // Primary server:
706       // (1) send create context message to secondary servers
707       // (2) initialize communication channels with secondary servers (create contextClient and contextServer)
[1021]708       if (serverLevel == 1)
[983]709       {
[1054]710         int i = 0, size;
711         CMessage msg;
712         int messageSize;
713         MPI_Comm_size(intraComm, &size) ;
714         for (std::list<MPI_Comm>::iterator it = interCommRight.begin(); it != interCommRight.end(); it++, ++i)
715         {
716           StdString str = contextId +"_server_" + boost::lexical_cast<string>(i);
[1077]717           msg<<str<<size<<rank_ ;
[1054]718           messageSize = msg.size() ;
719           buff = new char[messageSize] ;
720           CBufferOut buffer(buff,messageSize) ;
721           buffer<<msg ;
[1152]722           MPI_Send(buff, buffer.count(), MPI_CHAR, sndServerGlobalRanks[i], 1, CXios::globalComm) ;
[1071]723           MPI_Comm_dup(*it, &inter);
724           contextInterComms.push_back(inter);
725           MPI_Comm_dup(intraComm, &inter);
726           contextIntraComms.push_back(inter);
727           context->initClient(contextIntraComms.back(), contextInterComms.back()) ;
[1054]728           delete [] buff ;
729         }
[983]730       }
[490]731     }
732
[300]733     void CServer::contextEventLoop(void)
734     {
[1130]735       bool isFinalized ;
[300]736       map<string,CContext*>::iterator it ;
[983]737
[490]738       for(it=contextList.begin();it!=contextList.end();it++)
[300]739       {
[1130]740         isFinalized=it->second->isFinalized();
741         if (isFinalized)
[300]742         {
743           contextList.erase(it) ;
744           break ;
745         }
[1054]746         else
[1139]747           it->second->checkBuffersAndListen();
[300]748       }
749     }
[490]750
[1148]751     //! Get rank of the current process in the intraComm
[490]752     int CServer::getRank()
753     {
[1167]754       int rank;
755       MPI_Comm_rank(intraComm,&rank);
756       return rank;
[490]757     }
758
[1168]759     vector<int>& CServer::getSecondaryServerGlobalRanks()
760     {
761       return sndServerGlobalRanks;
762     }
763
[523]764    /*!
765    * Open a file specified by a suffix and an extension and use it for the given file buffer.
766    * The file name will be suffix+rank+extension.
767    *
768    * \param fileName[in] protype file name
769    * \param ext [in] extension of the file
770    * \param fb [in/out] the file buffer
771    */
772    void CServer::openStream(const StdString& fileName, const StdString& ext, std::filebuf* fb)
773    {
774      StdStringStream fileNameClient;
775      int numDigit = 0;
776      int size = 0;
[1021]777      int id;
[523]778      MPI_Comm_size(CXios::globalComm, &size);
779      while (size)
780      {
781        size /= 10;
782        ++numDigit;
783      }
[1167]784      id = rank_; //getRank();
[497]785
[1021]786      fileNameClient << fileName << "_" << std::setfill('0') << std::setw(numDigit) << id << ext;
[523]787      fb->open(fileNameClient.str().c_str(), std::ios::out);
788      if (!fb->is_open())
789        ERROR("void CServer::openStream(const StdString& fileName, const StdString& ext, std::filebuf* fb)",
790              << std::endl << "Can not open <" << fileNameClient << "> file to write the server log(s).");
791    }
[490]792
[523]793    /*!
794    * \brief Open a file stream to write the info logs
795    * Open a file stream with a specific file name suffix+rank
796    * to write the info logs.
797    * \param fileName [in] protype file name
798    */
799    void CServer::openInfoStream(const StdString& fileName)
800    {
801      std::filebuf* fb = m_infoStream.rdbuf();
802      openStream(fileName, ".out", fb);
[490]803
[523]804      info.write2File(fb);
805      report.write2File(fb);
806    }
[490]807
[523]808    //! Write the info logs to standard output
809    void CServer::openInfoStream()
810    {
811      info.write2StdOut();
812      report.write2StdOut();
813    }
[490]814
[523]815    //! Close the info logs file if it opens
816    void CServer::closeInfoStream()
817    {
818      if (m_infoStream.is_open()) m_infoStream.close();
819    }
820
821    /*!
822    * \brief Open a file stream to write the error log
823    * Open a file stream with a specific file name suffix+rank
824    * to write the error log.
825    * \param fileName [in] protype file name
826    */
827    void CServer::openErrorStream(const StdString& fileName)
828    {
829      std::filebuf* fb = m_errorStream.rdbuf();
830      openStream(fileName, ".err", fb);
831
832      error.write2File(fb);
833    }
834
835    //! Write the error log to standard error output
836    void CServer::openErrorStream()
837    {
838      error.write2StdErr();
839    }
840
841    //! Close the error log file if it opens
842    void CServer::closeErrorStream()
843    {
844      if (m_errorStream.is_open()) m_errorStream.close();
845    }
[300]846}
Note: See TracBrowser for help on using the repository browser.