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

Last change on this file since 1074 was 1074, checked in by yushan, 5 years ago

save modif

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