Loading...
Searching...
No Matches
XrdCl::XRootDMsgHandler Class Reference

Handle/Process/Forward XRootD messages. More...

`#include <XrdClXRootDMsgHandler.hh>`

Inheritance diagram for XrdCl::XRootDMsgHandler:
Collaboration diagram for XrdCl::XRootDMsgHandler:

## Public Member Functions

| XRootDMsgHandler (Message *msg, ResponseHandler *respHandler, const URL *url, std::shared_ptr< SIDManager > sidMgr, LocalFileHandler *lFileHandler)
| ~XRootDMsgHandler ()
| Destructor.
virtual uint16_t | Examine (std::shared_ptr< Message > &msg) override
time_t | GetExpiration () override
| Get a timestamp after which we give up.
const Message * | GetRequest () const
| Get the request pointer.
virtual uint16_t | GetSid () const override
virtual uint16_t | InspectStatusRsp () override
virtual bool | IsRaw () const override
| Are we a raw writer or not?
void | OnReadyToSend (Message *msg) override
virtual void | OnStatusReady (const Message *message, XRootDStatus status) override
| The requested action has been performed and the status is available.
virtual uint8_t | OnStreamEvent (StreamEvent event, XRootDStatus status) override
void | PartialReceived ()
virtual void | Process () override
| Process the message if it was "taken" by the examine action.
virtual XRootDStatus | ReadMessageBody (Message *msg, Socket *socket, uint32_t &bytesRead) override
void | SetChunkList (ChunkList *chunkList)
| Set the chunk list.
void | SetCrc32cDigests (std::vector< uint32_t > &&crc32cDigests)
void | SetExpiration (time_t expiration)
| Set a timestamp after which we give up.
void | SetFollowMetalink (bool followMetalink)
void | SetHostList (HostList *hostList)
| Set host list.
void | SetKernelBuffer (XrdSys::KernelBuffer *kbuff)
| Set the kernel buffer.
void | SetLoadBalancer (const HostInfo &loadBalancer)
| Set the load balancer.
void | SetOksofarAsAnswer (bool oksofarAsAnswer)
void | SetRedirectAsAnswer (bool redirectAsAnswer)
void | SetRedirectCounter (uint16_t redirectCounter)
| Set the redirect counter.
void | SetStateful (bool stateful)
void | WaitDone (time_t now)
XRootDStatus | WriteMessageBody (Socket *socket, uint32_t &bytesWritten) override
Public Member Functions inherited from XrdCl::MsgHandler
virtual | ~MsgHandler ()
| Event types that the message handler may receive.

## Friends

class | HandleRspJob

## Additional Inherited Members

Public Types inherited from XrdCl::MsgHandler
enum | Action { None = 0x0000 , Nop = 0x0001 , Ignore = 0x0002 , RemoveHandler = 0x0004 , Raw = 0x0008 , NoProcess = 0x0010 , Corrupted = 0x0020 , More = 0x0040 }
| Actions to be taken after a message is processed by the handler. More...
enum | StreamEvent { Ready = 1 , Broken = 2 , Timeout = 3 , FatalError = 4 }
| Events that may have occurred to the stream. More...

## Detailed Description

Handle/Process/Forward XRootD messages.

Definition at line 119 of file XrdClXRootDMsgHandler.hh.

## Constructor & Destructor Documentation

## XRootDMsgHandler()

XrdCl::XRootDMsgHandler::XRootDMsgHandler | ( | Message * | msg,
|  | ResponseHandler * | respHandler,
|  | const URL * | url,
|  | std::shared_ptr< SIDManager > | sidMgr,
|  | LocalFileHandler * | lFileHandler )

inline

Constructor

Parameters

  msg | message that has been sent out
  respHandler | response handler to be called then the final final response arrives
  url | the url the message has been sent to
  sidMgr | the sid manager used to allocate SID for the initial message

Definition at line 134 of file XrdClXRootDMsgHandler.hh.

:
pRequest( msg ),
pResponseHandler( respHandler ),
pUrl( *url ),
pEffectiveDataServerUrl( 0 ),
pSidMgr( sidMgr ),
pLFileHandler( lFileHandler ),
pExpiration( 0 ),
pRedirectAsAnswer( false ),
pOksofarAsAnswer( false ),
pHasLoadBalancer( false ),
pHasSessionId( false ),
pChunkList( 0 ),
pKBuff( 0 ),
pRedirectCounter( 0 ),
pNotAuthorizedCounter( 0 ),
pAsyncOffset( 0 ),
pAsyncChunkIndex( 0 ),
pPgWrtCksumBuff( 4 ),
pPgWrtCurrentPageOffset( 0 ),
pPgWrtCurrentPageNb( 0 ),
pOtherRawStarted( false ),
pFollowMetalink( false ),
pStateful( false ),
pAggregatedWaitTime( 0 ),
pMsgInFly( false ),
pSendingState( 0 ),
pTimeoutFence( false ),
pDirListStarted( false ),
pDirListWithStat( false ),
pCV( 0 ),
pSslErrCnt( 0 )
{
pPostMaster = DefaultEnv::GetPostMaster();
if( msg->GetSessionId() )
pHasSessionId = true;
Log *log = DefaultEnv::GetLog();
log->Debug( ExDbgMsg, "[%s] MsgHandler created: %p (message: %s ).",
pUrl.GetHostId().c_str(), (void*)this,
pRequest->GetObfuscatedDescription().c_str() );
ClientRequestHdr *hdr = (ClientRequestHdr*)pRequest->GetBuffer();
if( ntohs( hdr->requestid ) == kXR_pgread )
{
ClientPgReadRequest *pgrdreq = (ClientPgReadRequest*)pRequest->GetBuffer();
pCrc32cDigests.reserve( XrdOucPgrwUtils::csNum( ntohll( pgrdreq->offset ),
ntohl( pgrdreq->rlen ) ) );
}
//----------------------------------------------------------------------
// Pass the reader our pUrl, not *url. The latter is a reference, likely
// from FileStateHandler such as *pDataServer. Accessing that throughout
// our lifetime may lead to concurrent access. In the case of read-
// recovery the FileStateHandler may entirely reallocate the url object.
//----------------------------------------------------------------------
if( ntohs( hdr->requestid ) == kXR_readv )
pBodyReader.reset( new AsyncVectorReader( pUrl, *pRequest ) );
else if( ntohs( hdr->requestid ) == kXR_read )
pBodyReader.reset( new AsyncRawReader( pUrl, *pRequest ) );
else
pBodyReader.reset( new AsyncDiscardReader( pUrl, *pRequest ) );
}
ClientPgReadRequest::rlen
kXR_int32 rlen
Definition XProtocol.hh:511
ClientRequestHdr::requestid
kXR_unt16 requestid
Definition XProtocol.hh:157
kXR_read
@ kXR_read
Definition XProtocol.hh:125
kXR_readv
@ kXR_readv
Definition XProtocol.hh:137
kXR_pgread
@ kXR_pgread
Definition XProtocol.hh:142
ClientPgReadRequest::offset
kXR_int64 offset
Definition XProtocol.hh:510
XrdCl::DefaultEnv::GetLog
static Log * GetLog()
Get default log.
Definition XrdClDefaultEnv.cc:493
XrdCl::DefaultEnv::GetPostMaster
static PostMaster * GetPostMaster()
Get default post master.
Definition XrdClDefaultEnv.cc:453
XrdOucPgrwUtils::csNum
static int csNum(off_t offs, int count)
Compute the required size of a checksum vector based on offset & length.
Definition XrdOucPgrwUtils.cc:109
XrdCl::ExDbgMsg
const uint64_t ExDbgMsg
Definition XrdClConstants.hh:44
XrdGlobal::Log
XrdSysError Log
Definition XrdConfig.cc:113

References XrdOucPgrwUtils::csNum(), XrdCl::Log::Debug(), XrdCl::ExDbgMsg, XrdCl::DefaultEnv::GetLog(), XrdCl::DefaultEnv::GetPostMaster(), XrdCl::Message::GetSessionId(), kXR_pgread, kXR_read, kXR_readv, ClientPgReadRequest::offset, ClientRequestHdr::requestid, and ClientPgReadRequest::rlen.

Here is the call graph for this function:

## ~XRootDMsgHandler()

XrdCl::XRootDMsgHandler::~XRootDMsgHandler | ( |  | ) |

inline

Destructor.

Definition at line 216 of file XrdClXRootDMsgHandler.hh.

{
DumpRedirectTraceBack();
if( !pHasSessionId )
delete pRequest;
delete pEffectiveDataServerUrl;
pRequest = reinterpret_cast<Message*>( 0xDEADBEEF );
pResponseHandler = reinterpret_cast<ResponseHandler*>( 0xDEADBEEF );
pPostMaster = reinterpret_cast<PostMaster*>( 0xDEADBEEF );
pLFileHandler = reinterpret_cast<LocalFileHandler*>( 0xDEADBEEF );
pChunkList = reinterpret_cast<ChunkList*>( 0xDEADBEEF );
pEffectiveDataServerUrl = reinterpret_cast<URL*>( 0xDEADBEEF );
Log *log = DefaultEnv::GetLog();
log->Debug( ExDbgMsg, "[%s] Destroying MsgHandler: %p.",
pUrl.GetHostId().c_str(), (void*)this );
}
XrdCl::ChunkList
std::vector< ChunkInfo > ChunkList
List of chunks.
Definition XrdClXRootDResponses.hh:1056

References XrdCl::Log::Debug(), XrdCl::ExDbgMsg, and XrdCl::DefaultEnv::GetLog().

Here is the call graph for this function:

## Member Function Documentation

## Examine()

uint16_t XrdCl::XRootDMsgHandler::Examine | ( | std::shared_ptr< Message > & | msg | ) |

override virtual

Examine an incoming message, and decide on the action to be taken

Parameters

  msg | the message, may be zero if receive failed

Returns
  action type that needs to be take wrt the message and the handler

Implements XrdCl::MsgHandler.

Definition at line 109 of file XrdClXRootDMsgHandler.cc.

{
const int sst = pSendingState.fetch_or( kSawResp );
if( !( sst & kSendDone ) && !( sst & kSawResp ) )
{
// we must have been sent although we haven't got the OnStatusReady
// notification yet. Set the inflight notice.
Log *log = DefaultEnv::GetLog();
log->Dump( XRootDMsg, "[%s] Message %s reply received before notification "
"that it was sent, assuming it was sent ok.",
pUrl.GetHostId().c_str(),
pRequest->GetObfuscatedDescription().c_str() );
pMsgInFly = true;
}
//--------------------------------------------------------------------------
// if the MsgHandler is already being used to process another request
// (kXR_oksofar) we need to wait
//--------------------------------------------------------------------------
if( pOksofarAsAnswer )
{
XrdSysCondVarHelper lck( pCV );
while( pResponse ) pCV.Wait();
}
else
{
if( pResponse )
{
Log *log = DefaultEnv::GetLog();
log->Warning( ExDbgMsg, "[%s] MsgHandler is examining a response although "
"it already owns a response: %p (message: %s ).",
pUrl.GetHostId().c_str(), (void*)this,
pRequest->GetObfuscatedDescription().c_str() );
}
}
if( msg->GetSize() < 8 )
return Ignore;
ServerResponse *rsp = (ServerResponse *)msg->GetBuffer();
ClientRequest *req = (ClientRequest *)pRequest->GetBuffer();
uint16_t status = 0;
uint32_t dlen = 0;
//--------------------------------------------------------------------------
// We only care about async responses, but those are extracted now
// in the SocketHandler.
//--------------------------------------------------------------------------
if( rsp->hdr.status == kXR_attn )
{
return Ignore;
}
//--------------------------------------------------------------------------
// We got a sync message - check if it belongs to us
//--------------------------------------------------------------------------
else
{
if( rsp->hdr.streamid[0] != req->header.streamid[0] ||
rsp->hdr.streamid[1] != req->header.streamid[1] )
return Ignore;
status = rsp->hdr.status;
dlen = rsp->hdr.dlen;
}
//--------------------------------------------------------------------------
// We take the ownership of the message and decide what we will do
// with the handler itself, the options are:
// 1) we want to either read in raw mode (the Raw flag) or have the message
// body reconstructed for us by the TransportHandler by the time
// Process() is called (default, no extra flag)
// 2) we either got a full response in which case we don't want to be
// notified about anything anymore (RemoveHandler) or we got a partial
// answer and we need to wait for more (default, no extra flag)
//--------------------------------------------------------------------------
pResponse = msg;
pBodyReader->SetDataLength( dlen );
Log *log = DefaultEnv::GetLog();
switch( status )
{
//------------------------------------------------------------------------
// Handle the cached cases
//------------------------------------------------------------------------
case kXR_error:
case kXR_redirect:
case kXR_wait:
return RemoveHandler;
case kXR_waitresp:
{
log->Dump( XRootDMsg, "[%s] Got kXR_waitresp response to "
"message %s", pUrl.GetHostId().c_str(),
pRequest->GetObfuscatedDescription().c_str() );
pResponse.reset();
return Ignore; // This must be handled synchronously!
}
//------------------------------------------------------------------------
// Handle the potential raw cases
//------------------------------------------------------------------------
case kXR_ok:
{
//----------------------------------------------------------------------
// For kXR_read we read in raw mode
//----------------------------------------------------------------------
uint16_t reqId = ntohs( req->header.requestid );
if( reqId == kXR_read )
{
return Raw | RemoveHandler;
}
//----------------------------------------------------------------------
// kXR_readv is the same as kXR_read
//----------------------------------------------------------------------
if( reqId == kXR_readv )
{
return Raw | RemoveHandler;
}
//----------------------------------------------------------------------
// For everything else we just take what we got
//----------------------------------------------------------------------
return RemoveHandler;
}
//------------------------------------------------------------------------
// kXR_oksofars are special, they are not full responses, so we reset
// the response pointer to 0 and add the message to the partial list
//------------------------------------------------------------------------
case kXR_oksofar:
{
log->Dump( XRootDMsg, "[%s] Got a kXR_oksofar response to request "
"%s", pUrl.GetHostId().c_str(),
pRequest->GetObfuscatedDescription().c_str() );
if( !pOksofarAsAnswer )
{
pPartialResps.emplace_back( std::move( pResponse ) );
}
//----------------------------------------------------------------------
// For kXR_read we either read in raw mode if the message has not
// been fully reconstructed already, if it has, we adjust
// the buffer offset to prepare for the next one
//----------------------------------------------------------------------
uint16_t reqId = ntohs( req->header.requestid );
if( reqId == kXR_read )
{
pTimeoutFence.store( true, std::memory_order_relaxed );
return Raw | ( pOksofarAsAnswer ? None : NoProcess );
}
//----------------------------------------------------------------------
// kXR_readv is similar to read, except that the payload is different
//----------------------------------------------------------------------
if( reqId == kXR_readv )
{
pTimeoutFence.store( true, std::memory_order_relaxed );
return Raw | ( pOksofarAsAnswer ? None : NoProcess );
}
return ( pOksofarAsAnswer ? None : NoProcess );
}
case kXR_status:
{
log->Dump( XRootDMsg, "[%s] Got a kXR_status response to request "
"%s", pUrl.GetHostId().c_str(),
pRequest->GetObfuscatedDescription().c_str() );
uint16_t reqId = ntohs( req->header.requestid );
if( reqId == kXR_pgwrite )
{
//--------------------------------------------------------------------
// In case of pgwrite by definition this wont be a partial response
// so we can already remove the handler from the in-queue
//--------------------------------------------------------------------
return RemoveHandler;
}
//----------------------------------------------------------------------
// Otherwise (pgread), first of all we need to read the body of the
// kXR_status response, we can handle the raw data (if any) only after
// we have the whole kXR_status body
//----------------------------------------------------------------------
pTimeoutFence.store( true, std::memory_order_relaxed );
return None;
}
//------------------------------------------------------------------------
// Default
//------------------------------------------------------------------------
default:
return RemoveHandler;
}
return RemoveHandler;
}
ServerResponseHeader::status
kXR_unt16 status
Definition XProtocol.hh:915
ClientRequestHdr::streamid
kXR_char streamid[2]
Definition XProtocol.hh:156
ServerResponseHeader::streamid
kXR_char streamid[2]
Definition XProtocol.hh:914
kXR_waitresp
@ kXR_waitresp
Definition XProtocol.hh:906
kXR_redirect
@ kXR_redirect
Definition XProtocol.hh:904
kXR_oksofar
@ kXR_oksofar
Definition XProtocol.hh:900
kXR_status
@ kXR_status
Definition XProtocol.hh:907
kXR_ok
@ kXR_ok
Definition XProtocol.hh:899
kXR_attn
@ kXR_attn
Definition XProtocol.hh:901
kXR_wait
@ kXR_wait
Definition XProtocol.hh:905
kXR_error
@ kXR_error
Definition XProtocol.hh:903
ClientRequest::header
struct ClientRequestHdr header
Definition XProtocol.hh:846
kXR_pgwrite
@ kXR_pgwrite
Definition XProtocol.hh:138
ServerResponseHeader::dlen
kXR_int32 dlen
Definition XProtocol.hh:916
ServerResponse::hdr
ServerResponseHeader hdr
Definition XProtocol.hh:1288
XrdCl::MsgHandler::NoProcess
@ NoProcess
Definition XrdClPostMasterInterfaces.hh:66
XrdCl::MsgHandler::Raw
@ Raw
Definition XrdClPostMasterInterfaces.hh:63
XrdCl::MsgHandler::RemoveHandler
@ RemoveHandler
Definition XrdClPostMasterInterfaces.hh:61
XrdCl::MsgHandler::None
@ None
Definition XrdClPostMasterInterfaces.hh:58
XrdCl::MsgHandler::Ignore
@ Ignore
Ignore the message.
Definition XrdClPostMasterInterfaces.hh:60
XrdCl::XRootDMsg
const uint64_t XRootDMsg
Definition XrdClConstants.hh:39

References ServerResponseHeader::dlen, XrdCl::Log::Dump(), XrdCl::ExDbgMsg, XrdCl::DefaultEnv::GetLog(), ServerResponse::hdr, ClientRequest::header, XrdCl::MsgHandler::Ignore, kXR_attn, kXR_error, kXR_ok, kXR_oksofar, kXR_pgwrite, kXR_read, kXR_readv, kXR_redirect, kXR_status, kXR_wait, kXR_waitresp, XrdCl::MsgHandler::None, XrdCl::MsgHandler::NoProcess, XrdCl::MsgHandler::Raw, XrdCl::MsgHandler::RemoveHandler, ClientRequestHdr::requestid, ServerResponseHeader::status, ClientRequestHdr::streamid, ServerResponseHeader::streamid, XrdCl::Log::Warning(), and XrdCl::XRootDMsg.

Here is the call graph for this function:

## GetExpiration()

time_t XrdCl::XRootDMsgHandler::GetExpiration | ( |  | ) |

inline override virtual

Get a timestamp after which we give up.

Implements XrdCl::MsgHandler.

Definition at line 337 of file XrdClXRootDMsgHandler.hh.

{
return pExpiration;
}

## GetRequest()

const Message * XrdCl::XRootDMsgHandler::GetRequest | ( |  | ) | const

inline

Get the request pointer.

Definition at line 363 of file XrdClXRootDMsgHandler.hh.

{
return pRequest;
}

## GetSid()

uint16_t XrdCl::XRootDMsgHandler::GetSid | ( |  | ) | const

override virtual

Get handler sid

return sid of the corresponding request, otherwise 0

Implements XrdCl::MsgHandler.

Definition at line 405 of file XrdClXRootDMsgHandler.cc.

{
ClientRequest* req = (ClientRequest*) pRequest->GetBuffer();
return ((uint16_t)req->header.streamid[1] << 8) | (uint16_t)req->header.streamid[0];
}

References ClientRequest::header, and ClientRequestHdr::streamid.

## InspectStatusRsp()

uint16_t XrdCl::XRootDMsgHandler::InspectStatusRsp | ( |  | ) |

override virtual

Reexamine the incoming message, and decide on the action to be taken

In case of kXR_status the message can be only fully examined after reading the whole body (without raw data).

Parameters

  msg | the message, may be zero if receive failed

Returns
  action type that needs to be take wrt the message and the handler

Implements XrdCl::MsgHandler.

Definition at line 315 of file XrdClXRootDMsgHandler.cc.

{
if( !pResponse )
return 0;
Log *log = DefaultEnv::GetLog();
ServerResponse *rsp = (ServerResponse *)pResponse->GetBuffer();
//--------------------------------------------------------------------------
// Additional action is only required for kXR_status
//--------------------------------------------------------------------------
if( rsp->hdr.status != kXR_status ) return 0;
//--------------------------------------------------------------------------
// Ignore malformed status response
//--------------------------------------------------------------------------
if( pResponse->GetSize() < sizeof( ServerResponseStatus ) )
{
log->Error( XRootDMsg, "[%s] kXR_status: invalid message size.", pUrl.GetHostId().c_str() );
return Corrupted;
}
ClientRequest *req = (ClientRequest *)pRequest->GetBuffer();
uint16_t reqId = ntohs( req->header.requestid );
//--------------------------------------------------------------------------
// Unmarshal the status body
//--------------------------------------------------------------------------
XRootDStatus st = XRootDTransport::UnMarshalStatusBody( *pResponse, reqId );
if( !st.IsOK() && st.code == errDataError )
{
log->Error( XRootDMsg, "[%s] %s", pUrl.GetHostId().c_str(),
st.GetErrorMessage().c_str() );
return Corrupted;
}
if( !st.IsOK() )
{
log->Error( XRootDMsg, "[%s] Failed to unmarshall status body.",
pUrl.GetHostId().c_str() );
pStatus = st;
HandleRspOrQueue();
return Ignore;
}
//--------------------------------------------------------------------------
// Common handling for partial results
//--------------------------------------------------------------------------
ServerResponseV2 *rspst = (ServerResponseV2*)pResponse->GetBuffer();
if( rspst->status.bdy.resptype == XrdProto::kXR_PartialResult )
{
pPartialResps.push_back( std::move( pResponse ) );
}
//--------------------------------------------------------------------------
// Decide the actions that we need to take
//--------------------------------------------------------------------------
uint16_t action = 0;
if( reqId == kXR_pgread )
{
//----------------------------------------------------------------------
// The message contains only Status header and body but no raw data
//----------------------------------------------------------------------
if( !pPageReader )
pPageReader.reset( new AsyncPageReader( *pChunkList, pCrc32cDigests ) );
pPageReader->SetRsp( rspst );
action |= Raw;
if( rspst->status.bdy.resptype == XrdProto::kXR_PartialResult )
action |= NoProcess;
else
action |= RemoveHandler;
}
else if( reqId == kXR_pgwrite )
{
// if data corruption has been detected on the server side we will
// send some additional data pointing to the pages that need to be
// retransmitted
if( size_t( sizeof( ServerResponseHeader ) + rspst->status.hdr.dlen + rspst->status.bdy.dlen ) >
pResponse->GetCursor() )
action |= More;
}
return action;
}
ServerResponseV2::status
ServerResponseStatus status
Definition XProtocol.hh:1310
ServerResponseStatus::bdy
struct ServerResponseBody_Status bdy
Definition XProtocol.hh:1262
ServerResponseBody_Status::dlen
kXR_int32 dlen
Definition XProtocol.hh:1241
ServerResponseStatus::hdr
struct ServerResponseHeader hdr
Definition XProtocol.hh:1261
ServerResponseBody_Status::resptype
kXR_char resptype
Definition XProtocol.hh:1239
XrdCl::MsgHandler::More
@ More
there are more (non-raw) data to be read
Definition XrdClPostMasterInterfaces.hh:72
XrdCl::MsgHandler::Corrupted
@ Corrupted
Definition XrdClPostMasterInterfaces.hh:69
XrdCl::XRootDTransport::UnMarshalStatusBody
static XRootDStatus UnMarshalStatusBody(Message &msg, uint16_t reqType)
Unmarshall the body of the status response.
Definition XrdClXRootDTransport.cc:1358
XrdCl::errDataError
const uint16_t errDataError
data is corrupted
Definition XrdClStatus.hh:63
XrdProto::kXR_PartialResult
@ kXR_PartialResult
Definition XProtocol.hh:1251

References ServerResponseStatus::bdy, XrdCl::Status::code, XrdCl::MsgHandler::Corrupted, ServerResponseBody_Status::dlen, ServerResponseHeader::dlen, XrdCl::errDataError, XrdCl::Log::Error(), XrdCl::XRootDStatus::GetErrorMessage(), XrdCl::DefaultEnv::GetLog(), ServerResponse::hdr, ServerResponseStatus::hdr, ClientRequest::header, XrdCl::MsgHandler::Ignore, XrdCl::Status::IsOK(), XrdProto::kXR_PartialResult, kXR_pgread, kXR_pgwrite, kXR_status, XrdCl::MsgHandler::More, XrdCl::MsgHandler::NoProcess, XrdCl::MsgHandler::Raw, XrdCl::MsgHandler::RemoveHandler, ClientRequestHdr::requestid, ServerResponseBody_Status::resptype, ServerResponseHeader::status, ServerResponseV2::status, XrdCl::XRootDTransport::UnMarshalStatusBody(), and XrdCl::XRootDMsg.

Here is the call graph for this function:

## IsRaw()

bool XrdCl::XRootDMsgHandler::IsRaw | ( |  | ) | const

override virtual

Are we a raw writer or not?

Reimplemented from XrdCl::MsgHandler.

Definition at line 981 of file XrdClXRootDMsgHandler.cc.

{
ClientRequest *req = (ClientRequest *)pRequest->GetBuffer();
uint16_t reqId = ntohs( req->header.requestid );
if( reqId == kXR_write || reqId == kXR_writev || reqId == kXR_pgwrite )
return true;
// checkpoint + execute
if( reqId == kXR_chkpoint && req->chkpoint.opcode == kXR_ckpXeq )
{
ClientRequest *xeq = (ClientRequest*)pRequest->GetBuffer( sizeof( ClientRequest ) );
reqId = ntohs( xeq->header.requestid );
return reqId != kXR_truncate; // only checkpointed truncate does not have raw data
}
return false;
}
kXR_ckpXeq
static const int kXR_ckpXeq
Definition XProtocol.hh:216
ClientChkPointRequest::opcode
kXR_char opcode
Definition XProtocol.hh:206
kXR_writev
@ kXR_writev
Definition XProtocol.hh:143
kXR_write
@ kXR_write
Definition XProtocol.hh:131
kXR_truncate
@ kXR_truncate
Definition XProtocol.hh:140
kXR_chkpoint
@ kXR_chkpoint
Definition XProtocol.hh:124
ClientRequest::chkpoint
struct ClientChkPointRequest chkpoint
Definition XProtocol.hh:849

References ClientRequest::chkpoint, ClientRequest::header, kXR_chkpoint, kXR_ckpXeq, kXR_pgwrite, kXR_truncate, kXR_write, kXR_writev, ClientChkPointRequest::opcode, and ClientRequestHdr::requestid.

## OnReadyToSend()

void XrdCl::XRootDMsgHandler::OnReadyToSend | ( | Message * | msg | ) |

inline override virtual

Called just before the message is going to be sent through a valid connection, so that the user can still make some modifications that were impossible before (ie. protocol version dependent adjustments)

Parameters

  msg | message concerned

Reimplemented from XrdCl::MsgHandler.

Definition at line 439 of file XrdClXRootDMsgHandler.hh.

{
pSendingState |= kSawReadySend;
}

## OnStatusReady()

void XrdCl::XRootDMsgHandler::OnStatusReady | ( | const Message * | message,
|  | XRootDStatus | status )

override virtual

The requested action has been performed and the status is available.

Implements XrdCl::MsgHandler.

Definition at line 921 of file XrdClXRootDMsgHandler.cc.

{
Log *log = DefaultEnv::GetLog();
const int sst = pSendingState.fetch_or( kSendDone );
// if we have already seen a response we can not be in the out-queue
// anymore, so we should be getting notified of a successful send.
// But if not, log and do our best to recover.
if( !status.IsOK() && ( ( sst & kFinalResp ) || ( sst & kSawResp ) ) )
{
log->Error( XRootDMsg, "[%s] Unexpected error for message %s. Trying to "
"recover.", pUrl.GetHostId().c_str(),
message->GetObfuscatedDescription().c_str() );
HandleError( status );
return;
}
if( sst & kFinalResp )
{
log->Dump( XRootDMsg, "[%s] Got late notification that outgoing message %s was "
"sent, already have final response, queuing handler callback.",
pUrl.GetHostId().c_str(), message->GetObfuscatedDescription().c_str() );
HandleRspOrQueue();
return;
}
if( sst & kSawResp )
{
log->Dump( XRootDMsg, "[%s] Got late notification that message %s has "
"been successfully sent.",
pUrl.GetHostId().c_str(), message->GetObfuscatedDescription().c_str() );
return;
}
//--------------------------------------------------------------------------
// We were successful, so we now need to listen for a response
//--------------------------------------------------------------------------
if( status.IsOK() )
{
log->Dump( XRootDMsg, "[%s] Message %s has been successfully sent.",
pUrl.GetHostId().c_str(), message->GetObfuscatedDescription().c_str() );
pMsgInFly = true;
return;
}
//--------------------------------------------------------------------------
// We have failed, recover if possible
//--------------------------------------------------------------------------
log->Error( XRootDMsg, "[%s] Impossible to send message %s. Trying to "
"recover.", pUrl.GetHostId().c_str(),
message->GetObfuscatedDescription().c_str() );
HandleError( status );
}

References XrdCl::Log::Dump(), XrdCl::Log::Error(), XrdCl::DefaultEnv::GetLog(), XrdCl::Message::GetObfuscatedDescription(), XrdCl::Status::IsOK(), and XrdCl::XRootDMsg.

Here is the call graph for this function:

## OnStreamEvent()

uint8_t XrdCl::XRootDMsgHandler::OnStreamEvent | ( | StreamEvent | event,
|  | XRootDStatus | status )

override virtual

Handle an event other that a message arrival

Parameters

  event | type of the event
  status | status info

Reimplemented from XrdCl::MsgHandler.

Definition at line 882 of file XrdClXRootDMsgHandler.cc.

{
Log *log = DefaultEnv::GetLog();
log->Dump( XRootDMsg, "[%s] Stream event reported for msg %s",
pUrl.GetHostId().c_str(), pRequest->GetObfuscatedDescription().c_str() );
if( event == Ready )
return 0;
if( pTimeoutFence.load( std::memory_order_relaxed ) )
return 0;
HandleError( status );
return RemoveHandler;
}
XrdCl::MsgHandler::Ready
@ Ready
The stream has become connected.
Definition XrdClPostMasterInterfaces.hh:80

References XrdCl::Log::Dump(), XrdCl::DefaultEnv::GetLog(), XrdCl::MsgHandler::Ready, XrdCl::MsgHandler::RemoveHandler, and XrdCl::XRootDMsg.

Here is the call graph for this function:

## PartialReceived()

void XrdCl::XRootDMsgHandler::PartialReceived | ( |  | ) |

Bookkeeping after partial response has been received:

- take down the timeout fence after oksofar response has been handled
- reset status-response-body marshaled flag

Definition at line 1158 of file XrdClXRootDMsgHandler.cc.

{
pTimeoutFence.store( false, std::memory_order_relaxed ); // Take down the timeout fence
}

Referenced by XrdCl::Stream::ForceError(), XrdCl::Stream::OnError(), and XrdCl::Stream::OnIncoming().

Here is the caller graph for this function:

## Process()

void XrdCl::XRootDMsgHandler::Process | ( |  | ) |

override virtual

Process the message if it was "taken" by the examine action.

Process the message if it was "taken" by the examine action

Parameters

  msg | the message to be processed

Reimplemented from XrdCl::MsgHandler.

Definition at line 414 of file XrdClXRootDMsgHandler.cc.

{
Log *log = DefaultEnv::GetLog();
ServerResponse *rsp = (ServerResponse *)pResponse->GetBuffer();
ClientRequest *req = (ClientRequest *)pRequest->GetBuffer();
//--------------------------------------------------------------------------
// If it is a local file, it can be only a metalink redirector
//--------------------------------------------------------------------------
if( pUrl.IsLocalFile() && pUrl.IsMetalink() )
pHosts->back().protocol = kXR_PROTOCOLVERSION;
//--------------------------------------------------------------------------
// We got an answer, check who we were talking to
//--------------------------------------------------------------------------
else
{
AnyObject qryResult;
int *qryResponse = nullptr;
pPostMaster->QueryTransport( pUrl, XRootDQuery::ServerFlags, qryResult );
qryResult.Get( qryResponse );
if (qryResponse) {
pHosts->back().flags = *qryResponse;
delete qryResponse;
qryResponse = nullptr;
}
pPostMaster->QueryTransport( pUrl, XRootDQuery::ProtocolVersion, qryResult );
qryResult.Get( qryResponse );
if (qryResponse) {
pHosts->back().protocol = *qryResponse;
delete qryResponse;
}
}
//--------------------------------------------------------------------------
// Process the message
//--------------------------------------------------------------------------
Status st = XRootDTransport::UnMarshallBody( pResponse.get(), req->header.requestid );
if( !st.IsOK() )
{
pStatus = Status( stFatal, errInvalidMessage );
HandleResponse();
return;
}
//--------------------------------------------------------------------------
// we have an response for the message so it's not in fly anymore
//--------------------------------------------------------------------------
pMsgInFly = false;
//--------------------------------------------------------------------------
// Reset the aggregated wait (used to omit wait response in case of Metalink
// redirector)
//--------------------------------------------------------------------------
if( rsp->hdr.status != kXR_wait )
pAggregatedWaitTime = 0;
switch( rsp->hdr.status )
{
//------------------------------------------------------------------------
// kXR_ok - we're done here
//------------------------------------------------------------------------
case kXR_ok:
{
log->Dump( XRootDMsg, "[%s] Got a kXR_ok response to request %s",
pUrl.GetHostId().c_str(),
pRequest->GetObfuscatedDescription().c_str() );
pStatus = Status();
HandleResponse();
return;
}
case kXR_status:
{
log->Dump( XRootDMsg, "[%s] Got a kXR_status response to request %s",
pUrl.GetHostId().c_str(),
pRequest->GetObfuscatedDescription().c_str() );
pStatus = Status();
HandleResponse();
return;
}
//------------------------------------------------------------------------
// kXR_ok - we're serving partial result to the user
//------------------------------------------------------------------------
case kXR_oksofar:
{
log->Dump( XRootDMsg, "[%s] Got a kXR_oksofar response to request %s",
pUrl.GetHostId().c_str(),
pRequest->GetObfuscatedDescription().c_str() );
pStatus = Status( stOK, suContinue );
HandleResponse();
return;
}
//------------------------------------------------------------------------
// kXR_error - we've got a problem
//------------------------------------------------------------------------
case kXR_error:
{
char *errmsg = new char[rsp->hdr.dlen-3]; errmsg[rsp->hdr.dlen-4] = 0;
memcpy( errmsg, rsp->body.error.errmsg, rsp->hdr.dlen-4 );
log->Dump( XRootDMsg, "[%s] Got a kXR_error response to request %s "
"[%d] %s", pUrl.GetHostId().c_str(),
pRequest->GetObfuscatedDescription().c_str(), rsp->body.error.errnum,
errmsg );
delete [] errmsg;
HandleError( Status(stError, errErrorResponse, rsp->body.error.errnum) );
return;
}
//------------------------------------------------------------------------
// kXR_redirect - they tell us to go elsewhere
//------------------------------------------------------------------------
case kXR_redirect:
{
if( rsp->hdr.dlen <= 4 )
{
log->Error( XRootDMsg, "[%s] Got invalid redirect response.",
pUrl.GetHostId().c_str() );
pStatus = Status( stError, errInvalidResponse );
HandleResponse();
return;
}
char *urlInfoBuff = new char[rsp->hdr.dlen-3];
urlInfoBuff[rsp->hdr.dlen-4] = 0;
memcpy( urlInfoBuff, rsp->body.redirect.host, rsp->hdr.dlen-4 );
std::string urlInfo = urlInfoBuff;
delete [] urlInfoBuff;
log->Dump( XRootDMsg, "[%s] Got kXR_redirect response to "
"message %s: %s, port %d", pUrl.GetHostId().c_str(),
pRequest->GetObfuscatedDescription().c_str(), urlInfo.c_str(),
rsp->body.redirect.port );
//----------------------------------------------------------------------
// Check if we can proceed
//----------------------------------------------------------------------
if( !pRedirectCounter )
{
log->Warning( XRootDMsg, "[%s] Redirect limit has been reached for "
"message %s, the last known error is: %s",
pUrl.GetHostId().c_str(),
pRequest->GetObfuscatedDescription().c_str(),
pLastError.ToString().c_str() );
pStatus = Status( stFatal, errRedirectLimit );
HandleResponse();
return;
}
--pRedirectCounter;
//----------------------------------------------------------------------
// Keep the info about this server if we still need to find a load
// balancer
//----------------------------------------------------------------------
uint32_t flags = pHosts->back().flags;
if( !pHasLoadBalancer )
{
if( flags & kXR_isManager )
{
//------------------------------------------------------------------
// If the current server is a meta manager then it supersedes
// any existing load balancer, otherwise we assign a load-balancer
// only if it has not been already assigned
//------------------------------------------------------------------
if( ( flags & kXR_attrMeta ) || !pLoadBalancer.url.IsValid() )
{
pLoadBalancer = pHosts->back();
log->Dump( XRootDMsg, "[%s] Current server has been assigned "
"as a load-balancer for message %s",
pUrl.GetHostId().c_str(),
pRequest->GetObfuscatedDescription().c_str() );
HostList::iterator it;
for( it = pHosts->begin(); it != pHosts->end(); ++it )
it->loadBalancer = false;
pHosts->back().loadBalancer = true;
}
}
}
//----------------------------------------------------------------------
// If the redirect comes from a data server safe the URL because
// in case of a failure we will use it as the effective data server URL
// for the tried CGI opaque info
//----------------------------------------------------------------------
if( flags & kXR_isServer )
pEffectiveDataServerUrl = new URL( pHosts->back().url );
//----------------------------------------------------------------------
// Build the URL and check it's validity
//----------------------------------------------------------------------
std::vector<std::string> urlComponents;
std::string newCgi;
Utils::splitString( urlComponents, urlInfo, "?" );
std::ostringstream o;
o << urlComponents[0];
if( rsp->body.redirect.port > 0 )
o << ":" << rsp->body.redirect.port << "/";
else if( rsp->body.redirect.port < 0 )
{
//--------------------------------------------------------------------
// check if the manager wants to enforce write recovery at himself
// (beware we are dealing here with negative flags)
//--------------------------------------------------------------------
if( ~uint32_t( rsp->body.redirect.port ) & kXR_recoverWrts )
pHosts->back().flags |= kXR_recoverWrts;
//--------------------------------------------------------------------
// check if the manager wants to collapse the communication channel
// (the redirect host is to replace the current host)
//--------------------------------------------------------------------
if( ~uint32_t( rsp->body.redirect.port ) & kXR_collapseRedir )
{
std::string url( rsp->body.redirect.host, rsp->hdr.dlen-4 );
pPostMaster->CollapseRedirect( pUrl, url );
}
if( ~uint32_t( rsp->body.redirect.port ) & kXR_ecRedir )
{
std::string url( rsp->body.redirect.host, rsp->hdr.dlen-4 );
if( Utils::CheckEC( pRequest, url ) )
pRedirectAsAnswer = true;
}
}
URL newUrl = URL( o.str() );
if( !newUrl.IsValid() )
{
pStatus = Status( stError, errInvalidRedirectURL );
log->Error( XRootDMsg, "[%s] Got invalid redirection URL: %s",
pUrl.GetHostId().c_str(), urlInfo.c_str() );
HandleResponse();
return;
}
if( pUrl.GetUserName() != "" && newUrl.GetUserName() == "" )
newUrl.SetUserName( pUrl.GetUserName() );
if( pUrl.GetPassword() != "" && newUrl.GetPassword() == "" )
newUrl.SetPassword( pUrl.GetPassword() );
//----------------------------------------------------------------------
// Forward any "xrd.*" params from the original client request also to
// the new redirection url
// Also, we need to preserve any "xrdcl.*' as they are important for
// our internal workflows.
//----------------------------------------------------------------------
std::ostringstream ossXrd;
const URL::ParamsMap &urlParams = pUrl.GetParams();
for(URL::ParamsMap::const_iterator it = urlParams.begin();
it != urlParams.end(); ++it )
{
if( it->first.compare( 0, 4, "xrd." ) &&
it->first.compare( 0, 6, "xrdcl." ) )
continue;
ossXrd << it->first << '=' << it->second << '&';
}
std::string xrdCgi = ossXrd.str();
pRedirectUrl = newUrl.GetURL();
URL cgiURL;
if( urlComponents.size() > 1 )
{
pRedirectUrl += "?";
pRedirectUrl += urlComponents[1];
std::ostringstream o;
o << "fake://fake:111//fake?";
o << urlComponents[1];
if( urlComponents.size() == 3 )
o << '?' << urlComponents[2];
if (!xrdCgi.empty())
{
o << '&' << xrdCgi;
pRedirectUrl += '&';
pRedirectUrl += xrdCgi;
}
cgiURL = URL( o.str() );
}
else {
if (!xrdCgi.empty())
{
std::ostringstream o;
o << "fake://fake:111//fake?";
o << xrdCgi;
cgiURL = URL( o.str() );
pRedirectUrl += '?';
pRedirectUrl += xrdCgi;
}
}
//----------------------------------------------------------------------
// Check if we need to return the URL as a response
//----------------------------------------------------------------------
if( newUrl.GetProtocol() != "root" && newUrl.GetProtocol() != "xroot" &&
newUrl.GetProtocol() != "roots" && newUrl.GetProtocol() != "xroots" &&
!newUrl.IsLocalFile() )
pRedirectAsAnswer = true;
if( pRedirectAsAnswer )
{
pStatus = Status( stError, errRedirect );
HandleResponse();
return;
}
//----------------------------------------------------------------------
// Rewrite the message in a way required to send it to another server
//----------------------------------------------------------------------
newUrl.SetParams( cgiURL.GetParams() );
Status st = RewriteRequestRedirect( newUrl );
if( !st.IsOK() )
{
pStatus = st;
HandleResponse();
return;
}
//----------------------------------------------------------------------
// Make sure we don't change the protocol by accident (root vs roots)
//----------------------------------------------------------------------
if( ( pUrl.GetProtocol() == "roots" || pUrl.GetProtocol() == "xroots" ) &&
( newUrl.GetProtocol() == "root" || newUrl.GetProtocol() == "xroot" ) )
newUrl.SetProtocol( "roots" );
//----------------------------------------------------------------------
// Send the request to the new location
//----------------------------------------------------------------------
HandleError( RetryAtServer( newUrl, RedirectEntry::EntryRedirect ) );
return;
}
//------------------------------------------------------------------------
// kXR_wait - we wait, and re-issue the request later
//------------------------------------------------------------------------
case kXR_wait:
{
uint32_t waitSeconds = 0;
if( rsp->hdr.dlen >= 4 )
{
char *infoMsg = new char[rsp->hdr.dlen-3];
infoMsg[rsp->hdr.dlen-4] = 0;
memcpy( infoMsg, rsp->body.wait.infomsg, rsp->hdr.dlen-4 );
log->Dump( XRootDMsg, "[%s] Got kXR_wait response of %d seconds to "
"message %s: %s", pUrl.GetHostId().c_str(),
rsp->body.wait.seconds, pRequest->GetObfuscatedDescription().c_str(),
infoMsg );
delete [] infoMsg;
waitSeconds = rsp->body.wait.seconds;
}
else
{
log->Dump( XRootDMsg, "[%s] Got kXR_wait response of 0 seconds to "
"message %s", pUrl.GetHostId().c_str(),
pRequest->GetObfuscatedDescription().c_str() );
}
pAggregatedWaitTime += waitSeconds;
// We need a special case if the data node comes from metalink
// redirector. In this case it might make more sense to try the
// next entry in the Metalink than wait.
if( OmitWait( *pRequest, pLoadBalancer.url ) )
{
int maxWait = DefaultMaxMetalinkWait;
DefaultEnv::GetEnv()->GetInt( "MaxMetalinkWait", maxWait );
if( pAggregatedWaitTime > maxWait )
{
UpdateTriedCGI();
HandleError( RetryAtServer( pLoadBalancer.url, RedirectEntry::EntryRedirectOnWait ) );
return;
}
}
//----------------------------------------------------------------------
// Some messages require rewriting before they can be sent again
// after wait
//----------------------------------------------------------------------
Status st = RewriteRequestWait();
if( !st.IsOK() )
{
pStatus = st;
HandleResponse();
return;
}
//----------------------------------------------------------------------
// Register a task to resend the message in some seconds, if we still
// have time to do that, and report a timeout otherwise
//----------------------------------------------------------------------
time_t resendTime = ::time(0)+waitSeconds;
if( resendTime < pExpiration )
{
log->Debug( ExDbgMsg, "[%s] Scheduling WaitTask for MsgHandler: %p (message: %s ).",
pUrl.GetHostId().c_str(), (void*)this,
pRequest->GetObfuscatedDescription().c_str() );
TaskManager *taskMgr = pPostMaster->GetTaskManager();
taskMgr->RegisterTask( new WaitTask( this ), resendTime );
}
else
{
log->Debug( XRootDMsg, "[%s] Wait time is too long, timing out %s",
pUrl.GetHostId().c_str(),
pRequest->GetObfuscatedDescription().c_str() );
HandleError( Status( stError, errOperationExpired) );
}
return;
}
//------------------------------------------------------------------------
// kXR_waitresp - the response will be returned in some seconds as an
// unsolicited message. Currently all messages of this type are handled
// one step before in the XrdClStream::OnIncoming as they need to be
// processed synchronously.
//------------------------------------------------------------------------
case kXR_waitresp:
{
if( rsp->hdr.dlen < 4 )
{
log->Error( XRootDMsg, "[%s] Got invalid waitresp response.",
pUrl.GetHostId().c_str() );
pStatus = Status( stError, errInvalidResponse );
HandleResponse();
return;
}
log->Dump( XRootDMsg, "[%s] Got kXR_waitresp response of %d seconds to "
"message %s", pUrl.GetHostId().c_str(),
rsp->body.waitresp.seconds,
pRequest->GetObfuscatedDescription().c_str() );
return;
}
//------------------------------------------------------------------------
// Default - unrecognized/unsupported response, declare an error
//------------------------------------------------------------------------
default:
{
log->Dump( XRootDMsg, "[%s] Got unrecognized response %d to "
"message %s", pUrl.GetHostId().c_str(),
rsp->hdr.status, pRequest->GetObfuscatedDescription().c_str() );
pStatus = Status( stError, errInvalidResponse );
HandleResponse();
return;
}
}
return;
}
kXR_isManager
#define kXR_isManager
Definition XProtocol.hh:1156
kXR_collapseRedir
#define kXR_collapseRedir
Definition XProtocol.hh:1167
kXR_attrMeta
#define kXR_attrMeta
Definition XProtocol.hh:1159
ServerResponse::body
union ServerResponse::@040373375333017131300127053271011057331004327334 body
kXR_recoverWrts
#define kXR_recoverWrts
Definition XProtocol.hh:1166
kXR_isServer
#define kXR_isServer
Definition XProtocol.hh:1157
kXR_PROTOCOLVERSION
#define kXR_PROTOCOLVERSION
Definition XProtocol.hh:70
kXR_ecRedir
#define kXR_ecRedir
Definition XProtocol.hh:1168
XrdCl::DefaultEnv::GetEnv
static Env * GetEnv()
Get default client environment.
Definition XrdClDefaultEnv.cc:445
XrdCl::Env::GetInt
bool GetInt(const std::string &key, int &value)
Definition XrdClEnv.cc:89
XrdCl::URL::ParamsMap
std::map< std::string, std::string > ParamsMap
Definition XrdClURL.hh:33
XrdCl::Utils::splitString
static void splitString(Container &result, const std::string &input, const std::string &delimiter)
Split a string.
Definition XrdClUtils.hh:56
XrdCl::Utils::CheckEC
static bool CheckEC(const Message *req, const URL &url)
Check if this client can support given EC redirect.
Definition XrdClUtils.cc:703
XrdCl::XRootDTransport::UnMarshallBody
static XRootDStatus UnMarshallBody(Message *msg, uint16_t reqType)
Unmarshall the body of the incoming message.
Definition XrdClXRootDTransport.cc:1280
XrdCl::errRedirectLimit
const uint16_t errRedirectLimit
Definition XrdClStatus.hh:102
XrdCl::DefaultMaxMetalinkWait
const int DefaultMaxMetalinkWait
Definition XrdClConstants.hh:86
XrdCl::errErrorResponse
const uint16_t errErrorResponse
Definition XrdClStatus.hh:105
XrdCl::errOperationExpired
const uint16_t errOperationExpired
Definition XrdClStatus.hh:90
XrdCl::stFatal
const uint16_t stFatal
Fatal error, it's still an error.
Definition XrdClStatus.hh:33
XrdCl::stError
const uint16_t stError
An error occurred that could potentially be retried.
Definition XrdClStatus.hh:32
XrdCl::stOK
const uint16_t stOK
Everything went OK.
Definition XrdClStatus.hh:31
XrdCl::errInvalidResponse
const uint16_t errInvalidResponse
Definition XrdClStatus.hh:99
XrdCl::errInvalidRedirectURL
const uint16_t errInvalidRedirectURL
Definition XrdClStatus.hh:98
XrdCl::suContinue
const uint16_t suContinue
Definition XrdClStatus.hh:39
XrdCl::errRedirect
const uint16_t errRedirect
Definition XrdClStatus.hh:106
XrdCl::errInvalidMessage
const uint16_t errInvalidMessage
Definition XrdClStatus.hh:85
XrdCl::RedirectEntry::EntryRedirect
@ EntryRedirect
Definition XrdClXRootDMsgHandler.hh:74
XrdCl::RedirectEntry::EntryRedirectOnWait
@ EntryRedirectOnWait
Definition XrdClXRootDMsgHandler.hh:75
XrdCl::XRootDQuery::ServerFlags
static const uint16_t ServerFlags
returns server flags
Definition XrdClPostMasterInterfaces.hh:290
XrdCl::XRootDQuery::ProtocolVersion
static const uint16_t ProtocolVersion
returns the protocol version
Definition XrdClPostMasterInterfaces.hh:291

References ServerResponse::body, XrdCl::Utils::CheckEC(), XrdCl::Log::Debug(), XrdCl::DefaultMaxMetalinkWait, ServerResponseHeader::dlen, XrdCl::Log::Dump(), XrdCl::RedirectEntry::EntryRedirect, XrdCl::RedirectEntry::EntryRedirectOnWait, XrdCl::errErrorResponse, XrdCl::errInvalidMessage, XrdCl::errInvalidRedirectURL, XrdCl::errInvalidResponse, XrdCl::errOperationExpired, XrdCl::Log::Error(), XrdCl::errRedirect, XrdCl::errRedirectLimit, XrdCl::ExDbgMsg, XrdCl::AnyObject::Get(), XrdCl::DefaultEnv::GetEnv(), XrdCl::Env::GetInt(), XrdCl::DefaultEnv::GetLog(), XrdCl::URL::GetParams(), XrdCl::URL::GetPassword(), XrdCl::URL::GetProtocol(), XrdCl::URL::GetURL(), XrdCl::URL::GetUserName(), ServerResponse::hdr, ClientRequest::header, XrdCl::URL::IsLocalFile(), XrdCl::Status::IsOK(), XrdCl::URL::IsValid(), kXR_attrMeta, kXR_collapseRedir, kXR_ecRedir, kXR_error, kXR_isManager, kXR_isServer, kXR_ok, kXR_oksofar, kXR_PROTOCOLVERSION, kXR_recoverWrts, kXR_redirect, kXR_status, kXR_wait, kXR_waitresp, ClientRequest::protocol, XrdCl::XRootDQuery::ProtocolVersion, XrdCl::TaskManager::RegisterTask(), ClientRequestHdr::requestid, XrdCl::XRootDQuery::ServerFlags, XrdCl::URL::SetParams(), XrdCl::URL::SetPassword(), XrdCl::URL::SetProtocol(), XrdCl::URL::SetUserName(), XrdCl::Utils::splitString(), ServerResponseHeader::status, XrdCl::stError, XrdCl::stFatal, XrdCl::stOK, XrdCl::suContinue, XrdCl::XRootDTransport::UnMarshallBody(), XrdCl::Log::Warning(), and XrdCl::XRootDMsg.

Here is the call graph for this function:

## ReadMessageBody()

XRootDStatus XrdCl::XRootDMsgHandler::ReadMessageBody | ( | Message * | msg,
|  | Socket * | socket,
|  | uint32_t & | bytesRead )

override virtual

Read message body directly from a socket - called if Examine returns Raw flag - only socket related errors may be returned here

Parameters

  msg | the corresponding message header
  socket | the socket to read from
  bytesRead | number of bytes read by the method

Returns
  stOK & suDone if the whole body has been processed stOK & suRetry if more data is needed stError on failure

Reimplemented from XrdCl::MsgHandler.

Definition at line 902 of file XrdClXRootDMsgHandler.cc.

{
ClientRequest *req = (ClientRequest *)pRequest->GetBuffer();
uint16_t reqId = ntohs( req->header.requestid );
if( reqId == kXR_pgread )
return pPageReader->Read( *socket, bytesRead );
return pBodyReader->Read( *socket, bytesRead );
}

References ClientRequest::header, kXR_pgread, and ClientRequestHdr::requestid.

## SetChunkList()

void XrdCl::XRootDMsgHandler::SetChunkList | ( | ChunkList * | chunkList | ) |

inline

Set the chunk list.

Definition at line 390 of file XrdClXRootDMsgHandler.hh.

{
pChunkList = chunkList;
if( pBodyReader )
pBodyReader->SetChunkList( chunkList );
if( chunkList )
pChunkStatus.resize( chunkList->size() );
else
pChunkStatus.clear();
}

Referenced by XrdCl::MessageUtils::RedirectMessage(), and XrdCl::MessageUtils::SendMessage().

Here is the caller graph for this function:

## SetCrc32cDigests()

void XrdCl::XRootDMsgHandler::SetCrc32cDigests | ( | std::vector< uint32_t > && | crc32cDigests | ) |

inline

Definition at line 401 of file XrdClXRootDMsgHandler.hh.

{
pCrc32cDigests = std::move( crc32cDigests );
}

Referenced by XrdCl::MessageUtils::SendMessage().

Here is the caller graph for this function:

## SetExpiration()

void XrdCl::XRootDMsgHandler::SetExpiration | ( | time_t | expiration | ) |

inline

Set a timestamp after which we give up.

Definition at line 329 of file XrdClXRootDMsgHandler.hh.

{
pExpiration = expiration;
}

Referenced by XrdCl::MessageUtils::RedirectMessage(), and XrdCl::MessageUtils::SendMessage().

Here is the caller graph for this function:

## SetFollowMetalink()

void XrdCl::XRootDMsgHandler::SetFollowMetalink | ( | bool | followMetalink | ) |

inline

Definition at line 422 of file XrdClXRootDMsgHandler.hh.

{
pFollowMetalink = followMetalink;
}

Referenced by XrdCl::MessageUtils::RedirectMessage().

Here is the caller graph for this function:

## SetHostList()

void XrdCl::XRootDMsgHandler::SetHostList | ( | HostList * | hostList | ) |

inline

Set host list.

Definition at line 382 of file XrdClXRootDMsgHandler.hh.

{
pHosts.reset( hostList );
}

Referenced by XrdCl::MessageUtils::RedirectMessage(), and XrdCl::MessageUtils::SendMessage().

Here is the caller graph for this function:

## SetKernelBuffer()

void XrdCl::XRootDMsgHandler::SetKernelBuffer | ( | XrdSys::KernelBuffer * | kbuff | ) |

inline

Set the kernel buffer.

Definition at line 409 of file XrdClXRootDMsgHandler.hh.

{
pKBuff = kbuff;
}

Referenced by XrdCl::MessageUtils::SendMessage().

Here is the caller graph for this function:

## SetLoadBalancer()

void XrdCl::XRootDMsgHandler::SetLoadBalancer | ( | const HostInfo & | loadBalancer | ) |

inline

Set the load balancer.

Definition at line 371 of file XrdClXRootDMsgHandler.hh.

{
if( !loadBalancer.url.IsValid() )
return;
pLoadBalancer = loadBalancer;
pHasLoadBalancer = true;
}

References XrdCl::URL::IsValid(), and XrdCl::HostInfo::url.

Referenced by XrdCl::MessageUtils::RedirectMessage(), and XrdCl::MessageUtils::SendMessage().

Here is the call graph for this function:
Here is the caller graph for this function:

## SetOksofarAsAnswer()

void XrdCl::XRootDMsgHandler::SetOksofarAsAnswer | ( | bool | oksofarAsAnswer | ) |

inline

Treat the kXR_oksofar response as a valid answer to the message and notify the handler with the URL as a response

Definition at line 355 of file XrdClXRootDMsgHandler.hh.

{
pOksofarAsAnswer = oksofarAsAnswer;
}

Referenced by XrdCl::MessageUtils::RedirectMessage(), and XrdCl::MessageUtils::SendMessage().

Here is the caller graph for this function:

## SetRedirectAsAnswer()

void XrdCl::XRootDMsgHandler::SetRedirectAsAnswer | ( | bool | redirectAsAnswer | ) |

inline

Treat the kXR_redirect response as a valid answer to the message and notify the handler with the URL as a response

Definition at line 346 of file XrdClXRootDMsgHandler.hh.

{
pRedirectAsAnswer = redirectAsAnswer;
}

Referenced by XrdCl::MessageUtils::RedirectMessage(), and XrdCl::MessageUtils::SendMessage().

Here is the caller graph for this function:

## SetRedirectCounter()

void XrdCl::XRootDMsgHandler::SetRedirectCounter | ( | uint16_t | redirectCounter | ) |

inline

Set the redirect counter.

Definition at line 417 of file XrdClXRootDMsgHandler.hh.

{
pRedirectCounter = redirectCounter;
}

Referenced by XrdCl::MessageUtils::RedirectMessage(), and XrdCl::MessageUtils::SendMessage().

Here is the caller graph for this function:

## SetStateful()

void XrdCl::XRootDMsgHandler::SetStateful | ( | bool | stateful | ) |

inline

Definition at line 427 of file XrdClXRootDMsgHandler.hh.

{
pStateful = stateful;
}

Referenced by XrdCl::MessageUtils::SendMessage().

Here is the caller graph for this function:

## WaitDone()

void XrdCl::XRootDMsgHandler::WaitDone | ( | time_t | now | ) |

Called after the wait time for kXR_wait has elapsed

Parameters

  now | current timestamp

Definition at line 1150 of file XrdClXRootDMsgHandler.cc.

{
HandleError( RetryAtServer( pUrl, RedirectEntry::EntryWait ) );
}
XrdCl::RedirectEntry::EntryWait
@ EntryWait
Definition XrdClXRootDMsgHandler.hh:77

References XrdCl::RedirectEntry::EntryWait.

## WriteMessageBody()

XRootDStatus XrdCl::XRootDMsgHandler::WriteMessageBody | ( | Socket * | socket,
|  | uint32_t & | bytesWritten )

override virtual

Write message body directly to a socket - called if IsRaw returns true - only socket related errors may be returned here

Parameters

  socket | the socket to read from
  bytesWritten | number of bytes written by the method

Returns
  stOK & suDone if the whole body has been processed stOK & suRetry if more data needs to be written stError on failure

Reimplemented from XrdCl::MsgHandler.

Definition at line 1001 of file XrdClXRootDMsgHandler.cc.

{
//--------------------------------------------------------------------------
// First check if it is a PgWrite
//--------------------------------------------------------------------------
if( !pChunkList->empty() && !pCrc32cDigests.empty() )
{
//------------------------------------------------------------------------
// PgWrite will have just one chunk
//------------------------------------------------------------------------
ChunkInfo chunk = pChunkList->front();
//------------------------------------------------------------------------
// Calculate the size of the first and last page (in case the chunk is not
// 4KB aligned)
//------------------------------------------------------------------------
int fLen = 0, lLen = 0;
size_t nbpgs = XrdOucPgrwUtils::csNum( chunk.offset, chunk.length, fLen, lLen );
//------------------------------------------------------------------------
// Set the crc32c buffer if not ready yet
//------------------------------------------------------------------------
if( pPgWrtCksumBuff.GetCursor() == 0 )
{
uint32_t digest = htonl( pCrc32cDigests[pPgWrtCurrentPageNb] );
memcpy( pPgWrtCksumBuff.GetBuffer(), &digest, sizeof( uint32_t ) );
}
uint32_t btsLeft = chunk.length - pAsyncOffset;
uint32_t pglen = ( pPgWrtCurrentPageNb == 0 ? fLen : XrdSys::PageSize ) - pPgWrtCurrentPageOffset;
if( pglen > btsLeft ) pglen = btsLeft;
char* pgbuf = static_cast<char*>( chunk.buffer ) + pAsyncOffset;
while( btsLeft > 0 )
{
// first write the crc32c digest
while( pPgWrtCksumBuff.GetCursor() < sizeof( uint32_t ) )
{
uint32_t dgstlen = sizeof( uint32_t ) - pPgWrtCksumBuff.GetCursor();
char* dgstbuf = pPgWrtCksumBuff.GetBufferAtCursor();
int btswrt = 0;
Status st = socket->Send( dgstbuf, dgstlen, btswrt );
if( !st.IsOK() ) return st;
bytesWritten += btswrt;
pPgWrtCksumBuff.AdvanceCursor( btswrt );
if( st.code == suRetry ) return st;
}
// then write the raw data (one page)
int btswrt = 0;
Status st = socket->Send( pgbuf, pglen, btswrt );
if( !st.IsOK() ) return st;
pgbuf += btswrt;
pglen -= btswrt;
btsLeft -= btswrt;
bytesWritten += btswrt;
pAsyncOffset += btswrt; // update the offset to the raw data
if( st.code == suRetry ) return st;
// if we managed to write all the data ...
if( pglen == 0 )
{
// move to the next page
++pPgWrtCurrentPageNb;
if( pPgWrtCurrentPageNb < nbpgs )
{
// set the digest buffer
pPgWrtCksumBuff.SetCursor( 0 );
uint32_t digest = htonl( pCrc32cDigests[pPgWrtCurrentPageNb] );
memcpy( pPgWrtCksumBuff.GetBuffer(), &digest, sizeof( uint32_t ) );
}
// set the page length
pglen = XrdSys::PageSize;
if( pglen > btsLeft ) pglen = btsLeft;
// reset offset in the current page
pPgWrtCurrentPageOffset = 0;
}
else
// otherwise just adjust the offset in the current page
pPgWrtCurrentPageOffset += btswrt;
}
}
else if( !pChunkList->empty() )
{
size_t size = pChunkList->size();
for( size_t i = pAsyncChunkIndex ; i < size; ++i )
{
char *buffer = (char*)(*pChunkList)[i].buffer;
uint32_t size = (*pChunkList)[i].length;
size_t leftToBeWritten = size - pAsyncOffset;
while( leftToBeWritten )
{
int btswrt = 0;
Status st = socket->Send( buffer + pAsyncOffset, leftToBeWritten, btswrt );
bytesWritten += btswrt;
if( !st.IsOK() || st.code == suRetry ) return st;
pAsyncOffset += btswrt;
leftToBeWritten -= btswrt;
}
//----------------------------------------------------------------------
// Remember that we have moved to the next chunk, also clear the offset
// within the buffer as we are going to move to a new one
//----------------------------------------------------------------------
++pAsyncChunkIndex;
pAsyncOffset = 0;
}
}
else
{
Log *log = DefaultEnv::GetLog();
//------------------------------------------------------------------------
// If the socket is encrypted we cannot use a kernel buffer, we have to
// convert to user space buffer
//------------------------------------------------------------------------
if( socket->IsEncrypted() )
{
log->Debug( XRootDMsg, "[%s] Channel is encrypted: cannot use kernel buffer.",
pUrl.GetHostId().c_str() );
char *ubuff = 0;
ssize_t ret = XrdSys::Move( *pKBuff, ubuff );
if( ret < 0 ) return Status( stError, errInternal );
pChunkList->push_back( ChunkInfo( 0, ret, ubuff ) );
return WriteMessageBody( socket, bytesWritten );
}
//------------------------------------------------------------------------
// Send the data
//------------------------------------------------------------------------
while( !pKBuff->Empty() )
{
int btswrt = 0;
Status st = socket->Send( *pKBuff, btswrt );
bytesWritten += btswrt;
if( !st.IsOK() || st.code == suRetry ) return st;
}
log->Debug( XRootDMsg, "[%s] Request %s payload (kernel buffer) transferred to socket.",
pUrl.GetHostId().c_str(), pRequest->GetObfuscatedDescription().c_str() );
}
return Status();
}
XrdCl::XRootDMsgHandler::WriteMessageBody
XRootDStatus WriteMessageBody(Socket *socket, uint32_t &bytesWritten) override
Definition XrdClXRootDMsgHandler.cc:1001
XrdCl::suRetry
const uint16_t suRetry
Definition XrdClStatus.hh:40
XrdCl::errInternal
const uint16_t errInternal
Internal error.
Definition XrdClStatus.hh:56
XrdSys::PageSize
static const int PageSize
Definition XrdSysPageSize.hh:36
XrdSys::Move
ssize_t Move(KernelBuffer &kbuff, char *&ubuff)
Definition XrdSysKernelBuffer.hh:452

References XrdCl::ChunkInfo::buffer, XrdCl::Status::code, XrdOucPgrwUtils::csNum(), XrdCl::Log::Debug(), XrdCl::errInternal, XrdCl::DefaultEnv::GetLog(), XrdCl::Socket::IsEncrypted(), XrdCl::Status::IsOK(), XrdCl::ChunkInfo::length, XrdSys::Move(), XrdCl::ChunkInfo::offset, XrdSys::PageSize, XrdCl::Socket::Send(), XrdCl::stError, XrdCl::suRetry, WriteMessageBody(), and XrdCl::XRootDMsg.

Referenced by WriteMessageBody().

Here is the call graph for this function:
Here is the caller graph for this function:

## Friends And Related Symbol Documentation

## HandleRspJob

friend class HandleRspJob

friend

Definition at line 121 of file XrdClXRootDMsgHandler.hh.

References HandleRspJob.

Referenced by HandleRspJob.

The documentation for this class was generated from the following files:

- XrdClXRootDMsgHandler.hh
- XrdClXRootDMsgHandler.cc