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

Last change on this file since 1037 was 1037, checked in by yushan, 4 years ago

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