source: XIOS/dev/dev_trunk_omp/src/client.cpp @ 1601

Last change on this file since 1601 was 1601, checked in by yushan, 2 years ago

branch_openmp merged with trunk r1597

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