source: XIOS/dev/branch_openmp/src/event_scheduler.cpp @ 1520

Last change on this file since 1520 was 1520, checked in by yushan, 3 years ago

save dev. TO DO : test with xios

  • 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: 6.6 KB
Line 
1#include "event_scheduler.hpp"
2#include "xios_spl.hpp"
3#include "mpi.hpp"
4#include "tracer.hpp"
5
6using namespace ep_lib;
7
8namespace xios
9{
10 
11 
12  CEventScheduler::CEventScheduler(const MPI_Comm& comm) 
13  {
14    MPI_Comm_dup(comm, &communicator) ;
15    MPI_Comm_size(communicator,&mpiSize) ;
16    MPI_Comm_rank(communicator,&mpiRank);
17
18
19    int maxChild=1 ;
20
21    int m ;
22    do
23    {
24      m=1 ;
25      maxChild=maxChild+1 ;
26      for(int i=0;i<maxChild;i++) m=m*maxChild ;
27     } while(m<mpiSize) ;
28   
29   
30    int maxLevel=0 ;
31    for(int size=1; size<=mpiSize; size*=maxChild) maxLevel++ ; 
32
33    int begin, end, nb ;
34    int pos, n ;
35 
36    parent=vector<int>(maxLevel+1) ;
37    child=vector<vector<int> >(maxLevel+1,vector<int>(maxChild)) ;
38    nbChild=vector<int> (maxLevel+1) ;
39   
40    level=0 ;
41    begin=0 ;
42    end=mpiSize-1 ;     
43    nb=end-begin+1 ;
44     
45    do
46    {
47      n=0 ;
48      pos=begin ;
49      nbChild[level]=0 ;
50      parent[level+1]=begin ;
51      for(int i=0;i<maxChild && i<nb ;i++)
52      {
53        if (i<nb%maxChild) n = nb/maxChild + 1 ;
54        else n = nb/maxChild ;
55     
56        if (mpiRank>=pos && mpiRank<pos+n)
57        {
58          begin=pos ;
59          end=pos+n-1 ;
60        }
61        child[level][i]=pos ;
62        pos=pos+n ;
63        nbChild[level]++ ;
64      } 
65      nb=end-begin+1 ;
66      level=level+1 ;
67    } while (nb>1) ;
68
69   
70  }
71
72  CEventScheduler::~CEventScheduler()
73  {
74
75  } 
76
77  void CEventScheduler::registerEvent(const size_t timeLine, const size_t contextHashId)
78  {
79    registerEvent(timeLine, contextHashId, level) ;
80  }
81 
82  void CEventScheduler::registerEvent(const size_t timeLine, const size_t contextHashId, const size_t lev)
83  {
84       
85    traceOff() ;
86    SPendingRequest* sentRequest=new SPendingRequest ;
87    sentRequest->buffer[0]=timeLine ;
88    sentRequest->buffer[1]=contextHashId ;
89    sentRequest->buffer[2]=lev-1 ;
90
91    pendingSentParentRequest.push(sentRequest) ;
92    MPI_Isend(sentRequest->buffer,3, MPI_UNSIGNED_LONG, parent[lev], 0, communicator, &sentRequest->request) ;
93    traceOn() ;
94  } 
95
96  bool CEventScheduler::queryEvent(const size_t timeLine, const size_t contextHashId)
97  {
98
99    if (! eventStack.empty() && eventStack.front().first==timeLine && eventStack.front().second==contextHashId)
100    {
101      eventStack.pop() ;
102      return true ;
103    }
104    else return false ; 
105  } 
106 
107  void CEventScheduler::checkEvent(void)
108  {
109    traceOff() ;
110    checkChildRequest() ;
111    checkParentRequest() ;
112    traceOn() ;
113   
114  }
115 
116  void CEventScheduler::checkParentRequest(void)
117  {
118    int completed ;
119    MPI_Status status ;
120    int received ;
121    SPendingRequest* recvRequest ;
122    completed=true ;
123   
124    // check sent request to parent
125    while (! pendingSentParentRequest.empty() && completed)
126    {
127      MPI_Test( & pendingSentParentRequest.front()->request, &completed, &status) ;
128      if (completed) 
129      {
130        delete pendingSentParentRequest.front() ;
131        pendingSentParentRequest.pop() ;
132      }
133    }
134   
135    // probe if a message is coming from parent
136    received=true ;
137    while(received)
138    {
139      MPI_Iprobe(-2,1,communicator,&received, &status) ;
140      if (received)
141      {
142        printf("probed a message coming from parent\n"),
143        recvRequest=new SPendingRequest ;
144        MPI_Irecv(recvRequest->buffer, 3, MPI_UNSIGNED_LONG, -2, 1, communicator, &(recvRequest->request)) ;
145        pendingRecvParentRequest.push(recvRequest) ;
146      }
147    }
148   
149     // check sent request from parent
150    completed=true ;
151    while (! pendingRecvParentRequest.empty() && completed)
152    {
153      recvRequest=pendingRecvParentRequest.front() ;
154      MPI_Test( &(recvRequest->request), &completed, &status) ;
155      if (completed) 
156      {
157        size_t timeLine=recvRequest->buffer[0] ;
158        size_t hashId=recvRequest->buffer[1] ;
159        size_t lev=recvRequest->buffer[2] ;
160        delete recvRequest ;
161        pendingRecvParentRequest.pop() ;       
162 
163        if (lev==level) eventStack.push(pair<size_t,size_t>(timeLine,hashId)) ;
164        else  bcastEvent(timeLine, hashId, lev) ;
165      }
166    }   
167   
168  }
169
170  void CEventScheduler::checkChildRequest(void)
171  {
172// function call only by parent mpi process
173
174    MPI_Status status ; 
175    int received ;
176    received=true ;
177    SPendingRequest* recvRequest ;
178   
179    // check for posted requests and make the corresponding receive
180    while(received)
181    {
182      MPI_Iprobe(-2,0,communicator,&received, &status) ;
183      if (received)
184      {
185        recvRequest=new SPendingRequest ;
186        MPI_Irecv(recvRequest->buffer, 3, MPI_UNSIGNED_LONG, -2, 0, communicator, &recvRequest->request) ;
187        pendingRecvChildRequest.push_back(recvRequest) ;
188      }
189    }
190   
191    // check if receive request is achieved
192   
193    for(list<SPendingRequest*>::iterator it=pendingRecvChildRequest.begin(); it!=pendingRecvChildRequest.end() ; )
194    {
195      MPI_Test(&((*it)->request),&received,&status) ;
196      if (received)
197      {
198        size_t timeLine=(*it)->buffer[0] ;
199        size_t hashId=(*it)->buffer[1] ;
200        size_t lev=(*it)->buffer[2] ;
201       
202        SEvent event={timeLine,hashId,lev} ;
203        delete *it ; // free mem
204        it=pendingRecvChildRequest.erase(it) ; // get out of the list
205       
206        map< SEvent,int>::iterator itEvent=recvEvent.find(event) ;
207        if (itEvent==recvEvent.end()) 
208        {
209          itEvent=(recvEvent.insert(pair< SEvent ,int > (event,1))).first ;
210 
211        }
212        else (itEvent->second)++ ;
213        if (itEvent->second==nbChild[lev])
214        {
215          if (lev==0)
216          {
217            bcastEvent(timeLine,hashId,lev) ;
218            recvEvent.erase(itEvent) ;
219          }
220          else
221          {
222            registerEvent( timeLine,hashId,lev) ;
223          }
224        }
225      }
226      else ++it ;
227    }
228   
229    // check if bcast request is achieved
230
231    for(list<SPendingRequest*>::iterator it=pendingSentChildRequest.begin(); it!=pendingSentChildRequest.end() ; )
232    {
233      MPI_Test(&(*it)->request,&received,&status) ;
234      if (received)
235      {
236        delete *it ;    // free memory
237        it = pendingSentChildRequest.erase(it) ;          // get out of the list
238
239      }
240      else ++it ;
241       
242    }
243  }
244 
245  void CEventScheduler::bcastEvent(const size_t timeLine, const size_t contextHashId, const size_t lev)
246  {
247    SPendingRequest* sentRequest ;
248     
249   
250    for(int i=0; i<nbChild[lev];i++)
251    {
252      sentRequest=new SPendingRequest ;
253      sentRequest->buffer[0]=timeLine ;
254      sentRequest->buffer[1]=contextHashId ;
255      sentRequest->buffer[2]=lev+1 ;
256      MPI_Isend(sentRequest->buffer,3, MPI_UNSIGNED_LONG, child[lev][i], 1, communicator, & sentRequest->request) ;
257      pendingSentChildRequest.push_back(sentRequest) ;
258    }
259  }
260   
261
262}
Note: See TracBrowser for help on using the repository browser.