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

Last change on this file since 2259 was 2259, checked in by ymipsl, 3 years ago

Improvment of one-sided protocol.
Windows are now created in the flight for each client-server connection.
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.0 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   
251    if (hasWindows)
252    { 
253      MPI_Win_lock(MPI_LOCK_EXCLUSIVE, clientRank_, 0, windows_[0]) ;
254      MPI_Win_unlock(clientRank_, windows_[0]) ;
255
256      MPI_Win_lock(MPI_LOCK_EXCLUSIVE, clientRank_, 0, windows_[1]) ;
257      MPI_Win_unlock(clientRank_, windows_[1]) ;
258    }
259   
260    if (pending)
261    {
262      traceOff();
263      MPI_Test(&request, &flag, &status);
264      traceOn();
265      if (flag == true) pending = false;
266    }
267
268    if (!pending)
269    {
270      if (!send && resizingBufferStep_==0 ) return false ;
271
272      if (count > 0)
273      {
274        double time=MPI_Wtime() ;
275        if (time - lastCheckedWithNothing_ > latency_)
276        {
277          lockBuffer() ;
278          if (*bufferCount[current] > 0)
279          {
280            MPI_Issend(buffer[current], count, MPI_CHAR, serverRank, 20, interComm, &request);
281            if (resizingBufferStep_==4) resizingBufferStep_=0 ;
282            pending = true;
283            *firstTimeLine[current]=0 ;
284            *bufferCount[current]=0 ;
285
286             unlockBuffer() ;
287
288            if (current == 1) current = 0;
289            else current = 1;
290            count = 0;
291          }
292          else 
293          {
294            unlockBuffer() ;
295            lastCheckedWithNothing_ = time ;
296          }
297        }
298      }
299      else
300      {
301        if (resizingBufferStep_==1) resizeBufferNotify() ;
302        else if (resizingBufferStep_==2) isNotifiedChangeBufferSize() ;
303        else if (resizingBufferStep_==3) resizeBuffer(newBufferSize_) ;
304      }
305    }
306
307    return pending;
308  }
309
310  void CClientBuffer::resizeBufferNotify(void)
311  {
312    // notify server of changing buffers size
313    lockBuffer() ;
314    int size=sizeof(int)+sizeof(size_t) ;
315    CBufferOut* bufOut = this->getBuffer(timelineEventNotifyChangeBufferSize, size);
316    bufOut->put(size);
317    bufOut->put(timelineEventNotifyChangeBufferSize);
318    resizingBufferStep_ = 2 ;
319    unlockBuffer() ;
320  }
321
322  void CClientBuffer::resizeBuffer(size_t newSize)
323  {
324
325    if (hasWindows)
326    { 
327      MPI_Win_detach(windows_[0], bufferHeader[0]) ;
328      MPI_Win_detach(windows_[1], bufferHeader[1]) ;
329    }
330    MPI_Free_mem(bufferHeader[0]) ;
331    MPI_Free_mem(bufferHeader[1]) ;
332
333    bufferSize=newSize ;
334    MPI_Alloc_mem(bufferSize+headerSize_, MPI_INFO_NULL, &bufferHeader[0]) ;
335    MPI_Alloc_mem(bufferSize+headerSize_, MPI_INFO_NULL, &bufferHeader[1]) ;
336    buffer[0] = bufferHeader[0]+headerSize_ ;
337    buffer[1] = bufferHeader[1]+headerSize_ ;
338    firstTimeLine[0]=(size_t*)bufferHeader[0] + timeLineOffset_;
339    firstTimeLine[1]=(size_t*)bufferHeader[1] + timeLineOffset_;
340    bufferCount[0]=(size_t*)bufferHeader[0] + countOffset_ ;
341    bufferCount[1]=(size_t*)bufferHeader[1] + countOffset_ ;
342    control[0]=(size_t*)bufferHeader[0] + controlOffset_ ;  // control=0 => nothing ; control=1 => changeBufferSize
343    control[1]=(size_t*)bufferHeader[1] + controlOffset_ ;
344    notify[0]=(size_t*)bufferHeader[0] + notifyOffset_ ;
345    notify[1]=(size_t*)bufferHeader[1] + notifyOffset_ ;
346
347    *firstTimeLine[0]=0 ;
348    *firstTimeLine[1]=0 ;
349    *bufferCount[0]=0 ;
350    *bufferCount[1]=0 ;
351    *control[0]=0 ;
352    *control[1]=0 ;
353    *notify[0] = notifyNothing_ ;
354    *notify[1] = notifyNothing_ ;
355    winState[0]=false ;
356    winState[1]=false ;
357    current=0 ;
358   
359    if (hasWindows)
360    { 
361   
362      MPI_Win_attach(windows_[0], bufferHeader[0], bufferSize+headerSize_) ;
363      MPI_Win_attach(windows_[1], bufferHeader[1], bufferSize+headerSize_) ;
364         
365      MPI_Win_lock(MPI_LOCK_EXCLUSIVE, clientRank_, 0, windows_[0]) ;
366      MPI_Win_lock(MPI_LOCK_EXCLUSIVE, clientRank_, 0, windows_[1]) ;
367
368      MPI_Win_unlock(clientRank_, windows_[1]) ;
369      MPI_Win_unlock(clientRank_, windows_[0]) ;
370    } 
371
372    lockBuffer() ;
373 
374    int size=sizeof(int)+2*sizeof(size_t)+2*sizeof(MPI_Aint) ;
375    CBufferOut* bufOut = this->getBuffer(timelineEventChangeBufferSize, size);
376    bufOut->put(size);
377    bufOut->put(timelineEventChangeBufferSize);
378    bufOut->put(newBufferSize_);
379    bufOut->put(this->getWinAddress(0));
380    bufOut->put(this->getWinAddress(1));
381
382    resizingBufferStep_=4;
383    unlockBuffer() ;
384    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;
385  }
386
387  bool CClientBuffer::hasPendingRequest(void)
388  {
389   
390    lockBuffer() ;
391    count=*bufferCount[current] ;
392    unlockBuffer() ;
393
394    return (pending || count > 0);
395  }
396
397  bool CClientBuffer::isNotifiedChangeBufferSize(void)
398  {
399   
400    bool ret ;
401    lockBuffer() ;
402    ret=*notify[current] == notifyResizeBuffer_ ? true : false ;
403    if (ret) 
404    {
405      *notify[current] = notifyNothing_ ;
406      resizingBufferStep_=3; 
407    }
408    unlockBuffer() ;
409
410    return ret;
411  }
412
413  bool CClientBuffer::isNotifiedFinalized(void)
414  {
415    if (!isFinalized_)
416    {
417      double time=MPI_Wtime() ;
418//      if (time - lastCheckedNotify_ > latency_)
419      {
420        int flag ;
421        MPI_Iprobe(MPI_ANY_SOURCE, MPI_ANY_TAG, MPI_COMM_WORLD, &flag, MPI_STATUS_IGNORE);
422        lockBuffer() ;
423        isFinalized_=*notify[current] == notifyFinalize_ ? true : false ;
424        unlockBuffer() ;
425        lastCheckedNotify_=time ;
426      }
427    }
428    return isFinalized_ ;
429  }
430
431}
Note: See TracBrowser for help on using the repository browser.