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

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

save modif for local compilation

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