source: XIOS/trunk/src/server.cpp @ 1638

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

dev on ADA

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