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

Last change on this file since 724 was 724, checked in by rlacroix, 8 years ago

Fix: Add a failsafe in case of unexpected request for buffer.

  • 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.5 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)
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      else
54      {
55        int clientByServer = clientSize / serverSize;
56        int remain = clientSize % serverSize;
57
58        if (clientRank < (clientByServer + 1) * remain)
59        {
60          if (clientRank % (clientByServer + 1) == 0)
61            ranksServerLeader.push_back(clientRank / (clientByServer + 1));
62        }
63        else
64        {
65          int rank = clientRank - (clientByServer + 1) * remain;
66          if (rank % clientByServer == 0)
67            ranksServerLeader.push_back(remain + rank / clientByServer);
68        }
69      }
70
71      timeLine = 0;
72
73    }
74
75     
76    /*!
77    In case of attached mode, the current context must be reset to context for client
78    \param [in] event Event sent to server
79    */
80    void CContextClient::sendEvent(CEventClient& event)
81    {
82      list<int>::iterator itServer;
83      list<int> ranks;
84      list<int> sizes;
85      list<int>::iterator itSize;
86
87      ranks = event.getRanks();
88      if (!event.isEmpty())
89      {
90        sizes = event.getSizes();
91        CMessage msg;
92
93        msg << *(sizes.begin()) << timeLine;
94        for (list<int>::iterator it = sizes.begin(); it != sizes.end(); it++) *it += msg.size();
95        list<CBufferOut*> buffList = getBuffers(ranks, sizes);
96
97        list<CBufferOut*>::iterator it;
98        for (it = buffList.begin(), itSize = sizes.begin(); it != buffList.end(); ++it, ++itSize)
99        {
100          **it << *itSize << timeLine;
101        }
102        event.send(buffList);
103        checkBuffers(ranks);
104      }
105
106      if (isAttachedModeEnabled())
107      {
108        waitEvent(ranks);
109        CContext::setCurrent(context->getId());
110      }
111
112      timeLine++;
113    }
114
115    /*!
116    Special function to setup size of buffer not only on client side but also on server side
117    corresponding to the connection
118    */
119    void CContextClient::sendBufferSizeEvent()
120    {
121      std::map<int,CClientBuffer*>::iterator it, itE;
122      std::map<int,StdSize>::const_iterator itMap = mapBufferSize_.begin(), iteMap = mapBufferSize_.end();
123
124      if (itMap == iteMap)
125         ERROR("void CContextClient::sendBufferSizeEvent()",
126              <<"No information about server buffer, that should not happen...");
127
128      for (; itMap != iteMap; ++itMap)
129      {
130        if (buffers.end() == buffers.find(itMap->first))
131          newBuffer(itMap->first);
132      }
133
134      CBufferOut* bufOut(NULL);
135      itE = buffers.end();
136      for (it = buffers.begin(); it != itE; ++it)
137      {
138        bufOut = (it->second)->getBuffer(sizeof(StdSize));
139        bufOut->put(mapBufferSize_[it->first]);  // Stupid C++
140        (it->second)->checkBuffer();
141      }
142
143      if (isAttachedModeEnabled())
144      {
145        while (checkBuffers())
146        {
147          parentServer->server->listen();
148        }
149        CContext::setCurrent(context->getId());
150      }
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    Setup buffer for each connection to server and verify their state to put content into them
175    \param [in] serverList list of rank of connected server
176    \param [in] sizeList size of message corresponding to each connection
177    \return List of buffer input which event can be placed
178    */
179    list<CBufferOut*> CContextClient::getBuffers(list<int>& serverList, list<int>& sizeList)
180    {
181      list<int>::iterator itServer, itSize;
182      list<CClientBuffer*> bufferList;
183      map<int,CClientBuffer*>::iterator it;
184      list<CClientBuffer*>::iterator itBuffer;
185      list<CBufferOut*>  retBuffer;
186      bool free;
187
188      for (itServer = serverList.begin(); itServer != serverList.end(); itServer++)
189      {
190        it = buffers.find(*itServer);
191        if (it == buffers.end())
192        {
193          newBuffer(*itServer);
194          it = buffers.find(*itServer);
195        }
196        bufferList.push_back(it->second);
197      }
198      free = false;
199
200      CTimer::get("Blocking time").resume();
201      while (!free)
202      {
203        free = true;
204        for (itBuffer = bufferList.begin(), itSize = sizeList.begin(); itBuffer != bufferList.end(); itBuffer++, itSize++)
205        {
206          (*itBuffer)->checkBuffer();
207         free &= (*itBuffer)->isBufferFree(*itSize);
208        }
209        if (!free)
210          context->server->listen();
211      }
212      CTimer::get("Blocking time").suspend();
213
214      for (itBuffer = bufferList.begin(), itSize = sizeList.begin(); itBuffer != bufferList.end(); itBuffer++, itSize++)
215      {
216        retBuffer.push_back((*itBuffer)->getBuffer(*itSize));
217      }
218      return retBuffer;
219   }
220
221   /*!
222   Make a new buffer for a certain connection to server with specific rank
223   \param [in] rank rank of connected server
224   */
225   void CContextClient::newBuffer(int rank)
226   {
227      if (!mapBufferSize_.count(rank))
228      {
229        error(0) << "WARNING: Unexpected request for buffer to communicate with server " << rank << std::endl;
230        mapBufferSize_[rank] = CXios::minBufferSize;
231      }
232      buffers[rank] = new CClientBuffer(interComm, rank, mapBufferSize_[rank]);
233   }
234
235   /*!
236   Verify state of buffers. Buffer is under pending state if there is no message on it
237   \return state of buffers, pending(true), ready(false)
238   */
239   bool CContextClient::checkBuffers(void)
240   {
241      map<int,CClientBuffer*>::iterator itBuff;
242      bool pending = false;
243      for (itBuff = buffers.begin(); itBuff != buffers.end(); itBuff++) pending |= itBuff->second->checkBuffer();
244      return pending;
245   }
246
247   //! Release all buffers
248   void CContextClient::releaseBuffers(void)
249   {
250      map<int,CClientBuffer*>::iterator itBuff;
251      for (itBuff = buffers.begin(); itBuff != buffers.end(); itBuff++) delete itBuff->second;
252   }
253
254   /*!
255   Verify state of buffers corresponding to a connection
256   \param [in] ranks list rank of server to which client connects to
257   \return state of buffers, pending(true), ready(false)
258   */
259   bool CContextClient::checkBuffers(list<int>& ranks)
260   {
261      list<int>::iterator it;
262      bool pending = false;
263      for (it = ranks.begin(); it != ranks.end(); it++) pending |= buffers[*it]->checkBuffer();
264      return pending;
265   }
266
267   /*!
268   Set buffer size for each connection
269   \param [in] mapSize mapping rank of connected server to size of allocated buffer
270   */
271   void CContextClient::setBufferSize(const std::map<int,StdSize>& mapSize)
272   {
273     mapBufferSize_ = mapSize;
274     sendBufferSizeEvent();
275   }
276
277  /*!
278  Get leading server in the group of connected server
279  \return ranks of leading servers
280  */
281  const std::list<int>& CContextClient::getRanksServerLeader(void) const
282  {
283    return ranksServerLeader;
284  }
285
286  /*!
287  Check if client connects to leading server
288  \return connected(true), not connected (false)
289  */
290  bool CContextClient::isServerLeader(void) const
291  {
292    return !ranksServerLeader.empty();
293  }
294
295  /*!
296   * Check if the attached mode is used.
297   *
298   * \return true if and only if attached mode is used
299   */
300  bool CContextClient::isAttachedModeEnabled() const
301  {
302    return (parentServer != 0);
303  }
304
305   /*!
306   Finalize context client and do some reports
307   */
308   
309   void CContextClient::finalize(void)
310   {
311     map<int,CClientBuffer*>::iterator itBuff;
312     bool stop = true;
313     
314     CEventClient event(CContext::GetType(), CContext::EVENT_ID_CONTEXT_FINALIZE);
315     if (isServerLeader())
316     {
317       CMessage msg;
318       const std::list<int>& ranks = getRanksServerLeader();
319       for (std::list<int>::const_iterator itRank = ranks.begin(), itRankEnd = ranks.end(); itRank != itRankEnd; ++itRank)
320         event.push(*itRank, 1, msg);
321       sendEvent(event);
322     }
323     else sendEvent(event);
324
325     CTimer::get("Blocking time").resume();
326     while (stop)
327     {
328       checkBuffers();
329       stop = false;
330       for (itBuff = buffers.begin(); itBuff != buffers.end(); itBuff++) stop |= itBuff->second->hasPendingRequest();
331     }
332     CTimer::get("Blocking time").suspend();
333
334     std::map<int,StdSize>::const_iterator itbMap = mapBufferSize_.begin(),
335                                           iteMap = mapBufferSize_.end(), itMap;
336     StdSize totalBuf = 0;
337     for (itMap = itbMap; itMap != iteMap; ++itMap)
338     {
339       report(10) << " Memory report : Context <" << context->getId() << "> : client side : memory used for buffer of each connection to server" << endl
340                  << "  +) To server with rank " << itMap->first << " : " << itMap->second << " bytes " << endl;
341       totalBuf += itMap->second;
342     }
343     report(0) << " Memory report : Context <" << context->getId() << "> : client side : total memory used for buffer " << totalBuf << " bytes" << endl;
344
345     releaseBuffers();
346     
347   }
348}
Note: See TracBrowser for help on using the repository browser.