source: XIOS/trunk/src/client.cpp @ 597

Last change on this file since 597 was 597, checked in by rlacroix, 9 years ago

Add basic infrastructure for servers to clients communication.

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