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

Last change on this file since 884 was 884, checked in by rlacroix, 6 years ago

Fix a possible deadlock for clients communicating with at least two servers.

The event scheduler introduces a dependency between servers so clients must check all their buffers when waiting for a specific buffer to be free again.

This commit partially fixes the issue described in ticket #91.

  • 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: 9.1 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    In case of attached mode, the current context must be reset to context for client
77    \param [in] event Event sent to server
78    */
79    void CContextClient::sendEvent(CEventClient& event)
80    {
81      list<int> ranks = event.getRanks();
82      if (!event.isEmpty())
83      {
84        list<int> sizes = event.getSizes();
85
86        list<CBufferOut*> buffList = getBuffers(ranks, sizes);
87
88        event.send(timeLine, sizes, buffList);
89
90        checkBuffers(ranks);
91      }
92
93      if (isAttachedModeEnabled())
94      {
95        waitEvent(ranks);
96        CContext::setCurrent(context->getId());
97      }
98
99      timeLine++;
100    }
101
102    /*!
103    If client is also server (attached mode), after sending event, it should process right away
104    the incoming event.
105    \param [in] ranks list rank of server connected this client
106    */
107    void CContextClient::waitEvent(list<int>& ranks)
108    {
109      parentServer->server->setPendingEvent();
110      while (checkBuffers(ranks))
111      {
112        parentServer->server->listen();
113        parentServer->server->checkPendingRequest();
114      }
115
116      while (parentServer->server->hasPendingEvent())
117      {
118       parentServer->server->eventLoop();
119      }
120    }
121
122    /*!
123    Setup buffer for each connection to server and verify their state to put content into them
124    \param [in] serverList list of rank of connected server
125    \param [in] sizeList size of message corresponding to each connection
126    \return List of buffer input which event can be placed
127    */
128    list<CBufferOut*> CContextClient::getBuffers(list<int>& serverList, list<int>& sizeList)
129    {
130      list<int>::iterator itServer, itSize;
131      list<CClientBuffer*> bufferList;
132      map<int,CClientBuffer*>::iterator it;
133      list<CClientBuffer*>::iterator itBuffer;
134      list<CBufferOut*>  retBuffer;
135      bool areBuffersFree;
136
137      for (itServer = serverList.begin(); itServer != serverList.end(); itServer++)
138      {
139        it = buffers.find(*itServer);
140        if (it == buffers.end())
141        {
142          newBuffer(*itServer);
143          it = buffers.find(*itServer);
144        }
145        bufferList.push_back(it->second);
146      }
147
148      CTimer::get("Blocking time").resume();
149      do
150      {
151        areBuffersFree = true;
152        for (itBuffer = bufferList.begin(), itSize = sizeList.begin(); itBuffer != bufferList.end(); itBuffer++, itSize++)
153          areBuffersFree &= (*itBuffer)->isBufferFree(*itSize);
154
155        if (!areBuffersFree)
156        {
157          checkBuffers();
158          context->server->listen();
159        }
160      } while (!areBuffersFree);
161      CTimer::get("Blocking time").suspend();
162
163      for (itBuffer = bufferList.begin(), itSize = sizeList.begin(); itBuffer != bufferList.end(); itBuffer++, itSize++)
164      {
165        retBuffer.push_back((*itBuffer)->getBuffer(*itSize));
166      }
167      return retBuffer;
168   }
169
170   /*!
171   Make a new buffer for a certain connection to server with specific rank
172   \param [in] rank rank of connected server
173   */
174   void CContextClient::newBuffer(int rank)
175   {
176      if (!mapBufferSize_.count(rank))
177      {
178        error(0) << "WARNING: Unexpected request for buffer to communicate with server " << rank << std::endl;
179        mapBufferSize_[rank] = CXios::minBufferSize;
180      }
181      CClientBuffer* buffer = buffers[rank] = new CClientBuffer(interComm, rank, mapBufferSize_[rank]);
182      // Notify the server
183      CBufferOut* bufOut = buffer->getBuffer(sizeof(StdSize));
184      bufOut->put(mapBufferSize_[rank]); // Stupid C++
185      buffer->checkBuffer();
186   }
187
188   /*!
189   Verify state of buffers. Buffer is under pending state if there is no message on it
190   \return state of buffers, pending(true), ready(false)
191   */
192   bool CContextClient::checkBuffers(void)
193   {
194      map<int,CClientBuffer*>::iterator itBuff;
195      bool pending = false;
196      for (itBuff = buffers.begin(); itBuff != buffers.end(); itBuff++) pending |= itBuff->second->checkBuffer();
197      return pending;
198   }
199
200   //! Release all buffers
201   void CContextClient::releaseBuffers(void)
202   {
203      map<int,CClientBuffer*>::iterator itBuff;
204      for (itBuff = buffers.begin(); itBuff != buffers.end(); itBuff++) delete itBuff->second;
205   }
206
207   /*!
208   Verify state of buffers corresponding to a connection
209   \param [in] ranks list rank of server to which client connects to
210   \return state of buffers, pending(true), ready(false)
211   */
212   bool CContextClient::checkBuffers(list<int>& ranks)
213   {
214      list<int>::iterator it;
215      bool pending = false;
216      for (it = ranks.begin(); it != ranks.end(); it++) pending |= buffers[*it]->checkBuffer();
217      return pending;
218   }
219
220   /*!
221   Set buffer size for each connection
222   \param [in] mapSize mapping rank of connected server to size of allocated buffer
223   */
224   void CContextClient::setBufferSize(const std::map<int,StdSize>& mapSize)
225   {
226     mapBufferSize_ = mapSize;
227   }
228
229  /*!
230  Get leading server in the group of connected server
231  \return ranks of leading servers
232  */
233  const std::list<int>& CContextClient::getRanksServerLeader(void) const
234  {
235    return ranksServerLeader;
236  }
237
238  /*!
239  Check if client connects to leading server
240  \return connected(true), not connected (false)
241  */
242  bool CContextClient::isServerLeader(void) const
243  {
244    return !ranksServerLeader.empty();
245  }
246
247  /*!
248   * Check if the attached mode is used.
249   *
250   * \return true if and only if attached mode is used
251   */
252  bool CContextClient::isAttachedModeEnabled() const
253  {
254    return (parentServer != 0);
255  }
256
257   /*!
258   Finalize context client and do some reports
259   */
260   void CContextClient::finalize(void)
261   {
262     map<int,CClientBuffer*>::iterator itBuff;
263     bool stop = true;
264
265     CEventClient event(CContext::GetType(), CContext::EVENT_ID_CONTEXT_FINALIZE);
266     if (isServerLeader())
267     {
268       CMessage msg;
269       const std::list<int>& ranks = getRanksServerLeader();
270       for (std::list<int>::const_iterator itRank = ranks.begin(), itRankEnd = ranks.end(); itRank != itRankEnd; ++itRank)
271         event.push(*itRank, 1, msg);
272       sendEvent(event);
273     }
274     else sendEvent(event);
275
276     CTimer::get("Blocking time").resume();
277     while (stop)
278     {
279       checkBuffers();
280       stop = false;
281       for (itBuff = buffers.begin(); itBuff != buffers.end(); itBuff++) stop |= itBuff->second->hasPendingRequest();
282     }
283     CTimer::get("Blocking time").suspend();
284
285     std::map<int,StdSize>::const_iterator itbMap = mapBufferSize_.begin(),
286                                           iteMap = mapBufferSize_.end(), itMap;
287     StdSize totalBuf = 0;
288     for (itMap = itbMap; itMap != iteMap; ++itMap)
289     {
290       report(10) << " Memory report : Context <" << context->getId() << "> : client side : memory used for buffer of each connection to server" << endl
291                  << "  +) To server with rank " << itMap->first << " : " << itMap->second << " bytes " << endl;
292       totalBuf += itMap->second;
293     }
294     report(0) << " Memory report : Context <" << context->getId() << "> : client side : total memory used for buffer " << totalBuf << " bytes" << endl;
295
296     releaseBuffers();
297   }
298}
Note: See TracBrowser for help on using the repository browser.