source: XIOS/dev/dev_ym/XIOS_ONE_SIDED/src/context_client.cpp @ 1554

Last change on this file since 1554 was 1554, checked in by ymipsl, 6 years ago
  • Remove temporary buffered event
  • Some cleaning

YM

  • 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: 15.0 KB
Line 
1#include "xios_spl.hpp"
2#include "context_client.hpp"
3#include "context_server.hpp"
4#include "event_client.hpp"
5#include "buffer_out.hpp"
6#include "buffer_client.hpp"
7#include "type.hpp"
8#include "event_client.hpp"
9#include "context.hpp"
10#include "mpi.hpp"
11#include "timer.hpp"
12#include "cxios.hpp"
13#include "server.hpp"
14
15namespace xios
16{
17    /*!
18    \param [in] parent Pointer to context on client side
19    \param [in] intraComm_ communicator of group client
20    \param [in] interComm_ communicator of group server
21    \cxtSer [in] cxtSer Pointer to context of server side. (It is only used in case of attached mode).
22    */
23    CContextClient::CContextClient(CContext* parent, MPI_Comm intraComm_, MPI_Comm interComm_, CContext* cxtSer)
24     : mapBufferSize_(), parentServer(cxtSer), maxBufferedEvents(4)
25    {
26      pureOneSided=CXios::getin<bool>("pure_one_sided",false); // pure one sided communication (for test)
27      if (isAttachedModeEnabled()) pureOneSided=false ; // no one sided in attach mode
28     
29      context = parent;
30      intraComm = intraComm_;
31      interComm = interComm_;
32      MPI_Comm_rank(intraComm, &clientRank);
33      MPI_Comm_size(intraComm, &clientSize);
34
35      int flag;
36      MPI_Comm_test_inter(interComm, &flag);
37      if (flag) MPI_Comm_remote_size(interComm, &serverSize);
38      else  MPI_Comm_size(interComm, &serverSize);
39
40      computeLeader(clientRank, clientSize, serverSize, ranksServerLeader, ranksServerNotLeader);
41
42      if (flag) MPI_Intercomm_merge(interComm_,false,&interCommMerged) ;
43
44      MPI_Comm_split(intraComm_,clientRank,clientRank, &commSelf) ;
45
46      timeLine = 1;
47    }
48
49    void CContextClient::computeLeader(int clientRank, int clientSize, int serverSize,
50                                       std::list<int>& rankRecvLeader,
51                                       std::list<int>& rankRecvNotLeader)
52    {
53      if ((0 == clientSize) || (0 == serverSize)) return;
54
55      if (clientSize < serverSize)
56      {
57        int serverByClient = serverSize / clientSize;
58        int remain = serverSize % clientSize;
59        int rankStart = serverByClient * clientRank;
60
61        if (clientRank < remain)
62        {
63          serverByClient++;
64          rankStart += clientRank;
65        }
66        else
67          rankStart += remain;
68
69        for (int i = 0; i < serverByClient; i++)
70          rankRecvLeader.push_back(rankStart + i);
71
72        rankRecvNotLeader.resize(0);
73      }
74      else
75      {
76        int clientByServer = clientSize / serverSize;
77        int remain = clientSize % serverSize;
78
79        if (clientRank < (clientByServer + 1) * remain)
80        {
81          if (clientRank % (clientByServer + 1) == 0)
82            rankRecvLeader.push_back(clientRank / (clientByServer + 1));
83          else
84            rankRecvNotLeader.push_back(clientRank / (clientByServer + 1));
85        }
86        else
87        {
88          int rank = clientRank - (clientByServer + 1) * remain;
89          if (rank % clientByServer == 0)
90            rankRecvLeader.push_back(remain + rank / clientByServer);
91          else
92            rankRecvNotLeader.push_back(remain + rank / clientByServer);
93        }
94      }
95    }
96
97    /*!
98    In case of attached mode, the current context must be reset to context for client
99    \param [in] event Event sent to server
100    */
101    void CContextClient::sendEvent(CEventClient& event)
102    {
103      list<int> ranks = event.getRanks();
104
105      if (CXios::checkEventSync)
106      {
107        int typeId, classId, typeId_in, classId_in, timeLine_out;
108        typeId_in=event.getTypeId() ;
109        classId_in=event.getClassId() ;
110//        MPI_Allreduce(&timeLine,&timeLine_out, 1, MPI_UINT64_T, MPI_SUM, intraComm) ; // MPI_UINT64_T standardized by MPI 3
111        MPI_Allreduce(&timeLine,&timeLine_out, 1, MPI_LONG_LONG_INT, MPI_SUM, intraComm) ; 
112        MPI_Allreduce(&typeId_in,&typeId, 1, MPI_INT, MPI_SUM, intraComm) ;
113        MPI_Allreduce(&classId_in,&classId, 1, MPI_INT, MPI_SUM, intraComm) ;
114        if (typeId/clientSize!=event.getTypeId() || classId/clientSize!=event.getClassId() || timeLine_out/clientSize!=timeLine)
115        {
116           ERROR("void CContextClient::sendEvent(CEventClient& event)",
117               << "Event are not coherent between client.");
118        }
119      }
120
121      if (!event.isEmpty())
122      {
123        list<int> sizes = event.getSizes();
124
125         // We force the getBuffers call to be non-blocking on classical servers
126        list<CBufferOut*> buffList;
127        getBuffers(timeLine, ranks, sizes, buffList) ;
128
129        event.send(timeLine, sizes, buffList);
130        unlockBuffers(ranks) ;
131         
132        checkBuffers(ranks);
133
134        if (isAttachedModeEnabled()) // couldBuffer is always true in attached mode
135        {
136          waitEvent(ranks);
137          CContext::setCurrent(context->getId());
138        }
139      }
140
141      timeLine++;
142    }
143
144    /*!
145    If client is also server (attached mode), after sending event, it should process right away
146    the incoming event.
147    \param [in] ranks list rank of server connected this client
148    */
149    void CContextClient::waitEvent(list<int>& ranks)
150    {
151      parentServer->server->setPendingEvent();
152      while (checkBuffers(ranks))
153      {
154        parentServer->server->listen();
155        parentServer->server->checkPendingRequest();
156      }
157
158      while (parentServer->server->hasPendingEvent())
159      {
160       parentServer->server->eventLoop();
161      }
162    }
163
164    /*!
165     * Get buffers for each connection to the servers. This function blocks until there is enough room in the buffers unless
166     * it is explicitly requested to be non-blocking.
167     *
168     *
169     * \param [in] timeLine time line of the event which will be sent to servers
170     * \param [in] serverList list of rank of connected server
171     * \param [in] sizeList size of message corresponding to each connection
172     * \param [out] retBuffers list of buffers that can be used to store an event
173     * \param [in] nonBlocking whether this function should be non-blocking
174     * \return whether the already allocated buffers could be used
175    */
176    bool CContextClient::getBuffers(const size_t timeLine, const list<int>& serverList, const list<int>& sizeList, list<CBufferOut*>& retBuffers,
177                                    bool nonBlocking /*= false*/)
178    {
179      list<int>::const_iterator itServer, itSize;
180      list<CClientBuffer*> bufferList;
181      map<int,CClientBuffer*>::const_iterator it;
182      list<CClientBuffer*>::iterator itBuffer;
183      bool areBuffersFree;
184
185      for (itServer = serverList.begin(); itServer != serverList.end(); itServer++)
186      {
187        it = buffers.find(*itServer);
188        if (it == buffers.end())
189        {
190          newBuffer(*itServer);
191          it = buffers.find(*itServer);
192        }
193        bufferList.push_back(it->second);
194      }
195
196      CTimer::get("Blocking time").resume();
197      do
198      {
199        areBuffersFree = true;
200        for (itBuffer = bufferList.begin(), itSize = sizeList.begin(); itBuffer != bufferList.end(); itBuffer++, itSize++)
201        {
202          areBuffersFree &= (*itBuffer)->isBufferFree(*itSize);
203        }
204
205        if (!areBuffersFree)
206        {
207          for (itBuffer = bufferList.begin(); itBuffer != bufferList.end(); itBuffer++) (*itBuffer)->unlockBuffer();
208          checkBuffers();
209          if (CServer::serverLevel == 0)  context->server->listen();
210          else if (CServer::serverLevel == 1)
211          {
212            context->server->listen();
213            for (int i = 0; i < context->serverPrimServer.size(); ++i)  context->serverPrimServer[i]->listen();
214            CServer::contextEventLoop(false) ; // avoid dead-lock at finalize...
215          }
216
217          else if (CServer::serverLevel == 2) context->server->listen();
218
219        }
220      } while (!areBuffersFree && !nonBlocking);
221      CTimer::get("Blocking time").suspend();
222
223      if (areBuffersFree)
224      {
225        for (itBuffer = bufferList.begin(), itSize = sizeList.begin(); itBuffer != bufferList.end(); itBuffer++, itSize++)
226          retBuffers.push_back((*itBuffer)->getBuffer(timeLine, *itSize));
227      }
228      return areBuffersFree;
229   }
230
231   /*!
232   Make a new buffer for a certain connection to server with specific rank
233   \param [in] rank rank of connected server
234   */
235   void CContextClient::newBuffer(int rank)
236   {
237      if (!mapBufferSize_.count(rank))
238      {
239        error(0) << "WARNING: Unexpected request for buffer to communicate with server " << rank << std::endl;
240        mapBufferSize_[rank] = CXios::minBufferSize;
241        maxEventSizes[rank] = CXios::minBufferSize;
242      }
243      CClientBuffer* buffer = buffers[rank] = new CClientBuffer(interComm, rank, mapBufferSize_[rank], maxEventSizes[rank]);
244      // Notify the server
245      CBufferOut* bufOut = buffer->getBuffer(0, sizeof(StdSize));
246      bufOut->put(mapBufferSize_[rank]); // Stupid C++
247      buffer->checkBuffer(true);
248
249      if (!isAttachedModeEnabled()) // create windows only in server mode
250      {
251        MPI_Comm OneSidedInterComm, oneSidedComm ;
252        MPI_Intercomm_create(commSelf, 0, interCommMerged, clientSize+rank, 0, &OneSidedInterComm );
253        MPI_Intercomm_merge(OneSidedInterComm,false,&oneSidedComm);
254        buffer->createWindows(oneSidedComm) ;
255      }
256       
257   }
258
259   /*!
260   Verify state of buffers. Buffer is under pending state if there is no message on it
261   \return state of buffers, pending(true), ready(false)
262   */
263   bool CContextClient::checkBuffers(void)
264   {
265      map<int,CClientBuffer*>::iterator itBuff;
266      bool pending = false;
267      for (itBuff = buffers.begin(); itBuff != buffers.end(); itBuff++)
268        pending |= itBuff->second->checkBuffer(!pureOneSided);
269      return pending;
270   }
271
272   //! Release all buffers
273   void CContextClient::releaseBuffers()
274   {
275      map<int,CClientBuffer*>::iterator itBuff;
276      for (itBuff = buffers.begin(); itBuff != buffers.end(); itBuff++)
277      {
278         delete itBuff->second;
279      }
280      buffers.clear();
281   }
282
283  /*!
284   Lock the buffers for one sided communications
285   \param [in] ranks list rank of server to which client connects to
286   */
287   void CContextClient::lockBuffers(list<int>& ranks)
288   {
289      list<int>::iterator it;
290      for (it = ranks.begin(); it != ranks.end(); it++) buffers[*it]->lockBuffer();
291   }
292
293  /*!
294   Unlock the buffers for one sided communications
295   \param [in] ranks list rank of server to which client connects to
296   */
297   void CContextClient::unlockBuffers(list<int>& ranks)
298   {
299      list<int>::iterator it;
300      for (it = ranks.begin(); it != ranks.end(); it++) buffers[*it]->unlockBuffer();
301   }
302     
303   /*!
304   Verify state of buffers corresponding to a connection
305   \param [in] ranks list rank of server to which client connects to
306   \return state of buffers, pending(true), ready(false)
307   */
308   bool CContextClient::checkBuffers(list<int>& ranks)
309   {
310      list<int>::iterator it;
311      bool pending = false;
312      for (it = ranks.begin(); it != ranks.end(); it++) pending |= buffers[*it]->checkBuffer(!pureOneSided);
313      return pending;
314   }
315
316   /*!
317    * Set the buffer size for each connection. Warning: This function is collective.
318    *
319    * \param [in] mapSize maps the rank of the connected servers to the size of the correspoinding buffer
320    * \param [in] maxEventSize maps the rank of the connected servers to the size of the biggest event
321   */
322   void CContextClient::setBufferSize(const std::map<int,StdSize>& mapSize, const std::map<int,StdSize>& maxEventSize)
323   {
324     mapBufferSize_ = mapSize;
325     maxEventSizes = maxEventSize;
326   }
327
328  /*!
329  Get leading server in the group of connected server
330  \return ranks of leading servers
331  */
332  const std::list<int>& CContextClient::getRanksServerNotLeader(void) const
333  {
334    return ranksServerNotLeader;
335  }
336
337  /*!
338  Check if client connects to leading server
339  \return connected(true), not connected (false)
340  */
341  bool CContextClient::isServerNotLeader(void) const
342  {
343    return !ranksServerNotLeader.empty();
344  }
345
346  /*!
347  Get leading server in the group of connected server
348  \return ranks of leading servers
349  */
350  const std::list<int>& CContextClient::getRanksServerLeader(void) const
351  {
352    return ranksServerLeader;
353  }
354
355  /*!
356  Check if client connects to leading server
357  \return connected(true), not connected (false)
358  */
359  bool CContextClient::isServerLeader(void) const
360  {
361    return !ranksServerLeader.empty();
362  }
363
364  /*!
365   * Check if the attached mode is used.
366   *
367   * \return true if and only if attached mode is used
368   */
369  bool CContextClient::isAttachedModeEnabled() const
370  {
371    return (parentServer != 0);
372  }
373
374   /*!
375   * Finalize context client and do some reports. Function is non-blocking.
376   */
377  void CContextClient::finalize(void)
378  {
379    map<int,CClientBuffer*>::iterator itBuff;
380    std::list<int>::iterator ItServerLeader; 
381   
382    bool stop = false;
383
384    int* nbServerConnectionLocal  = new int[serverSize] ;
385    int* nbServerConnectionGlobal  = new int[serverSize] ;
386    for(int i=0;i<serverSize;++i) nbServerConnectionLocal[i]=0 ;
387    for (itBuff = buffers.begin(); itBuff != buffers.end(); itBuff++)  nbServerConnectionLocal[itBuff->first]=1 ;
388    for (ItServerLeader = ranksServerLeader.begin(); ItServerLeader != ranksServerLeader.end(); ItServerLeader++)  nbServerConnectionLocal[*ItServerLeader]=1 ;
389   
390    MPI_Allreduce(nbServerConnectionLocal, nbServerConnectionGlobal, serverSize, MPI_INT, MPI_SUM, intraComm);
391   
392    CEventClient event(CContext::GetType(), CContext::EVENT_ID_CONTEXT_FINALIZE);
393    CMessage msg;
394
395    for (int i=0;i<serverSize;++i) if (nbServerConnectionLocal[i]==1) event.push(i, nbServerConnectionGlobal[i], msg) ;
396    sendEvent(event);
397
398    delete[] nbServerConnectionLocal ;
399    delete[] nbServerConnectionGlobal ;
400/*   
401    if (isServerLeader())
402    {
403      CMessage msg;
404      const std::list<int>& ranks = getRanksServerLeader();
405      for (std::list<int>::const_iterator itRank = ranks.begin(), itRankEnd = ranks.end(); itRank != itRankEnd; ++itRank)
406      {
407        info(100)<<"DEBUG : Sent context Finalize event to rank "<<*itRank<<endl ;
408        event.push(*itRank, 1, msg);
409      }
410      sendEvent(event);
411    }
412    else sendEvent(event);
413*/
414
415    CTimer::get("Blocking time").resume();
416    checkBuffers();
417    CTimer::get("Blocking time").suspend();
418
419    std::map<int,StdSize>::const_iterator itbMap = mapBufferSize_.begin(),
420                                          iteMap = mapBufferSize_.end(), itMap;
421
422    StdSize totalBuf = 0;
423    for (itMap = itbMap; itMap != iteMap; ++itMap)
424    {
425      report(10) << " Memory report : Context <" << context->getId() << "> : client side : memory used for buffer of each connection to server" << endl
426                 << "  +) To server with rank " << itMap->first << " : " << itMap->second << " bytes " << endl;
427      totalBuf += itMap->second;
428    }
429    report(0) << " Memory report : Context <" << context->getId() << "> : client side : total memory used for buffer " << totalBuf << " bytes" << endl;
430
431    //releaseBuffers(); // moved to CContext::finalize()
432  }
433
434
435  /*!
436  */
437  bool CContextClient::havePendingRequests(void)
438  {
439    bool pending = false;
440    map<int,CClientBuffer*>::iterator itBuff;
441    for (itBuff = buffers.begin(); itBuff != buffers.end(); itBuff++)
442      pending |= itBuff->second->hasPendingRequest();
443    return pending;
444  }
445
446
447}
Note: See TracBrowser for help on using the repository browser.