source: XIOS/dev/dev_ym/XIOS_COUPLING/src/buffer_client.cpp @ 2298

Last change on this file since 2298 was 2298, checked in by ymipsl, 2 years ago

Fix dead-lock in attached mode
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: 12.9 KB
Line 
1#include "xios_spl.hpp"
2#include "exception.hpp"
3#include "log.hpp"
4#include "buffer_out.hpp"
5#include "buffer_client.hpp"
6#include "cxios.hpp"
7#include "mpi.hpp"
8#include "tracer.hpp"
9#include "timeline_events.hpp"
10#include "timer.hpp"
11
12namespace xios
13{
14  size_t CClientBuffer::maxRequestSize = 0;
15
16  CClientBuffer::CClientBuffer(MPI_Comm interComm, int serverRank, StdSize bufferSize, StdSize estimatedMaxEventSize)
17    : interComm(interComm)
18    , clientRank_(0)
19    , serverRank(serverRank)
20    , bufferSize(bufferSize)
21    , estimatedMaxEventSize(estimatedMaxEventSize)
22    , maxEventSize(0)
23    , current(0)
24    , count(0)
25    , pending(false)
26    , hasWindows(false) 
27  {
28     /*
29      if (windows[0]==MPI_WIN_NULL && windows[1]==MPI_WIN_NULL) hasWindows=false ;
30      else hasWindows=true ;
31     */
32
33      MPI_Alloc_mem(bufferSize+headerSize_, MPI_INFO_NULL, &bufferHeader[0]) ;
34      MPI_Alloc_mem(bufferSize+headerSize_, MPI_INFO_NULL, &bufferHeader[1]) ;
35      buffer[0] = bufferHeader[0]+headerSize_ ;
36      buffer[1] = bufferHeader[1]+headerSize_ ;
37      firstTimeLine[0]=(size_t*)bufferHeader[0] + timeLineOffset_ ;
38      firstTimeLine[1]=(size_t*)bufferHeader[1] + timeLineOffset_ ;
39      bufferCount[0]=(size_t*)bufferHeader[0] + countOffset_ ;
40      bufferCount[1]=(size_t*)bufferHeader[1] + countOffset_ ;
41      control[0]=(size_t*)bufferHeader[0] + controlOffset_ ;
42      control[1]=(size_t*)bufferHeader[1] + controlOffset_ ;
43      notify[0]=(size_t*)bufferHeader[0] + notifyOffset_ ;
44      notify[1]=(size_t*)bufferHeader[1] + notifyOffset_ ;
45
46      *firstTimeLine[0]=0 ;
47      *firstTimeLine[1]=0 ;
48      *bufferCount[0]=0 ;
49      *bufferCount[1]=0 ;
50      *control[0]=0 ;
51      *control[1]=0 ;
52      *notify[0]=notifyNothing_ ;
53      *notify[1]=notifyNothing_ ;
54      winState[0]=false ;
55      winState[1]=false ;
56
57
58    if (hasWindows)
59    { 
60   
61      MPI_Aint buffSize=bufferSize+headerSize_ ;
62      MPI_Win_attach(windows_[0], bufferHeader[0], buffSize) ;
63      MPI_Win_attach(windows_[1], bufferHeader[1], buffSize) ;
64   
65      MPI_Group group ;
66      int groupSize,groupRank ;
67      MPI_Win_get_group(windows_[0], &group) ;
68      MPI_Group_size(group, &groupSize) ;
69      MPI_Group_rank(group, &groupRank) ;
70      if (groupRank!=clientRank_) ERROR("CClientBuffer::CClientBuffer",<< " ClientRank != groupRank "<<clientRank_<<" "<<groupRank);
71
72      MPI_Win_get_group(windows_[1], &group) ;
73      MPI_Group_size(group, &groupSize) ;
74      MPI_Group_rank(group, &groupRank) ;
75      if (groupRank!=clientRank_) ERROR("CClientBuffer::CClientBuffer",<< " ClientRank != groupRank "<<clientRank_<<" "<<groupRank);
76
77      MPI_Win_lock(MPI_LOCK_EXCLUSIVE, clientRank_, 0, windows_[0]) ;
78      MPI_Win_lock(MPI_LOCK_EXCLUSIVE, clientRank_, 0, windows_[1]) ;
79
80      MPI_Win_unlock(clientRank_, windows_[1]) ;
81      MPI_Win_unlock(clientRank_, windows_[0]) ;
82    } 
83    retBuffer = new CBufferOut(buffer[current], bufferSize);
84    info(10) << "CClientBuffer: allocated 2 x " << bufferSize << " bytes for server " << serverRank << endl;
85  }
86
87  MPI_Aint CClientBuffer::getWinAddress(int i)
88  {
89    MPI_Aint address ;
90    MPI_Get_address(bufferHeader[i], &address) ;
91    return address ;
92  }
93
94  void CClientBuffer::attachWindows(vector<MPI_Win>& windows)
95  {
96    windows_=windows ;
97    if (windows_[0]==MPI_WIN_NULL && windows_[1]==MPI_WIN_NULL) hasWindows=false ;
98    else hasWindows=true ;
99
100    if (hasWindows)
101    { 
102      MPI_Aint buffSize=bufferSize+headerSize_ ;
103      MPI_Win_attach(windows_[0], bufferHeader[0], buffSize) ;
104      MPI_Win_attach(windows_[1], bufferHeader[1], buffSize) ;
105   
106      MPI_Group group ;
107      int groupSize,groupRank ;
108      MPI_Win_get_group(windows_[0], &group) ;
109      MPI_Group_size(group, &groupSize) ;
110      MPI_Group_rank(group, &groupRank) ;
111      if (groupRank!=clientRank_) ERROR("CClientBuffer::CClientBuffer",<< " ClientRank != groupRank "<<clientRank_<<" "<<groupRank);
112
113      MPI_Win_get_group(windows_[1], &group) ;
114      MPI_Group_size(group, &groupSize) ;
115      MPI_Group_rank(group, &groupRank) ;
116      if (groupRank!=clientRank_) ERROR("CClientBuffer::CClientBuffer",<< " ClientRank != groupRank "<<clientRank_<<" "<<groupRank);
117
118      MPI_Win_lock(MPI_LOCK_EXCLUSIVE, clientRank_, 0, windows_[0]) ;
119      MPI_Win_lock(MPI_LOCK_EXCLUSIVE, clientRank_, 0, windows_[1]) ;
120
121      MPI_Win_unlock(clientRank_, windows_[1]) ;
122      MPI_Win_unlock(clientRank_, windows_[0]) ;
123    } 
124
125  }
126
127
128  CClientBuffer::~CClientBuffer()
129  {
130     //freeWindows() ;
131     if (hasWindows)
132     {
133       MPI_Win_detach(windows_[0],bufferHeader[0]);
134       MPI_Win_detach(windows_[1],bufferHeader[1]);
135       MPI_Free_mem(bufferHeader[0]) ;
136       MPI_Free_mem(bufferHeader[1]) ;
137     }
138     delete retBuffer;
139  }
140
141  void CClientBuffer::lockBuffer(void)
142  {
143    CTimer::get("lock buffer").resume();
144    if (hasWindows)
145    {
146      MPI_Win_lock(MPI_LOCK_EXCLUSIVE,clientRank_, 0, windows_[current]) ;
147      winState[current]=true ;
148    }
149    CTimer::get("lock buffer").suspend();
150  }
151
152  void CClientBuffer::unlockBuffer(void)
153  {
154    CTimer::get("unlock buffer").resume();
155    if (hasWindows)
156    {
157      MPI_Win_unlock(clientRank_, windows_[current]) ;
158      winState[current]=false ;
159    }
160    CTimer::get("unlock buffer").suspend();
161  }
162
163  StdSize CClientBuffer::remain(void)
164  {
165    return bufferSize - count;
166  }
167
168  bool CClientBuffer::isBufferFree(StdSize size)
169  {
170 
171    lockBuffer();
172    count=*bufferCount[current] ;
173   
174    if (resizingBufferStep_ > 0 ) return false ;
175
176    if (size > bufferSize)
177    {
178      resizingBufferStep_=1 ;
179      *firstTimeLine[current]=0 ;
180      newBufferSize_=size ;
181      return false ;
182    }
183
184    if (size > maxEventSize)
185    {
186      maxEventSize = size;
187
188      if (size > estimatedMaxEventSize)
189        error(0) << "WARNING: Unexpected event of size " << size << " for server " << serverRank
190                 << " (estimated max event size = " << estimatedMaxEventSize << ")" << std::endl;
191
192      if (size > maxRequestSize) maxRequestSize = size;
193    }
194   
195    if (size > remain())
196    {
197      if (isGrowableBuffer_)
198      {
199        resizingBufferStep_ = 1 ;
200        *firstTimeLine[current]=0 ;
201        newBufferSize_ = (count+size)*growFactor_ ;
202      } 
203      return false ;
204    }
205    else return true ;
206  }
207
208
209  CBufferOut* CClientBuffer::getBuffer(size_t timeLine, StdSize size)
210  {
211    if (size <= remain())
212    {
213      retBuffer->realloc(buffer[current] + count, size);
214      count += size;
215      if (*firstTimeLine[current]==0) *firstTimeLine[current]=timeLine ;
216      *bufferCount[current]=count ;
217      return retBuffer;
218    }
219    else
220    {
221      ERROR("CBufferOut* CClientBuffer::getBuffer(StdSize size)",
222            << "Not enough space in buffer, this should not have happened...");
223      return NULL;
224    }
225  }
226
227  void CClientBuffer::infoBuffer(void)
228  {
229     
230      char checksum=0 ;
231      for(size_t i=0;i<*bufferCount[current];i++) checksum=checksum+buffer[current][i] ;
232 
233      char checksumFirst=0 ;
234      for(size_t i=5; i<10 && i<*bufferCount[current] ;i++) checksumFirst=checksumFirst+buffer[current][i] ;
235 
236      char checksumLast=0 ;
237      for(size_t i=(*bufferCount[current]<10)?0:*bufferCount[current]-10; i<*bufferCount[current] ; i++) checksumLast=checksumLast+buffer[current][i] ;
238 
239      info(45)<<"CClientBuffer::infoBuffer "<<" clientRank_ "<<clientRank_<<" serverRank "<<serverRank <<" current "<<current<<" WinState "<<winState[current]
240              <<" firstTimeLine "<<*firstTimeLine[current]<<" count "<<*bufferCount[current]<<" checksum "<<(int)checksum<<" "
241              <<(int)buffer[current][0]<<" "<<(int)buffer[current][1]<<" "<<(int)buffer[current][2]<<" "<<(int)buffer[current][3]<<" "<<(int)buffer[current][4]<<" "<<(int)buffer[current][5]<<" "
242              <<(int)buffer[current][6]<<" "<<(int)buffer[current][7]<<" "<<(int)buffer[current][8]<<" "<<(int)buffer[current][9]<<" "<<(int)buffer[current][10]<<" "<<(int)buffer[current][11]<<endl ;
243
244  }
245
246  bool CClientBuffer::checkBuffer(bool send)
247  {
248    MPI_Status status;
249    int flag;
250    MPI_Iprobe(MPI_ANY_SOURCE, MPI_ANY_TAG, MPI_COMM_WORLD, &flag, MPI_STATUS_IGNORE);
251 
252    if (pending)
253    {
254      traceOff();
255      MPI_Test(&request, &flag, &status);
256      traceOn();
257      if (flag == true) pending = false;
258    }
259
260    if (!pending)
261    {
262      if (!send && resizingBufferStep_==0 ) return false ;
263
264      if (count > 0)
265      {
266        double time=MPI_Wtime() ;
267        if (time - lastCheckedWithNothing_ > latency_)
268        {
269          lockBuffer() ;
270          if (*bufferCount[current] > 0)
271          {
272            MPI_Issend(buffer[current], count, MPI_CHAR, serverRank, 20, interComm, &request);
273            if (resizingBufferStep_==4) resizingBufferStep_=0 ;
274            pending = true;
275            *firstTimeLine[current]=0 ;
276            *bufferCount[current]=0 ;
277
278             unlockBuffer() ;
279
280            if (current == 1) current = 0;
281            else current = 1;
282            count = 0;
283          }
284          else 
285          {
286            unlockBuffer() ;
287            lastCheckedWithNothing_ = time ;
288          }
289        }
290      }
291      else
292      {
293        if (resizingBufferStep_==1) resizeBufferNotify() ;
294        else if (resizingBufferStep_==2) isNotifiedChangeBufferSize() ;
295        else if (resizingBufferStep_==3) resizeBuffer(newBufferSize_) ;
296      }
297    }
298
299    return pending;
300  }
301
302  void CClientBuffer::resizeBufferNotify(void)
303  {
304    // notify server of changing buffers size
305    lockBuffer() ;
306    int size=sizeof(int)+sizeof(size_t) ;
307    CBufferOut* bufOut = this->getBuffer(timelineEventNotifyChangeBufferSize, size);
308    bufOut->put(size);
309    bufOut->put(timelineEventNotifyChangeBufferSize);
310    resizingBufferStep_ = 2 ;
311    unlockBuffer() ;
312  }
313
314  void CClientBuffer::resizeBuffer(size_t newSize)
315  {
316
317    if (hasWindows)
318    { 
319      MPI_Win_detach(windows_[0], bufferHeader[0]) ;
320      MPI_Win_detach(windows_[1], bufferHeader[1]) ;
321    }
322    MPI_Free_mem(bufferHeader[0]) ;
323    MPI_Free_mem(bufferHeader[1]) ;
324
325    bufferSize=newSize ;
326    MPI_Alloc_mem(bufferSize+headerSize_, MPI_INFO_NULL, &bufferHeader[0]) ;
327    MPI_Alloc_mem(bufferSize+headerSize_, MPI_INFO_NULL, &bufferHeader[1]) ;
328    buffer[0] = bufferHeader[0]+headerSize_ ;
329    buffer[1] = bufferHeader[1]+headerSize_ ;
330    firstTimeLine[0]=(size_t*)bufferHeader[0] + timeLineOffset_;
331    firstTimeLine[1]=(size_t*)bufferHeader[1] + timeLineOffset_;
332    bufferCount[0]=(size_t*)bufferHeader[0] + countOffset_ ;
333    bufferCount[1]=(size_t*)bufferHeader[1] + countOffset_ ;
334    control[0]=(size_t*)bufferHeader[0] + controlOffset_ ;  // control=0 => nothing ; control=1 => changeBufferSize
335    control[1]=(size_t*)bufferHeader[1] + controlOffset_ ;
336    notify[0]=(size_t*)bufferHeader[0] + notifyOffset_ ;
337    notify[1]=(size_t*)bufferHeader[1] + notifyOffset_ ;
338
339    *firstTimeLine[0]=0 ;
340    *firstTimeLine[1]=0 ;
341    *bufferCount[0]=0 ;
342    *bufferCount[1]=0 ;
343    *control[0]=0 ;
344    *control[1]=0 ;
345    *notify[0] = notifyNothing_ ;
346    *notify[1] = notifyNothing_ ;
347    winState[0]=false ;
348    winState[1]=false ;
349    current=0 ;
350   
351    if (hasWindows)
352    { 
353   
354      MPI_Win_attach(windows_[0], bufferHeader[0], bufferSize+headerSize_) ;
355      MPI_Win_attach(windows_[1], bufferHeader[1], bufferSize+headerSize_) ;
356         
357      MPI_Win_lock(MPI_LOCK_EXCLUSIVE, clientRank_, 0, windows_[0]) ;
358      MPI_Win_lock(MPI_LOCK_EXCLUSIVE, clientRank_, 0, windows_[1]) ;
359
360      MPI_Win_unlock(clientRank_, windows_[1]) ;
361      MPI_Win_unlock(clientRank_, windows_[0]) ;
362    } 
363
364    lockBuffer() ;
365 
366    int size=sizeof(int)+2*sizeof(size_t)+2*sizeof(MPI_Aint) ;
367    CBufferOut* bufOut = this->getBuffer(timelineEventChangeBufferSize, size);
368    bufOut->put(size);
369    bufOut->put(timelineEventChangeBufferSize);
370    bufOut->put(newBufferSize_);
371    bufOut->put(this->getWinAddress(0));
372    bufOut->put(this->getWinAddress(1));
373
374    resizingBufferStep_=4;
375    unlockBuffer() ;
376    info(100)<<"CClientBuffer::resizeBuffer(size_t newSize) : resizing buffer of server "<<serverRank<<" ; new size : "<<newSize<<" ; winAdress[0] "<<this->getWinAddress(0)<<" winAdress[1] "<<this->getWinAddress(1)<<endl;
377  }
378
379  bool CClientBuffer::hasPendingRequest(void)
380  {
381   
382    lockBuffer() ;
383    count=*bufferCount[current] ;
384    unlockBuffer() ;
385
386    return (pending || count > 0);
387  }
388
389  bool CClientBuffer::isNotifiedChangeBufferSize(void)
390  {
391   
392    bool ret ;
393    lockBuffer() ;
394    ret=*notify[current] == notifyResizeBuffer_ ? true : false ;
395    if (ret || !hasWindows) 
396    {
397      *notify[current] = notifyNothing_ ;
398      resizingBufferStep_=3; 
399    }
400    unlockBuffer() ;
401
402    return ret;
403  }
404
405  bool CClientBuffer::isNotifiedFinalized(void)
406  {
407    if (!isFinalized_)
408    {
409      double time=MPI_Wtime() ;
410//      if (time - lastCheckedNotify_ > latency_)
411      {
412        int flag ;
413        MPI_Iprobe(MPI_ANY_SOURCE, MPI_ANY_TAG, MPI_COMM_WORLD, &flag, MPI_STATUS_IGNORE);
414        lockBuffer() ;
415        isFinalized_=*notify[current] == notifyFinalize_ ? true : false ;
416        unlockBuffer() ;
417        lastCheckedNotify_=time ;
418      }
419    }
420    return isFinalized_ ;
421  }
422
423}
Note: See TracBrowser for help on using the repository browser.