source: XIOS/dev/branch_openmp/src/context_client.cpp @ 1356

Last change on this file since 1356 was 1356, checked in by yushan, 3 years ago

unify MPI_Comm type

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