source: XIOS/dev/dev_olga/src/client.cpp @ 992

Last change on this file since 992 was 992, checked in by oabramkina, 6 years ago

First rebond on the secondary server pool. XIOS finalizes correctly.

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