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

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

save current modifications

  • 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.6 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            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      #pragma omp critical (_output)
176      printf("Client::registerContext context add = %p\n", &(*context));
177     
178     
179      StdString idServer(id);
180      idServer += "_server";
181
182      if (!CXios::isServer)  // server mode
183      {     
184        int size,rank,globalRank ;
185        //size_t message_size ;
186        //int leaderRank ;
187        MPI_Comm contextInterComm ;
188
189       
190        MPI_Comm_size(contextComm,&size) ;
191        MPI_Comm_rank(contextComm,&rank) ;
192        MPI_Comm_rank(CXios::globalComm,&globalRank) ;
193        if (rank!=0) globalRank=0 ;
194
195
196        CMessage msg ;
197        msg<<idServer<<size<<globalRank ;
198
199
200        int messageSize=msg.size() ;
201        void * buff = new char[messageSize] ;
202        CBufferOut buffer(buff,messageSize) ;
203        buffer<<msg ;
204       
205       
206
207        MPI_Send(buff,buffer.count(),MPI_CHAR,serverLeader,1,CXios::globalComm) ;
208        delete [] buff ;
209             
210
211        MPI_Intercomm_create(contextComm,0,CXios::globalComm,serverLeader,10+globalRank,&contextInterComm) ;
212       
213        #pragma omp critical (_output)
214        info(10)<<"Register new Context : "<<id<<endl ;
215                     
216
217        MPI_Comm inter ;
218        MPI_Intercomm_merge(contextInterComm,0,&inter) ;
219        MPI_Barrier(inter) ;
220
221        #pragma omp critical (_output)
222        printf("Client %d context=CContext::create(%s) OK, context.identifier = %d\n", getRank(), id, context->get_identifier());
223       
224        context->initClient(contextComm,contextInterComm) ;
225       
226        #pragma omp critical (_output)
227        printf("Client %d : context->initClient(contextComm,contextInterComm) OK \n", getRank()) ;
228       
229
230        contextInterComms.push_back(contextInterComm);
231        MPI_Comm_free(&inter);
232      }
233      else  // attached mode
234      {
235        MPI_Comm contextInterComm ;
236        MPI_Comm_dup(contextComm,&contextInterComm) ;
237        CContext* contextServer = CContext::create(idServer);
238       
239        // Firstly, initialize context on client side
240        context->initClient(contextComm,contextInterComm, contextServer);
241
242        // Secondly, initialize context on server side
243        contextServer->initServer(contextComm,contextInterComm, context);
244
245        // Finally, we should return current context to context client
246        CContext::setCurrent(id);
247
248        contextInterComms.push_back(contextInterComm);
249      }
250    }
251
252    void CClient::finalize(void)
253    {
254      int rank ;
255      int msg=0 ;
256
257      MPI_Comm_rank(intraComm,&rank) ;
258       
259      if (!CXios::isServer)
260      {
261        MPI_Comm_rank(intraComm,&rank) ;
262        if (rank==0)
263        {
264          MPI_Send(&msg,1,MPI_INT,0,0,interComm) ;
265        }
266      }
267
268      for (std::list<MPI_Comm>::iterator it = contextInterComms.begin(); it != contextInterComms.end(); ++it)
269        MPI_Comm_free(&(*it));
270     
271      MPI_Comm_free(&interComm);
272      MPI_Comm_free(&intraComm);
273
274      CTimer::get("XIOS finalize").suspend() ;
275      CTimer::get("XIOS").suspend() ;
276
277      if (!is_MPI_Initialized)
278      {
279        if (CXios::usingOasis) oasis_finalize();
280        else  MPI_Finalize(); 
281      }
282     
283     
284        //info(20) << "Client "<<rank<<" : Client side context is finalized "<< endl ;
285//         report(0) <<"     Performance report : total time spent for XIOS : "<< CTimer::get("XIOS").getCumulatedTime()<<" s"<<endl ;
286//         report(0)<< "     Performance report : time spent for waiting free buffer : "<< CTimer::get("Blocking time").getCumulatedTime()<<" s"<<endl ;
287//         report(0)<< "     Performance report : Ratio : "<< CTimer::get("Blocking time").getCumulatedTime()/CTimer::get("XIOS").getCumulatedTime()*100.<<" %"<<endl ;
288//         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 ;
289// //      report(0)<< "     Memory report : Current buffer_size : "<<CXios::bufferSize<<endl ;
290//         report(0)<< "     Memory report : Minimum buffer size required : " << CClientBuffer::maxRequestSize << " bytes" << endl ;
291//         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 ;
292     
293
294   }
295
296   int CClient::getRank()
297   {
298     return rank;
299   }
300
301    /*!
302    * Open a file specified by a suffix and an extension and use it for the given file buffer.
303    * The file name will be suffix+rank+extension.
304    *
305    * \param fileName[in] protype file name
306    * \param ext [in] extension of the file
307    * \param fb [in/out] the file buffer
308    */
309    void CClient::openStream(const StdString& fileName, const StdString& ext, std::filebuf* fb)
310    {
311      StdStringStream fileNameClient;
312      int numDigit = 0;
313      int size = 0;
314      MPI_Comm_size(CXios::globalComm, &size);
315      while (size)
316      {
317        size /= 10;
318        ++numDigit;
319      }
320
321      fileNameClient << fileName << "_" << std::setfill('0') << std::setw(numDigit) << getRank() << ext;
322      //printf("getrank() = %d, file name = %s\n", getRank(), fileNameClient.str().c_str());
323     
324        fb->open(fileNameClient.str().c_str(), std::ios::out);
325        if (!fb->is_open())
326          ERROR("void CClient::openStream(const StdString& fileName, const StdString& ext, std::filebuf* fb)",
327              << std::endl << "Can not open <" << fileNameClient << "> file to write the client log(s)."); 
328     
329     
330    }
331
332    /*!
333    * \brief Open a file stream to write the info logs
334    * Open a file stream with a specific file name suffix+rank
335    * to write the info logs.
336    * \param fileName [in] protype file name
337    */
338    void CClient::openInfoStream(const StdString& fileName)
339    {
340      std::filebuf* fb = m_infoStream.rdbuf();
341      openStream(fileName, ".out", fb);
342
343      info.write2File(fb);
344      report.write2File(fb);
345    }
346
347    //! Write the info logs to standard output
348    void CClient::openInfoStream()
349    {
350      info.write2StdOut();
351      report.write2StdOut();
352    }
353
354    //! Close the info logs file if it opens
355    void CClient::closeInfoStream()
356    {
357      if (m_infoStream.is_open()) m_infoStream.close();
358    }
359
360    /*!
361    * \brief Open a file stream to write the error log
362    * Open a file stream with a specific file name suffix+rank
363    * to write the error log.
364    * \param fileName [in] protype file name
365    */
366    void CClient::openErrorStream(const StdString& fileName)
367    {
368      std::filebuf* fb = m_errorStream.rdbuf();
369      openStream(fileName, ".err", fb);
370
371      error.write2File(fb);
372    }
373
374    //! Write the error log to standard error output
375    void CClient::openErrorStream()
376    {
377      error.write2StdErr();
378    }
379
380    //! Close the error log file if it opens
381    void CClient::closeErrorStream()
382    {
383      if (m_errorStream.is_open()) m_errorStream.close();
384    }
385}
Note: See TracBrowser for help on using the repository browser.