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

Last change on this file since 1054 was 1054, checked in by oabramkina, 7 years ago

dev: intermediate commit.

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