source: XIOS3/trunk/src/transport/one_sided_context_server.cpp @ 2628

Last change on this file since 2628 was 2628, checked in by jderouillat, 7 weeks ago

New timers integration/reporting
  • Property svn:eol-style set to native
  • Property svn:executable set to *
File size: 10.5 KB
Line 
1#include "one_sided_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#include "timeline_events.hpp"
23
24#include <boost/functional/hash.hpp>
25#include <random>
26#include <chrono>
27
28
29namespace xios
30{
31  using namespace std ;
32  extern CLogType logTimers ;
33  extern CLogType logProfile ;
34
35  COneSidedContextServer::COneSidedContextServer(CContext* parent,MPI_Comm intraComm_,MPI_Comm interComm_)
36                         : CContextServer(parent, intraComm_, interComm_), 
37                           isProcessingEvent_(false)
38  {
39   
40    xios::MPI_Comm_dup(intraComm, &processEventBarrier_) ;
41    CXios::getMpiGarbageCollector().registerCommunicator(processEventBarrier_) ;
42   
43    currentTimeLine=1;
44    scheduled=false;
45    finished=false;
46
47    xios::MPI_Intercomm_merge(interComm_,true,&interCommMerged_) ;
48    CXios::getMpiGarbageCollector().registerCommunicator(interCommMerged_) ;
49    xios::MPI_Comm_split(intraComm_, intraCommRank, intraCommRank, &commSelf_) ; // for windows
50    CXios::getMpiGarbageCollector().registerCommunicator(commSelf_) ;
51
52    itLastTimeLine=lastTimeLine.begin() ;
53
54    pureOneSided=CXios::getin<bool>("pure_one_sided",false); // pure one sided communication (for test)
55     
56  }
57
58  void COneSidedContextServer::setPendingEvent(void)
59  {
60    pendingEvent=true;
61  }
62
63  bool COneSidedContextServer::hasPendingEvent(void)
64  {
65    return ((pendingEvents_.size()!=0)||(completedEvents_.size()!=0));
66  }
67
68  bool COneSidedContextServer::hasFinished(void)
69  {
70    return finished;
71  }
72
73  bool COneSidedContextServer::eventLoop(bool enableEventsProcessing /*= true*/)
74  {
75    if (info.isActive(logTimers)) CTimer::get("listen request").resume();
76    listen();
77    if (info.isActive(logTimers)) CTimer::get("listen request").suspend();
78
79    if (info.isActive(logTimers)) CTimer::get("listen pending request").resume();
80    listenPendingRequest() ;
81    if (info.isActive(logTimers)) CTimer::get("listen pending request").suspend();
82
83    if (info.isActive(logTimers)) CTimer::get("check server Buffers").resume();
84    checkBuffers() ;
85    if (info.isActive(logTimers)) CTimer::get("check server Buffers").suspend();
86
87    if (info.isActive(logTimers)) CTimer::get("check event process").resume();
88    processEvents(enableEventsProcessing);
89    if (info.isActive(logTimers)) CTimer::get("check event process").suspend();
90    return finished;
91
92  }
93
94 void COneSidedContextServer::listen(void)
95  {
96    int rank;
97    int flag;
98    MPI_Status status;
99    flag=true ;
100
101    while(flag)
102    {
103      traceOff();
104      MPI_Iprobe(MPI_ANY_SOURCE, 20,interCommMerged_, &flag, &status);
105      traceOn();
106      if (flag==true)
107      {
108        int rank=status.MPI_SOURCE ;
109        auto& rankRequests = requests_[rank];
110        rankRequests.push_back(new CRequest(interCommMerged_, status)) ;
111        // Test 1st request of the list, request treatment must be ordered
112        if (rankRequests.front()->test()) 
113        {
114          processRequest(*(rankRequests.front())) ;
115          delete rankRequests.front();
116          rankRequests.pop_front() ;
117        }
118      }
119    }
120  }
121
122  void COneSidedContextServer::listenPendingRequest(void)
123  {
124    for(auto it_rank=requests_.begin() ; it_rank!=requests_.end() ; ++it_rank)
125    {
126      int rank = it_rank->first;
127      auto& rankRequests = it_rank->second;
128      while ( (!rankRequests.empty()) && (rankRequests.front()->test()) )
129      {
130        processRequest( *(rankRequests.front()) );
131        delete rankRequests.front();
132        rankRequests.pop_front() ;
133      }
134    }
135  }
136
137  void COneSidedContextServer::processRequest(CRequest& request)
138  {
139    int rank = request.getRank() ;
140    auto it=buffers_.find(rank);
141    if (it==buffers_.end())
142    {
143      buffers_[rank] = new COneSidedServerBuffer(rank, commSelf_, interCommMerged_, pendingEvents_, completedEvents_, request.getBuffer()) ;
144    }
145    else it->second->receivedRequest(request.getBuffer()) ;
146  }
147
148  void COneSidedContextServer::checkBuffers(void)
149  {
150    if (!pendingEvents_.empty())
151    {
152/*
153      SPendingEvent& nextEvent = pendingEvents_.begin()->second ;
154      for(auto& buffer : nextEvent.buffers ) buffer->eventLoop() ;
155      if (nextEvent.nbSenders==0) pendingEvents_.erase(pendingEvents_.begin()) ;
156*/
157      for(auto it=pendingEvents_.begin() ;  it!=pendingEvents_.end() ;)
158      {
159        SPendingEvent& nextEvent = it->second ;
160        for(auto& buffer : nextEvent.buffers ) buffer->eventLoop() ;
161        if (nextEvent.nbSenders==0) it=pendingEvents_.erase(it) ;
162        else ++it ;
163      }
164    }
165  }
166
167
168  void COneSidedContextServer::processEvents(bool enableEventsProcessing)
169  {
170 
171    if (isProcessingEvent_) return ;
172
173    auto it=completedEvents_.find(currentTimeLine);
174
175    if (it!=completedEvents_.end())
176    {
177      if (it->second.nbSenders == it->second.currentNbSenders)
178      {
179        if (!scheduled) 
180        {
181          eventScheduler_->registerEvent(currentTimeLine,hashId);
182          scheduled=true;
183        }
184        else if (eventScheduler_->queryEvent(currentTimeLine,hashId) )
185        {
186          //if (!enableEventsProcessing && isCollectiveEvent(event)) return ;
187
188          if (!eventScheduled_) 
189          {
190            MPI_Ibarrier(processEventBarrier_,&processEventRequest_) ;
191            eventScheduled_=true ;
192            return ;
193          }
194          else 
195          {
196            MPI_Status status ;
197            int flag ;
198            MPI_Test(&processEventRequest_, &flag, &status) ;
199            if (!flag) return ;
200            eventScheduled_=false ;
201          }
202
203          eventScheduler_->popEvent() ;
204
205          isProcessingEvent_=true ;
206          CEventServer event(this) ;
207          for(auto& buffer : it->second.buffers) buffer->fillEventServer(currentTimeLine, event) ;
208//          MPI_Barrier(intraComm) ;
209          CTimer::get("Process events").resume();
210          info(100)<<"Context id "<<context->getId()<<" : Process Event "<<currentTimeLine<<" of class "<<event.classId<<" of type "<<event.type<<endl ;
211          dispatchEvent(event);
212          CTimer::get("Process events").suspend();
213          isProcessingEvent_=false ;
214//         context->unsetProcessingEvent() ;
215          pendingEvent=false;
216          completedEvents_.erase(it);
217          currentTimeLine++;
218          scheduled = false;
219        }
220      }
221    }
222  }
223
224  COneSidedContextServer::~COneSidedContextServer()
225  {
226    for(auto& buffer : buffers_) delete buffer.second;
227    buffers_.clear() ;
228  }
229
230  void COneSidedContextServer::releaseBuffers()
231  {
232    //for(auto it=buffers.begin();it!=buffers.end();++it) delete it->second ;
233    //buffers.clear() ;
234    freeWindows() ;
235  }
236
237  void COneSidedContextServer::freeWindows()
238  {
239    //  for(auto& it : winComm_)
240    //  {
241    //    int rank = it.first ;
242    //    MPI_Win_free(&windows_[rank][0]);
243    //    MPI_Win_free(&windows_[rank][1]);
244    //    xios::MPI_Comm_free(&winComm_[rank]) ;
245    //  }
246  }
247
248  void COneSidedContextServer::notifyClientsFinalize(void)
249  {
250    for(auto it=buffers_.begin();it!=buffers_.end();++it)
251    {
252      it->second->notifyClientFinalize() ;
253    }
254  }
255
256  void COneSidedContextServer::dispatchEvent(CEventServer& event)
257  {
258    string contextName;
259    string buff;
260    int MsgSize;
261    int rank;
262    list<CEventServer::SSubEvent>::iterator it;
263    StdString ctxId = context->getId();
264    CContext::setCurrent(ctxId);
265    StdSize totalBuf = 0;
266
267    if (event.classId==CContext::GetType() && event.type==CContext::EVENT_ID_CONTEXT_FINALIZE)
268    {
269      if (info.isActive(logProfile)) CTimer::get("Context finalize").resume();
270      finished=true;
271      info(20)<<" COneSidedContextServer: Receive context <"<<context->getId()<<"> finalize."<<endl;
272      notifyClientsFinalize() ;
273      if (info.isActive(logTimers)) CTimer::get("receiving requests").suspend();
274      context->finalize();
275     
276      std::map<int, StdSize>::const_iterator itbMap = mapBufferSize_.begin(),
277                           iteMap = mapBufferSize_.end(), itMap;
278      for (itMap = itbMap; itMap != iteMap; ++itMap)
279      {
280        rank = itMap->first;
281        report(10)<< " Memory report : Context <"<<ctxId<<"> : server side : memory used for buffer of each connection to client" << endl
282            << "  +) With client of rank " << rank << " : " << itMap->second << " bytes " << endl;
283        totalBuf += itMap->second;
284      }
285      report(0)<< " Memory report : Context <"<<ctxId<<"> : server side : total memory used for buffer "<<totalBuf<<" bytes"<<endl;
286      if (info.isActive(logProfile)) CTimer::get("Context finalize").suspend();
287    }
288    else if (event.classId==CContext::GetType()) CContext::dispatchEvent(event);
289    else if (event.classId==CContextGroup::GetType()) CContextGroup::dispatchEvent(event);
290    else if (event.classId==CCalendarWrapper::GetType()) CCalendarWrapper::dispatchEvent(event);
291    else if (event.classId==CDomain::GetType()) CDomain::dispatchEvent(event);
292    else if (event.classId==CDomainGroup::GetType()) CDomainGroup::dispatchEvent(event);
293    else if (event.classId==CAxis::GetType()) CAxis::dispatchEvent(event);
294    else if (event.classId==CAxisGroup::GetType()) CAxisGroup::dispatchEvent(event);
295    else if (event.classId==CScalar::GetType()) CScalar::dispatchEvent(event);
296    else if (event.classId==CScalarGroup::GetType()) CScalarGroup::dispatchEvent(event);
297    else if (event.classId==CGrid::GetType()) CGrid::dispatchEvent(event);
298    else if (event.classId==CGridGroup::GetType()) CGridGroup::dispatchEvent(event);
299    else if (event.classId==CField::GetType()) 
300    {
301      if (event.type==CField::EVENT_ID_UPDATE_DATA) CField::dispatchEvent(event);
302      else CField::dispatchEvent(event);
303    }
304    else if (event.classId==CFieldGroup::GetType()) CFieldGroup::dispatchEvent(event);
305    else if (event.classId==CFile::GetType()) CFile::dispatchEvent(event);
306    else if (event.classId==CFileGroup::GetType()) CFileGroup::dispatchEvent(event);
307    else if (event.classId==CVariable::GetType()) CVariable::dispatchEvent(event);
308    else
309    {
310      ERROR("void COneSidedContextServer::dispatchEvent(CEventServer& event)",<<" Bad event class Id"<<endl);
311    }
312  }
313
314  bool COneSidedContextServer::isCollectiveEvent(CEventServer& event)
315  {
316    if (event.type>1000) return false ;
317    else return true ;
318  }
319}
Note: See TracBrowser for help on using the repository browser.