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

Last change on this file since 650 was 597, checked in by rlacroix, 9 years ago

Add basic infrastructure for servers to clients communication.

  • 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: 10.1 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 "message.hpp"
9#include "event_client.hpp"
10#include "context.hpp"
[382]11#include "mpi.hpp"
[347]12#include "timer.hpp"
[401]13#include "cxios.hpp"
[300]14
[335]15namespace xios
[300]16{
[512]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    */
[595]23    CContextClient::CContextClient(CContext* parent, MPI_Comm intraComm_, MPI_Comm interComm_, CContext* cxtSer)
[511]24     : mapBufferSize_(), parentServer(cxtSer)
[300]25    {
[595]26      context = parent;
27      intraComm = intraComm_;
28      interComm = interComm_;
29      MPI_Comm_rank(intraComm, &clientRank);
30      MPI_Comm_size(intraComm, &clientSize);
[509]31
[595]32      int flag;
33      MPI_Comm_test_inter(interComm, &flag);
34      if (flag) MPI_Comm_remote_size(interComm, &serverSize);
35      else  MPI_Comm_size(interComm, &serverSize);
[509]36
[595]37      if (clientSize < serverSize)
38      {
39        int serverByClient = serverSize / clientSize;
40        int remain = serverSize % clientSize;
41        int rankStart = serverByClient * clientRank;
[300]42
[595]43        if (clientRank < remain)
44        {
45          serverByClient++;
46          rankStart += clientRank;
47        }
48        else
49          rankStart += remain;
50
51        for (int i = 0; i < serverByClient; i++)
52          ranksServerLeader.push_back(rankStart + i);
53      }
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));
63        }
64        else
65        {
66          int rank = clientRank - (clientByServer + 1) * remain;
67          if (rank % clientByServer == 0)
68            ranksServerLeader.push_back(remain + rank / clientByServer);
69        }
70      }
71
72      timeLine = 0;
[300]73    }
74
[512]75    /*!
76    In case of attached mode, the current context must be reset to context for client
77    \param [in] event Event sent to server
78    */
[300]79    void CContextClient::sendEvent(CEventClient& event)
80    {
[595]81      list<int>::iterator itServer;
82      list<int> ranks;
83      list<int> sizes;
84      list<int>::iterator itSize;
[509]85
[595]86      ranks = event.getRanks();
87      if (!event.isEmpty())
[300]88      {
[595]89        sizes = event.getSizes();
90        CMessage msg;
[300]91
[595]92        msg << *(sizes.begin()) << timeLine;
93        for (list<int>::iterator it = sizes.begin(); it != sizes.end(); it++) *it += msg.size();
94        list<CBufferOut*> buffList = getBuffers(ranks, sizes);
[509]95
[595]96        list<CBufferOut*>::iterator it;
97        for (it = buffList.begin(), itSize = sizes.begin(); it != buffList.end(); ++it, ++itSize)
[300]98        {
[595]99          **it << *itSize << timeLine;
[300]100        }
[595]101        event.send(buffList);
102        checkBuffers(ranks);
[300]103      }
104
[596]105      if (0 != parentServer) // attached mode
[511]106      {
107        waitEvent(ranks);
108        CContext::setCurrent(context->getId());
109      }
110
[595]111      timeLine++;
[300]112    }
[509]113
[512]114    /*!
115    Special function to setup size of buffer not only on client side but also on server side
116    corresponding to the connection
117    */
[509]118    void CContextClient::sendBufferSizeEvent()
119    {
[595]120      std::map<int,CClientBuffer*>::iterator it, itE;
121      std::map<int,StdSize>::const_iterator itMap = mapBufferSize_.begin(), iteMap = mapBufferSize_.end();
[509]122
123      if (itMap == iteMap)
[595]124         ERROR("void CContextClient::sendBufferSizeEvent()",
[509]125              <<"No information about server buffer, that should not happen...");
126
[512]127      for (; itMap != iteMap; ++itMap)
[509]128      {
129        if (buffers.end() == buffers.find(itMap->first))
130          newBuffer(itMap->first);
131      }
132
133      CBufferOut* bufOut(NULL);
134      itE = buffers.end();
135      for (it = buffers.begin(); it != itE; ++it)
136      {
137        bufOut = (it->second)->getBuffer(sizeof(StdSize));
138        bufOut->put(mapBufferSize_[it->first]);  // Stupid C++
139        (it->second)->checkBuffer();
140      }
[596]141
142      if (0 != parentServer) // attached mode
143      {
144        while (checkBuffers())
145        {
146          parentServer->server->listen();
147        }
148        CContext::setCurrent(context->getId());
149      }
[509]150    }
151
[512]152    /*!
153    If client is also server (attached mode), after sending event, it should process right away
154    the incoming event.
155    \param [in] ranks list rank of server connected this client
156    */
[300]157    void CContextClient::waitEvent(list<int>& ranks)
158    {
[595]159      parentServer->server->setPendingEvent();
160      while (checkBuffers(ranks))
[300]161      {
[595]162        parentServer->server->listen();
163        parentServer->server->checkPendingRequest();
[300]164      }
[386]165
[595]166      while (parentServer->server->hasPendingEvent())
[386]167      {
[595]168       parentServer->server->eventLoop();
[386]169      }
[300]170    }
171
[512]172    /*!
173    Setup buffer for each connection to server and verify their state to put content into them
174    \param [in] serverList list of rank of connected server
175    \param [in] sizeList size of message corresponding to each connection
176    \return List of buffer input which event can be placed
177    */
[300]178    list<CBufferOut*> CContextClient::getBuffers(list<int>& serverList, list<int>& sizeList)
179    {
[595]180      list<int>::iterator itServer, itSize;
181      list<CClientBuffer*> bufferList;
182      map<int,CClientBuffer*>::iterator it;
183      list<CClientBuffer*>::iterator itBuffer;
184      list<CBufferOut*>  retBuffer;
185      bool free;
[300]186
[595]187      for (itServer = serverList.begin(); itServer != serverList.end(); itServer++)
[300]188      {
[595]189        it = buffers.find(*itServer);
190        if (it == buffers.end())
[300]191        {
[595]192          newBuffer(*itServer);
193          it = buffers.find(*itServer);
[509]194        }
[595]195        bufferList.push_back(it->second);
[300]196      }
[595]197      free = false;
[347]198
199      CTimer::get("Blocking time").resume();
[595]200      while (!free)
[300]201      {
[595]202        free = true;
203        for (itBuffer = bufferList.begin(), itSize = sizeList.begin(); itBuffer != bufferList.end(); itBuffer++, itSize++)
[300]204        {
[595]205          (*itBuffer)->checkBuffer();
206         free &= (*itBuffer)->isBufferFree(*itSize);
[300]207        }
[597]208        if (!free)
209          context->server->listen();
[300]210      }
[347]211      CTimer::get("Blocking time").suspend();
212
[595]213      for (itBuffer = bufferList.begin(), itSize = sizeList.begin(); itBuffer != bufferList.end(); itBuffer++, itSize++)
[300]214      {
[595]215        retBuffer.push_back((*itBuffer)->getBuffer(*itSize));
[300]216      }
[595]217      return retBuffer;
[300]218   }
[509]219
[512]220   /*!
221   Make a new buffer for a certain connection to server with specific rank
222   \param [in] rank rank of connected server
223   */
[300]224   void CContextClient::newBuffer(int rank)
225   {
[595]226      buffers[rank] = new CClientBuffer(interComm, rank, mapBufferSize_[rank]);
[509]227   }
[300]228
[512]229   /*!
230   Verify state of buffers. Buffer is under pending state if there is no message on it
231   \return state of buffers, pending(true), ready(false)
232   */
[300]233   bool CContextClient::checkBuffers(void)
234   {
[595]235      map<int,CClientBuffer*>::iterator itBuff;
236      bool pending = false;
237      for (itBuff = buffers.begin(); itBuff != buffers.end(); itBuff++) pending |= itBuff->second->checkBuffer();
238      return pending;
[509]239   }
[300]240
[512]241   //! Release all buffers
[300]242   void CContextClient::releaseBuffers(void)
243   {
[595]244      map<int,CClientBuffer*>::iterator itBuff;
245      for (itBuff = buffers.begin(); itBuff != buffers.end(); itBuff++) delete itBuff->second;
[509]246   }
[300]247
[512]248   /*!
249   Verify state of buffers corresponding to a connection
250   \param [in] ranks list rank of server to which client connects to
251   \return state of buffers, pending(true), ready(false)
252   */
[300]253   bool CContextClient::checkBuffers(list<int>& ranks)
254   {
[595]255      list<int>::iterator it;
256      bool pending = false;
257      for (it = ranks.begin(); it != ranks.end(); it++) pending |= buffers[*it]->checkBuffer();
258      return pending;
[509]259   }
[300]260
[512]261   /*!
262   Set buffer size for each connection
263   \param [in] mapSize mapping rank of connected server to size of allocated buffer
264   */
[595]265   void CContextClient::setBufferSize(const std::map<int,StdSize>& mapSize)
[509]266   {
267     mapBufferSize_ = mapSize;
268     sendBufferSizeEvent();
269   }
270
[595]271  /*!
272  Get leading server in the group of connected server
273  \return ranks of leading servers
274  */
275  const std::list<int>& CContextClient::getRanksServerLeader(void) const
276  {
277    return ranksServerLeader;
278  }
[509]279
[595]280  /*!
281  Check if client connects to leading server
282  \return connected(true), not connected (false)
283  */
284  bool CContextClient::isServerLeader(void) const
285  {
286    return !ranksServerLeader.empty();
287  }
[300]288
[512]289   /*!
290   Finalize context client and do some reports
291   */
[300]292   void CContextClient::finalize(void)
293   {
[595]294     map<int,CClientBuffer*>::iterator itBuff;
295     bool stop = true;
[300]296
[595]297     CEventClient event(CContext::GetType(), CContext::EVENT_ID_CONTEXT_FINALIZE);
[300]298     if (isServerLeader())
299     {
[595]300       CMessage msg;
301       const std::list<int>& ranks = getRanksServerLeader();
302       for (std::list<int>::const_iterator itRank = ranks.begin(), itRankEnd = ranks.end(); itRank != itRankEnd; ++itRank)
303         event.push(*itRank, 1, msg);
304       sendEvent(event);
[300]305     }
[595]306     else sendEvent(event);
[509]307
[347]308     CTimer::get("Blocking time").resume();
[595]309     while (stop)
[300]310     {
[595]311       checkBuffers();
312       stop = false;
313       for (itBuff = buffers.begin(); itBuff != buffers.end(); itBuff++) stop |= itBuff->second->hasPendingRequest();
[300]314     }
[347]315     CTimer::get("Blocking time").suspend();
[509]316
[595]317     std::map<int,StdSize>::const_iterator itbMap = mapBufferSize_.begin(),
318                                           iteMap = mapBufferSize_.end(), itMap;
[511]319     StdSize totalBuf = 0;
320     for (itMap = itbMap; itMap != iteMap; ++itMap)
321     {
[595]322       report(10) << " Memory report : Context <" << context->getId() << "> : client side : memory used for buffer of each connection to server" << endl
323                  << "  +) To server with rank " << itMap->first << " : " << itMap->second << " bytes " << endl;
[511]324       totalBuf += itMap->second;
325     }
[595]326     report(0) << " Memory report : Context <" << context->getId() << "> : client side : total memory used for buffer " << totalBuf << " bytes" << endl;
[511]327
[595]328     releaseBuffers();
[300]329   }
[509]330}
Note: See TracBrowser for help on using the repository browser.