Loading...
Searching...
No Matches
XrdClMetalinkRedirector.cc
Go to the documentation of this file.
/*
* XrdClMetalinkRedirector.cc
*
* Created on: May 2, 2016
* Author: simonm
*/
#include "XrdClMetalinkRedirector.hh"
#include "XrdCl/XrdClConstants.hh"
#include "XrdCl/XrdClDefaultEnv.hh"
#include "XrdCl/XrdClLog.hh"
#include "XrdCl/XrdClMessageUtils.hh"
#include "XrdCl/XrdClFile.hh"
#include "XrdCl/XrdClFileStateHandler.hh"
#include "XrdCl/XrdClURL.hh"
#include "XrdCl/XrdClJobManager.hh"
#include "XrdCl/XrdClUtils.hh"
#include "XrdCl/XrdClPostMasterInterfaces.hh"
#include "XrdCl/XrdClPostMaster.hh"
#include "XrdXml/XrdXmlMetaLink.hh"
#include "XProtocol/XProtocol.hh"
#include <arpa/inet.h>
#include <fcntl.h>
namespace XrdCl
{
void DeallocArgs( XRootDStatus *status, AnyObject *response,
HostList *hostList )
{
delete status;
delete response;
delete hostList;
}
//----------------------------------------------------------------------------
// Read metalink response handler.
//
// If the whole file has been read triggers parsing and
// initializes the Redirector, otherwise triggers another
// read.
//----------------------------------------------------------------------------
class MetalinkReadHandler: public ResponseHandler
{
public:
//----------------------------------------------------------------------------
// Constructor
//----------------------------------------------------------------------------
MetalinkReadHandler( MetalinkRedirector *mr, ResponseHandler *userHandler,
const std::string &content = "" ) :
pRedirector( mr ), pUserHandler( userHandler ), pBuffer(
new char[DefaultCPChunkSize] ), pContent( content )
{
}
//----------------------------------------------------------------------------
// Destructor
//----------------------------------------------------------------------------
virtual ~MetalinkReadHandler()
{
delete[] pBuffer;
}
//----------------------------------------------------------------------------
// Handle the response
//----------------------------------------------------------------------------
virtual void HandleResponse( XRootDStatus *status, AnyObject *response )
{
try
{
// check the status
if( !status->IsOK() )
throw status;
// we don't need it anymore
delete status;
// make sure we got a response
if( !response )
throw new XRootDStatus( stError, errInternal );
// make sure the response is not empty
ChunkInfo * info = 0;
response->Get( info );
if( !info )
throw new XRootDStatus( stError, errInternal );
uint32_t bytesRead = info->length;
uint64_t offset = info->offset + bytesRead;
pContent += std::string( pBuffer, bytesRead );
// are we done ?
if( bytesRead > 0 )
{
// lets try to read another chunk
MetalinkReadHandler * mrh = new MetalinkReadHandler( pRedirector,
pUserHandler, pContent );
XRootDStatus st = pRedirector->pFile->Read( offset,
DefaultCPChunkSize, mrh->GetBuffer(), mrh );
if( !st.IsOK() )
{
delete mrh;
throw new XRootDStatus( st );
}
}
else // we have the whole metalink file
{
// we don't need the File object anymore
delete pRedirector->pFile;
pRedirector->pFile = 0;
// now we can parse the metalink file
XRootDStatus st = pRedirector->Parse( pContent );
// now with the redirector fully initialized we can handle pending requests
pRedirector->FinalizeInitialization();
// we are done, pass the status to the user (whatever it is)
if( pUserHandler )
pUserHandler->HandleResponse( new XRootDStatus( st ), 0 );
}
// clean up
delete response;
}
catch( XRootDStatus *status )
{
pRedirector->FinalizeInitialization( *status );
// if we were not able to read from the metalink,
// propagate the error to the user handler
if( pUserHandler )
pUserHandler->HandleResponse( status, 0 );
else
DeallocArgs( status, response, 0 );
}
delete this;
}
//----------------------------------------------------------------------------
// Get the receive-buffer
//----------------------------------------------------------------------------
char * GetBuffer()
{
return pBuffer;
}
private:
MetalinkRedirector *pRedirector;
ResponseHandler *pUserHandler;
char *pBuffer;
std::string pContent;
};
//----------------------------------------------------------------------------
// Open metalink response handler.
//
// After successful open trrigers a read.
//----------------------------------------------------------------------------
class MetalinkOpenHandler: public ResponseHandler
{
public:
//----------------------------------------------------------------------------
// Constructor
//----------------------------------------------------------------------------
MetalinkOpenHandler( MetalinkRedirector *mr,
ResponseHandler *userHandler ) :
pRedirector( mr ), pUserHandler( userHandler )
{
}
//----------------------------------------------------------------------------
// Handle the response
//----------------------------------------------------------------------------
virtual void HandleResponseWithHosts( XRootDStatus *status,
AnyObject *response, HostList *hostList )
{
try
{
if( status->IsOK() )
{
delete status;
// download the content
MetalinkReadHandler *mrh = new MetalinkReadHandler( pRedirector,
pUserHandler );
XRootDStatus st = pRedirector->pFile->Read( 0, DefaultCPChunkSize,
mrh->GetBuffer(), mrh );
if( !st.IsOK() )
{
delete mrh;
throw new XRootDStatus( stError, errInternal );
} else
{
delete response;
delete hostList;
}
} else
throw status;
} catch( XRootDStatus *status )
{
pRedirector->FinalizeInitialization( *status );
// if we were not able to schedule a read
// pass an error to the user handler
if( pUserHandler )
pUserHandler->HandleResponseWithHosts( status, response, hostList );
else
DeallocArgs( status, response, hostList );
}
delete this;
}
private:
MetalinkRedirector *pRedirector;
ResponseHandler *pUserHandler;
};
//----------------------------------------------------------------------------
// Constructor
//----------------------------------------------------------------------------
MetalinkRedirector::MetalinkRedirector( const std::string & url ) :
pUrl( url ), pFile( new File( File::DisableVirtRedirect ) ), pReady(
false ), pFileSize( -1 )
{
}
//----------------------------------------------------------------------------
// Destructor
//----------------------------------------------------------------------------
MetalinkRedirector::~MetalinkRedirector()
{
delete pFile;
}
//----------------------------------------------------------------------------
// Initializes the object with the content of the metalink file
//----------------------------------------------------------------------------
XRootDStatus MetalinkRedirector::Load( ResponseHandler *userHandler )
{
MetalinkOpenHandler *handler = new MetalinkOpenHandler( this, userHandler );
XRootDStatus st = pFile->Open( pUrl, OpenFlags::Read, Access::None, handler,
0 );
if( !st.IsOK() )
delete handler;
return st;
}
//----------------------------------------------------------------------------
// Parses the metalink file
//----------------------------------------------------------------------------
XRootDStatus MetalinkRedirector::Parse( const std::string &metalink )
{
Log *log = DefaultEnv::GetLog();
Env *env = DefaultEnv::GetEnv();
std::string glfnRedirector;
env->GetString( "GlfnRedirector", glfnRedirector );
// parse the metalink
XrdXmlMetaLink parser( "root:xroot:roots:xroots:file:", "xroot:",
glfnRedirector.empty() ? 0 : glfnRedirector.c_str() );
int size = 0;
XrdOucFileInfo **fileInfos = parser.ConvertAll( metalink.c_str(), size,
metalink.size() );
if( !fileInfos )
{
int ecode;
const char * etxt = parser.GetStatus( ecode );
log->Error( UtilityMsg,
"Failed to parse the metalink file: %s (error code: %d)", etxt,
ecode );
return XRootDStatus( stError, errDataError, 0,
"Malformed or corrupted metalink file." );
}
// we are expecting just one file per metalink (as agreed with RUCIO)
if( size != 1 )
{
log->Error( UtilityMsg, "Expected only one file per metalink." );
return XRootDStatus( stError, errDataError );
}
InitCksum( fileInfos );
InitReplicas( fileInfos );
pTarget = fileInfos[0]->GetTargetName();
pFileSize = fileInfos[0]->GetSize();
XrdXmlMetaLink::DeleteAll( fileInfos, size );
return XRootDStatus();
}
//----------------------------------------------------------------------------
// Finalize the initialization process:
// - mark as ready
// - setup the status
// - and handle pending redirects
//----------------------------------------------------------------------------
void MetalinkRedirector::FinalizeInitialization( const XRootDStatus &status )
{
XrdSysMutexHelper scopedLock( pMutex );
pReady = true;
pStatus = status;
// Handle pending redirects (those that were
// submitted before the metalink has been loaded)
while( !pPendingRedirects.empty() )
{
const Message *msg = pPendingRedirects.front().first;
MsgHandler *handler = pPendingRedirects.front().second;
pPendingRedirects.pop_front();
if( !handler || !msg )
continue;
HandleRequestImpl( msg, handler );
}
}
//----------------------------------------------------------------------------
// Generates redirect response for the given request
//----------------------------------------------------------------------------
std::shared_ptr<Message> MetalinkRedirector::GetResponse( const Message *msg ) const
{
if( !pStatus.IsOK() )
return GetErrorMsg( msg, "Could not load the Metalink file.",
static_cast<XErrorCode>( XProtocol::mapError( pStatus.errNo ) ) );
const ClientRequestHdr *req =
reinterpret_cast<const ClientRequestHdr*>( msg->GetBuffer() );
// get the redirect location
std::string replica;
if( !GetReplica( *msg, replica ).IsOK() )
return GetErrorMsg( msg, "Metalink: no more replicas to try.", kXR_noReplicas );
auto resp = std::make_shared<Message>( sizeof(ServerResponse) );
ServerResponse* response =
reinterpret_cast<ServerResponse*>( resp->GetBuffer() );
response->hdr.status = kXR_redirect;
response->hdr.streamid[0] = req->streamid[0];
response->hdr.streamid[1] = req->streamid[1];
response->hdr.dlen = 4 + replica.size(); // 4 bytes are reserved for port number
response->body.redirect.port = -1; // this indicates that the full URL will be given in the 'host' field
memcpy( response->body.redirect.host, replica.c_str(), replica.size() );
return resp;
}
//----------------------------------------------------------------------------
// Generates error response for the given request
//----------------------------------------------------------------------------
std::shared_ptr<Message> MetalinkRedirector::GetErrorMsg( const Message *msg,
const std::string &errMsg, XErrorCode code ) const
{
const ClientRequestHdr *req =
reinterpret_cast<const ClientRequestHdr*>( msg->GetBuffer() );
auto resp = std::make_shared<Message>( sizeof(ServerResponse) );
ServerResponse* response =
reinterpret_cast<ServerResponse*>( resp->GetBuffer() );
response->hdr.status = kXR_error;
response->hdr.streamid[0] = req->streamid[0];
response->hdr.streamid[1] = req->streamid[1];
response->hdr.dlen = 4 + errMsg.size();
response->body.error.errnum = htonl( code );
memcpy( response->body.error.errmsg, errMsg.c_str(), errMsg.size() );
return resp;
}
//----------------------------------------------------------------------------
// Creates an instant redirect response for the given message
// or an error response if there are no more replicas to try.
// The virtual response is being handled by the given handler.
//----------------------------------------------------------------------------
XRootDStatus MetalinkRedirector::HandleRequestImpl( const Message *msg,
MsgHandler *handler )
{
auto resp = GetResponse( msg );
JobManager *jobMan = DefaultEnv::GetPostMaster()->GetJobManager();
RedirectJob *job = new RedirectJob( handler, std::move( resp ) );
jobMan->QueueJob( job );
return XRootDStatus();
}
//----------------------------------------------------------------------------
// If the MetalinkRedirector is initialized creates an instant
// redirect response, otherwise queues the request until initialization
// is done.
//----------------------------------------------------------------------------
XRootDStatus MetalinkRedirector::HandleRequest( const Message *msg,
MsgHandler *handler )
{
XrdSysMutexHelper scopedLck( pMutex );
// if the metalink data haven't been loaded yet, make it pending
if( !pReady )
{
pPendingRedirects.push_back(
std::make_pair( msg, handler ) );;
return XRootDStatus();
}
// otherwise generate a virtual response
return HandleRequestImpl( msg, handler );
}
//----------------------------------------------------------------------------
//----------------------------------------------------------------------------
int MetalinkRedirector::Count( Message &req ) const
{
ReplicaList::const_iterator itr = GetReplica( req );
return pReplicas.end() - itr;
}
//----------------------------------------------------------------------------
// Gets the file checksum if specified in the metalink
//----------------------------------------------------------------------------
void MetalinkRedirector::InitCksum( XrdOucFileInfo **fileInfos )
{
const char *chvalue = 0, *chtype = 0;
while( ( chtype = fileInfos[0]->GetDigest( chvalue ) ) )
{
pChecksums[chtype] = chvalue;
}
}
//----------------------------------------------------------------------------
// Initializes replica list
//----------------------------------------------------------------------------
void MetalinkRedirector::InitReplicas( XrdOucFileInfo **fileInfos )
{
URL replica;
const char *url = 0;
while( ( url = fileInfos[0]->GetUrl() ) )
{
replica = URL( url );
if( !replica.IsValid() || replica.GetURL().size() > 4096 )
continue; // this is the internal limit (defined in the protocol)
pReplicas.push_back( replica.GetURL() );
}
}
//----------------------------------------------------------------------------
//----------------------------------------------------------------------------
XRootDStatus MetalinkRedirector::GetReplica( const Message &msg,
std::string &replica ) const
{
ReplicaList::const_iterator itr = GetReplica( msg );
if( itr == pReplicas.end() )
return XRootDStatus( stError, errNotFound );
XrdCl::Env *env = XrdCl::DefaultEnv::GetEnv();
int tlsmtl = DefaultTlsMetalink;
env->GetInt( "TlsMetalink", tlsmtl );
URL url( *itr );
if( tlsmtl && ( url.GetProtocol() == "root" || url.GetProtocol() == "xroot" ) )
url.SetProtocol( "roots" );
replica = url.GetURL();
return XRootDStatus();
}
//----------------------------------------------------------------------------
//----------------------------------------------------------------------------
MetalinkRedirector::ReplicaList::const_iterator
MetalinkRedirector::GetReplica( const Message &msg ) const
{
if( pReplicas.empty() )
return pReplicas.cend();
std::string tried;
if( !GetCgiInfo( msg, "tried", tried ).IsOK() )
return pReplicas.cbegin();
ReplicaList triedList;
Utils::splitString( triedList, tried, "," );
std::set<std::string> triedSet( triedList.begin(), triedList.end() );
ReplicaList::const_iterator itr = pReplicas.begin();
for( ; itr != pReplicas.end(); ++itr )
{
URL url( *itr );
if( !triedSet.count( url.GetHostName() ) ) break;
}
return itr;
}
//----------------------------------------------------------------------------
//----------------------------------------------------------------------------
XRootDStatus MetalinkRedirector::GetCgiInfo( const Message &msg,
const std::string &key, std::string &value ) const
{
const ClientRequestHdr *req =
reinterpret_cast<const ClientRequestHdr*>( msg.GetBuffer() );
kXR_int32 dlen = msg.IsMarshalled() ? ntohl( req->dlen ) : req->dlen;
std::string url( msg.GetBuffer( 24 ), dlen );
size_t pos = url.find( '?' );
if( pos == std::string::npos )
return XRootDStatus( stError );
size_t start = url.find( key, pos );
if( start == std::string::npos )
return XRootDStatus( stError );
start += key.size() + 1; // the +1 stands for the '=' sign that is not part of the key
size_t end = url.find( '&', start );
if( end == std::string::npos )
end = url.size();
value = url.substr( start, end - start );
return XRootDStatus();
}
} /* namespace XrdCl */
XProtocol.hh
XErrorCode
XErrorCode
Definition XProtocol.hh:989
kXR_noReplicas
@ kXR_noReplicas
Definition XProtocol.hh:1019
ServerResponseHeader::status
kXR_unt16 status
Definition XProtocol.hh:915
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
kXR_redirect
@ kXR_redirect
Definition XProtocol.hh:904
kXR_error
@ kXR_error
Definition XProtocol.hh:903
ServerResponseHeader::dlen
kXR_int32 dlen
Definition XProtocol.hh:916
ClientRequestHdr::dlen
kXR_int32 dlen
Definition XProtocol.hh:159
ServerResponse::hdr
ServerResponseHeader hdr
Definition XProtocol.hh:1288
kXR_int32
int kXR_int32
Definition XPtypes.hh:89
XrdClConstants.hh
XrdClDefaultEnv.hh
XrdClFileStateHandler.hh
XrdClFile.hh
XrdClJobManager.hh
XrdClLog.hh
XrdClMessageUtils.hh
XrdClMetalinkRedirector.hh
XrdClPostMasterInterfaces.hh
XrdClPostMaster.hh
XrdClURL.hh
XrdClUtils.hh
XrdXmlMetaLink.hh
XProtocol::mapError
static int mapError(int rc)
Definition XProtocol.hh:1362
XrdCl::AnyObject
Definition XrdClAnyObject.hh:33
XrdCl::AnyObject::Get
void Get(Type &object)
Retrieve the object being held.
Definition XrdClAnyObject.hh:78
XrdCl::DefaultEnv::GetLog
static Log * GetLog()
Get default log.
Definition XrdClDefaultEnv.cc:493
XrdCl::DefaultEnv::GetPostMaster
static PostMaster * GetPostMaster()
Get default post master.
Definition XrdClDefaultEnv.cc:453
XrdCl::DefaultEnv::GetEnv
static Env * GetEnv()
Get default client environment.
Definition XrdClDefaultEnv.cc:445
XrdCl::Env
Definition XrdClEnv.hh:38
XrdCl::Env::GetString
bool GetString(const std::string &key, std::string &value)
Definition XrdClEnv.cc:31
XrdCl::Env::GetInt
bool GetInt(const std::string &key, int &value)
Definition XrdClEnv.cc:89
XrdCl::File
A file.
Definition XrdClFile.hh:46
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::Message
The message representation used throughout the system.
Definition XrdClMessage.hh:32
XrdCl::MetalinkOpenHandler::MetalinkOpenHandler
MetalinkOpenHandler(MetalinkRedirector *mr, ResponseHandler *userHandler)
Definition XrdClMetalinkRedirector.cc:162
XrdCl::MetalinkOpenHandler::HandleResponseWithHosts
virtual void HandleResponseWithHosts(XRootDStatus *status, AnyObject *response, HostList *hostList)
Definition XrdClMetalinkRedirector.cc:171
XrdCl::MetalinkReadHandler
Definition XrdClMetalinkRedirector.cc:48
XrdCl::MetalinkReadHandler::GetBuffer
char * GetBuffer()
Definition XrdClMetalinkRedirector.cc:138
XrdCl::MetalinkReadHandler::~MetalinkReadHandler
virtual ~MetalinkReadHandler()
Definition XrdClMetalinkRedirector.cc:63
XrdCl::MetalinkReadHandler::MetalinkReadHandler
MetalinkReadHandler(MetalinkRedirector *mr, ResponseHandler *userHandler, const std::string &content="")
Definition XrdClMetalinkRedirector.cc:53
XrdCl::MetalinkReadHandler::HandleResponse
virtual void HandleResponse(XRootDStatus *status, AnyObject *response)
Definition XrdClMetalinkRedirector.cc:71
XrdCl::MetalinkRedirector
Definition XrdClMetalinkRedirector.hh:32
XrdCl::MetalinkRedirector::MetalinkOpenHandler
friend class MetalinkOpenHandler
Definition XrdClMetalinkRedirector.hh:33
XrdCl::MetalinkRedirector::Load
XRootDStatus Load(ResponseHandler *userHandler)
Definition XrdClMetalinkRedirector.cc:235
XrdCl::MetalinkRedirector::~MetalinkRedirector
virtual ~MetalinkRedirector()
Destructor.
Definition XrdClMetalinkRedirector.cc:227
XrdCl::MetalinkRedirector::HandleRequest
XRootDStatus HandleRequest(const Message *msg, MsgHandler *handler)
Definition XrdClMetalinkRedirector.cc:381
XrdCl::MetalinkRedirector::MetalinkRedirector
MetalinkRedirector(const std::string &url)
Definition XrdClMetalinkRedirector.cc:218
XrdCl::MetalinkRedirector::Count
virtual int Count(Message &req) const
Count how many replicas do we have left to try for given request.
Definition XrdClMetalinkRedirector.cc:399
XrdCl::MsgHandler
Message handler.
Definition XrdClPostMasterInterfaces.hh:51
XrdCl::PostMaster::GetJobManager
JobManager * GetJobManager()
Get the job manager object user by the post master.
Definition XrdClPostMaster.cc:316
XrdCl::ResponseHandler
Handle an async response.
Definition XrdClXRootDResponses.hh:1127
XrdCl::Utils::splitString
static void splitString(Container &result, const std::string &input, const std::string &delimiter)
Split a string.
Definition XrdClUtils.hh:56
XrdCl::XRootDStatus
Request status.
Definition XrdClXRootDResponses.hh:219
XrdOucFileInfo
Definition XrdOucFileInfo.hh:46
XrdOucFileInfo::GetSize
long long GetSize()
Definition XrdOucFileInfo.hh:135
XrdOucFileInfo::GetTargetName
const char * GetTargetName()
Definition XrdOucFileInfo.hh:127
XrdSysMutexHelper
Definition XrdSysPthread.hh:263
XrdXmlMetaLink
Definition XrdXmlMetaLink.hh:47
XrdXmlMetaLink::DeleteAll
static void DeleteAll(XrdOucFileInfo **vecp, int vecn)
Definition XrdXmlMetaLink.cc:277
XrdCl
Definition XrdClAction.hh:34
XrdCl::DefaultCPChunkSize
const int DefaultCPChunkSize
Definition XrdClConstants.hh:61
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::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::DefaultTlsMetalink
const int DefaultTlsMetalink
Definition XrdClConstants.hh:92
XrdCl::UtilityMsg
const uint64_t UtilityMsg
Definition XrdClConstants.hh:33
XrdCl::DeallocArgs
void DeallocArgs(XRootDStatus *status, AnyObject *response, HostList *hostList)
Definition XrdClMetalinkRedirector.cc:32
XrdCl::value
Response NullRef< Response >::value
Definition XrdClOperationHandlers.hh:128
XrdCl::Access::None
@ None
Definition XrdClFileSystem.hh:123
XrdCl::ChunkInfo
Describe a data chunk for vector read.
Definition XrdClXRootDResponses.hh:918
XrdCl::ChunkInfo::length
uint32_t length
offset in the file
Definition XrdClXRootDResponses.hh:950
XrdCl::ChunkInfo::offset
uint64_t offset
Definition XrdClXRootDResponses.hh:949
XrdCl::OpenFlags::Read
@ Read
Open only for reading.
Definition XrdClFileSystem.hh:95
XrdCl::Status::IsOK
bool IsOK() const
We're fine.
Definition XrdClStatus.hh:124