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

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