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

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

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