source: XIOS3/branches/xios-3.0-beta/src/transport/legacy_context_client.cpp @ 2521

Last change on this file since 2521 was 2521, checked in by jderouillat, 12 months ago

Replace MPI probing on intercommunicator by probing on intracommunicator

  • Property svn:eol-style set to native
  • Property svn:executable set to *
File size: 15.1 KB
Line 
1#include "xios_spl.hpp"
2#include "legacy_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    CLegacyContextClient::CLegacyContextClient(CContext* parent, MPI_Comm intraComm_, MPI_Comm interComm_, CContext* cxtSer)
28                         : CContextClient(parent, intraComm_, interComm_, cxtSer),
29                           mapBufferSize_(),  maxBufferedEvents(4)
30    {
31      pureOneSided=CXios::getin<bool>("pure_one_sided",false); // pure one sided communication (for test)
32      if (isAttachedModeEnabled()) pureOneSided=false ; // no one sided in attach mode
33
34      if (!isAttachedModeEnabled()) MPI_Intercomm_merge(interComm_,false, &interCommMerged_) ;
35     
36      MPI_Comm_split(intraComm_,clientRank,clientRank, &commSelf_) ; // for windows
37
38      timeLine = 1;
39    }
40
41
42    /*!
43    In case of attached mode, the current context must be reset to context for client
44    \param [in] event Event sent to server
45    */
46    void CLegacyContextClient::sendEvent(CEventClient& event)
47    {
48      list<int> ranks = event.getRanks();
49 
50//      ostringstream str ;
51//      for(auto& rank : ranks) str<<rank<<" ; " ;
52//      info(100)<<"Event "<<timeLine<<" of context "<<context_->getId()<<"  for ranks : "<<str.str()<<endl ;
53
54      if (CXios::checkEventSync)
55      {
56        int typeId, classId, typeId_in, classId_in;
57        long long timeLine_out;
58        long long timeLine_in( timeLine );
59        typeId_in=event.getTypeId() ;
60        classId_in=event.getClassId() ;
61//        MPI_Allreduce(&timeLine,&timeLine_out, 1, MPI_UINT64_T, MPI_SUM, intraComm) ; // MPI_UINT64_T standardized by MPI 3
62        MPI_Allreduce(&timeLine_in,&timeLine_out, 1, MPI_LONG_LONG_INT, MPI_SUM, intraComm) ; 
63        MPI_Allreduce(&typeId_in,&typeId, 1, MPI_INT, MPI_SUM, intraComm) ;
64        MPI_Allreduce(&classId_in,&classId, 1, MPI_INT, MPI_SUM, intraComm) ;
65        if (typeId/clientSize!=event.getTypeId() || classId/clientSize!=event.getClassId() || timeLine_out/clientSize!=timeLine)
66        {
67           ERROR("void CLegacyContextClient::sendEvent(CEventClient& event)",
68               << "Event are not coherent between client for timeline = "<<timeLine);
69        }
70       
71        vector<int> servers(serverSize,0) ;
72        auto ranks=event.getRanks() ;
73        for(auto& rank : ranks) servers[rank]=1 ;
74        MPI_Allreduce(MPI_IN_PLACE, servers.data(), serverSize,MPI_INT,MPI_SUM,intraComm) ;
75        ostringstream osstr ;
76        for(int i=0;i<serverSize;i++)  if (servers[i]==0) osstr<<i<<" , " ;
77        if (!osstr.str().empty())
78        {
79          ERROR("void CLegacyContextClient::sendEvent(CEventClient& event)",
80                 <<" Some servers will not receive the message for timeline = "<<timeLine<<endl
81                 <<"Servers are : "<<osstr.str()) ;
82        }
83
84
85      }
86
87      if (!event.isEmpty())
88      {
89        list<int> sizes = event.getSizes();
90
91         // We force the getBuffers call to be non-blocking on classical servers
92        list<CBufferOut*> buffList;
93        getBuffers(timeLine, ranks, sizes, buffList) ;
94
95        event.send(timeLine, sizes, buffList);
96       
97        //for (auto itRank = ranks.begin(); itRank != ranks.end(); itRank++) buffers[*itRank]->infoBuffer() ;
98
99        unlockBuffers(ranks) ;
100        checkBuffers(ranks);
101       
102      }
103     
104      if (isAttachedModeEnabled()) // couldBuffer is always true in attached mode
105      {
106        while (checkBuffers(ranks)) callGlobalEventLoop() ;
107     
108        CXios::getDaemonsManager()->scheduleContext(hashId_) ;
109        while (CXios::getDaemonsManager()->isScheduledContext(hashId_)) callGlobalEventLoop() ;
110      }
111     
112      timeLine++;
113    }
114
115
116    /*!
117     * Get buffers for each connection to the servers. This function blocks until there is enough room in the buffers unless
118     * it is explicitly requested to be non-blocking.
119     *
120     *
121     * \param [in] timeLine time line of the event which will be sent to servers
122     * \param [in] serverList list of rank of connected server
123     * \param [in] sizeList size of message corresponding to each connection
124     * \param [out] retBuffers list of buffers that can be used to store an event
125     * \param [in] nonBlocking whether this function should be non-blocking
126     * \return whether the already allocated buffers could be used
127    */
128    bool CLegacyContextClient::getBuffers(const size_t timeLine, const list<int>& serverList, const list<int>& sizeList, list<CBufferOut*>& retBuffers,
129                                    bool nonBlocking /*= false*/)
130    {
131      list<int>::const_iterator itServer, itSize;
132      list<CClientBuffer*> bufferList;
133      map<int,CClientBuffer*>::const_iterator it;
134      list<CClientBuffer*>::iterator itBuffer;
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      double lastTimeBuffersNotFree=0. ;
149      double time ;
150      bool doUnlockBuffers ;
151      CTimer::get("Blocking time").resume();
152      do
153      {
154        areBuffersFree = true;
155        doUnlockBuffers=false ;
156        time=MPI_Wtime() ;
157        if (time-lastTimeBuffersNotFree > latency_)
158        {
159          for (itBuffer = bufferList.begin(), itSize = sizeList.begin(); itBuffer != bufferList.end(); itBuffer++, itSize++)
160          {
161            areBuffersFree &= (*itBuffer)->isBufferFree(*itSize);
162          }
163          if (!areBuffersFree)
164          {
165            lastTimeBuffersNotFree = time ;
166            doUnlockBuffers=true ;
167          }         
168        }
169        else areBuffersFree = false ;
170
171        if (!areBuffersFree)
172        {
173          if (doUnlockBuffers) for (itBuffer = bufferList.begin(); itBuffer != bufferList.end(); itBuffer++) (*itBuffer)->unlockBuffer();
174          checkBuffers();
175
176          callGlobalEventLoop() ;
177        }
178
179      } while (!areBuffersFree && !nonBlocking);
180      CTimer::get("Blocking time").suspend();
181
182      if (areBuffersFree)
183      {
184        for (itBuffer = bufferList.begin(), itSize = sizeList.begin(); itBuffer != bufferList.end(); itBuffer++, itSize++)
185          retBuffers.push_back((*itBuffer)->getBuffer(timeLine, *itSize));
186      }
187      return areBuffersFree;
188   }
189
190   void CLegacyContextClient::eventLoop(void)
191   {
192      if (!locked_) checkBuffers() ;
193   }
194
195   void CLegacyContextClient::callGlobalEventLoop(void)
196   {
197     locked_=true ;
198     context_->globalEventLoop() ;
199     locked_=false ;
200   }
201   /*!
202   Make a new buffer for a certain connection to server with specific rank
203   \param [in] rank rank of connected server
204   */
205   void CLegacyContextClient::newBuffer(int rank)
206   {
207      if (!mapBufferSize_.count(rank))
208      {
209        error(0) << "WARNING: Unexpected request for buffer to communicate with server " << rank << std::endl;
210        mapBufferSize_[rank] = CXios::minBufferSize;
211        maxEventSizes[rank] = CXios::minBufferSize;
212      }
213     
214      CClientBuffer* buffer = buffers[rank] = new CClientBuffer(interCommMerged_, clientSize+rank, mapBufferSize_[rank], maxEventSizes[rank]);
215      if (isGrowableBuffer_) buffer->setGrowableBuffer(1.2) ;
216      else buffer->fixBuffer() ;
217      // Notify the server
218      CBufferOut* bufOut = buffer->getBuffer(0, 4*sizeof(MPI_Aint));
219      MPI_Aint sendBuff[4] ;
220      sendBuff[0]=hashId_;
221      sendBuff[1]=mapBufferSize_[rank];
222      sendBuff[2]=buffers[rank]->getWinAddress(0); 
223      sendBuff[3]=buffers[rank]->getWinAddress(1); 
224      info(100)<<"CLegacyContextClient::newBuffer : rank "<<rank<<" winAdress[0] "<<buffers[rank]->getWinAddress(0)<<" winAdress[1] "<<buffers[rank]->getWinAddress(1)<<endl;
225      bufOut->put(sendBuff, 4); 
226      buffer->checkBuffer(true);
227     
228       // create windows dynamically for one-sided
229      if (!isAttachedModeEnabled())
230      { 
231        CTimer::get("create Windows").resume() ;
232        MPI_Comm interComm ;
233        MPI_Intercomm_create(commSelf_, 0, interCommMerged_, clientSize+rank, 0, &interComm) ;
234        MPI_Intercomm_merge(interComm, false, &winComm_[rank]) ;
235        CXios::getMpiGarbageCollector().registerCommunicator(winComm_[rank]) ;
236        MPI_Comm_free(&interComm) ;
237        windows_[rank].resize(2) ;
238       
239        MPI_Win_create_dynamic(MPI_INFO_NULL, winComm_[rank], &windows_[rank][0]);
240        CXios::getMpiGarbageCollector().registerWindow(windows_[rank][0]) ;
241       
242        MPI_Win_create_dynamic(MPI_INFO_NULL, winComm_[rank], &windows_[rank][1]);   
243        CXios::getMpiGarbageCollector().registerWindow(windows_[rank][1]) ;
244
245        CTimer::get("create Windows").suspend() ;
246      }
247      else
248      {
249        winComm_[rank] = MPI_COMM_NULL ;
250        windows_[rank].resize(2) ;
251        windows_[rank][0] = MPI_WIN_NULL ;
252        windows_[rank][1] = MPI_WIN_NULL ;
253      }
254      buffer->attachWindows(windows_[rank]) ;
255      if (!isAttachedModeEnabled()) MPI_Barrier(winComm_[rank]) ;
256       
257   }
258
259   /*!
260   Verify state of buffers. Buffer is under pending state if there is no message on it
261   \return state of buffers, pending(true), ready(false)
262   */
263   bool CLegacyContextClient::checkBuffers(void)
264   {
265      map<int,CClientBuffer*>::iterator itBuff;
266      bool pending = false;
267      for (itBuff = buffers.begin(); itBuff != buffers.end(); itBuff++)
268        pending |= itBuff->second->checkBuffer(!pureOneSided);
269      return pending;
270   }
271
272   //! Release all buffers
273   void CLegacyContextClient::releaseBuffers()
274   {
275      map<int,CClientBuffer*>::iterator itBuff;
276      for (itBuff = buffers.begin(); itBuff != buffers.end(); itBuff++)
277      {
278         delete itBuff->second;
279      }
280      buffers.clear();
281
282// don't know when release windows
283
284      //if (!isAttachedModeEnabled())
285      //{ 
286      //  for(auto& it : winComm_)
287      //  {
288      //    int rank = it.first ;
289      //    MPI_Win_free(&windows_[rank][0]);
290      //    MPI_Win_free(&windows_[rank][1]);
291      //    MPI_Comm_free(&winComm_[rank]) ;
292      //  }
293      //}
294   }
295
296     
297  /*!
298   Lock the buffers for one sided communications
299   \param [in] ranks list rank of server to which client connects to
300   */
301   void CLegacyContextClient::lockBuffers(list<int>& ranks)
302   {
303      list<int>::iterator it;
304      for (it = ranks.begin(); it != ranks.end(); it++) buffers[*it]->lockBuffer();
305   }
306
307  /*!
308   Unlock the buffers for one sided communications
309   \param [in] ranks list rank of server to which client connects to
310   */
311   void CLegacyContextClient::unlockBuffers(list<int>& ranks)
312   {
313      list<int>::iterator it;
314      for (it = ranks.begin(); it != ranks.end(); it++) buffers[*it]->unlockBuffer();
315   }
316     
317   /*!
318   Verify state of buffers corresponding to a connection
319   \param [in] ranks list rank of server to which client connects to
320   \return state of buffers, pending(true), ready(false)
321   */
322   bool CLegacyContextClient::checkBuffers(list<int>& ranks)
323   {
324      list<int>::iterator it;
325      bool pending = false;
326      for (it = ranks.begin(); it != ranks.end(); it++) pending |= buffers[*it]->checkBuffer(!pureOneSided);
327      return pending;
328   }
329
330   /*!
331    * Set the buffer size for each connection. Warning: This function is collective.
332    *
333    * \param [in] mapSize maps the rank of the connected servers to the size of the correspoinding buffer
334    * \param [in] maxEventSize maps the rank of the connected servers to the size of the biggest event
335   */
336   void CLegacyContextClient::setBufferSize(const std::map<int,StdSize>& mapSize)
337   {
338     setFixedBuffer() ;
339     for(auto& it : mapSize)
340     {
341      size_t size=std::max(CXios::minBufferSize*1.0,std::min(it.second*CXios::bufferSizeFactor*1.01,CXios::maxBufferSize*1.0)) ;
342      mapBufferSize_[it.first]=size ;
343      if (buffers.count(it.first)>0) buffers[it.first]->fixBufferSize(size);
344     }
345   }
346
347   /*!
348   * Finalize context client and do some reports. Function is non-blocking.
349   */
350  void CLegacyContextClient::finalize(void)
351  {
352    map<int,CClientBuffer*>::iterator itBuff;
353    std::list<int>::iterator ItServerLeader; 
354   
355    bool stop = false;
356
357    int* nbServerConnectionLocal  = new int[serverSize] ;
358    int* nbServerConnectionGlobal  = new int[serverSize] ;
359    for(int i=0;i<serverSize;++i) nbServerConnectionLocal[i]=0 ;
360    for (itBuff = buffers.begin(); itBuff != buffers.end(); itBuff++)  nbServerConnectionLocal[itBuff->first]=1 ;
361    for (ItServerLeader = ranksServerLeader.begin(); ItServerLeader != ranksServerLeader.end(); ItServerLeader++)  nbServerConnectionLocal[*ItServerLeader]=1 ;
362   
363    MPI_Allreduce(nbServerConnectionLocal, nbServerConnectionGlobal, serverSize, MPI_INT, MPI_SUM, intraComm);
364   
365    CEventClient event(CContext::GetType(), CContext::EVENT_ID_CONTEXT_FINALIZE);
366    CMessage msg;
367
368    for (int i=0;i<serverSize;++i) if (nbServerConnectionLocal[i]==1) event.push(i, nbServerConnectionGlobal[i], msg) ;
369    sendEvent(event);
370
371    delete[] nbServerConnectionLocal ;
372    delete[] nbServerConnectionGlobal ;
373
374
375    CTimer::get("Blocking time").resume();
376    checkBuffers();
377    CTimer::get("Blocking time").suspend();
378
379    std::map<int,StdSize>::const_iterator itbMap = mapBufferSize_.begin(),
380                                          iteMap = mapBufferSize_.end(), itMap;
381
382    StdSize totalBuf = 0;
383    for (itMap = itbMap; itMap != iteMap; ++itMap)
384    {
385      report(10) << " Memory report : Context <" << context_->getId() << "> : client side : memory used for buffer of each connection to server" << endl
386                 << "  +) To server with rank " << itMap->first << " : " << itMap->second << " bytes " << endl;
387      totalBuf += itMap->second;
388    }
389    report(0) << " Memory report : Context <" << context_->getId() << "> : client side : total memory used for buffer " << totalBuf << " bytes" << endl;
390
391  }
392
393
394  /*!
395  */
396  bool CLegacyContextClient::havePendingRequests(void)
397  {
398    bool pending = false;
399    map<int,CClientBuffer*>::iterator itBuff;
400    for (itBuff = buffers.begin(); itBuff != buffers.end(); itBuff++)
401      pending |= itBuff->second->hasPendingRequest();
402    return pending;
403  }
404 
405  bool CLegacyContextClient::havePendingRequests(list<int>& ranks)
406  {
407      list<int>::iterator it;
408      bool pending = false;
409      for (it = ranks.begin(); it != ranks.end(); it++) pending |= buffers[*it]->hasPendingRequest();
410      return pending;
411  }
412
413  bool CLegacyContextClient::isNotifiedFinalized(void)
414  {
415    if (isAttachedModeEnabled()) return true ;
416
417    bool finalized = true;
418    map<int,CClientBuffer*>::iterator itBuff;
419    for (itBuff = buffers.begin(); itBuff != buffers.end(); itBuff++)
420      finalized &= itBuff->second->isNotifiedFinalized();
421    return finalized;
422  }
423
424}
Note: See TracBrowser for help on using the repository browser.