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

Last change on this file since 1482 was 1482, checked in by yushan, 6 years ago

Branch EP merged with Dev_cmip6 @r1481

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