source: XIOS/dev/branch_yushan/src/client.cpp @ 1080

Last change on this file since 1080 was 1080, checked in by yushan, 5 years ago

Under development : save modification

  • 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: 12.2 KB
Line 
1#include "globalScopeData.hpp"
2#include "xios_spl.hpp"
3#include "cxios.hpp"
4#include "client.hpp"
5#include <boost/functional/hash.hpp>
6#include "type.hpp"
7#include "context.hpp"
8#include "context_client.hpp"
9#include "oasis_cinterface.hpp"
10#include "mpi.hpp"
11#include "timer.hpp"
12#include "buffer_client.hpp"
13
14namespace xios
15{
16
17    MPI_Comm CClient::intraComm ;
18    MPI_Comm CClient::interComm ;
19    std::list<MPI_Comm> CClient::contextInterComms;
20    int CClient::serverLeader ;
21    bool CClient::is_MPI_Initialized ;
22    int CClient::rank = INVALID_RANK;
23    StdOFStream CClient::m_infoStream;
24    StdOFStream CClient::m_errorStream;
25
26    void CClient::initialize(const string& codeId, MPI_Comm& localComm, MPI_Comm& returnComm)
27    {
28      int initialized ;
29      MPI_Initialized(&initialized) ;
30      if (initialized) is_MPI_Initialized=true ;
31      else is_MPI_Initialized=false ;
32
33
34     
35// don't use OASIS
36      if (!CXios::usingOasis)
37      {
38// localComm doesn't given
39
40        if (localComm == MPI_COMM_NULL)
41        {
42          if (!is_MPI_Initialized)
43          {
44            //MPI_Init(NULL, NULL);
45            int return_level;
46            #ifdef _intelmpi
47            MPI_Init_thread(NULL, NULL, 3, &return_level);
48            assert(return_level == 3);
49            #elif _openmpi
50            MPI_Init_thread(NULL, NULL, 2, &return_level);
51            assert(return_level == 2);
52            #endif
53          }
54          CTimer::get("XIOS").resume() ;
55          CTimer::get("XIOS init").resume() ;
56          boost::hash<string> hashString ;
57
58          unsigned long hashClient=hashString(codeId) ;
59          unsigned long hashServer=hashString(CXios::xiosCodeId) ;
60          //hashServer=hashString("xios.x") ;
61          unsigned long* hashAll ;
62          int size ;
63          int myColor ;
64          int i,c ;
65
66          MPI_Comm_size(CXios::globalComm,&size);
67          MPI_Comm_rank(CXios::globalComm,&rank);
68
69          printf("client init : rank = %d, size = %d\n", rank, size);
70         
71
72          hashAll=new unsigned long[size] ;
73
74          MPI_Allgather(&hashClient,1,MPI_LONG,hashAll,1,MPI_LONG,CXios::globalComm) ;
75
76          map<unsigned long, int> colors ;
77          map<unsigned long, int> leaders ;
78
79          for(i=0,c=0;i<size;i++)
80          {
81            if (colors.find(hashAll[i])==colors.end())
82            {
83              colors[hashAll[i]] =c ;
84              leaders[hashAll[i]]=i ;
85              c++ ;
86            }
87          }
88
89          // Verify whether we are on server mode or not
90          CXios::setNotUsingServer();
91          for (i=0; i < size; ++i)
92          {
93            if (hashServer == hashAll[i])
94            {
95              CXios::setUsingServer();
96              break;
97            }
98          }
99
100          myColor=colors[hashClient] ;
101
102          MPI_Comm_split(CXios::globalComm,myColor,rank,&intraComm) ;
103       
104
105          if (CXios::usingServer)
106          {
107            //int clientLeader=leaders[hashClient] ;
108            serverLeader=leaders[hashServer] ;
109
110            int intraCommSize, intraCommRank ;
111            MPI_Comm_size(intraComm,&intraCommSize) ;
112            MPI_Comm_rank(intraComm,&intraCommRank) ;
113            #pragma omp critical(_output)
114            {
115              info(50)<<"intercommCreate::client "<<rank<<" intraCommSize : "<<intraCommSize
116                 <<" intraCommRank :"<<intraCommRank<<"  serverLeader "<< serverLeader
117                 <<" globalComm : "<< &(CXios::globalComm) << endl ; 
118            }
119           
120            MPI_Intercomm_create(intraComm,0,CXios::globalComm,serverLeader,0,&interComm) ;
121
122            int interCommSize, interCommRank ;
123            MPI_Comm_size(interComm,&interCommSize) ;
124            MPI_Comm_rank(interComm,&interCommRank) ;
125
126            #pragma omp critical(_output)
127            {
128              info(50)<<" interCommRank :"<<interCommRank
129                 <<" interCommSize : "<< interCommSize << endl ; 
130            }
131
132
133          }
134          else
135          {
136            MPI_Comm_dup(intraComm,&interComm) ;
137          }
138          delete [] hashAll ;
139        }
140        // localComm argument is given
141        else
142        {
143          if (CXios::usingServer)
144          {
145            //ERROR("void CClient::initialize(const string& codeId,MPI_Comm& localComm,MPI_Comm& returnComm)", << " giving a local communictor is not compatible with using server mode") ;
146          }
147          else
148          {
149            MPI_Comm_dup(localComm,&intraComm) ;
150            MPI_Comm_dup(intraComm,&interComm) ;
151          }
152        }
153      }
154      // using OASIS
155      else
156      {
157        // localComm doesn't given
158        if (localComm == MPI_COMM_NULL)
159        {
160          if (!is_MPI_Initialized) oasis_init(codeId) ;
161          oasis_get_localcomm(localComm) ;
162        }
163        MPI_Comm_dup(localComm,&intraComm) ;
164
165        CTimer::get("XIOS").resume() ;
166        CTimer::get("XIOS init").resume() ;
167
168        if (CXios::usingServer)
169        {
170          MPI_Status status ;
171          MPI_Comm_rank(intraComm,&rank) ;
172
173          oasis_get_intercomm(interComm,CXios::xiosCodeId) ;
174          if (rank==0) MPI_Recv(&serverLeader,1, MPI_INT, 0, 0, interComm, &status) ;
175          MPI_Bcast(&serverLeader,1,MPI_INT,0,intraComm) ;
176
177        }
178        else MPI_Comm_dup(intraComm,&interComm) ;
179      }
180
181      MPI_Comm_dup(intraComm,&returnComm) ;
182
183    }
184
185
186    void CClient::registerContext(const string& id,MPI_Comm contextComm)
187    {
188      //info(50) << "Client "<<getRank() << " start registerContext using info output" << endl;
189      printf("Client %d start registerContext\n", getRank());
190      //printf("Client start registerContext\n");
191
192      CContext::setCurrent(id) ;
193      printf("Client %d CContext::setCurrent OK\n", getRank());
194      CContext* context = CContext::create(id);
195      printf("Client %d context=CContext::create(%s) OK, *context = %p\n", getRank(), id, &(*context));
196     
197      StdString idServer(id);
198      idServer += "_server";
199
200      if (!CXios::isServer)  // server mode
201      {     
202        int size,rank,globalRank ;
203        size_t message_size ;
204        int leaderRank ;
205        MPI_Comm contextInterComm ;
206
207       
208        MPI_Comm_size(contextComm,&size) ;
209        MPI_Comm_rank(contextComm,&rank) ;
210        MPI_Comm_rank(CXios::globalComm,&globalRank) ;
211        if (rank!=0) globalRank=0 ;
212
213
214        CMessage msg ;
215        msg<<idServer<<size<<globalRank ;
216
217
218        int messageSize=msg.size() ;
219        void * buff = new char[messageSize] ;
220        CBufferOut buffer(buff,messageSize) ;
221        buffer<<msg ;
222       
223       
224
225        MPI_Send(buff,buffer.count(),MPI_CHAR,serverLeader,1,CXios::globalComm) ;
226        delete [] buff ;
227             
228
229        MPI_Intercomm_create(contextComm,0,CXios::globalComm,serverLeader,10+globalRank,&contextInterComm) ;
230       
231        #pragma omp critical (_output)
232        info(10)<<"Register new Context : "<<id<<endl ;
233                     
234
235        MPI_Comm inter ;
236        MPI_Intercomm_merge(contextInterComm,0,&inter) ;
237        MPI_Barrier(inter) ;
238
239        #pragma omp critical (_output)
240        printf("Client %d : MPI_Intercomm_merge OK \n", getRank()) ;
241       
242       
243        context->initClient(contextComm,contextInterComm) ;
244        #pragma omp critical (_output)
245        printf("Client %d : context->initClient(contextComm,contextInterComm) OK \n", getRank()) ;
246       
247
248        contextInterComms.push_back(contextInterComm);
249        MPI_Comm_free(&inter);
250      }
251      else  // attached mode
252      {
253        MPI_Comm contextInterComm ;
254        MPI_Comm_dup(contextComm,&contextInterComm) ;
255        CContext* contextServer = CContext::create(idServer);
256       
257        // Firstly, initialize context on client side
258        context->initClient(contextComm,contextInterComm, contextServer);
259
260        // Secondly, initialize context on server side
261        contextServer->initServer(contextComm,contextInterComm, context);
262
263        // Finally, we should return current context to context client
264        CContext::setCurrent(id);
265
266        contextInterComms.push_back(contextInterComm);
267      }
268    }
269
270    void CClient::finalize(void)
271    {
272      int rank ;
273      int msg=0 ;
274
275      MPI_Comm_rank(intraComm,&rank) ;
276       
277      if (!CXios::isServer)
278      {
279        MPI_Comm_rank(intraComm,&rank) ;
280        if (rank==0)
281        {
282          MPI_Send(&msg,1,MPI_INT,0,0,interComm) ;
283        }
284      }
285
286      for (std::list<MPI_Comm>::iterator it = contextInterComms.begin(); it != contextInterComms.end(); ++it)
287        MPI_Comm_free(&(*it));
288     
289      MPI_Comm_free(&interComm);
290      MPI_Comm_free(&intraComm);
291
292      CTimer::get("XIOS finalize").suspend() ;
293      CTimer::get("XIOS").suspend() ;
294
295      if (!is_MPI_Initialized)
296      {
297        if (CXios::usingOasis) oasis_finalize();
298        else  MPI_Finalize(); 
299      }
300     
301      info(20) << "Client side context is finalized"<<endl ;
302      report(0) <<" Performance report : total time spent for XIOS : "<< CTimer::get("XIOS").getCumulatedTime()<<" s"<<endl ;
303      report(0)<< " Performance report : time spent for waiting free buffer : "<< CTimer::get("Blocking time").getCumulatedTime()<<" s"<<endl ;
304      report(0)<< " Performance report : Ratio : "<< CTimer::get("Blocking time").getCumulatedTime()/CTimer::get("XIOS").getCumulatedTime()*100.<<" %"<<endl ;
305      report(0)<< " Performance report : This ratio must be close to zero. Otherwise it may be usefull to increase buffer size or numbers of server"<<endl ;
306//      report(0)<< " Memory report : Current buffer_size : "<<CXios::bufferSize<<endl ;
307      report(0)<< " Memory report : Minimum buffer size required : " << CClientBuffer::maxRequestSize << " bytes" << endl ;
308      report(0)<< " Memory report : increasing it by a factor will increase performance, depending of the volume of data wrote in file at each time step of the file"<<endl ;
309   }
310
311   int CClient::getRank()
312   {
313     return rank;
314   }
315
316    /*!
317    * Open a file specified by a suffix and an extension and use it for the given file buffer.
318    * The file name will be suffix+rank+extension.
319    *
320    * \param fileName[in] protype file name
321    * \param ext [in] extension of the file
322    * \param fb [in/out] the file buffer
323    */
324    void CClient::openStream(const StdString& fileName, const StdString& ext, std::filebuf* fb)
325    {
326      StdStringStream fileNameClient;
327      int numDigit = 0;
328      int size = 0;
329      MPI_Comm_size(CXios::globalComm, &size);
330      while (size)
331      {
332        size /= 10;
333        ++numDigit;
334      }
335
336      fileNameClient << fileName << "_" << std::setfill('0') << std::setw(numDigit) << getRank() << ext;
337      printf("getrank() = %d, file name = %s\n", getRank(), fileNameClient.str().c_str());
338     
339        fb->open(fileNameClient.str().c_str(), std::ios::out);
340        if (!fb->is_open())
341          ERROR("void CClient::openStream(const StdString& fileName, const StdString& ext, std::filebuf* fb)",
342              << std::endl << "Can not open <" << fileNameClient << "> file to write the client log(s)."); 
343     
344     
345    }
346
347    /*!
348    * \brief Open a file stream to write the info logs
349    * Open a file stream with a specific file name suffix+rank
350    * to write the info logs.
351    * \param fileName [in] protype file name
352    */
353    void CClient::openInfoStream(const StdString& fileName)
354    {
355      std::filebuf* fb = m_infoStream.rdbuf();
356      openStream(fileName, ".out", fb);
357
358      info.write2File(fb);
359      report.write2File(fb);
360    }
361
362    //! Write the info logs to standard output
363    void CClient::openInfoStream()
364    {
365      info.write2StdOut();
366      report.write2StdOut();
367    }
368
369    //! Close the info logs file if it opens
370    void CClient::closeInfoStream()
371    {
372      if (m_infoStream.is_open()) m_infoStream.close();
373    }
374
375    /*!
376    * \brief Open a file stream to write the error log
377    * Open a file stream with a specific file name suffix+rank
378    * to write the error log.
379    * \param fileName [in] protype file name
380    */
381    void CClient::openErrorStream(const StdString& fileName)
382    {
383      std::filebuf* fb = m_errorStream.rdbuf();
384      openStream(fileName, ".err", fb);
385
386      error.write2File(fb);
387    }
388
389    //! Write the error log to standard error output
390    void CClient::openErrorStream()
391    {
392      error.write2StdErr();
393    }
394
395    //! Close the error log file if it opens
396    void CClient::closeErrorStream()
397    {
398      if (m_errorStream.is_open()) m_errorStream.close();
399    }
400}
Note: See TracBrowser for help on using the repository browser.