XRootD
Loading...
Searching...
No Matches
XrdCl::AsyncVectorReader Class Reference

Object for reading out data from the VectorRead response. More...

#include <XrdClAsyncVectorReader.hh>

Inheritance diagram for XrdCl::AsyncVectorReader:
Collaboration diagram for XrdCl::AsyncVectorReader:

Public Member Functions

 AsyncVectorReader (const URL &url, const Message &request)
XRootDStatus GetResponse (AnyObject *&response)
 Get the response.
XRootDStatus Read (Socket &socket, uint32_t &btsret)
Public Member Functions inherited from XrdCl::AsyncRawReaderIntfc
 AsyncRawReaderIntfc (const URL &url, const Message &request)
virtual ~AsyncRawReaderIntfc ()
 Destructor.
void SetChunkList (ChunkList *chunks)
 Sets the chunk list with user buffers.
void SetDataLength (int dlen)
 Sets response data length.

Additional Inherited Members

Protected Types inherited from XrdCl::AsyncRawReaderIntfc
using buffer_t = std::vector<char>
enum  Stage {
  ReadStart ,
  ReadRdLst ,
  ReadRaw ,
  ReadDiscard ,
  ReadDone
}
 Stages of reading out a response from the socket. More...
Protected Member Functions inherited from XrdCl::AsyncRawReaderIntfc
XRootDStatus ReadBytesAsync (Socket &socket, char *buffer, uint32_t toBeRead, uint32_t &bytesRead)
Protected Attributes inherited from XrdCl::AsyncRawReaderIntfc
size_t chidx
size_t chlen
size_t choff
std::vector< ChunkStatus > chstatus
ChunkList * chunks
bool dataerr
buffer_t discardbuff
uint32_t dlen
uint32_t msgbtsrd
uint32_t rawbtsrd
Stage readstage
const Message & request
const URL & url

Detailed Description

Object for reading out data from the VectorRead response.

Definition at line 32 of file XrdClAsyncVectorReader.hh.

Constructor & Destructor Documentation

◆ AsyncVectorReader()

XrdCl::AsyncVectorReader::AsyncVectorReader ( const URL & url,
const Message & request )
inline

Constructor

Parameters
url: channel URL

Definition at line 40 of file XrdClAsyncVectorReader.hh.

40 :
41 AsyncRawReaderIntfc( url, request ),
42 rdlstoff( 0 ),
43 rdlstlen( 0 )
44 {
45 memset( &rdlst, 0, sizeof( readahead_list ) );
46 }
XrdCl::AsyncRawReaderIntfc::AsyncRawReaderIntfc
AsyncRawReaderIntfc(const URL &url, const Message &request)
Definition XrdClAsyncRawReaderIntfc.hh:38

References XrdCl::AsyncRawReaderIntfc::AsyncRawReaderIntfc(), XrdCl::AsyncRawReaderIntfc::request, and XrdCl::AsyncRawReaderIntfc::url.

Here is the call graph for this function:

Member Function Documentation

◆ GetResponse()

XRootDStatus XrdCl::AsyncVectorReader::GetResponse ( AnyObject *& response)
inlinevirtual

Get the response.

Implements XrdCl::AsyncRawReaderIntfc.

Definition at line 249 of file XrdClAsyncVectorReader.hh.

250 {
251 //--------------------------------------------------------------------------
252 // See if all the chunks are OK and put them in the response
253 //--------------------------------------------------------------------------
254 std::unique_ptr<VectorReadInfo> ptr( new VectorReadInfo() );
255 for( uint32_t i = 0; i < chunks->size(); ++i )
256 {
257 if( !chstatus[i].done )
258 return Status( stFatal, errInvalidResponse );
259 ptr->GetChunks().emplace_back( ( *chunks )[i].offset,
260 ( *chunks )[i].length, ( *chunks )[i].buffer );
261 }
262 ptr->SetSize( rawbtsrd );
263 response = new AnyObject();
264 response->Set( ptr.release() );
265 return XRootDStatus();
266 }
XrdCl::AsyncRawReaderIntfc::chstatus
std::vector< ChunkStatus > chstatus
Definition XrdClAsyncRawReaderIntfc.hh:158
XrdCl::stFatal
const uint16_t stFatal
Fatal error, it's still an error.
Definition XrdClStatus.hh:33
XrdCl::errInvalidResponse
const uint16_t errInvalidResponse
Definition XrdClStatus.hh:99

References XrdCl::AsyncRawReaderIntfc::chstatus, XrdCl::AsyncRawReaderIntfc::chunks, XrdCl::errInvalidResponse, XrdCl::AsyncRawReaderIntfc::rawbtsrd, XrdCl::AnyObject::Set(), and XrdCl::stFatal.

Here is the call graph for this function:

◆ Read()

XRootDStatus XrdCl::AsyncVectorReader::Read ( Socket & socket,
uint32_t & btsret )
inlinevirtual

Readout raw data from socket

Parameters
socket: the socket
btsret: number of bytes read
Returns
: operation status

Implements XrdCl::AsyncRawReaderIntfc.

Definition at line 55 of file XrdClAsyncVectorReader.hh.

56 {
57 Log *log = DefaultEnv::GetLog();
58
59 while( true )
60 {
61 switch( readstage )
62 {
63 //------------------------------------------------------------------
64 // Prepare to readout a new response
65 //------------------------------------------------------------------
66 case ReadStart:
67 {
68 msgbtsrd = 0;
69 rdlstoff = 0;
70 rdlstlen = sizeof( readahead_list );
71 readstage = ReadRdLst;
72 continue;
73 }
74
75 //------------------------------------------------------------------
76 // Readout the read_list
77 //------------------------------------------------------------------
78 case ReadRdLst:
79 {
80 //----------------------------------------------------------------
81 // We cannot afford to read the next header from the stream
82 // because we will cross the message boundary
83 //----------------------------------------------------------------
84 if( msgbtsrd + rdlstlen > dlen )
85 {
86 uint32_t btsleft = dlen - msgbtsrd;
87 log->Error( XRootDMsg, "[%s] VectorReader: No enough data to read "
88 "another chunk header. Discarding %d bytes.",
89 url.GetHostId().c_str(), btsleft );
90 readstage = ReadDiscard;
91 continue;
92 }
93
94 //----------------------------------------------------------------
95 // Let's readout the read list record from the socket
96 //----------------------------------------------------------------
97 uint32_t btsrd = 0;
98 char *buff = reinterpret_cast<char*>( &rdlst );
99 Status st = ReadBytesAsync( socket, buff + rdlstoff, rdlstlen, btsrd );
100 rdlstoff += btsrd;
101 rdlstlen -= btsrd;
102 msgbtsrd += btsrd;
103 btsret += btsrd;
104
105 if( !st.IsOK() || st.code == suRetry )
106 return st;
107
108 //----------------------------------------------------------------
109 // We have a complete read list record, now we need to marshal it
110 //----------------------------------------------------------------
111 rdlst.rlen = ntohl( rdlst.rlen );
112 rdlst.offset = ntohll( rdlst.offset );
113 choff = 0;
114 chlen = rdlst.rlen;
115
116 //----------------------------------------------------------------
117 // Find the buffer corresponding to the chunk
118 //----------------------------------------------------------------
119 bool chfound = false;
120 for( size_t i = 0; i < chunks->size(); ++i )
121 {
122 if( ( *chunks )[i].offset == uint64_t( rdlst.offset ) &&
123 ( *chunks )[i].length == uint32_t( rdlst.rlen ) )
124 {
125 chfound = true;
126 chidx = i;
127 break;
128 }
129 }
130
131 //----------------------------------------------------------------
132 // If the chunk was not found this is a bogus response, switch
133 // to discard mode
134 //----------------------------------------------------------------
135 if( !chfound )
136 {
137 log->Error( XRootDMsg, "[%s] VectorReader: Impossible to find chunk "
138 "buffer corresponding to %d bytes at %lld",
139 url.GetHostId().c_str(), rdlst.rlen, rdlst.offset );
140 readstage = ReadDiscard;
141 continue;
142 }
143
144 readstage = ReadRaw;
145 continue;
146 }
147
148 //------------------------------------------------------------------
149 // Readout the raw data
150 //------------------------------------------------------------------
151 case ReadRaw:
152 {
153 //----------------------------------------------------------------
154 // The chunk was found, but reading all the data will cross the
155 // message boundary
156 //----------------------------------------------------------------
157 if( msgbtsrd + chlen > dlen )
158 {
159 uint32_t btsleft = dlen - msgbtsrd;
160 log->Error( XRootDMsg, "[%s] VectorReader: Malformed chunk header: "
161 "reading %d bytes from message would cross the message "
162 "boundary, discarding %d bytes.", url.GetHostId().c_str(),
163 rdlst.rlen, btsleft );
164 chstatus[chidx].sizeerr = true;
165 readstage = ReadDiscard;
166 continue;
167 }
168
169 //----------------------------------------------------------------
170 // Readout the raw data from the socket
171 //----------------------------------------------------------------
172 uint32_t btsrd = 0;
173 char *buff = static_cast<char*>( ( *chunks )[chidx].buffer );
174 Status st = ReadBytesAsync( socket, buff + choff, chlen, btsrd );
175 choff += btsrd;
176 chlen -= btsrd;
177 msgbtsrd += btsrd;
178 rawbtsrd += btsrd;
179 btsret += btsrd;
180
181 if( !st.IsOK() || st.code == suRetry )
182 return st;
183
184 log->Dump( XRootDMsg, "[%s] VectorReader: read buffer for chunk %d@%lld",
185 url.GetHostId().c_str(), rdlst.rlen, rdlst.offset );
186
187 //----------------------------------------------------------------
188 // Mark chunk as done
189 //----------------------------------------------------------------
190 chstatus[chidx].done = true;
191
192 //----------------------------------------------------------------
193 // There is still data to be read, we need to readout the next
194 // read list record.
195 //----------------------------------------------------------------
196 if( msgbtsrd < dlen )
197 {
198 rdlstoff = 0;
199 rdlstlen = sizeof( readahead_list );
200 readstage = ReadRdLst;
201 continue;
202 }
203
204 readstage = ReadDone;
205 continue;
206 }
207
208 //------------------------------------------------------------------
209 // We've had an error and we are in the discarding mode
210 //------------------------------------------------------------------
211 case ReadDiscard:
212 {
213 // Just drop the connection, we don't know if the stream is sane
214 // anymore. Recover with a reconnect.
215 return XRootDStatus( stError, errCorruptedHeader );
216 }
217
218 //------------------------------------------------------------------
219 // Finalize the read
220 //------------------------------------------------------------------
221 case ReadDone:
222 {
223 chidx = 0;
224 choff = 0;
225 chlen = 0;
226 rdlstoff = 0;
227 rdlstlen = 0;
228 break;
229 }
230
231 //------------------------------------------------------------------
232 // Others should not happen
233 //------------------------------------------------------------------
234 default : return XRootDStatus( stError, errInternal );
235 }
236
237 // just in case
238 break;
239 }
240 //----------------------------------------------------------------------
241 // We are done
242 //----------------------------------------------------------------------
243 return XRootDStatus();
244 }
XrdCl::AsyncRawReaderIntfc::ReadBytesAsync
XRootDStatus ReadBytesAsync(Socket &socket, char *buffer, uint32_t toBeRead, uint32_t &bytesRead)
Definition XrdClAsyncRawReaderIntfc.hh:98
XrdCl::DefaultEnv::GetLog
static Log * GetLog()
Get default log.
Definition XrdClDefaultEnv.cc:493
XrdCl::suRetry
const uint16_t suRetry
Definition XrdClStatus.hh:40
XrdCl::stError
const uint16_t stError
An error occurred that could potentially be retried.
Definition XrdClStatus.hh:32
XrdCl::XRootDMsg
const uint64_t XRootDMsg
Definition XrdClConstants.hh:39
XrdCl::errInternal
const uint16_t errInternal
Internal error.
Definition XrdClStatus.hh:56
XrdCl::errCorruptedHeader
const uint16_t errCorruptedHeader
Definition XrdClStatus.hh:103
XrdGlobal::Log
XrdSysError Log
Definition XrdConfig.cc:113

References XrdCl::AsyncRawReaderIntfc::chidx, XrdCl::AsyncRawReaderIntfc::chlen, XrdCl::AsyncRawReaderIntfc::choff, XrdCl::AsyncRawReaderIntfc::chstatus, XrdCl::AsyncRawReaderIntfc::chunks, XrdCl::Status::code, XrdCl::AsyncRawReaderIntfc::dlen, XrdCl::Log::Dump(), XrdCl::errCorruptedHeader, XrdCl::errInternal, XrdCl::Log::Error(), XrdCl::DefaultEnv::GetLog(), XrdCl::Status::IsOK(), XrdCl::AsyncRawReaderIntfc::msgbtsrd, XrdCl::AsyncRawReaderIntfc::rawbtsrd, XrdCl::AsyncRawReaderIntfc::ReadBytesAsync(), XrdCl::AsyncRawReaderIntfc::ReadDiscard, XrdCl::AsyncRawReaderIntfc::ReadDone, XrdCl::AsyncRawReaderIntfc::ReadRaw, XrdCl::AsyncRawReaderIntfc::ReadRdLst, XrdCl::AsyncRawReaderIntfc::readstage, XrdCl::AsyncRawReaderIntfc::ReadStart, XrdCl::stError, XrdCl::suRetry, XrdCl::AsyncRawReaderIntfc::url, and XrdCl::XRootDMsg.

Here is the call graph for this function:
The documentation for this class was generated from the following file: