XRootD
Loading...
Searching...
No Matches
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"
26
27#include <sys/uio.h>
28#include <memory>
29#include <arpa/inet.h>
30
31namespace 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 //--------------------------------------------------------------------------
73 {
74 }
75
76 //--------------------------------------------------------------------------
78 //--------------------------------------------------------------------------
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_ */
union ServerResponseV2::@13 info
ServerResponseStatus status
struct ServerResponseBody_Status bdy
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.
XRootDStatus ReadV(iovec *iov, int iocnt, int &bytesRead)
static int csNum(off_t offs, int count)
Compute the required size of a checksum vector based on offset & length.
const uint16_t suRetry
std::vector< ChunkInfo > ChunkList
List of chunks.
static const int PageSize
int getIovMax()