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

Last change on this file since 1077 was 1077, checked in by oabramkina, 4 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
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 in case of attached mode).
21    */
22    CContextClient::CContextClient(CContext* parent, MPI_Comm intraComm_, MPI_Comm interComm_, CContext* cxtSer)
23     : mapBufferSize_(), parentServer(cxtSer), maxBufferedEvents(4)
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        ranksServerNotLeader.resize(0);      }
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          else
64            ranksServerNotLeader.push_back(clientRank / (clientByServer + 1));
65        }
66        else
67        {
68          int rank = clientRank - (clientByServer + 1) * remain;
69          if (rank % clientByServer == 0)
70            ranksServerLeader.push_back(remain + rank / clientByServer);
71          else
72            ranksServerNotLeader.push_back(remain + rank / clientByServer);
73        }
74      }
75
76      timeLine = 0;
77    }
78
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    */
83    void CContextClient::sendEvent(CEventClient& event)
84    {
85      list<int> ranks = event.getRanks();
86
87      if (!event.isEmpty())
88      {
89        list<int> sizes = event.getSizes();
90
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);
94        bool couldBuffer = getBuffers(ranks, sizes, buffList, false);
95
96        if (couldBuffer)
97        {
98          event.send(timeLine, sizes, buffList);
99
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        }
118      }
119
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())
133      {
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        }
148      }
149
150      return couldSendTmpBufferedEvent;
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    /*!
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
183    */
184    bool CContextClient::getBuffers(const list<int>& serverList, const list<int>& sizeList, list<CBufferOut*>& retBuffers,
185                                    bool nonBlocking /*= false*/)
186    {
187      list<int>::const_iterator itServer, itSize;
188      list<CClientBuffer*> bufferList;
189      map<int,CClientBuffer*>::const_iterator it;
190      list<CClientBuffer*>::iterator itBuffer;
191      bool areBuffersFree;
192
193      for (itServer = serverList.begin(); itServer != serverList.end(); itServer++)
194      {
195        it = buffers.find(*itServer);
196        if (it == buffers.end())
197        {
198          newBuffer(*itServer);
199          it = buffers.find(*itServer);
200        }
201        bufferList.push_back(it->second);
202      }
203
204      CTimer::get("Blocking time").resume();
205      do
206      {
207        areBuffersFree = true;
208        for (itBuffer = bufferList.begin(), itSize = sizeList.begin(); itBuffer != bufferList.end(); itBuffer++, itSize++)
209          areBuffersFree &= (*itBuffer)->isBufferFree(*itSize);
210
211        if (!areBuffersFree)
212        {
213          checkBuffers();
214
215         // WHY DO WE PUT HERE SERVER INTO LISTENING LOOP AT ALL????
216//            context->server->listen();
217//            for (int i = 0; i < context->serverPrimServer.size(); ++i)
218//              context->serverPrimServer[i]->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     }
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();
248   }
249
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   */
254   bool CContextClient::checkBuffers(void)
255   {
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;
260   }
261
262   //! Release all buffers
263   void CContextClient::releaseBuffers()
264   {
265      map<int,CClientBuffer*>::iterator itBuff;
266      for (itBuff = buffers.begin(); itBuff != buffers.end(); itBuff++)
267          delete itBuff->second;
268      buffers.clear();
269   }
270
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   */
276//   bool CContextClient::checkBuffers(list<int>& ranks)
277   bool CContextClient::checkBuffers(list<int>& ranks)
278   {
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;
283   }
284
285   /*!
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
290   */
291   void CContextClient::setBufferSize(const std::map<int,StdSize>& mapSize, const std::map<int,StdSize>& maxEventSize)
292   {
293     mapBufferSize_ = mapSize;
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
311   }
312
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
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  }
339
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  }
348
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  }
358
359   /*!
360   Finalize context client and do some reports
361   */
362//  void CContextClient::finalize(void)
363  void CContextClient::finalize()
364  {
365    map<int,CClientBuffer*>::iterator itBuff;
366    bool stop = false;
367
368    CTimer::get("Blocking time").resume();
369    while (hasTemporarilyBufferedEvent())
370    {
371      checkBuffers();
372      sendTemporarilyBufferedEvent();
373    }
374    CTimer::get("Blocking time").suspend();
375
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);
386
387    CTimer::get("Blocking time").resume();
388    while (!stop)
389    {
390      checkBuffers();
391      if (hasTemporarilyBufferedEvent())
392        sendTemporarilyBufferedEvent();
393
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;
401
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  }
413}
Note: See TracBrowser for help on using the repository browser.