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

Last change on this file since 983 was 983, checked in by oabramkina, 5 years ago

My branch

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