Loading...
Searching...
No Matches
XrdClAsyncRawReader.hh
Go to the documentation of this file.
//------------------------------------------------------------------------------
// Copyright (c) 2011-2012 by European Organization for Nuclear Research (CERN)
// Author: Michal Simon <michal.simon@cern.ch>
//------------------------------------------------------------------------------
// XRootD is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// XRootD is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with XRootD. If not, see <http://www.gnu.org/licenses/>.
//------------------------------------------------------------------------------
#ifndef SRC_XRDCL_XRDCLASYNCRAWREADER_HH_
#define SRC_XRDCL_XRDCLASYNCRAWREADER_HH_
#include "XrdCl/XrdClXRootDResponses.hh"
#include "XrdCl/XrdClSocket.hh"
#include "XrdCl/XrdClStream.hh"
#include "XrdClAsyncRawReaderIntfc.hh"
namespace XrdCl
{
//----------------------------------------------------------------------------
//----------------------------------------------------------------------------
class AsyncRawReader : public AsyncRawReaderIntfc
{
public:
//------------------------------------------------------------------------
//------------------------------------------------------------------------
AsyncRawReader( const URL &url, const Message &request ) :
AsyncRawReaderIntfc( url, request )
{
}
//------------------------------------------------------------------------
//------------------------------------------------------------------------
XRootDStatus Read( Socket &socket, uint32_t &btsret )
{
while( true )
{
switch( readstage )
{
//------------------------------------------------------------------
// Prepare to readout a new response
//------------------------------------------------------------------
case ReadStart:
{
msgbtsrd = 0;
chlen = ( *chunks )[0].length;
readstage = ReadRaw;
continue;
}
//------------------------------------------------------------------
// Readout the raw data
//------------------------------------------------------------------
case ReadRaw:
{
//----------------------------------------------------------------
// Make sure we are not reading past the end of the read response
//----------------------------------------------------------------
if( msgbtsrd + chlen > dlen )
chlen = dlen - msgbtsrd;
//----------------------------------------------------------------
// Readout the raw data from the socket
//----------------------------------------------------------------
uint32_t btsrd = 0;
char *buff = static_cast<char*>( ( *chunks )[chidx].buffer );
Status st = ReadBytesAsync( socket, buff + choff, chlen, btsrd );
choff += btsrd;
chlen -= btsrd;
msgbtsrd += btsrd;
rawbtsrd += btsrd;
btsret += btsrd;
if( !st.IsOK() || st.code == suRetry )
return st;
//----------------------------------------------------------------
// If the chunk is full, move to the next buffer
//----------------------------------------------------------------
if( choff == ( *chunks )[chidx].length )
{
++chidx;
choff = 0;
chlen = ( chidx < chunks->size() ? ( *chunks )[chidx].length : 0 );
}
//----------------------------------------------------------------
// Check if there are some data left in the response to be readout
// from the socket.
//----------------------------------------------------------------
if( msgbtsrd < dlen )
{
//--------------------------------------------------------------
// We run out of space, the server has send too much data
//--------------------------------------------------------------
if( chidx >= chunks->size() )
{
readstage = ReadDiscard;
continue;
}
readstage = ReadRaw;
continue;
}
//----------------------------------------------------------------
// We are done
//----------------------------------------------------------------
readstage = ReadDone;
continue;
}
//------------------------------------------------------------------
// We've had an error and we are in the discarding mode
//------------------------------------------------------------------
case ReadDiscard:
{
DefaultEnv::GetLog()->Error( XRootDMsg, "[%s] RawReader: Handling "
"response to %s: user supplied buffer is "
"too small for the received data.",
url.GetHostId().c_str(),
request.GetObfuscatedDescription().c_str() );
// Just drop the connection, we don't know if the stream is sane
// anymore. Recover with a reconnect.
return XRootDStatus( stError, errCorruptedHeader );
}
//------------------------------------------------------------------
// Finalize the read
//------------------------------------------------------------------
case ReadDone:
{
break;
}
//------------------------------------------------------------------
// Others should not happen
//------------------------------------------------------------------
default : return XRootDStatus( stError, errInternal );
}
// just in case
break;
}
//----------------------------------------------------------------------
// We are done
//----------------------------------------------------------------------
return XRootDStatus();
}
//------------------------------------------------------------------------
//------------------------------------------------------------------------
XRootDStatus GetResponse( AnyObject *&response )
{
if( dataerr )
return XRootDStatus( stError, errInvalidResponse );
std::unique_ptr<AnyObject> rsp( new AnyObject() );
if( request.GetVirtReqID() == kXR_virtReadv )
rsp->Set( GetVectorReadInfo() );
else
rsp->Set( GetChunkInfo() );
response = rsp.release();
return XRootDStatus();
}
private:
inline ChunkInfo* GetChunkInfo()
{
ChunkInfo *info = new ChunkInfo( chunks->front() );
info->length = rawbtsrd;
return info;
}
inline VectorReadInfo* GetVectorReadInfo()
{
VectorReadInfo *info = new VectorReadInfo();
info->SetSize( rawbtsrd );
int btsleft = rawbtsrd;
for( auto &chunk : *chunks )
{
int length = uint32_t( btsleft ) >= chunk.length ? chunk.length : btsleft;
info->GetChunks().emplace_back( chunk.offset, length, chunk.buffer );
btsleft -= length;
}
return info;
}
};
} /* namespace XrdCl */
#endif /* SRC_XRDCL_XRDCLASYNCVECTORREADER_HH_ */
kXR_virtReadv
@ kXR_virtReadv
Definition XProtocol.hh:150
XrdClAsyncRawReaderIntfc.hh
XrdClSocket.hh
XrdClStream.hh
XrdClXRootDResponses.hh
XrdCl::AnyObject
Definition XrdClAnyObject.hh:33
XrdCl::AsyncRawReaderIntfc::chidx
size_t chidx
Definition XrdClAsyncRawReaderIntfc.hh:163
XrdCl::AsyncRawReaderIntfc::dataerr
bool dataerr
Definition XrdClAsyncRawReaderIntfc.hh:168
XrdCl::AsyncRawReaderIntfc::chlen
size_t chlen
Definition XrdClAsyncRawReaderIntfc.hh:165
XrdCl::AsyncRawReaderIntfc::readstage
Stage readstage
Definition XrdClAsyncRawReaderIntfc.hh:149
XrdCl::AsyncRawReaderIntfc::ReadBytesAsync
XRootDStatus ReadBytesAsync(Socket &socket, char *buffer, uint32_t toBeRead, uint32_t &bytesRead)
Definition XrdClAsyncRawReaderIntfc.hh:98
XrdCl::AsyncRawReaderIntfc::request
const Message & request
Definition XrdClAsyncRawReaderIntfc.hh:155
XrdCl::AsyncRawReaderIntfc::choff
size_t choff
Definition XrdClAsyncRawReaderIntfc.hh:164
XrdCl::AsyncRawReaderIntfc::url
const URL & url
Definition XrdClAsyncRawReaderIntfc.hh:154
XrdCl::AsyncRawReaderIntfc::AsyncRawReaderIntfc
AsyncRawReaderIntfc(const URL &url, const Message &request)
Definition XrdClAsyncRawReaderIntfc.hh:38
XrdCl::AsyncRawReaderIntfc::dlen
uint32_t dlen
Definition XrdClAsyncRawReaderIntfc.hh:159
XrdCl::AsyncRawReaderIntfc::rawbtsrd
uint32_t rawbtsrd
Definition XrdClAsyncRawReaderIntfc.hh:161
XrdCl::AsyncRawReaderIntfc::msgbtsrd
uint32_t msgbtsrd
Definition XrdClAsyncRawReaderIntfc.hh:160
XrdCl::AsyncRawReaderIntfc::chunks
ChunkList * chunks
Definition XrdClAsyncRawReaderIntfc.hh:157
XrdCl::AsyncRawReaderIntfc::ReadDiscard
@ ReadDiscard
Definition XrdClAsyncRawReaderIntfc.hh:142
XrdCl::AsyncRawReaderIntfc::ReadStart
@ ReadStart
Definition XrdClAsyncRawReaderIntfc.hh:139
XrdCl::AsyncRawReaderIntfc::ReadDone
@ ReadDone
Definition XrdClAsyncRawReaderIntfc.hh:143
XrdCl::AsyncRawReaderIntfc::ReadRaw
@ ReadRaw
Definition XrdClAsyncRawReaderIntfc.hh:141
XrdCl::AsyncRawReader::Read
XRootDStatus Read(Socket &socket, uint32_t &btsret)
Definition XrdClAsyncRawReader.hh:55
XrdCl::AsyncRawReader::GetResponse
XRootDStatus GetResponse(AnyObject *&response)
Get the response.
Definition XrdClAsyncRawReader.hh:172
XrdCl::AsyncRawReader::AsyncRawReader
AsyncRawReader(const URL &url, const Message &request)
Definition XrdClAsyncRawReader.hh:43
XrdCl::DefaultEnv::GetLog
static Log * GetLog()
Get default log.
Definition XrdClDefaultEnv.cc:493
XrdCl::Log::Error
void Error(uint64_t topic, const char *format,...)
Report an error.
Definition XrdClLog.cc:231
XrdCl::Message
The message representation used throughout the system.
Definition XrdClMessage.hh:32
XrdCl::Socket
A network socket.
Definition XrdClSocket.hh:43
XrdCl::URL
URL representation.
Definition XrdClURL.hh:31
XrdCl::XRootDStatus
Request status.
Definition XrdClXRootDResponses.hh:219
XrdCl
Definition XrdClAction.hh:34
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::errInvalidResponse
const uint16_t errInvalidResponse
Definition XrdClStatus.hh:99
XrdCl::errCorruptedHeader
const uint16_t errCorruptedHeader
Definition XrdClStatus.hh:103
XrdCl::ChunkInfo
Describe a data chunk for vector read.
Definition XrdClXRootDResponses.hh:918
XrdCl::ChunkInfo::length
uint32_t length
offset in the file
Definition XrdClXRootDResponses.hh:950
XrdCl::Status
Procedure execution status.
Definition XrdClStatus.hh:115
XrdCl::Status::code
uint16_t code
Error type, or additional hints on what to do.
Definition XrdClStatus.hh:147
XrdCl::Status::IsOK
bool IsOK() const
We're fine.
Definition XrdClStatus.hh:124