source: XIOS/trunk/src/context_client.cpp @ 1192

Last change on this file since 1192 was 1192, checked in by rlacroix, 5 years ago

Display a warning message if an event has a size greater than the estimated maximum event size.

This should not happen in theory and must be fixed.

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