source: XIOS/dev/dev_ym/XIOS_COUPLING/src/context_server.cpp @ 1853

Last change on this file since 1853 was 1853, checked in by ymipsl, 4 years ago

Coupling branch : replace hasServer and hasClient combination by the name of correct service : CLIENT, GATHERER or OUT_SERVER.

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: 13.1 KB
Line 
1#include "context_server.hpp"
2#include "buffer_in.hpp"
3#include "type.hpp"
4#include "context.hpp"
5#include "object_template.hpp"
6#include "group_template.hpp"
7#include "attribute_template.hpp"
8#include "domain.hpp"
9#include "field.hpp"
10#include "file.hpp"
11#include "grid.hpp"
12#include "mpi.hpp"
13#include "tracer.hpp"
14#include "timer.hpp"
15#include "cxios.hpp"
16#include "event_scheduler.hpp"
17#include "server.hpp"
18#include "servers_ressource.hpp"
19#include "pool_ressource.hpp"
20#include "services.hpp"
21#include "contexts_manager.hpp"
22
23#include <boost/functional/hash.hpp>
24#include <random>
25#include <chrono>
26
27
28namespace xios
29{
30  using namespace std ;
31
32  CContextServer::CContextServer(CContext* parent,MPI_Comm intraComm_,MPI_Comm interComm_) 
33    : eventScheduler_(nullptr), isProcessingEvent_(false), associatedClient_(nullptr)
34  {
35    context=parent;
36    intraComm=intraComm_;
37    MPI_Comm_size(intraComm,&intraCommSize);
38    MPI_Comm_rank(intraComm,&intraCommRank);
39
40    interComm=interComm_;
41    int flag;
42    MPI_Comm_test_inter(interComm,&flag);
43
44    if (flag) attachedMode=false ;
45    else  attachedMode=true ;
46   
47    if (flag) MPI_Comm_remote_size(interComm,&commSize);
48    else  MPI_Comm_size(interComm,&commSize);
49
50   
51    SRegisterContextInfo contextInfo ;
52    CXios::getContextsManager()->getContextInfo(context->getId(), contextInfo, intraComm) ;
53
54    if (contextInfo.serviceType != CServicesManager::CLIENT) // we must have an event scheduler => to be retrieve from the associated services
55    {
56      if (!isAttachedModeEnabled()) eventScheduler_=CXios::getPoolRessource()->getService(contextInfo.serviceId,contextInfo.partitionId)->getEventScheduler() ;
57    }
58
59
60    currentTimeLine=1;
61    scheduled=false;
62    finished=false;
63
64    // generate unique hash for server
65    auto time=chrono::system_clock::now().time_since_epoch().count() ;
66    std::default_random_engine rd(time); // not reproducible from a run to another
67    std::uniform_int_distribution<size_t> dist;
68    hashId=dist(rd) ;
69    MPI_Bcast(&hashId,1,MPI_SIZE_T,0,intraComm) ; // Bcast to all server of the context
70
71
72    if (!isAttachedModeEnabled())
73    {
74      MPI_Intercomm_merge(interComm_,true,&interCommMerged) ;
75// create windows for one sided comm
76      int interCommMergedRank;
77      MPI_Comm winComm ;
78      MPI_Comm_rank(intraComm, &interCommMergedRank);
79      windows.resize(2) ;
80      for(int rank=commSize; rank<commSize+intraCommSize; rank++)
81      {
82        if (rank==commSize+interCommMergedRank) 
83        {
84          MPI_Comm_split(interCommMerged, interCommMergedRank, rank, &winComm);
85          int myRank ;
86          MPI_Comm_rank(winComm,&myRank);
87          MPI_Win_create_dynamic(MPI_INFO_NULL, winComm, &windows[0]);
88          MPI_Win_create_dynamic(MPI_INFO_NULL, winComm, &windows[1]);     
89        }
90        else MPI_Comm_split(interCommMerged, interCommMergedRank, rank, &winComm);
91        MPI_Comm_free(&winComm) ;
92      }
93    }
94    else 
95    {
96      windows.resize(2) ;
97      windows[0]=MPI_WIN_NULL ;
98      windows[1]=MPI_WIN_NULL ;
99    }
100
101
102   
103    MPI_Comm_split(intraComm_,intraCommRank,intraCommRank, &commSelf) ;
104    itLastTimeLine=lastTimeLine.begin() ;
105
106    pureOneSided=CXios::getin<bool>("pure_one_sided",false); // pure one sided communication (for test)
107    if (isAttachedModeEnabled()) pureOneSided=false ; // no one sided in attach mode
108     
109  }
110
111//! Attached mode is used ?
112//! \return true if attached mode is used, false otherwise
113  bool CContextServer::isAttachedModeEnabled() const
114  {
115    return attachedMode ;
116  }
117 
118  void CContextServer::setPendingEvent(void)
119  {
120    pendingEvent=true;
121  }
122
123  bool CContextServer::hasPendingEvent(void)
124  {
125    return pendingEvent;
126  }
127
128  bool CContextServer::hasFinished(void)
129  {
130    return finished;
131  }
132
133  bool CContextServer::eventLoop(bool enableEventsProcessing /*= true*/)
134  {
135    listen();
136    checkPendingRequest();
137    if (enableEventsProcessing)  processEvents();
138    return finished;
139  }
140
141  void CContextServer::listen(void)
142  {
143    int rank;
144    int flag;
145    int count;
146    char * addr;
147    MPI_Status status;
148    map<int,CServerBuffer*>::iterator it;
149    bool okLoop;
150
151    traceOff();
152    MPI_Iprobe(MPI_ANY_SOURCE, 20,interComm,&flag,&status);
153    traceOn();
154
155    if (flag==true)
156    {
157      rank=status.MPI_SOURCE ;
158      okLoop = true;
159      if (pendingRequest.find(rank)==pendingRequest.end())
160        okLoop = !listenPendingRequest(status) ;
161      if (okLoop)
162      {
163        for(rank=0;rank<commSize;rank++)
164        {
165          if (pendingRequest.find(rank)==pendingRequest.end())
166          {
167
168            traceOff();
169            MPI_Iprobe(rank, 20,interComm,&flag,&status);
170            traceOn();
171            if (flag==true) listenPendingRequest(status) ;
172          }
173        }
174      }
175    }
176  }
177
178  bool CContextServer::listenPendingRequest(MPI_Status& status)
179  {
180    int count;
181    char * addr;
182    map<int,CServerBuffer*>::iterator it;
183    int rank=status.MPI_SOURCE ;
184
185    it=buffers.find(rank);
186    if (it==buffers.end()) // Receive the buffer size and allocate the buffer
187    {
188       MPI_Aint recvBuff[3] ;
189       MPI_Recv(recvBuff, 3, MPI_AINT, rank, 20, interComm, &status);
190       StdSize buffSize = recvBuff[0];
191       vector<MPI_Aint> winAdress(2) ;
192       winAdress[0]=recvBuff[1] ; winAdress[1]=recvBuff[2] ;
193       mapBufferSize_.insert(std::make_pair(rank, buffSize));
194       it=(buffers.insert(pair<int,CServerBuffer*>(rank,new CServerBuffer(windows, winAdress, rank, buffSize)))).first;
195     
196       lastTimeLine[rank]=0 ;
197       itLastTimeLine=lastTimeLine.begin() ;
198
199       return true;
200    }
201    else
202    {
203      MPI_Get_count(&status,MPI_CHAR,&count);
204      if (it->second->isBufferFree(count))
205      {
206         addr=(char*)it->second->getBuffer(count);
207         MPI_Irecv(addr,count,MPI_CHAR,rank,20,interComm,&pendingRequest[rank]);
208         bufferRequest[rank]=addr;
209         return true;
210       }
211      else
212        return false;
213    }
214  }
215
216
217  void CContextServer::checkPendingRequest(void)
218  {
219    map<int,MPI_Request>::iterator it;
220    list<int> recvRequest;
221    list<int>::iterator itRecv;
222    int rank;
223    int flag;
224    int count;
225    MPI_Status status;
226
227    for(it=pendingRequest.begin();it!=pendingRequest.end();it++)
228    {
229      rank=it->first;
230      traceOff();
231      MPI_Test(& it->second, &flag, &status);
232      traceOn();
233      if (flag==true)
234      {
235        buffers[rank]->updateCurrentWindows() ;
236        recvRequest.push_back(rank);
237        MPI_Get_count(&status,MPI_CHAR,&count);
238        processRequest(rank,bufferRequest[rank],count);
239      }
240    }
241
242    for(itRecv=recvRequest.begin();itRecv!=recvRequest.end();itRecv++)
243    {
244      pendingRequest.erase(*itRecv);
245      bufferRequest.erase(*itRecv);
246    }
247  }
248
249  void CContextServer::getBufferFromClient(size_t timeLine)
250  {
251    if (!isAttachedModeEnabled()) // one sided desactivated in attached mode
252    { 
253      int rank ;
254      char *buffer ;
255      size_t count ; 
256
257      if (itLastTimeLine==lastTimeLine.end()) itLastTimeLine=lastTimeLine.begin() ;
258      for(;itLastTimeLine!=lastTimeLine.end();++itLastTimeLine)
259      {
260        rank=itLastTimeLine->first ;
261        if (itLastTimeLine->second < timeLine &&  pendingRequest.count(rank)==0)
262        {
263          if (buffers[rank]->getBufferFromClient(timeLine, buffer, count))
264          {
265            processRequest(rank, buffer, count);
266            break ;
267          }
268        }
269      }
270    }
271  }
272         
273       
274  void CContextServer::processRequest(int rank, char* buff,int count)
275  {
276
277    CBufferIn buffer(buff,count);
278    char* startBuffer,endBuffer;
279    int size, offset;
280    size_t timeLine=0;
281    map<size_t,CEventServer*>::iterator it;
282
283   
284    CTimer::get("Process request").resume();
285    while(count>0)
286    {
287      char* startBuffer=(char*)buffer.ptr();
288      CBufferIn newBuffer(startBuffer,buffer.remain());
289      newBuffer>>size>>timeLine;
290      it=events.find(timeLine);
291      if (it==events.end()) it=events.insert(pair<int,CEventServer*>(timeLine,new CEventServer(this))).first;
292      it->second->push(rank,buffers[rank],startBuffer,size);
293
294      buffer.advance(size);
295      count=buffer.remain();
296    }
297
298    if (timeLine>0) lastTimeLine[rank]=timeLine ;
299   
300    CTimer::get("Process request").suspend();
301  }
302
303  void CContextServer::processEvents(void)
304  {
305    map<size_t,CEventServer*>::iterator it;
306    CEventServer* event;
307   
308//    if (context->isProcessingEvent()) return ;
309    if (isProcessingEvent_) return ;
310
311    it=events.find(currentTimeLine);
312    if (it!=events.end())
313    {
314      event=it->second;
315
316      if (event->isFull())
317      {
318        if (!scheduled && eventScheduler_) // Skip event scheduling for attached mode and reception on client side
319        {
320          eventScheduler_->registerEvent(currentTimeLine,hashId);
321          scheduled=true;
322        }
323        else if (!eventScheduler_ || eventScheduler_->queryEvent(currentTimeLine,hashId) )
324        {
325         // When using attached mode, synchronise the processes to avoid that differents event be scheduled by differents processes
326         // The best way to properly solve this problem will be to use the event scheduler also in attached mode
327         // for now just set up a MPI barrier
328         if (!eventScheduler_ && CXios::isServer) MPI_Barrier(intraComm) ;
329
330//         context->setProcessingEvent() ;
331         isProcessingEvent_=true ;
332         CTimer::get("Process events").resume();
333         dispatchEvent(*event);
334         CTimer::get("Process events").suspend();
335         isProcessingEvent_=false ;
336//         context->unsetProcessingEvent() ;
337         pendingEvent=false;
338         delete event;
339         events.erase(it);
340         currentTimeLine++;
341         scheduled = false;
342        }
343      }
344      else getBufferFromClient(currentTimeLine) ;
345    }
346    else if (pureOneSided) getBufferFromClient(currentTimeLine) ; // if pure one sided check buffer even if no event recorded at current time line
347  }
348
349  CContextServer::~CContextServer()
350  {
351    map<int,CServerBuffer*>::iterator it;
352    for(it=buffers.begin();it!=buffers.end();++it) delete it->second;
353  }
354
355  void CContextServer::releaseBuffers()
356  {
357    map<int,CServerBuffer*>::iterator it;
358    bool out ;
359    do
360    {
361      out=true ;
362      for(it=buffers.begin();it!=buffers.end();++it)
363      {
364//        out = out && it->second->freeWindows() ;
365
366      }
367    } while (! out) ; 
368  }
369
370  void CContextServer::notifyClientsFinalize(void)
371  {
372    for(auto it=buffers.begin();it!=buffers.end();++it)
373    {
374      it->second->notifyClientFinalize() ;
375    }
376  }
377
378  void CContextServer::dispatchEvent(CEventServer& event)
379  {
380    string contextName;
381    string buff;
382    int MsgSize;
383    int rank;
384    list<CEventServer::SSubEvent>::iterator it;
385    StdString ctxId = context->getId();
386    CContext::setCurrent(ctxId);
387    StdSize totalBuf = 0;
388
389    if (event.classId==CContext::GetType() && event.type==CContext::EVENT_ID_CONTEXT_FINALIZE)
390    {
391      finished=true;
392      info(20)<<" CContextServer: Receive context <"<<context->getId()<<"> finalize."<<endl;
393//      releaseBuffers() ;
394      notifyClientsFinalize() ;
395      context->finalize();
396
397/* don't know where release windows
398      MPI_Win_free(&windows[0]) ;
399      MPI_Win_free(&windows[1]) ;
400*/     
401      std::map<int, StdSize>::const_iterator itbMap = mapBufferSize_.begin(),
402                           iteMap = mapBufferSize_.end(), itMap;
403      for (itMap = itbMap; itMap != iteMap; ++itMap)
404      {
405        rank = itMap->first;
406        report(10)<< " Memory report : Context <"<<ctxId<<"> : server side : memory used for buffer of each connection to client" << endl
407            << "  +) With client of rank " << rank << " : " << itMap->second << " bytes " << endl;
408        totalBuf += itMap->second;
409      }
410      report(0)<< " Memory report : Context <"<<ctxId<<"> : server side : total memory used for buffer "<<totalBuf<<" bytes"<<endl;
411    }
412    else if (event.classId==CContext::GetType()) CContext::dispatchEvent(event);
413    else if (event.classId==CContextGroup::GetType()) CContextGroup::dispatchEvent(event);
414    else if (event.classId==CCalendarWrapper::GetType()) CCalendarWrapper::dispatchEvent(event);
415    else if (event.classId==CDomain::GetType()) CDomain::dispatchEvent(event);
416    else if (event.classId==CDomainGroup::GetType()) CDomainGroup::dispatchEvent(event);
417    else if (event.classId==CAxis::GetType()) CAxis::dispatchEvent(event);
418    else if (event.classId==CAxisGroup::GetType()) CAxisGroup::dispatchEvent(event);
419    else if (event.classId==CScalar::GetType()) CScalar::dispatchEvent(event);
420    else if (event.classId==CScalarGroup::GetType()) CScalarGroup::dispatchEvent(event);
421    else if (event.classId==CGrid::GetType()) CGrid::dispatchEvent(event);
422    else if (event.classId==CGridGroup::GetType()) CGridGroup::dispatchEvent(event);
423    else if (event.classId==CField::GetType()) CField::dispatchEvent(event);
424    else if (event.classId==CFieldGroup::GetType()) CFieldGroup::dispatchEvent(event);
425    else if (event.classId==CFile::GetType()) CFile::dispatchEvent(event);
426    else if (event.classId==CFileGroup::GetType()) CFileGroup::dispatchEvent(event);
427    else if (event.classId==CVariable::GetType()) CVariable::dispatchEvent(event);
428    else
429    {
430      ERROR("void CContextServer::dispatchEvent(CEventServer& event)",<<" Bad event class Id"<<endl);
431    }
432  }
433}
Note: See TracBrowser for help on using the repository browser.