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  //--------------------------------------------------------------------------
46  //--------------------------------------------------------------------------
48  std::vector<uint32_t> &digests ) :
49  chunks( chunks ),
50  digests( digests ),
51  dlen( 0 ),
52  rspoff( 0 ),
53  chindex( 0 ),
54  choff( 0 ),
55  dgindex( 0 ),
56  dgoff( 0 ),
57  iovcnt( 0 ),
58  iovindex( 0 )
59  {
60  uint64_t rdoff = chunks.front().offset;
61  uint32_t rdlen = 0;
62  for( auto &ch : chunks )
63  rdlen += ch.length;
64  int fpglen, lpglen;
65  int pgcnt = XrdOucPgrwUtils::csNum( rdoff, rdlen, fpglen, lpglen);
66  digests.resize( pgcnt );
67  }
68 
69  //--------------------------------------------------------------------------
71  //--------------------------------------------------------------------------
72  virtual ~AsyncPageReader()
73  {
74  }
75 
76  //--------------------------------------------------------------------------
78  //--------------------------------------------------------------------------
79  void SetRsp( ServerResponseV2 *rsp )
80  {
81  dlen = rsp->status.bdy.dlen;
82  rspoff = rsp->info.pgread.offset;
83 
84  uint64_t bufoff = rspoff - chunks[0].offset;
85  chindex = 0;
86 
87  for( chindex = 0; chindex < chunks.size(); ++chindex )
88  {
89  if( chunks[chindex].length < bufoff )
90  {
91  bufoff -= chunks[chindex].length;
92  continue;
93  }
94  break;
95  }
96  choff = bufoff;
97  }
98 
99  //--------------------------------------------------------------------------
104  //--------------------------------------------------------------------------
105  XRootDStatus Read( Socket &socket, uint32_t &btsread )
106  {
107  if( dlen == 0 || chindex >= chunks.size() )
108  return XRootDStatus();
109  btsread = 0;
110  int nbbts = 0;
111  do
112  {
113  // Prepare the IO vector for receiving the data
114  if( iov.empty() )
115  InitIOV();
116  // read the data into the buffer
117  nbbts = 0;
118  auto st = socket.ReadV( iov.data() + iovindex, iovcnt, nbbts );
119  if( !st.IsOK() )
120  return st;
121  btsread += nbbts;
122  dlen -= nbbts;
123  ShiftIOV( nbbts );
124  if( st.code == suRetry )
125  return st;
126  }
127  while( nbbts > 0 && dlen > 0 && chindex < chunks.size() );
128 
129  return XRootDStatus();
130  }
131 
132  private:
133 
134  //--------------------------------------------------------------------------
136  //--------------------------------------------------------------------------
137  inline static int max_iovcnt()
138  {
139  // make sure it is an even number
140  static const int iovmax = XrdSys::getIovMax() & ~1;
141  return iovmax;
142  }
143 
144  //--------------------------------------------------------------------------
146  //--------------------------------------------------------------------------
147  inline void addiov( char *&buf, size_t len )
148  {
149  iov.emplace_back();
150  iov.back().iov_base = buf;
151  iov.back().iov_len = len;
152  buf += len;
153  ++iovcnt;
154  }
155 
156  //--------------------------------------------------------------------------
158  //--------------------------------------------------------------------------
159  inline void addiov( char *&buf, uint32_t len, uint32_t &dleft )
160  {
161  if( len > dleft ) len = dleft;
162  addiov( buf, len );
163  dleft -= len;
164  }
165 
166  //--------------------------------------------------------------------------
169  //--------------------------------------------------------------------------
170  inline static uint32_t CalcIOVSize( uint32_t dleft )
171  {
172  uint32_t ret = ( dleft / PageWithDigest + 2 ) * 2;
173  return ( ret > uint32_t( max_iovcnt() ) ? max_iovcnt() : ret );
174  }
175 
176  //--------------------------------------------------------------------------
178  //--------------------------------------------------------------------------
179  uint32_t CalcRdSize()
180  {
181  // data size in the server response (including digests)
182  uint32_t dleft = dlen;
183  // space in our page buffer
184  uint32_t pgspace = chunks[chindex].length - choff;
185  // space in our digest buffer
186  uint32_t dgspace = sizeof( uint32_t ) * (digests.size() - dgindex ) - dgoff;
187  if( dleft > pgspace + dgspace )
188  dleft = pgspace + dgspace;
189  return dleft;
190  }
191 
192  //--------------------------------------------------------------------------
194  //--------------------------------------------------------------------------
195  void InitIOV()
196  {
197  iovindex = 0;
198  // figure out the number of data we can read in one go
199  uint32_t dleft = CalcRdSize();
200  // and reset the I/O vector
201  iov.clear();
202  iovcnt = 0;
203  iov.reserve( CalcIOVSize( dleft ) );
204  // now prepare the page and digest buffers
205  ChunkInfo ch = chunks[chindex];
206  char* pgbuf = static_cast<char*>( ch.buffer ) + choff;
207  uint64_t rdoff = ch.offset + choff;
208  char* dgbuf = reinterpret_cast<char*>( digests.data() + dgindex ) + dgoff;
209  // handle the first digest
210  uint32_t fdglen = sizeof( uint32_t ) - dgoff;
211  addiov( dgbuf, fdglen, dleft );
212  if( dleft == 0 || iovcnt >= max_iovcnt() )
213  return;
214  // handle the first page
215  uint32_t fpglen = XrdSys::PageSize - rdoff % XrdSys::PageSize;
216  addiov( pgbuf, fpglen, dleft );
217  if( dleft == 0 || iovcnt >= max_iovcnt() )
218  return;
219  // handle all the subsequent aligned pages
220  size_t fullpgs = dleft / PageWithDigest;
221  for( size_t i = 0; i < fullpgs; ++i )
222  {
223  addiov( dgbuf, sizeof( uint32_t ), dleft );
224  if( dleft == 0 || iovcnt >= max_iovcnt() )
225  return;
226  addiov( pgbuf, XrdSys::PageSize, dleft );
227  if( dleft == 0 || iovcnt >= max_iovcnt() )
228  return;
229  }
230  // handle the last digest
231  uint32_t ldglen = sizeof( uint32_t );
232  addiov( dgbuf, ldglen, dleft );
233  if( dleft == 0 || iovcnt >= max_iovcnt() )
234  return;
235  // handle the last page
236  addiov( pgbuf, dleft, dleft );
237  }
238 
239  //--------------------------------------------------------------------------
241  //--------------------------------------------------------------------------
242  inline void shift( void *&buffer, size_t nbbts )
243  {
244  char *buf = static_cast<char*>( buffer );
245  buf += nbbts;
246  buffer = buf;
247  }
248 
249  //--------------------------------------------------------------------------
253  //--------------------------------------------------------------------------
254  inline void shiftdgbuf( uint32_t &btsread )
255  {
256  if( iov[iovindex].iov_len > btsread )
257  {
258  iov[iovindex].iov_len -= btsread;
259  shift( iov[iovindex].iov_base, btsread );
260  dgoff += btsread;
261  btsread = 0;
262  return;
263  }
264 
265  btsread -= iov[iovindex].iov_len;
266  iov[iovindex].iov_len = 0;
267  dgoff = 0;
268  digests[dgindex] = ntohl( digests[dgindex] );
269  ++dgindex;
270  ++iovindex;
271  --iovcnt;
272  }
273 
274  //--------------------------------------------------------------------------
278  //--------------------------------------------------------------------------
279  inline void shiftpgbuf( uint32_t &btsread )
280  {
281  if( iov[iovindex].iov_len > btsread )
282  {
283  iov[iovindex].iov_len -= btsread;
284  shift( iov[iovindex].iov_base, btsread );
285  choff += btsread;
286  btsread = 0;
287  return;
288  }
289 
290  btsread -= iov[iovindex].iov_len;
291  choff += iov[iovindex].iov_len;
292  iov[iovindex].iov_len = 0;
293  ++iovindex;
294  --iovcnt;
295  }
296 
297  //--------------------------------------------------------------------------
299  //--------------------------------------------------------------------------
300  void ShiftIOV( uint32_t btsread )
301  {
302  // if iovindex is even it point to digest, otherwise it points to a page
303  if( iovindex % 2 == 0 )
304  shiftdgbuf( btsread );
305  // adjust as many I/O buffers as necessary
306  while( btsread > 0 )
307  {
308  // handle page
309  shiftpgbuf( btsread );
310  if( btsread == 0 ) break;
311  // handle digest
312  shiftdgbuf( btsread );
313  }
314  // if we filled the buffer, move to the next one
315  if( iovcnt == 0 )
316  iov.clear();
317  // do we need to move to the next chunk?
318  if( choff >= chunks[chindex].length )
319  {
320  ++chindex;
321  choff = 0;
322  }
323  }
324 
325  ChunkList &chunks; //< list of data chunks to be filled with user data
326  std::vector<uint32_t> &digests; //< list of crc32c digests for every 4KB page of data
327  uint32_t dlen; //< size of the data in the message
328  uint64_t rspoff; //< response offset
329 
330  size_t chindex; //< index of the current data buffer
331  size_t choff; //< offset within the current buffer
332  size_t dgindex; //< index of the current digest buffer
333  size_t dgoff; //< offset within the current digest buffer
334 
335  std::vector<iovec> iov; //< I/O vector
336  int iovcnt; //< size of the I/O vector
337  size_t iovindex; //< index of the first valid element in the I/O vector
338 
339  static const int PageWithDigest = XrdSys::PageSize + sizeof( uint32_t );
340 };
341 
342 } /* namespace XrdEc */
343 
344 #endif /* SRC_XRDCL_XRDCLASYNCPAGEREADER_HH_ */
ServerResponseStatus status
Definition: XProtocol.hh:1309
struct ServerResponseBody_Status bdy
Definition: XProtocol.hh:1261
union ServerResponseV2::@1 info
Object for reading out data from the PgRead response.
void SetRsp(ServerResponseV2 *rsp)
Sets message data size.
virtual ~AsyncPageReader()
Destructor.
AsyncPageReader(ChunkList &chunks, std::vector< uint32_t > &digests)
XRootDStatus Read(Socket &socket, uint32_t &btsread)
A network socket.
Definition: XrdClSocket.hh:43
XRootDStatus ReadV(iovec *iov, int iocnt, int &bytesRead)
Definition: XrdClSocket.cc:761
static int csNum(off_t offs, int count)
Compute the required size of a checksum vector based on offset & length.
const uint16_t suRetry
Definition: XrdClStatus.hh:40
std::vector< ChunkInfo > ChunkList
List of chunks.
static const int PageSize
int getIovMax()
char * data
Definition: XrdOucIOVec.hh:45