source: XIOS/dev/dev_trunk_omp/src/context_client.cpp @ 1630

Last change on this file since 1630 was 1630, checked in by yushan, 4 years ago

working branch @1608 with bug fix @1628

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