XRootD
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. More...
 
XRootDStatus Read (Socket &socket, uint32_t &btsret)
 
- Public Member Functions inherited from XrdCl::AsyncRawReaderIntfc
 AsyncRawReaderIntfc (const URL &url, const Message &request)
 
virtual ~AsyncRawReaderIntfc ()
 Destructor. More...
 
void SetChunkList (ChunkList *chunks)
 Sets the chunk list with user buffers. More...
 
void SetDataLength (int dlen)
 Sets response data length. More...
 

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< ChunkStatuschstatus
 
ChunkListchunks
 
bool dataerr
 
buffer_t discardbuff
 
uint32_t dlen
 
uint32_t msgbtsrd
 
uint32_t rawbtsrd
 
Stage readstage
 
const Messagerequest
 
const URLurl
 

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  :
42  rdlstoff( 0 ),
43  rdlstlen( 0 )
44  {
45  memset( &rdlst, 0, sizeof( readahead_list ) );
46  }
AsyncRawReaderIntfc(const URL &url, const Message &request)

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  }
std::vector< ChunkStatus > chstatus
const uint16_t stFatal
Fatal error, it's still an error.
Definition: XrdClStatus.hh:33
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 );
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 );
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 );
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;
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 );
201  continue;
202  }
203 
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  }
kXR_int32 rlen
Definition: XProtocol.hh:660
kXR_int64 offset
Definition: XProtocol.hh:661
XRootDStatus ReadBytesAsync(Socket &socket, char *buffer, uint32_t toBeRead, uint32_t &bytesRead)
static Log * GetLog()
Get default log.
std::string GetHostId() const
Get the host part of the URL (user:password@host:port)
Definition: XrdClURL.hh:99
const uint16_t suRetry
Definition: XrdClStatus.hh:40
const uint16_t stError
An error occurred that could potentially be retried.
Definition: XrdClStatus.hh:32
const uint64_t XRootDMsg
const uint16_t errInternal
Internal error.
Definition: XrdClStatus.hh:56
const uint16_t errCorruptedHeader
Definition: XrdClStatus.hh:103
XrdSysError Log
Definition: XrdConfig.cc:112

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::URL::GetHostId(), XrdCl::DefaultEnv::GetLog(), XrdCl::Status::IsOK(), XrdCl::AsyncRawReaderIntfc::msgbtsrd, readahead_list::offset, XrdCl::AsyncRawReaderIntfc::rawbtsrd, XrdCl::AsyncRawReaderIntfc::ReadBytesAsync(), XrdCl::AsyncRawReaderIntfc::ReadDiscard, XrdCl::AsyncRawReaderIntfc::ReadDone, XrdCl::AsyncRawReaderIntfc::ReadRaw, XrdCl::AsyncRawReaderIntfc::ReadRdLst, XrdCl::AsyncRawReaderIntfc::readstage, XrdCl::AsyncRawReaderIntfc::ReadStart, readahead_list::rlen, 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: