xrootd
XrdClAsyncPageReader.hh
Go to the documentation of this file.
1 //------------------------------------------------------------------------------
2 // Copyright (c) 2011-2012 by European Organization for Nuclear Research (CERN)
3 // Author: Michal Simon <michal.simon@cern.ch>
4 //------------------------------------------------------------------------------
5 // XRootD is free software: you can redistribute it and/or modify
6 // it under the terms of the GNU Lesser General Public License as published by
7 // the Free Software Foundation, either version 3 of the License, or
8 // (at your option) any later version.
9 //
10 // XRootD is distributed in the hope that it will be useful,
11 // but WITHOUT ANY WARRANTY; without even the implied warranty of
12 // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13 // GNU General Public License for more details.
14 //
15 // You should have received a copy of the GNU Lesser General Public License
16 // along with XRootD. If not, see <http://www.gnu.org/licenses/>.
17 //------------------------------------------------------------------------------
18 
19 #ifndef SRC_XRDCL_XRDCLASYNCPAGEREADER_HH_
20 #define SRC_XRDCL_XRDCLASYNCPAGEREADER_HH_
21 
23 #include "XrdCl/XrdClSocket.hh"
25 #include "XrdSys/XrdSysPageSize.hh"
26 
27 #include <sys/uio.h>
28 #include <memory>
29 #include <arpa/inet.h>
30 
31 namespace XrdCl
32 {
33 
34 //------------------------------------------------------------------------------
36 //------------------------------------------------------------------------------
38 {
39  public:
40 
41  //--------------------------------------------------------------------------
49  //--------------------------------------------------------------------------
51  std::vector<uint32_t> &digests ) :
52  chunks( chunks ),
53  digests( digests ),
54  dlen( 0 ),
55  rspoff( 0 ),
56  chindex( 0 ),
57  choff( 0 ),
58  dgindex( 0 ),
59  dgoff( 0 ),
60  iovcnt( 0 ),
61  iovindex( 0 )
62  {
63  uint64_t rdoff = chunks.front().offset;
64  uint32_t rdlen = 0;
65  for( auto &ch : chunks )
66  rdlen += ch.length;
67  int fpglen, lpglen;
68  int pgcnt = XrdOucPgrwUtils::csNum( rdoff, rdlen, fpglen, lpglen);
69  digests.resize( pgcnt );
70  }
71 
72  //--------------------------------------------------------------------------
74  //--------------------------------------------------------------------------
75  virtual ~AsyncPageReader()
76  {
77  }
78 
79  //--------------------------------------------------------------------------
81  //--------------------------------------------------------------------------
82  void SetRsp( ServerResponseV2 *rsp )
83  {
84  dlen = rsp->status.bdy.dlen;
85  rspoff = rsp->info.pgread.offset;
86 
87  uint64_t bufoff = rspoff - chunks[0].offset;
88  chindex = 0;
89 
90  for( chindex = 0; chindex < chunks.size(); ++chindex )
91  {
92  if( chunks[chindex].length < bufoff )
93  {
94  bufoff -= chunks[chindex].length;
95  continue;
96  }
97  break;
98  }
99  choff = bufoff;
100  }
101 
102  //--------------------------------------------------------------------------
106  //--------------------------------------------------------------------------
107  XRootDStatus Read( Socket &socket, uint32_t &btsread )
108  {
109  if( dlen == 0 || chindex >= chunks.size() )
110  return XRootDStatus();
111  btsread = 0;
112  int nbbts = 0;
113  do
114  {
115  // Prepare the IO vector for receiving the data
116  if( iov.empty() )
117  InitIOV();
118  // read the data into the buffer
119  nbbts = 0;
120  auto st = socket.ReadV( iov.data() + iovindex, iovcnt, nbbts );
121  if( !st.IsOK() ) return st;
122  btsread += nbbts;
123  dlen -= nbbts;
124  ShiftIOV( nbbts );
125  if( st.code == suRetry ) return st;
126  }
127  while( nbbts > 0 && dlen > 0 && chindex < chunks.size() );
128 
129  return XRootDStatus();
130  }
131 
132  private:
133 
134  //--------------------------------------------------------------------------
136  //--------------------------------------------------------------------------
137  struct iovmax_t
138  {
140  {
141 #ifdef _SC_IOV_MAX
142  value = sysconf(_SC_IOV_MAX);
143  if (value == -1)
144 #endif
145 #ifdef IOV_MAX
146  value = IOV_MAX;
147 #else
148  value = 1024;
149 #endif
150  value &= ~uint32_t( 1 ); // make sure it is an even number
151  }
152  int32_t value;
153  };
154 
155  //--------------------------------------------------------------------------
157  //--------------------------------------------------------------------------
158  inline static int max_iovcnt()
159  {
160  static iovmax_t iovmax;
161  return iovmax.value;
162  }
163 
164  //--------------------------------------------------------------------------
166  //--------------------------------------------------------------------------
167  inline void addiov( char *&buf, size_t len )
168  {
169  iov.emplace_back();
170  iov.back().iov_base = buf;
171  iov.back().iov_len = len;
172  buf += len;
173  ++iovcnt;
174  }
175 
176  //--------------------------------------------------------------------------
178  //--------------------------------------------------------------------------
179  inline void addiov( char *&buf, uint32_t len, uint32_t &dleft )
180  {
181  if( len > dleft ) len = dleft;
182  addiov( buf, len );
183  dleft -= len;
184  }
185 
186  //--------------------------------------------------------------------------
189  //--------------------------------------------------------------------------
190  inline static uint32_t CalcIOVSize( uint32_t dleft )
191  {
192  return ( dleft / PageWithDigest + 2 ) * 2;
193  }
194 
195  //--------------------------------------------------------------------------
197  //--------------------------------------------------------------------------
198  uint32_t CalcRdSize()
199  {
200  // data size in the server response (including digests)
201  uint32_t dleft = dlen;
202  // space in our page buffer
203  uint32_t pgspace = chunks[chindex].length - choff;
204  // space in our digest buffer
205  uint32_t dgspace = sizeof( uint32_t ) * (digests.size() - dgindex ) - dgoff;
206  if( dleft > pgspace + dgspace ) dleft = pgspace + dgspace;
207  return dleft;
208  }
209 
210  //--------------------------------------------------------------------------
212  //--------------------------------------------------------------------------
213  void InitIOV()
214  {
215  iovindex = 0;
216  // figure out the number of data we can read in one go
217  uint32_t dleft = CalcRdSize();
218  // and reset the I/O vector
219  iov.clear();
220  iovcnt = 0;
221  iov.reserve( CalcIOVSize( dleft ) );
222  // now prepare the page and digest buffers
223  ChunkInfo ch = chunks[chindex];
224  char* pgbuf = static_cast<char*>( ch.buffer ) + choff;
225  uint64_t rdoff = ch.offset + choff;
226  char* dgbuf = reinterpret_cast<char*>( digests.data() + dgindex ) + dgoff;
227  // handle the first digest
228  uint32_t fdglen = sizeof( uint32_t ) - dgoff;
229  addiov( dgbuf, fdglen, dleft );
230  if( dleft == 0 || iovcnt >= max_iovcnt() ) return;
231  // handle the first page
232  uint32_t fpglen = XrdSys::PageSize - rdoff % XrdSys::PageSize;
233  addiov( pgbuf, fpglen, dleft );
234  if( dleft == 0 || iovcnt >= max_iovcnt() ) return;
235  // handle all the subsequent aligned pages
236  size_t fullpgs = dleft / PageWithDigest;
237  for( size_t i = 0; i < fullpgs; ++i )
238  {
239  addiov( dgbuf, sizeof( uint32_t ) );
240  addiov( pgbuf, XrdSys::PageSize );
241  }
242  dleft -= fullpgs * PageWithDigest;
243  if( dleft == 0 || iovcnt >= max_iovcnt() ) return;
244  // handle the last digest
245  uint32_t ldglen = sizeof( uint32_t );
246  addiov( dgbuf, ldglen, dleft );
247  if( dleft == 0 || iovcnt >= max_iovcnt() ) return;
248  // handle the last page
249  addiov( pgbuf, dleft );
250  }
251 
252  //--------------------------------------------------------------------------
254  //--------------------------------------------------------------------------
255  inline void shift( void *&buffer, size_t nbbts )
256  {
257  char *buf = static_cast<char*>( buffer );
258  buf += nbbts;
259  buffer = buf;
260  }
261 
262  //--------------------------------------------------------------------------
266  //--------------------------------------------------------------------------
267  inline void shiftdgbuf( uint32_t &btsread )
268  {
269  if( iov[iovindex].iov_len > btsread )
270  {
271  iov[iovindex].iov_len -= btsread;
272  shift( iov[iovindex].iov_base, btsread );
273  dgoff += btsread;
274  btsread = 0;
275  return;
276  }
277 
278  btsread -= iov[iovindex].iov_len;
279  iov[iovindex].iov_len = 0;
280  dgoff = 0;
281  digests[dgindex] = ntohl( digests[dgindex] );
282  ++dgindex;
283  ++iovindex;
284  --iovcnt;
285  }
286 
287  //--------------------------------------------------------------------------
291  //--------------------------------------------------------------------------
292  inline void shiftpgbuf( uint32_t &btsread )
293  {
294  if( iov[iovindex].iov_len > btsread )
295  {
296  iov[iovindex].iov_len -= btsread;
297  shift( iov[iovindex].iov_base, btsread );
298  choff += btsread;
299  btsread = 0;
300  return;
301  }
302 
303  btsread -= iov[iovindex].iov_len;
304  choff += iov[iovindex].iov_len;
305  iov[iovindex].iov_len = 0;
306  ++iovindex;
307  --iovcnt;
308  }
309 
310  //--------------------------------------------------------------------------
312  //--------------------------------------------------------------------------
313  void ShiftIOV( uint32_t btsread )
314  {
315  // if iovindex is even it point to digest, otherwise it points to a page
316  if( iovindex % 2 == 0 )
317  shiftdgbuf( btsread );
318  // adjust as many I/O buffers as necessary
319  while( btsread > 0 )
320  {
321  // handle page
322  shiftpgbuf( btsread );
323  if( btsread == 0 ) break;
324  // handle digest
325  shiftdgbuf( btsread );
326  }
327  // if we filled the buffer, move to the next one
328  if( iovcnt == 0 )
329  iov.clear();
330  // do we need to move to the next chunk?
331  if( choff >= chunks[chindex].length )
332  {
333  ++chindex;
334  choff = 0;
335  }
336  }
337 
338  ChunkList &chunks; //< list of data chunks to be filled with user data
339  std::vector<uint32_t> &digests; //< list of crc32c digests for every 4KB page of data
340  uint32_t dlen; //< size of the data in the message
341  uint64_t rspoff; //< response offset
342 
343  size_t chindex; //< index of the current data buffer
344  size_t choff; //< offset within the current buffer
345  size_t dgindex; //< index of the current digest buffer
346  size_t dgoff; //< offset within the current digest buffer
347 
348  std::vector<iovec> iov; //< I/O vector
349  int iovcnt; //< size of the I/O vector
350  size_t iovindex; //< index of the first valid element in the I/O vector
351 
352  static const int PageWithDigest = XrdSys::PageSize + sizeof( uint32_t );
353 };
354 
355 } /* namespace XrdEc */
356 
357 #endif /* SRC_XRDCL_XRDCLASYNCPAGEREADER_HH_ */
Object for reading out data from the PgRead response.
Definition: XrdClAsyncPageReader.hh:38
std::vector< uint32_t > & digests
Definition: XrdClAsyncPageReader.hh:339
static int max_iovcnt()
Definition: XrdClAsyncPageReader.hh:158
void SetRsp(ServerResponseV2 *rsp)
Sets message data size.
Definition: XrdClAsyncPageReader.hh:82
size_t choff
Definition: XrdClAsyncPageReader.hh:344
static uint32_t CalcIOVSize(uint32_t dleft)
Definition: XrdClAsyncPageReader.hh:190
void shiftdgbuf(uint32_t &btsread)
Definition: XrdClAsyncPageReader.hh:267
static const int PageWithDigest
Definition: XrdClAsyncPageReader.hh:352
uint64_t rspoff
Definition: XrdClAsyncPageReader.hh:341
size_t iovindex
Definition: XrdClAsyncPageReader.hh:350
void addiov(char *&buf, size_t len)
Add I/O buffer to the vector.
Definition: XrdClAsyncPageReader.hh:167
size_t dgindex
Definition: XrdClAsyncPageReader.hh:345
virtual ~AsyncPageReader()
Destructor.
Definition: XrdClAsyncPageReader.hh:75
std::vector< iovec > iov
Definition: XrdClAsyncPageReader.hh:348
ChunkList & chunks
Definition: XrdClAsyncPageReader.hh:338
size_t dgoff
Definition: XrdClAsyncPageReader.hh:346
uint32_t CalcRdSize()
Calculate the size of the data to be read.
Definition: XrdClAsyncPageReader.hh:198
void shift(void *&buffer, size_t nbbts)
Shift buffer by a number of bytes.
Definition: XrdClAsyncPageReader.hh:255
void InitIOV()
Initialize the I/O vector.
Definition: XrdClAsyncPageReader.hh:213
void ShiftIOV(uint32_t btsread)
shift the I/O vector by the number of bytes read
Definition: XrdClAsyncPageReader.hh:313
AsyncPageReader(ChunkList &chunks, std::vector< uint32_t > &digests)
Definition: XrdClAsyncPageReader.hh:50
int iovcnt
Definition: XrdClAsyncPageReader.hh:349
void addiov(char *&buf, uint32_t len, uint32_t &dleft)
Add I/O buffer to the vector and update number of bytes left to be read.
Definition: XrdClAsyncPageReader.hh:179
XRootDStatus Read(Socket &socket, uint32_t &btsread)
Definition: XrdClAsyncPageReader.hh:107
uint32_t dlen
Definition: XrdClAsyncPageReader.hh:340
size_t chindex
Definition: XrdClAsyncPageReader.hh:343
void shiftpgbuf(uint32_t &btsread)
Definition: XrdClAsyncPageReader.hh:292
A network socket.
Definition: XrdClSocket.hh:43
XRootDStatus ReadV(iovec *iov, int iocnt, int &bytesRead)
Request status.
Definition: XrdClXRootDResponses.hh:219
static int csNum(off_t offs, int count)
Compute the required size of a checksum vector based on offset & length.
Definition: XrdClAnyObject.hh:26
const uint16_t suRetry
Definition: XrdClStatus.hh:40
std::vector< ChunkInfo > ChunkList
List of chunks.
Definition: XrdClXRootDResponses.hh:1046
static const int PageSize
Definition: XrdSysPageSize.hh:36
kXR_int32 dlen
Definition: XProtocol.hh:1231
kXR_int64 offset
Definition: XProtocol.hh:1051
struct ServerResponseBody_Status bdy
Definition: XProtocol.hh:1252
Definition: XProtocol.hh:1299
ServerResponseStatus status
Definition: XProtocol.hh:1300
union ServerResponseV2::@1 info
ServerResponseBody_pgRead pgread
Definition: XProtocol.hh:1303
Helper class for retrieving the maximum size of the I/O vector.
Definition: XrdClAsyncPageReader.hh:138
int32_t value
Definition: XrdClAsyncPageReader.hh:152
iovmax_t()
Definition: XrdClAsyncPageReader.hh:139
Describe a data chunk for vector read.
Definition: XrdClXRootDResponses.hh:908
void * buffer
length of the chunk
Definition: XrdClXRootDResponses.hh:941
uint64_t offset
Definition: XrdClXRootDResponses.hh:939
char * data
Definition: XrdOucIOVec.hh:45
Definition: XrdOucIOVec.hh:65