source: XIOS/dev/branch_openmp/src/client.cpp @ 1374

Last change on this file since 1374 was 1374, checked in by yushan, 6 years ago

unify type : MPI_Request

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