source: XIOS/dev/dev_ym/XIOS_COUPLING/src/context_client.cpp @ 2229

Last change on this file since 2229 was 2229, checked in by ymipsl, 4 months ago

When checkEvent is set, add a new diagnoistic :
Check if every server receive the event.
YM

  • 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: 18.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#include "server.hpp"
14#include "services.hpp"
15#include <boost/functional/hash.hpp>
16#include <random>
17#include <chrono>
18
19namespace xios
20{
21    /*!
22    \param [in] parent Pointer to context on client side
23    \param [in] intraComm_ communicator of group client
24    \param [in] interComm_ communicator of group server
25    \cxtSer [in] cxtSer Pointer to context of server side. (It is only used in case of attached mode).
26    */
27    CContextClient::CContextClient(CContext* parent, MPI_Comm intraComm_, MPI_Comm interComm_, CContext* cxtSer)
28     : mapBufferSize_(), parentServer(cxtSer), maxBufferedEvents(4), associatedServer_(nullptr)
29    {
30     
31      context_ = parent;
32      intraComm = intraComm_;
33      interComm = interComm_;
34      MPI_Comm_rank(intraComm, &clientRank);
35      MPI_Comm_size(intraComm, &clientSize);
36
37      int flag;
38      MPI_Comm_test_inter(interComm, &flag);
39      if (flag) isAttached_=false ;
40      else  isAttached_=true ;
41
42      pureOneSided=CXios::getin<bool>("pure_one_sided",false); // pure one sided communication (for test)
43      if (isAttachedModeEnabled()) pureOneSided=false ; // no one sided in attach mode
44     
45
46
47      if (flag) MPI_Comm_remote_size(interComm, &serverSize);
48      else  MPI_Comm_size(interComm, &serverSize);
49
50      computeLeader(clientRank, clientSize, serverSize, ranksServerLeader, ranksServerNotLeader);
51
52      if (flag) 
53      {
54        MPI_Intercomm_merge(interComm_,false, &interCommMerged) ;
55        int interCommMergedRank;
56        MPI_Comm_rank(interComm_, &interCommMergedRank);
57        MPI_Comm_rank(interCommMerged, &interCommMergedRank);
58        MPI_Comm_rank(intraComm, &interCommMergedRank);
59      }
60     
61      if (!isAttachedModeEnabled())
62      { 
63        windows.resize(serverSize) ;
64        MPI_Comm winComm ;
65        for(int rank=0; rank<serverSize; rank++)
66        {
67          windows[rank].resize(2) ;
68          MPI_Comm_split(interCommMerged, rank, clientRank, &winComm);
69          int myRank ;
70          MPI_Comm_rank(winComm,&myRank);
71          MPI_Win_create_dynamic(MPI_INFO_NULL, winComm, &windows[rank][0]);
72          MPI_Win_create_dynamic(MPI_INFO_NULL, winComm, &windows[rank][1]);
73//       ym : Warning : intelMPI doesn't support that communicator of windows be deallocated before the windows deallocation, crash at MPI_Win_lock
74//            Bug or not ?         
75//        MPI_Comm_free(&winComm) ;
76        }
77      }
78
79      MPI_Comm_split(intraComm_,clientRank,clientRank, &commSelf) ;
80
81      auto time=chrono::system_clock::now().time_since_epoch().count() ;
82      std::default_random_engine rd(time); // not reproducible from a run to another
83      std::uniform_int_distribution<size_t> dist;
84      hashId_=dist(rd) ;
85      MPI_Bcast(&hashId_,1,MPI_SIZE_T,0,intraComm) ; // Bcast to all server of the context
86
87      timeLine = 1;
88    }
89
90    void CContextClient::computeLeader(int clientRank, int clientSize, int serverSize,
91                                       std::list<int>& rankRecvLeader,
92                                       std::list<int>& rankRecvNotLeader)
93    {
94      if ((0 == clientSize) || (0 == serverSize)) return;
95
96      if (clientSize < serverSize)
97      {
98        int serverByClient = serverSize / clientSize;
99        int remain = serverSize % clientSize;
100        int rankStart = serverByClient * clientRank;
101
102        if (clientRank < remain)
103        {
104          serverByClient++;
105          rankStart += clientRank;
106        }
107        else
108          rankStart += remain;
109
110        for (int i = 0; i < serverByClient; i++)
111          rankRecvLeader.push_back(rankStart + i);
112
113        rankRecvNotLeader.resize(0);
114      }
115      else
116      {
117        int clientByServer = clientSize / serverSize;
118        int remain = clientSize % serverSize;
119
120        if (clientRank < (clientByServer + 1) * remain)
121        {
122          if (clientRank % (clientByServer + 1) == 0)
123            rankRecvLeader.push_back(clientRank / (clientByServer + 1));
124          else
125            rankRecvNotLeader.push_back(clientRank / (clientByServer + 1));
126        }
127        else
128        {
129          int rank = clientRank - (clientByServer + 1) * remain;
130          if (rank % clientByServer == 0)
131            rankRecvLeader.push_back(remain + rank / clientByServer);
132          else
133            rankRecvNotLeader.push_back(remain + rank / clientByServer);
134        }
135      }
136    }
137
138    /*!
139    In case of attached mode, the current context must be reset to context for client
140    \param [in] event Event sent to server
141    */
142    void CContextClient::sendEvent(CEventClient& event)
143    {
144      list<int> ranks = event.getRanks();
145      info(100)<<"Event "<<timeLine<<" of context "<<context_->getId()<<endl ;
146      if (CXios::checkEventSync)
147      {
148        int typeId, classId, typeId_in, classId_in;
149        long long timeLine_out;
150        long long timeLine_in( timeLine );
151        typeId_in=event.getTypeId() ;
152        classId_in=event.getClassId() ;
153//        MPI_Allreduce(&timeLine,&timeLine_out, 1, MPI_UINT64_T, MPI_SUM, intraComm) ; // MPI_UINT64_T standardized by MPI 3
154        MPI_Allreduce(&timeLine_in,&timeLine_out, 1, MPI_LONG_LONG_INT, MPI_SUM, intraComm) ; 
155        MPI_Allreduce(&typeId_in,&typeId, 1, MPI_INT, MPI_SUM, intraComm) ;
156        MPI_Allreduce(&classId_in,&classId, 1, MPI_INT, MPI_SUM, intraComm) ;
157        if (typeId/clientSize!=event.getTypeId() || classId/clientSize!=event.getClassId() || timeLine_out/clientSize!=timeLine)
158        {
159           ERROR("void CContextClient::sendEvent(CEventClient& event)",
160               << "Event are not coherent between client for timeline = "<<timeLine);
161        }
162       
163        vector<int> servers(serverSize,0) ;
164        auto ranks=event.getRanks() ;
165        for(auto& rank : ranks) servers[rank]=1 ;
166        MPI_Allreduce(MPI_IN_PLACE, servers.data(), serverSize,MPI_INT,MPI_SUM,intraComm) ;
167        ostringstream osstr ;
168        for(int i=0;i<serverSize;i++)  if (servers[i]==0) osstr<<i<<" , " ;
169        if (!osstr.str().empty())
170        {
171          ERROR("void CContextClient::sendEvent(CEventClient& event)",
172                 <<" Some servers will not receive the message for timeline = "<<timeLine<<endl
173                 <<"Servers are : "<<osstr.str()) ;
174        }
175
176
177      }
178
179      if (!event.isEmpty())
180      {
181        list<int> sizes = event.getSizes();
182
183         // We force the getBuffers call to be non-blocking on classical servers
184        list<CBufferOut*> buffList;
185        getBuffers(timeLine, ranks, sizes, buffList) ;
186
187        event.send(timeLine, sizes, buffList);
188       
189        //for (auto itRank = ranks.begin(); itRank != ranks.end(); itRank++) buffers[*itRank]->infoBuffer() ;
190
191        unlockBuffers(ranks) ;
192        info(100)<<"Event "<<timeLine<<" of context "<<context_->getId()<<"  sent"<<endl ;
193         
194        checkBuffers(ranks);
195      }
196     
197      if (isAttachedModeEnabled()) // couldBuffer is always true in attached mode
198      {
199        while (checkBuffers(ranks)) context_->globalEventLoop() ;
200     
201        CXios::getDaemonsManager()->scheduleContext(hashId_) ;
202        while (CXios::getDaemonsManager()->isScheduledContext(hashId_)) context_->globalEventLoop() ;
203      }
204     
205      timeLine++;
206    }
207
208    /*!
209    If client is also server (attached mode), after sending event, it should process right away
210    the incoming event.
211    \param [in] ranks list rank of server connected this client
212    */
213    void CContextClient::waitEvent(list<int>& ranks)
214    {
215      while (checkBuffers(ranks))
216      {
217        context_->eventLoop() ;
218      }
219
220      MPI_Request req ;
221      MPI_Status status ;
222
223      MPI_Ibarrier(intraComm,&req) ;
224      int flag=false ;
225
226      do 
227      {
228        CXios::getDaemonsManager()->eventLoop() ;
229        MPI_Test(&req,&flag,&status) ;
230      } while (!flag) ;
231
232
233    }
234
235
236    void CContextClient::waitEvent_old(list<int>& ranks)
237    {
238      parentServer->server->setPendingEvent();
239      while (checkBuffers(ranks))
240      {
241        parentServer->server->listen();
242        parentServer->server->checkPendingRequest();
243      }
244
245      while (parentServer->server->hasPendingEvent())
246      {
247       parentServer->server->eventLoop();
248      }
249    }
250
251    /*!
252     * Get buffers for each connection to the servers. This function blocks until there is enough room in the buffers unless
253     * it is explicitly requested to be non-blocking.
254     *
255     *
256     * \param [in] timeLine time line of the event which will be sent to servers
257     * \param [in] serverList list of rank of connected server
258     * \param [in] sizeList size of message corresponding to each connection
259     * \param [out] retBuffers list of buffers that can be used to store an event
260     * \param [in] nonBlocking whether this function should be non-blocking
261     * \return whether the already allocated buffers could be used
262    */
263    bool CContextClient::getBuffers(const size_t timeLine, const list<int>& serverList, const list<int>& sizeList, list<CBufferOut*>& retBuffers,
264                                    bool nonBlocking /*= false*/)
265    {
266      list<int>::const_iterator itServer, itSize;
267      list<CClientBuffer*> bufferList;
268      map<int,CClientBuffer*>::const_iterator it;
269      list<CClientBuffer*>::iterator itBuffer;
270      bool areBuffersFree;
271
272      for (itServer = serverList.begin(); itServer != serverList.end(); itServer++)
273      {
274        it = buffers.find(*itServer);
275        if (it == buffers.end())
276        {
277          newBuffer(*itServer);
278          it = buffers.find(*itServer);
279        }
280        bufferList.push_back(it->second);
281      }
282
283      CTimer::get("Blocking time").resume();
284      do
285      {
286        areBuffersFree = true;
287        for (itBuffer = bufferList.begin(), itSize = sizeList.begin(); itBuffer != bufferList.end(); itBuffer++, itSize++)
288        {
289          areBuffersFree &= (*itBuffer)->isBufferFree(*itSize);
290        }
291
292        if (!areBuffersFree)
293        {
294          for (itBuffer = bufferList.begin(); itBuffer != bufferList.end(); itBuffer++) (*itBuffer)->unlockBuffer();
295          checkBuffers();
296/*         
297          context->server->listen();
298
299          if (context->serverPrimServer.size()>0)
300          {
301            for (int i = 0; i < context->serverPrimServer.size(); ++i)  context->serverPrimServer[i]->listen();
302 //ym           CServer::contextEventLoop(false) ; // avoid dead-lock at finalize...
303            context->globalEventLoop() ;
304          }
305*/
306           context_->globalEventLoop() ;
307        }
308
309      } while (!areBuffersFree && !nonBlocking);
310      CTimer::get("Blocking time").suspend();
311
312      if (areBuffersFree)
313      {
314        for (itBuffer = bufferList.begin(), itSize = sizeList.begin(); itBuffer != bufferList.end(); itBuffer++, itSize++)
315          retBuffers.push_back((*itBuffer)->getBuffer(timeLine, *itSize));
316      }
317      return areBuffersFree;
318   }
319
320   /*!
321   Make a new buffer for a certain connection to server with specific rank
322   \param [in] rank rank of connected server
323   */
324   void CContextClient::newBuffer(int rank)
325   {
326      if (!mapBufferSize_.count(rank))
327      {
328        error(0) << "WARNING: Unexpected request for buffer to communicate with server " << rank << std::endl;
329        mapBufferSize_[rank] = CXios::minBufferSize;
330        maxEventSizes[rank] = CXios::minBufferSize;
331      }
332     
333      vector<MPI_Win> Wins(2,MPI_WIN_NULL) ;
334      if (!isAttachedModeEnabled()) Wins=windows[rank] ;
335 
336      CClientBuffer* buffer = buffers[rank] = new CClientBuffer(interComm, Wins, clientRank, rank, mapBufferSize_[rank], maxEventSizes[rank]);
337      if (isGrowableBuffer_) buffer->setGrowableBuffer(1.2) ;
338      else buffer->fixBuffer() ;
339      // Notify the server
340      CBufferOut* bufOut = buffer->getBuffer(0, 4*sizeof(MPI_Aint));
341      MPI_Aint sendBuff[4] ;
342      sendBuff[0]=hashId_;
343      sendBuff[1]=mapBufferSize_[rank];
344      sendBuff[2]=buffers[rank]->getWinAddress(0); 
345      sendBuff[3]=buffers[rank]->getWinAddress(1); 
346      info(100)<<"CContextClient::newBuffer : rank "<<rank<<" winAdress[0] "<<buffers[rank]->getWinAddress(0)<<" winAdress[1] "<<buffers[rank]->getWinAddress(1)<<endl;
347      bufOut->put(sendBuff, 4); 
348      buffer->checkBuffer(true);
349
350   }
351
352   /*!
353   Verify state of buffers. Buffer is under pending state if there is no message on it
354   \return state of buffers, pending(true), ready(false)
355   */
356   bool CContextClient::checkBuffers(void)
357   {
358      map<int,CClientBuffer*>::iterator itBuff;
359      bool pending = false;
360      for (itBuff = buffers.begin(); itBuff != buffers.end(); itBuff++)
361        pending |= itBuff->second->checkBuffer(!pureOneSided);
362      return pending;
363   }
364
365   //! Release all buffers
366   void CContextClient::releaseBuffers()
367   {
368      map<int,CClientBuffer*>::iterator itBuff;
369      for (itBuff = buffers.begin(); itBuff != buffers.end(); itBuff++)
370      {
371         delete itBuff->second;
372      }
373      buffers.clear();
374
375/* don't know when release windows
376
377      if (!isAttachedModeEnabled())
378      { 
379        for(int rank=0; rank<serverSize; rank++)
380        {
381          MPI_Win_free(&windows[rank][0]);
382          MPI_Win_free(&windows[rank][1]);
383        }
384      }
385*/
386   }
387
388     
389  /*!
390   Lock the buffers for one sided communications
391   \param [in] ranks list rank of server to which client connects to
392   */
393   void CContextClient::lockBuffers(list<int>& ranks)
394   {
395      list<int>::iterator it;
396      for (it = ranks.begin(); it != ranks.end(); it++) buffers[*it]->lockBuffer();
397   }
398
399  /*!
400   Unlock the buffers for one sided communications
401   \param [in] ranks list rank of server to which client connects to
402   */
403   void CContextClient::unlockBuffers(list<int>& ranks)
404   {
405      list<int>::iterator it;
406      for (it = ranks.begin(); it != ranks.end(); it++) buffers[*it]->unlockBuffer();
407   }
408     
409   /*!
410   Verify state of buffers corresponding to a connection
411   \param [in] ranks list rank of server to which client connects to
412   \return state of buffers, pending(true), ready(false)
413   */
414   bool CContextClient::checkBuffers(list<int>& ranks)
415   {
416      list<int>::iterator it;
417      bool pending = false;
418      for (it = ranks.begin(); it != ranks.end(); it++) pending |= buffers[*it]->checkBuffer(!pureOneSided);
419      return pending;
420   }
421
422   /*!
423    * Set the buffer size for each connection. Warning: This function is collective.
424    *
425    * \param [in] mapSize maps the rank of the connected servers to the size of the correspoinding buffer
426    * \param [in] maxEventSize maps the rank of the connected servers to the size of the biggest event
427   */
428   void CContextClient::setBufferSize(const std::map<int,StdSize>& mapSize)
429   {
430     for(auto& it : mapSize) 
431      buffers[it.first]->fixBufferSize(std::max(CXios::minBufferSize*1.0,std::min(it.second*CXios::bufferSizeFactor*1.01,CXios::maxBufferSize*1.0)));
432   }
433
434  /*!
435  Get leading server in the group of connected server
436  \return ranks of leading servers
437  */
438  const std::list<int>& CContextClient::getRanksServerNotLeader(void) const
439  {
440    return ranksServerNotLeader;
441  }
442
443  /*!
444  Check if client connects to leading server
445  \return connected(true), not connected (false)
446  */
447  bool CContextClient::isServerNotLeader(void) const
448  {
449    return !ranksServerNotLeader.empty();
450  }
451
452  /*!
453  Get leading server in the group of connected server
454  \return ranks of leading servers
455  */
456  const std::list<int>& CContextClient::getRanksServerLeader(void) const
457  {
458    return ranksServerLeader;
459  }
460
461  /*!
462  Check if client connects to leading server
463  \return connected(true), not connected (false)
464  */
465  bool CContextClient::isServerLeader(void) const
466  {
467    return !ranksServerLeader.empty();
468  }
469
470   /*!
471   * Finalize context client and do some reports. Function is non-blocking.
472   */
473  void CContextClient::finalize(void)
474  {
475    map<int,CClientBuffer*>::iterator itBuff;
476    std::list<int>::iterator ItServerLeader; 
477   
478    bool stop = false;
479
480    int* nbServerConnectionLocal  = new int[serverSize] ;
481    int* nbServerConnectionGlobal  = new int[serverSize] ;
482    for(int i=0;i<serverSize;++i) nbServerConnectionLocal[i]=0 ;
483    for (itBuff = buffers.begin(); itBuff != buffers.end(); itBuff++)  nbServerConnectionLocal[itBuff->first]=1 ;
484    for (ItServerLeader = ranksServerLeader.begin(); ItServerLeader != ranksServerLeader.end(); ItServerLeader++)  nbServerConnectionLocal[*ItServerLeader]=1 ;
485   
486    MPI_Allreduce(nbServerConnectionLocal, nbServerConnectionGlobal, serverSize, MPI_INT, MPI_SUM, intraComm);
487   
488    CEventClient event(CContext::GetType(), CContext::EVENT_ID_CONTEXT_FINALIZE);
489    CMessage msg;
490
491    for (int i=0;i<serverSize;++i) if (nbServerConnectionLocal[i]==1) event.push(i, nbServerConnectionGlobal[i], msg) ;
492    sendEvent(event);
493
494    delete[] nbServerConnectionLocal ;
495    delete[] nbServerConnectionGlobal ;
496
497
498    CTimer::get("Blocking time").resume();
499    checkBuffers();
500    CTimer::get("Blocking time").suspend();
501
502    std::map<int,StdSize>::const_iterator itbMap = mapBufferSize_.begin(),
503                                          iteMap = mapBufferSize_.end(), itMap;
504
505    StdSize totalBuf = 0;
506    for (itMap = itbMap; itMap != iteMap; ++itMap)
507    {
508      report(10) << " Memory report : Context <" << context_->getId() << "> : client side : memory used for buffer of each connection to server" << endl
509                 << "  +) To server with rank " << itMap->first << " : " << itMap->second << " bytes " << endl;
510      totalBuf += itMap->second;
511    }
512    report(0) << " Memory report : Context <" << context_->getId() << "> : client side : total memory used for buffer " << totalBuf << " bytes" << endl;
513
514  }
515
516
517  /*!
518  */
519  bool CContextClient::havePendingRequests(void)
520  {
521    bool pending = false;
522    map<int,CClientBuffer*>::iterator itBuff;
523    for (itBuff = buffers.begin(); itBuff != buffers.end(); itBuff++)
524      pending |= itBuff->second->hasPendingRequest();
525    return pending;
526  }
527 
528  bool CContextClient::isNotifiedFinalized(void)
529  {
530    if (isAttachedModeEnabled()) return true ;
531
532    bool finalized = true;
533    map<int,CClientBuffer*>::iterator itBuff;
534    for (itBuff = buffers.begin(); itBuff != buffers.end(); itBuff++)
535      finalized &= itBuff->second->isNotifiedFinalized();
536    return finalized;
537  }
538
539}
Note: See TracBrowser for help on using the repository browser.