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	OnWaitingToSend (Message *msg) override
	Called to indicate the message is waiting to be sent.
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 ),
 
        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 ) );
      }

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 215 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 );
      }

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 )

overridevirtual

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() );
    }
 
    //--------------------------------------------------------------------------
    // 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;
  }

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 ( )

inlineoverridevirtual

Get a timestamp after which we give up.

Implements XrdCl::MsgHandler.

Definition at line 336 of file XrdClXRootDMsgHandler.hh.

      {
        return pExpiration;
      }

GetRequest()

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

inline

Get the request pointer.

Definition at line 362 of file XrdClXRootDMsgHandler.hh.

      {
        return pRequest;
      }

GetSid()

uint16_t XrdCl::XRootDMsgHandler::GetSid ( ) const

overridevirtual

Get handler sid

return sid of the corresponding request, otherwise 0

Implements XrdCl::MsgHandler.

Definition at line 403 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 ( )

overridevirtual

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 313 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;
  }

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

overridevirtual

Are we a raw writer or not?

Reimplemented from XrdCl::MsgHandler.

Definition at line 996 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;
  }

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 )

inlineoverridevirtual

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 438 of file XrdClXRootDMsgHandler.hh.

      {
        pSendingState |= kSawReadySend;
      }

OnStatusReady()

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

overridevirtual

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

Implements XrdCl::MsgHandler.

Definition at line 919 of file XrdClXRootDMsgHandler.cc.

  {
    Log *log = DefaultEnv::GetLog();
 
    if( status.IsOK() )
    {
      log->Dump( XRootDMsg, "[%s] Got notification that outgoing message %s "
                 "was sent successfully.", pUrl.GetHostId().c_str(),
                 message->GetObfuscatedDescription().c_str() );
    }
 
    // After setting kSendDone processing of this object may continue in
    // another thread. Unless we're in an error condition our object may
    // be modified or even destroyed after this point.
    const int sst = pSendingState.fetch_or( kSendDone );
 
    // ignore if we're already in this state
    if( status.IsOK() && ( sst & kSendDone ) ) return;
 
    // if we have already seen a response 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 )
    {
      // late notification and we already have final response for user,
      // need to queue handler callback.
      HandleRspOrQueue();
      return;
    }
 
    if( sst & kRetryAtSrv )
    {
       // late notification and we already received a response and know
       // we need to retry at differnt server.
       HandleError( RetryAtServer( pRetryAtUrl, pRetryAtEntryType ) );
       return;
    }
 
    if( sst & kSawResp )
    {
      // late notification, response processing may be happening in another
      // thread.
      return;
    }
 
    //--------------------------------------------------------------------------
    // We were successful, so we now need to listen for a response
    //--------------------------------------------------------------------------
    if( status.IsOK() )
    {
      // this is the expcted order, we got the notificaiton but no response
      // received yet. However another thread is liable to be processing
      // one or sending a final response and deleting us at any point now.
      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 )

overridevirtual

Handle an event other that a message arrival

Parameters

event	type of the event
status	status info

Reimplemented from XrdCl::MsgHandler.

Definition at line 880 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;
  }

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

Here is the call graph for this function:

OnWaitingToSend()

void XrdCl::XRootDMsgHandler::OnWaitingToSend ( Message * msg )

inlineoverridevirtual

Called to indicate the message is waiting to be sent.

Reimplemented from XrdCl::MsgHandler.

Definition at line 443 of file XrdClXRootDMsgHandler.hh.

      {
        pSendingState = 0;
      }

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 1173 of file XrdClXRootDMsgHandler.cc.

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

Referenced by XrdCl::Stream::OnIncoming().

Here is the caller graph for this function:

Process()

void XrdCl::XRootDMsgHandler::Process ( )

overridevirtual

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 412 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
    //--------------------------------------------------------------------------
    pSendingState.fetch_or( kInFlyDone );
 
    //--------------------------------------------------------------------------
    // 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;
  }

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 )

overridevirtual

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 900 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 389 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 400 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 328 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 421 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 381 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 408 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 370 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 354 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 345 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 416 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 426 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 1165 of file XrdClXRootDMsgHandler.cc.

  {
    HandleError( RetryAtServer( pUrl, RedirectEntry::EntryWait ) );
  }

References XrdCl::RedirectEntry::EntryWait.

WriteMessageBody()

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

overridevirtual

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 1016 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();
  }

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