Loading...
Searching...
No Matches
XrdClXRootDMsgHandler.cc
Go to the documentation of this file.
//------------------------------------------------------------------------------
// Copyright (c) 2011-2014 by European Organization for Nuclear Research (CERN)
// Author: Lukasz Janyst <ljanyst@cern.ch>
//------------------------------------------------------------------------------
// This file is part of the XRootD software suite.
//
// XRootD is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// XRootD is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with XRootD. If not, see <http://www.gnu.org/licenses/>.
//
// In applying this licence, CERN does not waive the privileges and immunities
// granted to it by virtue of its status as an Intergovernmental Organization
// or submit itself to any jurisdiction.
//------------------------------------------------------------------------------
#include "XrdCl/XrdClXRootDMsgHandler.hh"
#include "XrdCl/XrdClLog.hh"
#include "XrdCl/XrdClDefaultEnv.hh"
#include "XrdCl/XrdClConstants.hh"
#include "XrdCl/XrdClXRootDTransport.hh"
#include "XrdCl/XrdClMessage.hh"
#include "XrdCl/XrdClURL.hh"
#include "XrdCl/XrdClUtils.hh"
#include "XrdCl/XrdClTaskManager.hh"
#include "XrdCl/XrdClJobManager.hh"
#include "XrdCl/XrdClSIDManager.hh"
#include "XrdCl/XrdClMessageUtils.hh"
#include "XrdCl/XrdClLocalFileHandler.hh"
#include "XrdCl/XrdClRedirectorRegistry.hh"
#include "XrdCl/XrdClSocket.hh"
#include "XrdCl/XrdClTls.hh"
#include "XrdCl/XrdClOptimizers.hh"
#include "XrdOuc/XrdOucCRC.hh"
#include "XrdOuc/XrdOucPrivateUtils.hh"
#include "XrdSys/XrdSysPlatform.hh" // same as above
#include "XrdSys/XrdSysAtomics.hh"
#include "XrdSys/XrdSysPthread.hh"
#include <memory>
#include <sstream>
#include <numeric>
namespace
{
//----------------------------------------------------------------------------
// We need an extra task what will run the handler in the future, because
// tasks get deleted and we need the handler
//----------------------------------------------------------------------------
class WaitTask: public XrdCl::Task
{
public:
WaitTask( XrdCl::XRootDMsgHandler *handler ): pHandler( handler )
{
std::ostringstream o;
o << "WaitTask for: 0x" << handler->GetRequest();
SetName( o.str() );
}
virtual time_t Run( time_t now )
{
pHandler->WaitDone( now );
return 0;
}
private:
XrdCl::XRootDMsgHandler *pHandler;
};
}
namespace XrdCl
{
//----------------------------------------------------------------------------
// Delegate the response handling to the thread-pool
//----------------------------------------------------------------------------
class HandleRspJob: public XrdCl::Job
{
public:
HandleRspJob( XrdCl::XRootDMsgHandler *handler ): pHandler( handler )
{
}
virtual ~HandleRspJob()
{
}
virtual void Run( void *arg )
{
pHandler->HandleResponse();
delete this;
}
private:
XrdCl::XRootDMsgHandler *pHandler;
};
//----------------------------------------------------------------------------
// Examine an incoming message, and decide on the action to be taken
//----------------------------------------------------------------------------
uint16_t XRootDMsgHandler::Examine( std::shared_ptr<Message> &msg )
{
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;
}
//----------------------------------------------------------------------------
// Reexamine the incoming message, and decide on the action to be taken
//----------------------------------------------------------------------------
uint16_t XRootDMsgHandler::InspectStatusRsp()
{
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;
}
//----------------------------------------------------------------------------
// Get handler sid
//----------------------------------------------------------------------------
uint16_t XRootDMsgHandler::GetSid() const
{
ClientRequest* req = (ClientRequest*) pRequest->GetBuffer();
return ((uint16_t)req->header.streamid[1] << 8) | (uint16_t)req->header.streamid[0];
}
//----------------------------------------------------------------------------
//----------------------------------------------------------------------------
void XRootDMsgHandler::Process()
{
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;
}
//----------------------------------------------------------------------------
// Handle an event other that a message arrival - may be timeout
//----------------------------------------------------------------------------
uint8_t XRootDMsgHandler::OnStreamEvent( StreamEvent event,
XRootDStatus status )
{
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;
}
//----------------------------------------------------------------------------
// Read message body directly from a socket
//----------------------------------------------------------------------------
XRootDStatus XRootDMsgHandler::ReadMessageBody( Message*,
Socket *socket,
uint32_t &bytesRead )
{
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 );
}
//----------------------------------------------------------------------------
// We're here when we requested sending something over the wire
// or other status update on this action.
// We can be called when message is still in out-queue, with an
// error status indicating message will not be sent.
//----------------------------------------------------------------------------
void XRootDMsgHandler::OnStatusReady( const Message *message,
XRootDStatus status )
{
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 );
}
//----------------------------------------------------------------------------
// Are we a raw writer or not?
//----------------------------------------------------------------------------
bool XRootDMsgHandler::IsRaw() const
{
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;
}
//----------------------------------------------------------------------------
// Write the message body
//----------------------------------------------------------------------------
XRootDStatus XRootDMsgHandler::WriteMessageBody( Socket *socket,
uint32_t &bytesWritten )
{
//--------------------------------------------------------------------------
// 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();
}
//----------------------------------------------------------------------------
// We're here when we got a time event. We needed to re-issue the request
// in some time in the future, and that moment has arrived
//----------------------------------------------------------------------------
void XRootDMsgHandler::WaitDone( time_t )
{
HandleError( RetryAtServer( pUrl, RedirectEntry::EntryWait ) );
}
//----------------------------------------------------------------------------
// Bookkeeping after partial response has been received.
//----------------------------------------------------------------------------
void XRootDMsgHandler::PartialReceived()
{
pTimeoutFence.store( false, std::memory_order_relaxed ); // Take down the timeout fence
}
//----------------------------------------------------------------------------
// Unpack the message and call the response handler
//----------------------------------------------------------------------------
void XRootDMsgHandler::HandleResponse()
{
//--------------------------------------------------------------------------
// Is it a final response?
//--------------------------------------------------------------------------
bool finalrsp = !( pStatus.IsOK() && pStatus.code == suContinue );
if( finalrsp )
{
// Do not do final processing of the response if we haven't had
// confirmation the original request was sent (via OnStatusReady).
// The final processing will be triggered when we get the confirm.
const int sst = pSendingState.fetch_or( kFinalResp );
if( ( sst & kSawReadySend ) && !( sst & kSendDone ) )
return;
}
//--------------------------------------------------------------------------
// Process the response and notify the listener
//--------------------------------------------------------------------------
XRootDTransport::UnMarshallRequest( pRequest );
XRootDStatus *status = ProcessStatus();
AnyObject *response = 0;
Log *log = DefaultEnv::GetLog();
log->Debug( ExDbgMsg, "[%s] Calling MsgHandler: %p (message: %s ) "
"with status: %s.",
pUrl.GetHostId().c_str(), (void*)this,
pRequest->GetObfuscatedDescription().c_str(),
status->ToString().c_str() );
if( status->IsOK() )
{
Status st = ParseResponse( response );
if( !st.IsOK() )
{
delete status;
delete response;
status = new XRootDStatus( st );
response = 0;
}
}
//--------------------------------------------------------------------------
// Close the redirect entry if necessary
//--------------------------------------------------------------------------
if( pRdirEntry )
{
pRdirEntry->status = *status;
pRedirectTraceBack.push_back( std::move( pRdirEntry ) );
}
//--------------------------------------------------------------------------
// Release the stream id
//--------------------------------------------------------------------------
if( pSidMgr && finalrsp )
{
ClientRequest *req = (ClientRequest *)pRequest->GetBuffer();
if( status->IsOK() || !pMsgInFly ||
!( status->code == errOperationExpired || status->code == errOperationInterrupted ) )
pSidMgr->ReleaseSID( req->header.streamid );
}
HostList *hosts = pHosts.release();
if( !finalrsp )
pHosts.reset( new HostList( *hosts ) );
pResponseHandler->HandleResponseWithHosts( status, response, hosts );
//--------------------------------------------------------------------------
// if it is the final response there is nothing more to do ...
//--------------------------------------------------------------------------
if( finalrsp )
delete this;
//--------------------------------------------------------------------------
// on the other hand if it is not the final response, we have to keep the
// MsgHandler and delete the current response
//--------------------------------------------------------------------------
else
{
XrdSysCondVarHelper lck( pCV );
pResponse.reset();
pTimeoutFence.store( false, std::memory_order_relaxed );
pCV.Broadcast();
}
}
//----------------------------------------------------------------------------
// Extract the status information from the stuff that we got
//----------------------------------------------------------------------------
XRootDStatus *XRootDMsgHandler::ProcessStatus()
{
XRootDStatus *st = new XRootDStatus( pStatus );
ServerResponse *rsp = 0;
if( pResponse )
rsp = (ServerResponse *)pResponse->GetBuffer();
if( !pStatus.IsOK() && rsp )
{
if( pStatus.code == errErrorResponse )
{
st->errNo = rsp->body.error.errnum;
// omit the last character as the string returned from the server
// (acording to protocol specs) should be null-terminated
std::string errmsg( rsp->body.error.errmsg, rsp->hdr.dlen-5 );
if( st->errNo == kXR_noReplicas && !pLastError.IsOK() )
errmsg += " Last seen error: " + pLastError.ToString();
st->SetErrorMessage( errmsg );
}
else if( pStatus.code == errRedirect )
st->SetErrorMessage( pRedirectUrl );
}
return st;
}
//------------------------------------------------------------------------
// Parse the response and put it in an object that could be passed to
// the user
//------------------------------------------------------------------------
Status XRootDMsgHandler::ParseResponse( AnyObject *&response )
{
if( !pResponse )
return Status();
ServerResponse *rsp = (ServerResponse *)pResponse->GetBuffer();
ClientRequest *req = (ClientRequest *)pRequest->GetBuffer();
Log *log = DefaultEnv::GetLog();
//--------------------------------------------------------------------------
// Handle redirect as an answer
//--------------------------------------------------------------------------
if( rsp->hdr.status == kXR_redirect )
{
log->Error( XRootDMsg, "Internal Error: unable to process redirect" );
return 0;
}
Buffer buff;
uint32_t length = 0;
char *buffer = 0;
//--------------------------------------------------------------------------
// We don't have any partial answers so pass what we have
//--------------------------------------------------------------------------
if( pPartialResps.empty() )
{
buffer = rsp->body.buffer.data;
length = rsp->hdr.dlen;
}
//--------------------------------------------------------------------------
// Partial answers, we need to glue them together before parsing
//--------------------------------------------------------------------------
else if( req->header.requestid != kXR_read &&
req->header.requestid != kXR_readv )
{
for( uint32_t i = 0; i < pPartialResps.size(); ++i )
{
ServerResponse *part = (ServerResponse*)pPartialResps[i]->GetBuffer();
length += part->hdr.dlen;
}
length += rsp->hdr.dlen;
buff.Allocate( length );
uint32_t offset = 0;
for( uint32_t i = 0; i < pPartialResps.size(); ++i )
{
ServerResponse *part = (ServerResponse*)pPartialResps[i]->GetBuffer();
buff.Append( part->body.buffer.data, part->hdr.dlen, offset );
offset += part->hdr.dlen;
}
buff.Append( rsp->body.buffer.data, rsp->hdr.dlen, offset );
buffer = buff.GetBuffer();
}
//--------------------------------------------------------------------------
// Right, but what was the question?
//--------------------------------------------------------------------------
switch( req->header.requestid )
{
//------------------------------------------------------------------------
// kXR_mv, kXR_truncate, kXR_rm, kXR_mkdir, kXR_rmdir, kXR_chmod,
// kXR_ping, kXR_close, kXR_write, kXR_sync
//------------------------------------------------------------------------
case kXR_mv:
case kXR_truncate:
case kXR_rm:
case kXR_mkdir:
case kXR_rmdir:
case kXR_chmod:
case kXR_ping:
case kXR_close:
case kXR_write:
case kXR_writev:
case kXR_sync:
case kXR_chkpoint:
return Status();
//------------------------------------------------------------------------
// kXR_locate
//------------------------------------------------------------------------
case kXR_locate:
{
AnyObject *obj = new AnyObject();
char *nullBuffer = new char[length+1];
nullBuffer[length] = 0;
memcpy( nullBuffer, buffer, length );
log->Dump( XRootDMsg, "[%s] Parsing the response to %s as "
"LocateInfo: %s", pUrl.GetHostId().c_str(),
pRequest->GetObfuscatedDescription().c_str(), nullBuffer );
LocationInfo *data = new LocationInfo();
if( data->ParseServerResponse( nullBuffer ) == false )
{
delete obj;
delete data;
delete [] nullBuffer;
return Status( stError, errInvalidResponse );
}
delete [] nullBuffer;
obj->Set( data );
response = obj;
return Status();
}
//------------------------------------------------------------------------
// kXR_stat
//------------------------------------------------------------------------
case kXR_stat:
{
AnyObject *obj = new AnyObject();
//----------------------------------------------------------------------
// Virtual File System stat (kXR_vfs)
//----------------------------------------------------------------------
if( req->stat.options & kXR_vfs )
{
StatInfoVFS *data = new StatInfoVFS();
char *nullBuffer = new char[length+1];
nullBuffer[length] = 0;
memcpy( nullBuffer, buffer, length );
log->Dump( XRootDMsg, "[%s] Parsing the response to %s as "
"StatInfoVFS: %s", pUrl.GetHostId().c_str(),
pRequest->GetObfuscatedDescription().c_str(), nullBuffer );
if( data->ParseServerResponse( nullBuffer ) == false )
{
delete obj;
delete data;
delete [] nullBuffer;
return Status( stError, errInvalidResponse );
}
delete [] nullBuffer;
obj->Set( data );
}
//----------------------------------------------------------------------
// Normal stat
//----------------------------------------------------------------------
else
{
StatInfo *data = new StatInfo();
char *nullBuffer = new char[length+1];
nullBuffer[length] = 0;
memcpy( nullBuffer, buffer, length );
log->Dump( XRootDMsg, "[%s] Parsing the response to %s as StatInfo: "
"%s", pUrl.GetHostId().c_str(),
pRequest->GetObfuscatedDescription().c_str(), nullBuffer );
if( data->ParseServerResponse( nullBuffer ) == false )
{
delete obj;
delete data;
delete [] nullBuffer;
return Status( stError, errInvalidResponse );
}
delete [] nullBuffer;
obj->Set( data );
}
response = obj;
return Status();
}
//------------------------------------------------------------------------
// kXR_protocol
//------------------------------------------------------------------------
case kXR_protocol:
{
log->Dump( XRootDMsg, "[%s] Parsing the response to %s as ProtocolInfo",
pUrl.GetHostId().c_str(),
pRequest->GetObfuscatedDescription().c_str() );
if( rsp->hdr.dlen < 8 )
{
log->Error( XRootDMsg, "[%s] Got invalid redirect response.",
pUrl.GetHostId().c_str() );
return Status( stError, errInvalidResponse );
}
AnyObject *obj = new AnyObject();
ProtocolInfo *data = new ProtocolInfo( rsp->body.protocol.pval,
rsp->body.protocol.flags );
obj->Set( data );
response = obj;
return Status();
}
//------------------------------------------------------------------------
// kXR_dirlist
//------------------------------------------------------------------------
case kXR_dirlist:
{
AnyObject *obj = new AnyObject();
log->Dump( XRootDMsg, "[%s] Parsing the response to %s as "
"DirectoryList", pUrl.GetHostId().c_str(),
pRequest->GetObfuscatedDescription().c_str() );
char *path = new char[req->dirlist.dlen+1];
path[req->dirlist.dlen] = 0;
memcpy( path, pRequest->GetBuffer(24), req->dirlist.dlen );
DirectoryList *data = new DirectoryList();
data->SetParentName( path );
delete [] path;
char *nullBuffer = new char[length+1];
nullBuffer[length] = 0;
memcpy( nullBuffer, buffer, length );
bool invalidrsp = false;
if( !pDirListStarted )
{
pDirListWithStat = DirectoryList::HasStatInfo( nullBuffer );
pDirListStarted = true;
invalidrsp = !data->ParseServerResponse( pUrl.GetHostId(), nullBuffer );
}
else
invalidrsp = !data->ParseServerResponse( pUrl.GetHostId(), nullBuffer, pDirListWithStat );
if( invalidrsp )
{
delete data;
delete obj;
delete [] nullBuffer;
return Status( stError, errInvalidResponse );
}
delete [] nullBuffer;
obj->Set( data );
response = obj;
return Status();
}
//------------------------------------------------------------------------
// kXR_open - if we got the statistics, otherwise return 0
//------------------------------------------------------------------------
case kXR_open:
{
log->Dump( XRootDMsg, "[%s] Parsing the response to %s as OpenInfo",
pUrl.GetHostId().c_str(),
pRequest->GetObfuscatedDescription().c_str() );
if( rsp->hdr.dlen < 4 )
{
log->Error( XRootDMsg, "[%s] Got invalid open response.",
pUrl.GetHostId().c_str() );
return Status( stError, errInvalidResponse );
}
AnyObject *obj = new AnyObject();
StatInfo *statInfo = 0;
//----------------------------------------------------------------------
// Handle StatInfo if requested
//----------------------------------------------------------------------
if( req->open.options & kXR_retstat )
{
log->Dump( XRootDMsg, "[%s] Parsing StatInfo in response to %s",
pUrl.GetHostId().c_str(),
pRequest->GetObfuscatedDescription().c_str() );
if( rsp->hdr.dlen >= 12 )
{
char *nullBuffer = new char[rsp->hdr.dlen-11];
nullBuffer[rsp->hdr.dlen-12] = 0;
memcpy( nullBuffer, buffer+12, rsp->hdr.dlen-12 );
statInfo = new StatInfo();
if( statInfo->ParseServerResponse( nullBuffer ) == false )
{
delete statInfo;
statInfo = 0;
}
delete [] nullBuffer;
}
if( rsp->hdr.dlen < 12 || !statInfo )
{
log->Error( XRootDMsg, "[%s] Unable to parse StatInfo in response "
"to %s", pUrl.GetHostId().c_str(),
pRequest->GetObfuscatedDescription().c_str() );
delete obj;
return Status( stError, errInvalidResponse );
}
}
OpenInfo *data = new OpenInfo( (uint8_t*)buffer,
pResponse->GetSessionId(),
statInfo );
obj->Set( data );
response = obj;
return Status();
}
//------------------------------------------------------------------------
// kXR_read
//------------------------------------------------------------------------
case kXR_read:
{
log->Dump( XRootDMsg, "[%s] Parsing the response to %s as ChunkInfo",
pUrl.GetHostId().c_str(),
pRequest->GetObfuscatedDescription().c_str() );
for( uint32_t i = 0; i < pPartialResps.size(); ++i )
{
//--------------------------------------------------------------------
// we are expecting to have only the header in the message, the raw
// data have been readout into the user buffer
//--------------------------------------------------------------------
if( pPartialResps[i]->GetSize() > 8 )
return Status( stOK, errInternal );
}
//----------------------------------------------------------------------
// we are expecting to have only the header in the message, the raw
// data have been readout into the user buffer
//----------------------------------------------------------------------
if( pResponse->GetSize() > 8 )
return Status( stOK, errInternal );
//----------------------------------------------------------------------
// Get the response for the end user
//----------------------------------------------------------------------
return pBodyReader->GetResponse( response );
}
//------------------------------------------------------------------------
// kXR_pgread
//------------------------------------------------------------------------
case kXR_pgread:
{
log->Dump( XRootDMsg, "[%s] Parsing the response to %s as PageInfo",
pUrl.GetHostId().c_str(),
pRequest->GetObfuscatedDescription().c_str() );
//----------------------------------------------------------------------
// Glue in the cached responses if necessary
//----------------------------------------------------------------------
ChunkInfo chunk = pChunkList->front();
bool sizeMismatch = false;
uint32_t currentOffset = 0;
for( uint32_t i = 0; i < pPartialResps.size(); ++i )
{
ServerResponseV2 *part = (ServerResponseV2*)pPartialResps[i]->GetBuffer();
//--------------------------------------------------------------------
// the actual size of the raw data without the crc32c checksums
//--------------------------------------------------------------------
size_t datalen = part->status.bdy.dlen - NbPgPerRsp( part->info.pgread.offset,
part->status.bdy.dlen ) * CksumSize;
if( currentOffset + datalen > chunk.length )
{
sizeMismatch = true;
break;
}
currentOffset += datalen;
}
ServerResponseV2 *rspst = (ServerResponseV2*)pResponse->GetBuffer();
size_t datalen = rspst->status.bdy.dlen - NbPgPerRsp( rspst->info.pgread.offset,
rspst->status.bdy.dlen ) * CksumSize;
if( currentOffset + datalen <= chunk.length )
currentOffset += datalen;
else
sizeMismatch = true;
//----------------------------------------------------------------------
// Overflow
//----------------------------------------------------------------------
if( pChunkStatus.front().sizeError || sizeMismatch )
{
log->Error( XRootDMsg, "[%s] Handling response to %s: user supplied "
"buffer is too small for the received data.",
pUrl.GetHostId().c_str(),
pRequest->GetObfuscatedDescription().c_str() );
return Status( stError, errInvalidResponse );
}
AnyObject *obj = new AnyObject();
PageInfo *pgInfo = new PageInfo( chunk.offset, currentOffset, chunk.buffer,
std::move( pCrc32cDigests) );
obj->Set( pgInfo );
response = obj;
return Status();
}
//------------------------------------------------------------------------
// kXR_pgwrite
//------------------------------------------------------------------------
case kXR_pgwrite:
{
std::vector<std::tuple<uint64_t, uint32_t>> retries;
ServerResponseV2 *rsp = (ServerResponseV2*)pResponse->GetBuffer();
if( rsp->status.bdy.dlen > 0 )
{
ServerResponseBody_pgWrCSE *cse = (ServerResponseBody_pgWrCSE*)pResponse->GetBuffer( sizeof( ServerResponseV2 ) );
size_t pgcnt = ( rsp->status.bdy.dlen - 8 ) / sizeof( kXR_int64 );
retries.reserve( pgcnt );
kXR_int64 *pgoffs = (kXR_int64*)pResponse->GetBuffer( sizeof( ServerResponseV2 ) +
sizeof( ServerResponseBody_pgWrCSE ) );
for( size_t i = 0; i < pgcnt; ++i )
{
uint32_t len = XrdSys::PageSize;
if( i == 0 ) len = cse->dlFirst;
else if( i == pgcnt - 1 ) len = cse->dlLast;
retries.push_back( std::make_tuple( pgoffs[i], len ) );
}
}
RetryInfo *info = new RetryInfo( std::move( retries ) );
AnyObject *obj = new AnyObject();
obj->Set( info );
response = obj;
return Status();
}
//------------------------------------------------------------------------
// kXR_readv - we need to pass the length of the buffer to the user code
//------------------------------------------------------------------------
case kXR_readv:
{
log->Dump( XRootDMsg, "[%s] Parsing the response to %s as "
"VectorReadInfo", pUrl.GetHostId().c_str(),
pRequest->GetObfuscatedDescription().c_str() );
for( uint32_t i = 0; i < pPartialResps.size(); ++i )
{
//--------------------------------------------------------------------
// we are expecting to have only the header in the message, the raw
// data have been readout into the user buffer
//--------------------------------------------------------------------
if( pPartialResps[i]->GetSize() > 8 )
return Status( stOK, errInternal );
}
//----------------------------------------------------------------------
// we are expecting to have only the header in the message, the raw
// data have been readout into the user buffer
//----------------------------------------------------------------------
if( pResponse->GetSize() > 8 )
return Status( stOK, errInternal );
//----------------------------------------------------------------------
// Get the response for the end user
//----------------------------------------------------------------------
return pBodyReader->GetResponse( response );
}
//------------------------------------------------------------------------
// kXR_fattr
//------------------------------------------------------------------------
case kXR_fattr:
{
int len = rsp->hdr.dlen;
char* data = rsp->body.buffer.data;
return ParseXAttrResponse( data, len, response );
}
//------------------------------------------------------------------------
// kXR_query
//------------------------------------------------------------------------
case kXR_query:
case kXR_set:
case kXR_prepare:
default:
{
AnyObject *obj = new AnyObject();
log->Dump( XRootDMsg, "[%s] Parsing the response to %s as BinaryData",
pUrl.GetHostId().c_str(),
pRequest->GetObfuscatedDescription().c_str() );
BinaryDataInfo *data = new BinaryDataInfo();
data->Allocate( length );
data->Append( buffer, length );
obj->Set( data );
response = obj;
return Status();
}
};
return Status( stError, errInvalidMessage );
}
//------------------------------------------------------------------------
// Parse the response to kXR_fattr request and put it in an object that
// could be passed to the user
//------------------------------------------------------------------------
Status XRootDMsgHandler::ParseXAttrResponse( char *data, size_t len,
AnyObject *&response )
{
ClientRequest *req = (ClientRequest *)pRequest->GetBuffer();
// Log *log = DefaultEnv::GetLog(); //TODO
switch( req->fattr.subcode )
{
case kXR_fattrDel:
case kXR_fattrSet:
{
Status status;
kXR_char nerrs = 0;
if( !( status = ReadFromBuffer( data, len, nerrs ) ).IsOK() )
return status;
kXR_char nattr = 0;
if( !( status = ReadFromBuffer( data, len, nattr ) ).IsOK() )
return status;
std::vector<XAttrStatus> resp;
// read the namevec
for( kXR_char i = 0; i < nattr; ++i )
{
kXR_unt16 rc = 0;
if( !( status = ReadFromBuffer( data, len, rc ) ).IsOK() )
return status;
rc = ntohs( rc );
// count errors
if( rc ) --nerrs;
std::string name;
if( !( status = ReadFromBuffer( data, len, name ) ).IsOK() )
return status;
XRootDStatus st = rc ? XRootDStatus( stError, errErrorResponse, rc ) :
XRootDStatus();
resp.push_back( XAttrStatus( name, st ) );
}
// check if we read all the data and if the error count is OK
if( len != 0 || nerrs != 0 ) return Status( stError, errDataError );
// set up the response object
response = new AnyObject();
response->Set( new std::vector<XAttrStatus>( std::move( resp ) ) );
return Status();
}
case kXR_fattrGet:
{
Status status;
kXR_char nerrs = 0;
if( !( status = ReadFromBuffer( data, len, nerrs ) ).IsOK() )
return status;
kXR_char nattr = 0;
if( !( status = ReadFromBuffer( data, len, nattr ) ).IsOK() )
return status;
std::vector<XAttr> resp;
resp.reserve( nattr );
// read the name vec
for( kXR_char i = 0; i < nattr; ++i )
{
kXR_unt16 rc = 0;
if( !( status = ReadFromBuffer( data, len, rc ) ).IsOK() )
return status;
rc = ntohs( rc );
// count errors
if( rc ) --nerrs;
std::string name;
if( !( status = ReadFromBuffer( data, len, name ) ).IsOK() )
return status;
XRootDStatus st = rc ? XRootDStatus( stError, errErrorResponse, rc ) :
XRootDStatus();
resp.push_back( XAttr( name, st ) );
}
// read the value vec
for( kXR_char i = 0; i < nattr; ++i )
{
kXR_int32 vlen = 0;
if( !( status = ReadFromBuffer( data, len, vlen ) ).IsOK() )
return status;
vlen = ntohl( vlen );
std::string value;
if( !( status = ReadFromBuffer( data, len, vlen, value ) ).IsOK() )
return status;
resp[i].value.swap( value );
}
// check if we read all the data and if the error count is OK
if( len != 0 || nerrs != 0 ) return Status( stError, errDataError );
// set up the response object
response = new AnyObject();
response->Set( new std::vector<XAttr>( std::move( resp ) ) );
return Status();
}
case kXR_fattrList:
{
Status status;
std::vector<XAttr> resp;
while( len > 0 )
{
std::string name;
if( !( status = ReadFromBuffer( data, len, name ) ).IsOK() )
return status;
kXR_int32 vlen = 0;
if( !( status = ReadFromBuffer( data, len, vlen ) ).IsOK() )
return status;
vlen = ntohl( vlen );
std::string value;
if( !( status = ReadFromBuffer( data, len, vlen, value ) ).IsOK() )
return status;
resp.push_back( XAttr( name, value ) );
}
// set up the response object
response = new AnyObject();
response->Set( new std::vector<XAttr>( std::move( resp ) ) );
return Status();
}
default:
return Status( stError, errDataError );
}
}
//----------------------------------------------------------------------------
// Perform the changes to the original request needed by the redirect
// procedure - allocate new streamid, append redirection data and such
//----------------------------------------------------------------------------
Status XRootDMsgHandler::RewriteRequestRedirect( const URL &newUrl )
{
Log *log = DefaultEnv::GetLog();
Status st;
// Append any "xrd.*" parameters present in newCgi so that any authentication
// requirements are properly enforced
const URL::ParamsMap &newCgi = newUrl.GetParams();
std::string xrdCgi = "";
std::ostringstream ossXrd;
for(URL::ParamsMap::const_iterator it = newCgi.begin(); it != newCgi.end(); ++it )
{
if( it->first.compare( 0, 4, "xrd." ) )
continue;
ossXrd << it->first << '=' << it->second << '&';
}
xrdCgi = ossXrd.str();
// Redirection URL containing also any original xrd.* opaque parameters
XrdCl::URL authUrl;
if (xrdCgi.empty())
{
authUrl = newUrl;
}
else
{
std::string surl = newUrl.GetURL();
(surl.find('?') == std::string::npos) ? (surl += '?') :
((*surl.rbegin() != '&') ? (surl += '&') : (surl += ""));
surl += xrdCgi;
if (!authUrl.FromString(surl))
{
std::string surlLog = surl;
if( unlikely( log->GetLevel() >= Log::ErrorMsg ) ) {
surlLog = obfuscateAuth(surlLog);
}
log->Error( XRootDMsg, "[%s] Failed to build redirection URL from data: %s",
newUrl.GetHostId().c_str(), surl.c_str());
return Status(stError, errInvalidRedirectURL);
}
}
//--------------------------------------------------------------------------
// Rewrite particular requests
//--------------------------------------------------------------------------
XRootDTransport::UnMarshallRequest( pRequest );
MessageUtils::RewriteCGIAndPath( pRequest, newCgi, true, newUrl.GetPath() );
XRootDTransport::MarshallRequest( pRequest );
return Status();
}
//----------------------------------------------------------------------------
// Some requests need to be rewritten also after getting kXR_wait
//----------------------------------------------------------------------------
Status XRootDMsgHandler::RewriteRequestWait()
{
ClientRequest *req = (ClientRequest *)pRequest->GetBuffer();
XRootDTransport::UnMarshallRequest( pRequest );
//------------------------------------------------------------------------
// For kXR_locate and kXR_open request the kXR_refresh bit needs to be
// turned off after wait
//------------------------------------------------------------------------
switch( req->header.requestid )
{
case kXR_locate:
{
uint16_t refresh = kXR_refresh;
req->locate.options &= (~refresh);
break;
}
case kXR_open:
{
uint16_t refresh = kXR_refresh;
req->locate.options &= (~refresh);
break;
}
}
XRootDTransport::SetDescription( pRequest );
XRootDTransport::MarshallRequest( pRequest );
return Status();
}
//----------------------------------------------------------------------------
// Recover error
//----------------------------------------------------------------------------
void XRootDMsgHandler::HandleError( XRootDStatus status )
{
//--------------------------------------------------------------------------
// If there was no error then do nothing
//--------------------------------------------------------------------------
if( status.IsOK() )
return;
if( pSidMgr && pMsgInFly && (
status.code == errOperationExpired ||
status.code == errOperationInterrupted ) )
{
ClientRequest *req = (ClientRequest *)pRequest->GetBuffer();
pSidMgr->TimeOutSID( req->header.streamid );
}
bool noreplicas = ( status.code == errErrorResponse &&
status.errNo == kXR_noReplicas );
if( !noreplicas ) pLastError = status;
Log *log = DefaultEnv::GetLog();
log->Debug( XRootDMsg, "[%s] Handling error while processing %s: %s.",
pUrl.GetHostId().c_str(), pRequest->GetObfuscatedDescription().c_str(),
status.ToString().c_str() );
//--------------------------------------------------------------------------
// Check if it is a fatal TLS error that has been marked as potentially
// recoverable, if yes check if we can downgrade from fatal to error.
//--------------------------------------------------------------------------
if( status.IsFatal() && status.code == errTlsError && status.errNo == EAGAIN )
{
if( pSslErrCnt < MaxSslErrRetry )
{
status.status &= ~stFatal; // switch off fatal&error bits
status.status |= stError; // switch on error bit
}
++pSslErrCnt; // count number of consecutive SSL errors
}
else
pSslErrCnt = 0;
//--------------------------------------------------------------------------
// We have got an error message, we can recover it at the load balancer if:
// 1) we haven't got it from the load balancer
// 2) we have a load balancer assigned
// 3) the error is either one of: kXR_FSError, kXR_IOError, kXR_ServerError,
// kXR_NotFound
// 4) in the case of kXR_NotFound a kXR_refresh flags needs to be set
//--------------------------------------------------------------------------
if( status.code == errErrorResponse )
{
if( RetriableErrorResponse( status ) )
{
UpdateTriedCGI(status.errNo);
if( status.errNo == kXR_NotFound || status.errNo == kXR_Overloaded )
SwitchOnRefreshFlag();
HandleError( RetryAtServer( pLoadBalancer.url, RedirectEntry::EntryRetry ) );
return;
}
else
{
pStatus = status;
HandleRspOrQueue();
return;
}
}
//--------------------------------------------------------------------------
// Nothing can be done if:
// 1) a user timeout has occurred
// 2) has a non-zero session id
// 3) if another error occurred and the validity of the message expired
//--------------------------------------------------------------------------
if( status.code == errOperationExpired || pRequest->GetSessionId() ||
status.code == errOperationInterrupted || time(0) >= pExpiration )
{
log->Error( XRootDMsg, "[%s] Unable to get the response to request %s",
pUrl.GetHostId().c_str(),
pRequest->GetObfuscatedDescription().c_str() );
pStatus = status;
HandleRspOrQueue();
return;
}
//--------------------------------------------------------------------------
// At this point we're left with connection errors, we recover them
// at a load balancer if we have one and if not on the current server
// until we get a response, an unrecoverable error or a timeout
//--------------------------------------------------------------------------
if( pLoadBalancer.url.IsValid() &&
pLoadBalancer.url.GetLocation() != pUrl.GetLocation() )
{
UpdateTriedCGI( kXR_ServerError );
HandleError( RetryAtServer( pLoadBalancer.url, RedirectEntry::EntryRetry ) );
return;
}
else
{
if( !status.IsFatal() && IsRetriable() )
{
log->Info( XRootDMsg, "[%s] Retrying request: %s.",
pUrl.GetHostId().c_str(),
pRequest->GetObfuscatedDescription().c_str() );
UpdateTriedCGI( kXR_ServerError );
HandleError( RetryAtServer( pUrl, RedirectEntry::EntryRetry ) );
return;
}
pStatus = status;
HandleRspOrQueue();
return;
}
}
//----------------------------------------------------------------------------
// Retry the message at another server
//----------------------------------------------------------------------------
Status XRootDMsgHandler::RetryAtServer( const URL &url, RedirectEntry::Type entryType )
{
// prepare to possibly be requeued in the out-queue for a different channel,
// so reset sendingstate.
pSendingState = 0;
pResponse.reset();
Log *log = DefaultEnv::GetLog();
//--------------------------------------------------------------------------
// Set up a redirect entry
//--------------------------------------------------------------------------
if( pRdirEntry ) pRedirectTraceBack.push_back( std::move( pRdirEntry ) );
pRdirEntry.reset( new RedirectEntry( pUrl.GetLocation(), url.GetLocation(), entryType ) );
if( pUrl.GetLocation() != url.GetLocation() )
{
pHosts->push_back( url );
//------------------------------------------------------------------------
// Assign a new stream id to the message
//------------------------------------------------------------------------
// first release the old stream id
// (though it could be a redirect from a local
// metalink file, in this case there's no SID)
ClientRequestHdr *req = (ClientRequestHdr*)pRequest->GetBuffer();
if( pSidMgr )
{
pSidMgr->ReleaseSID( req->streamid );
pSidMgr.reset();
}
// then get the new SIDManager
// (again this could be a redirect to a local
// file and in this case there is no SID)
if( !url.IsLocalFile() )
{
pSidMgr = SIDMgrPool::Instance().GetSIDMgr( url );
Status st = pSidMgr->AllocateSID( req->streamid );
if( !st.IsOK() )
{
log->Error( XRootDMsg, "[%s] Impossible to send message %s.",
pUrl.GetHostId().c_str(),
pRequest->GetObfuscatedDescription().c_str() );
return st;
}
}
pUrl = url;
}
if( pUrl.IsMetalink() && pFollowMetalink )
{
log->Debug( ExDbgMsg, "[%s] Metaling redirection for MsgHandler: %p (message: %s ).",
pUrl.GetHostId().c_str(), (void*)this,
pRequest->GetObfuscatedDescription().c_str() );
return pPostMaster->Redirect( pUrl, pRequest, this );
}
else if( pUrl.IsLocalFile() )
{
HandleLocalRedirect( &pUrl );
return Status();
}
else
{
log->Debug( ExDbgMsg, "[%s] Retry at server MsgHandler: %p (message: %s ).",
pUrl.GetHostId().c_str(), (void*)this,
pRequest->GetObfuscatedDescription().c_str() );
return pPostMaster->Send( pUrl, pRequest, this, true, pExpiration );
}
}
//----------------------------------------------------------------------------
// Update the "tried=" part of the CGI of the current message
//----------------------------------------------------------------------------
void XRootDMsgHandler::UpdateTriedCGI(uint32_t errNo)
{
URL::ParamsMap cgi;
std::string tried;
//--------------------------------------------------------------------------
// In case a data server responded with a kXR_redirect and we fail at the
// node where we were redirected to, the original data server should be
// included in the tried CGI opaque info (instead of the current one).
//--------------------------------------------------------------------------
if( pEffectiveDataServerUrl )
{
tried = pEffectiveDataServerUrl->GetHostName();
delete pEffectiveDataServerUrl;
pEffectiveDataServerUrl = 0;
}
//--------------------------------------------------------------------------
// Otherwise use the current URL.
//--------------------------------------------------------------------------
else
tried = pUrl.GetHostName();
// Report the reason for the failure to the next location
//
if (errNo)
{ if (errNo == kXR_NotFound) cgi["triedrc"] = "enoent";
else if (errNo == kXR_IOError) cgi["triedrc"] = "ioerr";
else if (errNo == kXR_FSError) cgi["triedrc"] = "fserr";
else if (errNo == kXR_ServerError) cgi["triedrc"] = "srverr";
}
//--------------------------------------------------------------------------
// If our current load balancer is a metamanager and we failed either
// at a diskserver or at an unidentified node we also exclude the last
// known manager
//--------------------------------------------------------------------------
if( pLoadBalancer.url.IsValid() && (pLoadBalancer.flags & kXR_attrMeta) )
{
HostList::reverse_iterator it;
for( it = pHosts->rbegin()+1; it != pHosts->rend(); ++it )
{
if( it->loadBalancer )
break;
tried += "," + it->url.GetHostName();
if( it->flags & kXR_isManager )
break;
}
}
cgi["tried"] = tried;
XRootDTransport::UnMarshallRequest( pRequest );
MessageUtils::RewriteCGIAndPath( pRequest, cgi, false, "" );
XRootDTransport::MarshallRequest( pRequest );
}
//----------------------------------------------------------------------------
// Switch on the refresh flag for some requests
//----------------------------------------------------------------------------
void XRootDMsgHandler::SwitchOnRefreshFlag()
{
XRootDTransport::UnMarshallRequest( pRequest );
ClientRequest *req = (ClientRequest *)pRequest->GetBuffer();
switch( req->header.requestid )
{
case kXR_locate:
{
req->locate.options |= kXR_refresh;
break;
}
case kXR_open:
{
req->locate.options |= kXR_refresh;
break;
}
}
XRootDTransport::SetDescription( pRequest );
XRootDTransport::MarshallRequest( pRequest );
}
//------------------------------------------------------------------------
// If the current thread is a worker thread from our thread-pool
// handle the response, otherwise submit a new task to the thread-pool
//------------------------------------------------------------------------
void XRootDMsgHandler::HandleRspOrQueue()
{
//--------------------------------------------------------------------------
// Is it a final response?
//--------------------------------------------------------------------------
bool finalrsp = !( pStatus.IsOK() && pStatus.code == suContinue );
if( finalrsp )
{
// Do not do final processing of the response if we haven't had
// confirmation the original request was sent (via OnStatusReady).
// The final processing will be triggered when we get the confirm.
const int sst = pSendingState.fetch_or( kFinalResp );
if( ( sst & kSawReadySend ) && !( sst & kSendDone ) )
return;
}
JobManager *jobMgr = pPostMaster->GetJobManager();
if( jobMgr->IsWorker() )
HandleResponse();
else
{
Log *log = DefaultEnv::GetLog();
log->Debug( ExDbgMsg, "[%s] Passing to the thread-pool MsgHandler: %p (message: %s ).",
pUrl.GetHostId().c_str(), (void*)this,
pRequest->GetObfuscatedDescription().c_str() );
jobMgr->QueueJob( new HandleRspJob( this ), 0 );
}
}
//------------------------------------------------------------------------
// Notify the FileStateHandler to retry Open() with new URL
//------------------------------------------------------------------------
void XRootDMsgHandler::HandleLocalRedirect( URL *url )
{
Log *log = DefaultEnv::GetLog();
log->Debug( ExDbgMsg, "[%s] Handling local redirect - MsgHandler: %p (message: %s ).",
pUrl.GetHostId().c_str(), (void*)this,
pRequest->GetObfuscatedDescription().c_str() );
if( !pLFileHandler )
{
HandleError( XRootDStatus( stFatal, errNotSupported ) );
return;
}
AnyObject *resp = 0;
pLFileHandler->SetHostList( *pHosts );
XRootDStatus st = pLFileHandler->Open( url, pRequest, resp );
if( !st.IsOK() )
{
HandleError( st );
return;
}
pResponseHandler->HandleResponseWithHosts( new XRootDStatus(),
resp,
pHosts.release() );
delete this;
return;
}
//------------------------------------------------------------------------
// Check if it is OK to retry this request
//------------------------------------------------------------------------
bool XRootDMsgHandler::IsRetriable()
{
std::string value;
DefaultEnv::GetEnv()->GetString( "OpenRecovery", value );
if( value == "true" ) return true;
// check if it is a mutable open (open + truncate or open + create)
ClientRequest *req = reinterpret_cast<ClientRequest*>( pRequest->GetBuffer() );
if( req->header.requestid == htons( kXR_open ) )
{
bool _mutable = ( req->open.options & htons( kXR_delete ) ) ||
( req->open.options & htons( kXR_new ) );
if( _mutable )
{
Log *log = DefaultEnv::GetLog();
log->Debug( XRootDMsg,
"[%s] Not allowed to retry open request (OpenRecovery disabled): %s.",
pUrl.GetHostId().c_str(),
pRequest->GetObfuscatedDescription().c_str() );
// disallow retry if it is a mutable open
return false;
}
}
return true;
}
//------------------------------------------------------------------------
// Check if for given request and Metalink redirector it is OK to omit
// the kXR_wait and proceed straight to the next entry in the Metalink file
//------------------------------------------------------------------------
bool XRootDMsgHandler::OmitWait( Message &request, const URL &url )
{
// we can omit kXR_wait only if we have a Metalink redirector
if( !url.IsMetalink() )
return false;
// we can omit kXR_wait only for requests that can be redirected
// (kXR_read is the only stateful request that can be redirected)
ClientRequest *req = reinterpret_cast<ClientRequest*>( request.GetBuffer() );
if( pStateful && req->header.requestid != kXR_read )
return false;
// we can only omit kXR_wait if the Metalink redirect has more
// replicas
RedirectorRegistry &registry = RedirectorRegistry::Instance();
VirtualRedirector *redirector = registry.Get( url );
// we need more than one server as the current one is not reflected
// in tried CGI
if( redirector->Count( request ) > 1 )
return true;
return false;
}
//------------------------------------------------------------------------
// Checks if the given error returned by server is retriable.
//------------------------------------------------------------------------
bool XRootDMsgHandler::RetriableErrorResponse( const Status &status )
{
// we can only retry error response if we have a valid load-balancer and
// it is not our current URL
if( !( pLoadBalancer.url.IsValid() &&
pUrl.GetLocation() != pLoadBalancer.url.GetLocation() ) )
return false;
// following errors are retriable at any load-balancer
if( status.errNo == kXR_FSError || status.errNo == kXR_IOError ||
status.errNo == kXR_ServerError || status.errNo == kXR_NotFound ||
status.errNo == kXR_Overloaded || status.errNo == kXR_NoMemory )
return true;
// check if the load-balancer is a meta-manager, if yes there are
// more errors that can be recovered
if( !( pLoadBalancer.flags & kXR_attrMeta ) ) return false;
// those errors are retriable for meta-managers
if( status.errNo == kXR_Unsupported || status.errNo == kXR_FileLocked )
return true;
// in case of not-authorized error there is an imposed upper limit
// on how many times we can retry this error
if( status.errNo == kXR_NotAuthorized )
{
int limit = DefaultNotAuthorizedRetryLimit;
DefaultEnv::GetEnv()->GetInt( "NotAuthorizedRetryLimit", limit );
bool ret = pNotAuthorizedCounter < limit;
++pNotAuthorizedCounter;
if( !ret )
{
Log *log = DefaultEnv::GetLog();
log->Error( XRootDMsg,
"[%s] Reached limit of NotAuthorized retries!",
pUrl.GetHostId().c_str() );
}
return ret;
}
// check if the load-balancer is a virtual (metalink) redirector,
// if yes there are even more errors that can be recovered
if( !( pLoadBalancer.flags & kXR_attrVirtRdr ) ) return false;
// those errors are retriable for virtual (metalink) redirectors
if( status.errNo == kXR_noserver || status.errNo == kXR_ArgTooLong )
return true;
// otherwise it is a non-retriable error
return false;
}
//------------------------------------------------------------------------
// Dump the redirect-trace-back into the log file
//------------------------------------------------------------------------
void XRootDMsgHandler::DumpRedirectTraceBack()
{
if( pRedirectTraceBack.empty() ) return;
std::stringstream sstrm;
sstrm << "Redirect trace-back:\n";
int counter = 0;
auto itr = pRedirectTraceBack.begin();
sstrm << '\t' << counter << ". " << (*itr)->ToString() << '\n';
auto prev = itr;
++itr;
++counter;
for( ; itr != pRedirectTraceBack.end(); ++itr, ++prev, ++counter )
sstrm << '\t' << counter << ". "
<< (*itr)->ToString( (*prev)->status.IsOK() ) << '\n';
int authlimit = DefaultNotAuthorizedRetryLimit;
DefaultEnv::GetEnv()->GetInt( "NotAuthorizedRetryLimit", authlimit );
bool warn = !pStatus.IsOK() &&
( pStatus.code == errNotFound ||
pStatus.code == errRedirectLimit ||
( pStatus.code == errAuthFailed && pNotAuthorizedCounter >= authlimit ) );
Log *log = DefaultEnv::GetLog();
if( warn )
log->Warning( XRootDMsg, "%s", sstrm.str().c_str() );
else
log->Debug( XRootDMsg, "%s", sstrm.str().c_str() );
}
// Read data from buffer
//------------------------------------------------------------------------
template<typename T>
Status XRootDMsgHandler::ReadFromBuffer( char *&buffer, size_t &buflen, T& result )
{
if( sizeof( T ) > buflen ) return Status( stError, errDataError );
memcpy(&result, buffer, sizeof(T));
buffer += sizeof( T );
buflen -= sizeof( T );
return Status();
}
//------------------------------------------------------------------------
// Read a string from buffer
//------------------------------------------------------------------------
Status XRootDMsgHandler::ReadFromBuffer( char *&buffer, size_t &buflen, std::string &result )
{
Status status;
char c = 0;
while( true )
{
if( !( status = ReadFromBuffer( buffer, buflen, c ) ).IsOK() )
return status;
if( c == 0 ) break;
result += c;
}
return status;
}
//------------------------------------------------------------------------
// Read a string from buffer
//------------------------------------------------------------------------
Status XRootDMsgHandler::ReadFromBuffer( char *&buffer, size_t &buflen,
size_t size, std::string &result )
{
Status status;
if( size > buflen ) return Status( stError, errDataError );
result.append( buffer, size );
buffer += size;
buflen -= size;
return status;
}
}
kXR_NotAuthorized
@ kXR_NotAuthorized
Definition XProtocol.hh:1000
kXR_NotFound
@ kXR_NotFound
Definition XProtocol.hh:1001
kXR_FileLocked
@ kXR_FileLocked
Definition XProtocol.hh:993
kXR_noReplicas
@ kXR_noReplicas
Definition XProtocol.hh:1019
kXR_Unsupported
@ kXR_Unsupported
Definition XProtocol.hh:1003
kXR_ServerError
@ kXR_ServerError
Definition XProtocol.hh:1002
kXR_Overloaded
@ kXR_Overloaded
Definition XProtocol.hh:1014
kXR_ArgTooLong
@ kXR_ArgTooLong
Definition XProtocol.hh:992
kXR_noserver
@ kXR_noserver
Definition XProtocol.hh:1004
kXR_IOError
@ kXR_IOError
Definition XProtocol.hh:997
kXR_FSError
@ kXR_FSError
Definition XProtocol.hh:995
kXR_NoMemory
@ kXR_NoMemory
Definition XProtocol.hh:998
kXR_isManager
#define kXR_isManager
Definition XProtocol.hh:1156
kXR_fattrDel
@ kXR_fattrDel
Definition XProtocol.hh:270
kXR_fattrSet
@ kXR_fattrSet
Definition XProtocol.hh:273
kXR_fattrList
@ kXR_fattrList
Definition XProtocol.hh:272
kXR_fattrGet
@ kXR_fattrGet
Definition XProtocol.hh:271
ServerResponseBody_pgWrCSE::dlFirst
kXR_int16 dlFirst
Definition XProtocol.hh:1077
ClientRequest::fattr
struct ClientFattrRequest fattr
Definition XProtocol.hh:854
kXR_collapseRedir
#define kXR_collapseRedir
Definition XProtocol.hh:1167
ServerResponseHeader::status
kXR_unt16 status
Definition XProtocol.hh:915
ServerResponseV2::status
ServerResponseStatus status
Definition XProtocol.hh:1310
kXR_attrMeta
#define kXR_attrMeta
Definition XProtocol.hh:1159
ServerResponse::body
union ServerResponse::@040373375333017131300127053271011057331004327334 body
ClientRequestHdr::streamid
kXR_char streamid[2]
Definition XProtocol.hh:156
ServerResponseHeader::streamid
kXR_char streamid[2]
Definition XProtocol.hh:914
ClientOpenRequest::options
kXR_unt16 options
Definition XProtocol.hh:481
ClientRequest::dirlist
struct ClientDirlistRequest dirlist
Definition XProtocol.hh:852
kXR_ckpXeq
static const int kXR_ckpXeq
Definition XProtocol.hh:216
ClientChkPointRequest::opcode
kXR_char opcode
Definition XProtocol.hh:206
kXR_delete
@ kXR_delete
Definition XProtocol.hh:453
kXR_refresh
@ kXR_refresh
Definition XProtocol.hh:459
kXR_new
@ kXR_new
Definition XProtocol.hh:455
kXR_retstat
@ kXR_retstat
Definition XProtocol.hh:463
ServerResponseBody_pgWrCSE::dlLast
kXR_int16 dlLast
Definition XProtocol.hh:1078
ClientRequest::open
struct ClientOpenRequest open
Definition XProtocol.hh:860
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
ServerResponseStatus::bdy
struct ServerResponseBody_Status bdy
Definition XProtocol.hh:1262
ClientRequest::header
struct ClientRequestHdr header
Definition XProtocol.hh:846
kXR_recoverWrts
#define kXR_recoverWrts
Definition XProtocol.hh:1166
ServerResponseV2::info
union ServerResponseV2::@207342300141235315373173036347114307032363217365 info
ClientRequestHdr::requestid
kXR_unt16 requestid
Definition XProtocol.hh:157
kXR_read
@ kXR_read
Definition XProtocol.hh:125
kXR_open
@ kXR_open
Definition XProtocol.hh:122
kXR_writev
@ kXR_writev
Definition XProtocol.hh:143
kXR_readv
@ kXR_readv
Definition XProtocol.hh:137
kXR_mkdir
@ kXR_mkdir
Definition XProtocol.hh:120
kXR_sync
@ kXR_sync
Definition XProtocol.hh:128
kXR_chmod
@ kXR_chmod
Definition XProtocol.hh:114
kXR_dirlist
@ kXR_dirlist
Definition XProtocol.hh:116
kXR_fattr
@ kXR_fattr
Definition XProtocol.hh:132
kXR_rm
@ kXR_rm
Definition XProtocol.hh:126
kXR_query
@ kXR_query
Definition XProtocol.hh:113
kXR_write
@ kXR_write
Definition XProtocol.hh:131
kXR_set
@ kXR_set
Definition XProtocol.hh:130
kXR_rmdir
@ kXR_rmdir
Definition XProtocol.hh:127
kXR_truncate
@ kXR_truncate
Definition XProtocol.hh:140
kXR_protocol
@ kXR_protocol
Definition XProtocol.hh:118
kXR_mv
@ kXR_mv
Definition XProtocol.hh:121
kXR_ping
@ kXR_ping
Definition XProtocol.hh:123
kXR_stat
@ kXR_stat
Definition XProtocol.hh:129
kXR_pgread
@ kXR_pgread
Definition XProtocol.hh:142
kXR_chkpoint
@ kXR_chkpoint
Definition XProtocol.hh:124
kXR_locate
@ kXR_locate
Definition XProtocol.hh:139
kXR_close
@ kXR_close
Definition XProtocol.hh:115
kXR_pgwrite
@ kXR_pgwrite
Definition XProtocol.hh:138
kXR_prepare
@ kXR_prepare
Definition XProtocol.hh:133
ServerResponseBody_Status::dlen
kXR_int32 dlen
Definition XProtocol.hh:1241
ClientDirlistRequest::dlen
kXR_int32 dlen
Definition XProtocol.hh:249
kXR_isServer
#define kXR_isServer
Definition XProtocol.hh:1157
kXR_attrVirtRdr
#define kXR_attrVirtRdr
Definition XProtocol.hh:1162
ClientRequest::chkpoint
struct ClientChkPointRequest chkpoint
Definition XProtocol.hh:849
ServerResponseStatus::hdr
struct ServerResponseHeader hdr
Definition XProtocol.hh:1261
ClientLocateRequest::options
kXR_unt16 options
Definition XProtocol.hh:346
ServerResponseHeader::dlen
kXR_int32 dlen
Definition XProtocol.hh:916
kXR_PROTOCOLVERSION
#define kXR_PROTOCOLVERSION
Definition XProtocol.hh:70
kXR_vfs
@ kXR_vfs
Definition XProtocol.hh:763
ClientRequest::stat
struct ClientStatRequest stat
Definition XProtocol.hh:873
ClientStatRequest::options
kXR_char options
Definition XProtocol.hh:769
ClientRequest::protocol
struct ClientProtocolRequest protocol
Definition XProtocol.hh:865
kXR_ecRedir
#define kXR_ecRedir
Definition XProtocol.hh:1168
ClientRequest::locate
struct ClientLocateRequest locate
Definition XProtocol.hh:856
ServerResponseBody_Status::resptype
kXR_char resptype
Definition XProtocol.hh:1239
ServerResponse::hdr
ServerResponseHeader hdr
Definition XProtocol.hh:1288
ClientRequest
Definition XProtocol.hh:845
ServerResponseHeader
Definition XProtocol.hh:913
ServerResponse
Definition XProtocol.hh:1287
ServerResponseV2
Definition XProtocol.hh:1309
ServerResponseStatus
Definition XProtocol.hh:1260
ClientRequestHdr
Definition XProtocol.hh:155
kXR_int64
long long kXR_int64
Definition XPtypes.hh:98
kXR_int32
int kXR_int32
Definition XPtypes.hh:89
kXR_unt16
unsigned short kXR_unt16
Definition XPtypes.hh:67
kXR_char
unsigned char kXR_char
Definition XPtypes.hh:65
XrdClConstants.hh
XrdClDefaultEnv.hh
XrdClJobManager.hh
XrdClLocalFileHandler.hh
XrdClLog.hh
XrdClMessageUtils.hh
XrdClMessage.hh
XrdClOptimizers.hh
XrdClRedirectorRegistry.hh
XrdClSIDManager.hh
XrdClSocket.hh
XrdClTaskManager.hh
XrdClTls.hh
XrdClURL.hh
XrdClUtils.hh
XrdClXRootDMsgHandler.hh
XrdClXRootDTransport.hh
XrdOucCRC.hh
XrdOucPrivateUtils.hh
XrdSysAtomics.hh
XrdSysPlatform.hh
XrdSysPthread.hh
XrdCl::AnyObject
Definition XrdClAnyObject.hh:33
XrdCl::AnyObject::Get
void Get(Type &object)
Retrieve the object being held.
Definition XrdClAnyObject.hh:78
XrdCl::AsyncPageReader
Object for reading out data from the PgRead response.
Definition XrdClAsyncPageReader.hh:38
XrdCl::DefaultEnv::GetLog
static Log * GetLog()
Get default log.
Definition XrdClDefaultEnv.cc:493
XrdCl::DefaultEnv::GetEnv
static Env * GetEnv()
Get default client environment.
Definition XrdClDefaultEnv.cc:445
XrdCl::DirectoryList::HasStatInfo
static bool HasStatInfo(const char *data)
Returns true if data contain stat info.
Definition XrdClXRootDResponses.cc:594
XrdCl::Env::GetInt
bool GetInt(const std::string &key, int &value)
Definition XrdClEnv.cc:89
XrdCl::HandleRspJob::Run
virtual void Run(void *arg)
The job logic.
Definition XrdClXRootDMsgHandler.cc:97
XrdCl::HandleRspJob::~HandleRspJob
virtual ~HandleRspJob()
Definition XrdClXRootDMsgHandler.cc:92
XrdCl::HandleRspJob::HandleRspJob
HandleRspJob(XrdCl::XRootDMsgHandler *handler)
Definition XrdClXRootDMsgHandler.cc:87
XrdCl::Job
Interface for a job to be run by the job manager.
Definition XrdClJobManager.hh:34
XrdCl::Log
Handle diagnostics.
Definition XrdClLog.hh:101
XrdCl::Log::Error
void Error(uint64_t topic, const char *format,...)
Report an error.
Definition XrdClLog.cc:231
XrdCl::Log::Warning
void Warning(uint64_t topic, const char *format,...)
Report a warning.
Definition XrdClLog.cc:248
XrdCl::Log::Dump
void Dump(uint64_t topic, const char *format,...)
Print a dump message.
Definition XrdClLog.cc:299
XrdCl::Log::Debug
void Debug(uint64_t topic, const char *format,...)
Print a debug message.
Definition XrdClLog.cc:282
XrdCl::MessageUtils::RewriteCGIAndPath
static void RewriteCGIAndPath(Message *msg, const URL::ParamsMap &newCgi, bool replace, const std::string &newPath)
Append cgi to the one already present in the message.
Definition XrdClMessageUtils.cc:250
XrdCl::Message
The message representation used throughout the system.
Definition XrdClMessage.hh:32
XrdCl::Message::GetObfuscatedDescription
const std::string & GetObfuscatedDescription() const
Get the description of the message with authz parameter obfuscated.
Definition XrdClMessage.hh:106
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::More
@ More
there are more (non-raw) data to be read
Definition XrdClPostMasterInterfaces.hh:72
XrdCl::MsgHandler::Ignore
@ Ignore
Ignore the message.
Definition XrdClPostMasterInterfaces.hh:60
XrdCl::MsgHandler::Corrupted
@ Corrupted
Definition XrdClPostMasterInterfaces.hh:69
XrdCl::MsgHandler::StreamEvent
StreamEvent
Events that may have occurred to the stream.
Definition XrdClPostMasterInterfaces.hh:79
XrdCl::MsgHandler::Ready
@ Ready
The stream has become connected.
Definition XrdClPostMasterInterfaces.hh:80
XrdCl::Socket
A network socket.
Definition XrdClSocket.hh:43
XrdCl::Socket::Send
virtual XRootDStatus Send(const char *buffer, size_t size, int &bytesWritten)
Definition XrdClSocket.cc:461
XrdCl::Socket::IsEncrypted
bool IsEncrypted()
Definition XrdClSocket.cc:867
XrdCl::TaskManager
Definition XrdClTaskManager.hh:76
XrdCl::TaskManager::RegisterTask
void RegisterTask(Task *task, time_t time, bool own=true)
Definition XrdClTaskManager.cc:127
XrdCl::Task
Interface for a task to be run by the TaskManager.
Definition XrdClTaskManager.hh:36
XrdCl::URL
URL representation.
Definition XrdClURL.hh:31
XrdCl::URL::GetHostId
std::string GetHostId() const
Get the host part of the URL (user:password@host:port).
Definition XrdClURL.hh:99
XrdCl::URL::GetPassword
const std::string & GetPassword() const
Get the password.
Definition XrdClURL.hh:153
XrdCl::URL::ParamsMap
std::map< std::string, std::string > ParamsMap
Definition XrdClURL.hh:33
XrdCl::URL::FromString
bool FromString(const std::string &url)
Parse a string and fill the URL fields.
Definition XrdClURL.cc:62
XrdCl::URL::SetPassword
void SetPassword(const std::string &password)
Set the password.
Definition XrdClURL.hh:161
XrdCl::URL::SetParams
void SetParams(const std::string &params)
Set params.
Definition XrdClURL.cc:402
XrdCl::URL::GetUserName
const std::string & GetUserName() const
Get the username.
Definition XrdClURL.hh:135
XrdCl::URL::GetURL
std::string GetURL() const
Get the URL.
Definition XrdClURL.hh:86
XrdCl::URL::IsLocalFile
bool IsLocalFile() const
Definition XrdClURL.cc:474
XrdCl::URL::SetProtocol
void SetProtocol(const std::string &protocol)
Set protocol.
Definition XrdClURL.hh:126
XrdCl::URL::GetParams
const ParamsMap & GetParams() const
Get the URL params.
Definition XrdClURL.hh:244
XrdCl::URL::GetProtocol
const std::string & GetProtocol() const
Get the protocol.
Definition XrdClURL.hh:118
XrdCl::URL::IsValid
bool IsValid() const
Is the url valid.
Definition XrdClURL.cc:452
XrdCl::URL::SetUserName
void SetUserName(const std::string &userName)
Set the username.
Definition XrdClURL.hh:143
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::XRootDMsgHandler
Handle/Process/Forward XRootD messages.
Definition XrdClXRootDMsgHandler.hh:120
XrdCl::XRootDMsgHandler::InspectStatusRsp
virtual uint16_t InspectStatusRsp() override
Definition XrdClXRootDMsgHandler.cc:315
XrdCl::XRootDMsgHandler::OnStatusReady
virtual void OnStatusReady(const Message *message, XRootDStatus status) override
The requested action has been performed and the status is available.
Definition XrdClXRootDMsgHandler.cc:921
XrdCl::XRootDMsgHandler::GetRequest
const Message * GetRequest() const
Get the request pointer.
Definition XrdClXRootDMsgHandler.hh:363
XrdCl::XRootDMsgHandler::Examine
virtual uint16_t Examine(std::shared_ptr< Message > &msg) override
Definition XrdClXRootDMsgHandler.cc:109
XrdCl::XRootDMsgHandler::PartialReceived
void PartialReceived()
Definition XrdClXRootDMsgHandler.cc:1158
XrdCl::XRootDMsgHandler::WaitDone
void WaitDone(time_t now)
Definition XrdClXRootDMsgHandler.cc:1150
XrdCl::XRootDMsgHandler::Process
virtual void Process() override
Process the message if it was "taken" by the examine action.
Definition XrdClXRootDMsgHandler.cc:414
XrdCl::XRootDMsgHandler::ReadMessageBody
virtual XRootDStatus ReadMessageBody(Message *msg, Socket *socket, uint32_t &bytesRead) override
Definition XrdClXRootDMsgHandler.cc:902
XrdCl::XRootDMsgHandler::WriteMessageBody
XRootDStatus WriteMessageBody(Socket *socket, uint32_t &bytesWritten) override
Definition XrdClXRootDMsgHandler.cc:1001
XrdCl::XRootDMsgHandler::OnStreamEvent
virtual uint8_t OnStreamEvent(StreamEvent event, XRootDStatus status) override
Definition XrdClXRootDMsgHandler.cc:882
XrdCl::XRootDMsgHandler::GetSid
virtual uint16_t GetSid() const override
Definition XrdClXRootDMsgHandler.cc:405
XrdCl::XRootDMsgHandler::IsRaw
virtual bool IsRaw() const override
Are we a raw writer or not?
Definition XrdClXRootDMsgHandler.cc:981
XrdCl::XRootDStatus
Request status.
Definition XrdClXRootDResponses.hh:219
XrdCl::XRootDStatus::GetErrorMessage
const std::string & GetErrorMessage() const
Get error message.
Definition XrdClXRootDResponses.hh:242
XrdCl::XRootDTransport::UnMarshallBody
static XRootDStatus UnMarshallBody(Message *msg, uint16_t reqType)
Unmarshall the body of the incoming message.
Definition XrdClXRootDTransport.cc:1280
XrdCl::XRootDTransport::UnMarshallRequest
static XRootDStatus UnMarshallRequest(Message *msg)
Definition XrdClXRootDTransport.cc:1259
XrdCl::XRootDTransport::UnMarshalStatusBody
static XRootDStatus UnMarshalStatusBody(Message &msg, uint16_t reqType)
Unmarshall the body of the status response.
Definition XrdClXRootDTransport.cc:1358
XrdCl::XRootDTransport::MarshallRequest
static XRootDStatus MarshallRequest(Message *msg)
Marshal the outgoing message.
Definition XrdClXRootDTransport.hh:175
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
XrdSysCondVarHelper
Definition XrdSysPthread.hh:129
XrdCl
Definition XrdClAction.hh:34
XrdCl::suRetry
const uint16_t suRetry
Definition XrdClStatus.hh:40
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::errNotFound
const uint16_t errNotFound
Definition XrdClStatus.hh:100
XrdCl::XRootDMsg
const uint64_t XRootDMsg
Definition XrdClConstants.hh:39
XrdCl::HostList
std::vector< HostInfo > HostList
Definition XrdClXRootDResponses.hh:1121
XrdCl::errDataError
const uint16_t errDataError
data is corrupted
Definition XrdClStatus.hh:63
XrdCl::errInternal
const uint16_t errInternal
Internal error.
Definition XrdClStatus.hh:56
XrdCl::stOK
const uint16_t stOK
Everything went OK.
Definition XrdClStatus.hh:31
XrdCl::ExDbgMsg
const uint64_t ExDbgMsg
Definition XrdClConstants.hh:44
XrdCl::errInvalidResponse
const uint16_t errInvalidResponse
Definition XrdClStatus.hh:99
XrdCl::errInvalidRedirectURL
const uint16_t errInvalidRedirectURL
Definition XrdClStatus.hh:98
XrdCl::BinaryDataInfo
Buffer BinaryDataInfo
Binary buffer.
Definition XrdClXRootDResponses.hh:336
XrdCl::errOperationInterrupted
const uint16_t errOperationInterrupted
Definition XrdClStatus.hh:91
XrdCl::suContinue
const uint16_t suContinue
Definition XrdClStatus.hh:39
XrdCl::DefaultNotAuthorizedRetryLimit
const int DefaultNotAuthorizedRetryLimit
Definition XrdClConstants.hh:88
XrdCl::errRedirect
const uint16_t errRedirect
Definition XrdClStatus.hh:106
XrdCl::value
Response NullRef< Response >::value
Definition XrdClOperationHandlers.hh:128
XrdCl::errAuthFailed
const uint16_t errAuthFailed
Definition XrdClStatus.hh:88
XrdCl::errInvalidMessage
const uint16_t errInvalidMessage
Definition XrdClStatus.hh:85
XrdGlobal::Log
XrdSysError Log
Definition XrdConfig.cc:113
XrdProto::kXR_PartialResult
@ kXR_PartialResult
Definition XProtocol.hh:1251
XrdSys::PageSize
static const int PageSize
Definition XrdSysPageSize.hh:36
XrdSys::Move
ssize_t Move(KernelBuffer &kbuff, char *&ubuff)
Definition XrdSysKernelBuffer.hh:452
ClientFattrRequest::subcode
kXR_char subcode
Definition XProtocol.hh:289
XrdCl::ChunkInfo
Describe a data chunk for vector read.
Definition XrdClXRootDResponses.hh:918
XrdCl::ChunkInfo::buffer
void * buffer
length of the chunk
Definition XrdClXRootDResponses.hh:951
XrdCl::ChunkInfo::length
uint32_t length
offset in the file
Definition XrdClXRootDResponses.hh:950
XrdCl::ChunkInfo::offset
uint64_t offset
Definition XrdClXRootDResponses.hh:949
XrdCl::RedirectEntry::EntryRedirect
@ EntryRedirect
Definition XrdClXRootDMsgHandler.hh:74
XrdCl::RedirectEntry::EntryRedirectOnWait
@ EntryRedirectOnWait
Definition XrdClXRootDMsgHandler.hh:75
XrdCl::RedirectEntry::EntryWait
@ EntryWait
Definition XrdClXRootDMsgHandler.hh:77
XrdCl::Status
Procedure execution status.
Definition XrdClStatus.hh:115
XrdCl::Status::code
uint16_t code
Error type, or additional hints on what to do.
Definition XrdClStatus.hh:147
XrdCl::Status::IsOK
bool IsOK() const
We're fine.
Definition XrdClStatus.hh:124
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