source: XIOS/dev/dev_olga/src/context_client.cpp @ 1077

Last change on this file since 1077 was 1077, checked in by oabramkina, 7 years ago

dev: Intermediate commit

Done:
Registries: ok with two server levels

To do (on a single server level):
test_complete
reading

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