XrdCl::XRootDTransport Class Reference

XRootD transport handler. More...

#include <XrdClXRootDTransport.hh>

Inheritance diagram for XrdCl::XRootDTransport:

Collaboration diagram for XrdCl::XRootDTransport:

Public Member Functions
	XRootDTransport ()
	Constructor.
	~XRootDTransport ()
	Destructor.
virtual void	DecFileInstCnt (AnyObject &channelData)
	Decrement file object instance count bound to this channel.
virtual void	Disconnect (AnyObject &channelData, uint16_t subStreamId)
	The stream has been disconnected, do the cleanups.
virtual void	FinalizeChannel (AnyObject &channelData)
	Finalize channel.
virtual URL	GetBindPreference (const URL &url, AnyObject &channelData)
	Get bind preference for the next data stream.
virtual XRootDStatus	GetBody (Message &message, Socket *socket)
virtual XRootDStatus	GetHeader (Message &message, Socket *socket)
virtual XRootDStatus	GetMore (Message &message, Socket *socket)
virtual Status	GetSignature (Message *toSign, Message *&sign, AnyObject &channelData)
	Get signature for given message.
virtual Status	GetSignature (Message *toSign, Message *&sign, XRootDChannelInfo *info)
	Get signature for given message.
virtual XRootDStatus	HandShake (HandShakeData *handShakeData, AnyObject &channelData)
	HandShake.
virtual bool	HandShakeDone (HandShakeData *handShakeData, AnyObject &channelData)
virtual void	InitializeChannel (const URL &url, AnyObject &channelData)
	Initialize channel.
virtual Status	IsStreamBroken (time_t inactiveTime, AnyObject &channelData)
virtual bool	IsStreamTTLElapsed (time_t time, AnyObject &channelData)
	Check if the stream should be disconnected.
virtual uint32_t	MessageReceived (Message &msg, uint16_t subStream, AnyObject &channelData)
	Check if the message invokes a stream action.
virtual void	MessageSent (Message *msg, uint16_t subStream, uint32_t bytesSent, AnyObject &channelData)
	Notify the transport about a message having been sent.
virtual PathID	Multiplex (Message *msg, AnyObject &channelData, PathID *hint=0)
virtual PathID	MultiplexSubStream (Message *msg, AnyObject &channelData, PathID *hint=0)
virtual bool	NeedControlConnection ()
virtual bool	NeedEncryption (HandShakeData *handShakeData, AnyObject &channelData)
virtual Status	Query (uint16_t query, AnyObject &result, AnyObject &channelData)
	Query the channel.
virtual uint16_t	SubStreamNumber (AnyObject &channelData)
	Return a number of substreams per stream that should be created.
virtual void	WaitBeforeExit ()
	Wait until the program can safely exit.
Public Member Functions inherited from XrdCl::TransportHandler
virtual	~TransportHandler ()

Static Public Member Functions
static void	GenerateDescription (char *msg, std::ostringstream &o)
	Get the description of a message.
static void	LogErrorResponse (const Message &msg)
	Log server error response.
static XRootDStatus	MarshallRequest (char *msg)
	Marshal the outgoing message.
static XRootDStatus	MarshallRequest (Message *msg)
	Marshal the outgoing message.
static uint16_t	NbConnectedStrm (AnyObject &channelData)
	Number of currently connected data streams.
static void	SetDescription (Message *msg)
	Get the description of a message.
static XRootDStatus	UnMarchalStatusMore (Message &msg)
	Unmarshall the correction-segment of the status response for pgwrite.
static XRootDStatus	UnMarshallBody (Message *msg, uint16_t reqType)
	Unmarshall the body of the incoming message.
static void	UnMarshallHeader (Message &msg)
	Unmarshall the header incoming message.
static XRootDStatus	UnMarshallRequest (Message *msg)
static XRootDStatus	UnMarshalStatusBody (Message &msg, uint16_t reqType)
	Unmarshall the body of the status response.

Friends
struct	PluginUnloadHandler

Additional Inherited Members
Public Types inherited from XrdCl::TransportHandler
enum	StreamAction {   NoAction = 0x0000 ,   DigestMsg = 0x0001 ,   AbortStream = 0x0002 ,   CloseStream = 0x0004 ,   ResumeStream = 0x0008 ,   HoldStream = 0x0010 ,   RequestClose = 0x0020 }
	Stream actions that may be triggered by incoming control messages. More...

Detailed Description

XRootD transport handler.

Definition at line 47 of file XrdClXRootDTransport.hh.

Constructor & Destructor Documentation

XRootDTransport()

XrdCl::XRootDTransport::XRootDTransport

(

)

Constructor.

Definition at line 290 of file XrdClXRootDTransport.cc.

                                  :
    pSecUnloadHandler( new PluginUnloadHandler() )
  {
  }

References PluginUnloadHandler.

Here is the call graph for this function:

~XRootDTransport()

XrdCl::XRootDTransport::~XRootDTransport

(

)

Destructor.

Definition at line 298 of file XrdClXRootDTransport.cc.

  {
    delete pSecUnloadHandler; pSecUnloadHandler = 0;
  }

Member Function Documentation

DecFileInstCnt()

void XrdCl::XRootDTransport::DecFileInstCnt ( AnyObject & channelData )

virtual

Decrement file object instance count bound to this channel.

Implements XrdCl::TransportHandler.

Definition at line 1824 of file XrdClXRootDTransport.cc.

  {
    XRootDChannelInfo *info = 0;
    channelData.Get( info );
    if( info->finstcnt.load( std::memory_order_relaxed ) > 0 )
      info->finstcnt.fetch_sub( 1, std::memory_order_relaxed );
  }

References XrdCl::XRootDChannelInfo::finstcnt, and XrdCl::AnyObject::Get().

Here is the call graph for this function:

Disconnect()

void XrdCl::XRootDTransport::Disconnect ( AnyObject & channelData, uint16_t subStreamId )

virtual

The stream has been disconnected, do the cleanups.

Implements XrdCl::TransportHandler.

Definition at line 1554 of file XrdClXRootDTransport.cc.

  {
    XRootDChannelInfo *info = 0;
    channelData.Get( info );
 
    if (!info) {
      DefaultEnv::GetLog()->Error(XRootDTransportMsg, "Internal error: no channel info");
      return;
    }
 
    XrdSysMutexHelper scopedLock( info->mutex );
 
    CleanUpProtection( info );
 
    if( !info->stream.empty() )
    {
      XRootDStreamInfo &sInfo = info->stream[subStreamId];
      sInfo.status = XRootDStreamInfo::Disconnected;
    }
 
    if( subStreamId == 0 )
    {
      info->sidManager->ReleaseAllTimedOut();
      info->sentOpens.clear();
      info->sentCloses.clear();
      info->openFiles   = 0;
      info->waitBarrier = 0;
    }
  }

References XrdCl::XRootDStreamInfo::Disconnected, XrdCl::Log::Error(), XrdCl::AnyObject::Get(), XrdCl::DefaultEnv::GetLog(), XrdCl::XRootDChannelInfo::mutex, XrdCl::XRootDChannelInfo::openFiles, XrdCl::XRootDChannelInfo::sentCloses, XrdCl::XRootDChannelInfo::sentOpens, XrdCl::XRootDChannelInfo::sidManager, XrdCl::XRootDStreamInfo::status, XrdCl::XRootDChannelInfo::stream, XrdCl::XRootDChannelInfo::waitBarrier, and XrdCl::XRootDTransportMsg.

Here is the call graph for this function:

FinalizeChannel()

void XrdCl::XRootDTransport::FinalizeChannel ( AnyObject & channelData )

virtual

Finalize channel.

Implements XrdCl::TransportHandler.

Definition at line 470 of file XrdClXRootDTransport.cc.

  {
  }

GenerateDescription()

void XrdCl::XRootDTransport::GenerateDescription ( char * msg, std::ostringstream & o )

static

Get the description of a message.

Definition at line 3003 of file XrdClXRootDTransport.cc.

  {
    Log *log = DefaultEnv::GetLog();
    if( log->GetLevel() < Log::ErrorMsg )
      return;
 
    ClientRequestHdr *req = (ClientRequestHdr *)msg;
    switch( req->requestid )
    {
      //------------------------------------------------------------------------
      // kXR_open
      //------------------------------------------------------------------------
      case kXR_open:
      {
        ClientOpenRequest *sreq = (ClientOpenRequest *)msg;
        o << "kXR_open (";
        char *fn = GetDataAsString( msg );
        o << "file: " << fn << ", ";
        delete [] fn;
        o << "mode: 0" << std::setbase(8) << sreq->mode << ", ";
        o << std::setbase(10);
        o << "flags: ";
        if( sreq->options == 0 )
          o << "none";
        else
        {
          if( sreq->options & kXR_compress )
            o << "kXR_compress ";
          if( sreq->options & kXR_delete )
            o << "kXR_delete ";
          if( sreq->options & kXR_force )
            o << "kXR_force ";
          if( sreq->options & kXR_mkpath )
            o << "kXR_mkpath ";
          if( sreq->options & kXR_new )
            o << "kXR_new ";
          if( sreq->options & kXR_nowait )
            o << "kXR_nowait ";
          if( sreq->options & kXR_open_apnd )
            o << "kXR_open_apnd ";
          if( sreq->options & kXR_open_read )
            o << "kXR_open_read ";
          if( sreq->options & kXR_open_updt )
            o << "kXR_open_updt ";
          if( sreq->options & kXR_open_wrto )
            o << "kXR_open_wrto ";
          if( sreq->options & kXR_posc )
            o << "kXR_posc ";
          if( sreq->options & kXR_prefname )
            o << "kXR_prefname ";
          if( sreq->options & kXR_refresh )
            o << "kXR_refresh ";
          if( sreq->options & kXR_4dirlist )
            o << "kXR_4dirlist ";
          if( sreq->options & kXR_replica )
            o << "kXR_replica ";
          if( sreq->options & kXR_seqio )
            o << "kXR_seqio ";
          if( sreq->options & kXR_async )
            o << "kXR_async ";
          if( sreq->options & kXR_retstat )
            o << "kXR_retstat ";
        }
        o << ")";
        break;
      }
 
      //------------------------------------------------------------------------
      // kXR_close
      //------------------------------------------------------------------------
      case kXR_close:
      {
        ClientCloseRequest *sreq = (ClientCloseRequest *)msg;
        o << "kXR_close (";
        o << "handle: " << FileHandleToStr( sreq->fhandle );
        o << ")";
        break;
      }
 
      //------------------------------------------------------------------------
      // kXR_stat
      //------------------------------------------------------------------------
      case kXR_stat:
      {
        ClientStatRequest *sreq = (ClientStatRequest *)msg;
        o << "kXR_stat (";
        if( sreq->dlen )
        {
          char *fn = GetDataAsString( msg );;
          o << "path: " << fn << ", ";
          delete [] fn;
        }
        else
        {
          o << "handle: " << FileHandleToStr( sreq->fhandle );
          o << ", ";
        }
        o << "flags: ";
        if( sreq->options == 0 )
          o << "none";
        else
        {
          if( sreq->options & kXR_vfs )
            o << "kXR_vfs";
        }
        o << ")";
        break;
      }
 
      //------------------------------------------------------------------------
      // kXR_read
      //------------------------------------------------------------------------
      case kXR_read:
      {
        ClientReadRequest *sreq = (ClientReadRequest *)msg;
        o << "kXR_read (";
        o << "handle: " << FileHandleToStr( sreq->fhandle );
        o << std::setbase(10);
        o << ", ";
        o << "offset: " << sreq->offset << ", ";
        o << "size: " << sreq->rlen << ")";
        break;
      }
 
      //------------------------------------------------------------------------
      // kXR_pgread
      //------------------------------------------------------------------------
      case kXR_pgread:
      {
        ClientPgReadRequest *sreq = (ClientPgReadRequest *)msg;
        o << "kXR_pgread (";
        o << "handle: " << FileHandleToStr( sreq->fhandle );
        o << std::setbase(10);
        o << ", ";
        o << "offset: " << sreq->offset << ", ";
        o << "size: " << sreq->rlen << ")";
        break;
      }
 
      //------------------------------------------------------------------------
      // kXR_write
      //------------------------------------------------------------------------
      case kXR_write:
      {
        ClientWriteRequest *sreq = (ClientWriteRequest *)msg;
        o << "kXR_write (";
        o << "handle: " << FileHandleToStr( sreq->fhandle );
        o << std::setbase(10);
        o << ", ";
        o << "offset: " << sreq->offset << ", ";
        o << "size: " << sreq->dlen << ")";
        break;
      }
 
      //------------------------------------------------------------------------
      // kXR_pgwrite
      //------------------------------------------------------------------------
      case kXR_pgwrite:
      {
        ClientPgWriteRequest *sreq = (ClientPgWriteRequest *)msg;
        o << "kXR_pgwrite (";
        o << "handle: " << FileHandleToStr( sreq->fhandle );
        o << std::setbase(10);
        o << ", ";
        o << "offset: " << sreq->offset << ", ";
        o << "size: " << sreq->dlen << ")";
        break;
      }
 
      //------------------------------------------------------------------------
      // kXR_fattr
      //------------------------------------------------------------------------
      case kXR_fattr:
      {
        ClientFattrRequest *sreq = (ClientFattrRequest *)msg;
        int nattr = sreq->numattr;
        int options = sreq->options;
        o << "kXR_fattr";
        switch (sreq->subcode) {
          case kXR_fattrGet:
            o << "Get";
            break;
          case kXR_fattrSet:
            o << "Set";
            break;
          case kXR_fattrList:
            o << "List";
            break;
          case kXR_fattrDel:
            o << "Delete";
            break;
          default:
            o << " unknown subcode: " << sreq->subcode;
            break;
        }
        o << " (handle: " << FileHandleToStr( sreq->fhandle );
        o << std::setbase(10);
        if (nattr)
          o << ", numattr: " << nattr;
        if (options) {
          o << ", options: ";
          if (options & 0x01)
            o << "new";
          if (options & 0x10)
            o << "list values";
        }
        o << ", total size: " << req->dlen << ")";
        break;
      }
 
      //------------------------------------------------------------------------
      // kXR_sync
      //------------------------------------------------------------------------
      case kXR_sync:
      {
        ClientSyncRequest *sreq = (ClientSyncRequest *)msg;
        o << "kXR_sync (";
        o << "handle: " << FileHandleToStr( sreq->fhandle );
        o << ")";
        break;
      }
 
      //------------------------------------------------------------------------
      // kXR_truncate
      //------------------------------------------------------------------------
      case kXR_truncate:
      {
        ClientTruncateRequest *sreq = (ClientTruncateRequest *)msg;
        o << "kXR_truncate (";
        if( !sreq->dlen )
          o << "handle: " << FileHandleToStr( sreq->fhandle );
        else
        {
          char *fn = GetDataAsString( msg );
          o << "file: " << fn;
          delete [] fn;
        }
        o << std::setbase(10);
        o << ", ";
        o << "offset: " << sreq->offset;
        o << ")";
        break;
      }
 
      //------------------------------------------------------------------------
      // kXR_readv
      //------------------------------------------------------------------------
      case kXR_readv:
      {
        unsigned char *fhandle = 0;
        o << "kXR_readv (";
 
        o << "handle: ";
        readahead_list *dataChunk = (readahead_list*)(msg + 24 );
        fhandle = dataChunk[0].fhandle;
        if( fhandle )
          o << FileHandleToStr( fhandle );
        else
          o << "unknown";
        o << ", ";
        o << std::setbase(10);
        o << "chunks: [";
        uint64_t size      = 0;
        for( size_t i = 0; i < req->dlen/sizeof(readahead_list); ++i )
        {
          size += dataChunk[i].rlen;
          o << "(offset: " << dataChunk[i].offset;
          o << ", size: " << dataChunk[i].rlen << "); ";
        }
        o << "], ";
        o << "total size: " << size << ")";
        break;
      }
 
      //------------------------------------------------------------------------
      // kXR_writev
      //------------------------------------------------------------------------
      case kXR_writev:
      {
        unsigned char *fhandle = 0;
        o << "kXR_writev (";
 
        XrdProto::write_list *wrtList =
            reinterpret_cast<XrdProto::write_list*>( msg + 24 );
        uint64_t size      = 0;
        uint32_t numChunks = 0;
        for( size_t i = 0; i < req->dlen/sizeof(XrdProto::write_list); ++i )
        {
          fhandle = wrtList[i].fhandle;
          size   += wrtList[i].wlen;
          ++numChunks;
        }
        o << "handle: ";
        if( fhandle )
          o << FileHandleToStr( fhandle );
        else
          o << "unknown";
        o << ", ";
        o << std::setbase(10);
        o << "chunks: " << numChunks << ", ";
        o << "total size: " << size << ")";
        break;
      }
 
      //------------------------------------------------------------------------
      // kXR_locate
      //------------------------------------------------------------------------
      case kXR_locate:
      {
        ClientLocateRequest *sreq = (ClientLocateRequest *)msg;
        char *fn = GetDataAsString( msg );;
        o << "kXR_locate (";
        o << "path: " << fn << ", ";
        delete [] fn;
        o << "flags: ";
        if( sreq->options == 0 )
          o << "none";
        else
        {
          if( sreq->options & kXR_refresh )
            o << "kXR_refresh ";
          if( sreq->options & kXR_prefname )
            o << "kXR_prefname ";
          if( sreq->options & kXR_nowait )
            o << "kXR_nowait ";
          if( sreq->options & kXR_force )
            o << "kXR_force ";
          if( sreq->options & kXR_compress )
            o << "kXR_compress ";
        }
        o << ")";
        break;
      }
 
      //------------------------------------------------------------------------
      // kXR_mv
      //------------------------------------------------------------------------
      case kXR_mv:
      {
        ClientMvRequest *sreq = (ClientMvRequest *)msg;
        o << "kXR_mv (";
        o << "source: ";
        o.write( msg + sizeof( ClientMvRequest ), sreq->arg1len );
        o << ", ";
        o << "destination: ";
        o.write( msg + sizeof( ClientMvRequest ) + sreq->arg1len + 1, sreq->dlen - sreq->arg1len - 1 );
        o << ")";
        break;
      }
 
      //------------------------------------------------------------------------
      // kXR_query
      //------------------------------------------------------------------------
      case kXR_query:
      {
        ClientQueryRequest *sreq = (ClientQueryRequest *)msg;
        o << "kXR_query (";
        o << "code: ";
        switch( sreq->infotype )
        {
          case kXR_Qconfig: o << "kXR_Qconfig"; break;
          case kXR_Qckscan: o << "kXR_Qckscan"; break;
          case kXR_Qcksum:  o << "kXR_Qcksum"; break;
          case kXR_Qopaque: o << "kXR_Qopaque"; break;
          case kXR_Qopaquf: o << "kXR_Qopaquf"; break;
          case kXR_Qopaqug: o << "kXR_Qopaqug"; break;
          case kXR_QPrep:   o << "kXR_QPrep"; break;
          case kXR_Qspace:  o << "kXR_Qspace"; break;
          case kXR_QStats:  o << "kXR_QStats"; break;
          case kXR_Qvisa:   o << "kXR_Qvisa"; break;
          case kXR_Qxattr:  o << "kXR_Qxattr"; break;
          default: o << sreq->infotype; break;
        }
        o << ", ";          
 
        if( sreq->infotype == kXR_Qopaqug || sreq->infotype == kXR_Qvisa )
        {
          o << "handle: " << FileHandleToStr( sreq->fhandle );
          o << ", ";
        }
 
        o << "arg length: " << sreq->dlen << ")";
        break;
      }
 
      //------------------------------------------------------------------------
      // kXR_rm
      //------------------------------------------------------------------------
      case kXR_rm:
      {
        o << "kXR_rm (";
        char *fn = GetDataAsString( msg );;
        o << "path: " << fn << ")";
        delete [] fn;
        break;
      }
 
      //------------------------------------------------------------------------
      // kXR_mkdir
      //------------------------------------------------------------------------
      case kXR_mkdir:
      {
        ClientMkdirRequest *sreq = (ClientMkdirRequest *)msg;
        o << "kXR_mkdir (";
        char *fn = GetDataAsString( msg );
        o << "path: " << fn << ", ";
        delete [] fn;
        o << "mode: 0" << std::setbase(8) << sreq->mode << ", ";
        o << std::setbase(10);
        o << "flags: ";
        if( sreq->options[0] == 0 )
          o << "none";
        else
        {
          if( sreq->options[0] & kXR_mkdirpath )
            o << "kXR_mkdirpath";
        }
        o << ")";
        break;
      }
 
      //------------------------------------------------------------------------
      // kXR_rmdir
      //------------------------------------------------------------------------
      case kXR_rmdir:
      {
        o << "kXR_rmdir (";
        char *fn = GetDataAsString( msg );
        o << "path: " << fn << ")";
        delete [] fn;
        break;
      }
 
      //------------------------------------------------------------------------
      // kXR_chmod
      //------------------------------------------------------------------------
      case kXR_chmod:
      {
        ClientChmodRequest *sreq = (ClientChmodRequest *)msg;
        o << "kXR_chmod (";
        char *fn = GetDataAsString( msg );
        o << "path: " << fn << ", ";
        delete [] fn;
        o << "mode: 0" << std::setbase(8) << sreq->mode << ")";
        break;
      }
 
      //------------------------------------------------------------------------
      // kXR_ping
      //------------------------------------------------------------------------
      case kXR_ping:
      {
        o << "kXR_ping ()";
        break;
      }
 
      //------------------------------------------------------------------------
      // kXR_protocol
      //------------------------------------------------------------------------
      case kXR_protocol:
      {
        ClientProtocolRequest *sreq = (ClientProtocolRequest *)msg;
        o << "kXR_protocol (";
        o << "clientpv: 0x" << std::setbase(16) << sreq->clientpv << ")";
        break;
      }
 
      //------------------------------------------------------------------------
      // kXR_dirlist
      //------------------------------------------------------------------------
      case kXR_dirlist:
      {
        o << "kXR_dirlist (";
        char *fn = GetDataAsString( msg );;
        o << "path: " << fn << ")";
        delete [] fn;
        break;
      }
 
      //------------------------------------------------------------------------
      // kXR_set
      //------------------------------------------------------------------------
      case kXR_set:
      {
        o << "kXR_set (";
        char *fn = GetDataAsString( msg );;
        o << "data: " << fn << ")";
        delete [] fn;
        break;
      }
 
      //------------------------------------------------------------------------
      // kXR_prepare
      //------------------------------------------------------------------------
      case kXR_prepare:
      {
        ClientPrepareRequest *sreq = (ClientPrepareRequest *)msg;
        o << "kXR_prepare (";
        o << "flags: ";
 
        if( sreq->options == 0 )
          o << "none";
        else
        {
          if( sreq->options & kXR_stage )
            o << "kXR_stage ";
          if( sreq->options & kXR_wmode )
            o << "kXR_wmode ";
          if( sreq->options & kXR_coloc )
            o << "kXR_coloc ";
          if( sreq->options & kXR_fresh )
            o << "kXR_fresh ";
        }
 
        o << ", priority: " << (int) sreq->prty << ", ";
 
        char *fn = GetDataAsString( msg );
        char *cursor;
        for( cursor = fn; *cursor; ++cursor )
          if( *cursor == '\n' ) *cursor = ' ';
 
        o << "paths: " << fn << ")";
        delete [] fn;
        break;
      }
 
      case kXR_chkpoint:
      {
        ClientChkPointRequest *sreq = (ClientChkPointRequest*)msg;
        o << "kXR_chkpoint (";
        o << "opcode: ";
        if( sreq->opcode == kXR_ckpBegin )         o << "kXR_ckpBegin)";
        else if( sreq->opcode == kXR_ckpCommit )   o << "kXR_ckpCommit)";
        else if( sreq->opcode == kXR_ckpQuery )    o << "kXR_ckpQuery)";
        else if( sreq->opcode == kXR_ckpRollback ) o << "kXR_ckpRollback)";
        else if( sreq->opcode == kXR_ckpXeq )
        {
          o << "kXR_ckpXeq) ";
          // In this case our request body will be one of kXR_pgwrite,
          // kXR_truncate, kXR_write, or kXR_writev request.
          GenerateDescription( msg + sizeof( ClientChkPointRequest ), o );
        }
 
        break;
      }
 
      //------------------------------------------------------------------------
      // Default
      //------------------------------------------------------------------------
      default:
      {
        o << "kXR_unknown (length: " << req->dlen << ")";
        break;
      }
    };
  }

References ClientMvRequest::arg1len, ClientProtocolRequest::clientpv, ClientMvRequest::dlen, ClientPgWriteRequest::dlen, ClientQueryRequest::dlen, ClientRequestHdr::dlen, ClientStatRequest::dlen, ClientTruncateRequest::dlen, ClientWriteRequest::dlen, XrdCl::Log::ErrorMsg, ClientCloseRequest::fhandle, ClientFattrRequest::fhandle, ClientPgReadRequest::fhandle, ClientPgWriteRequest::fhandle, ClientQueryRequest::fhandle, ClientReadRequest::fhandle, ClientStatRequest::fhandle, ClientSyncRequest::fhandle, ClientTruncateRequest::fhandle, ClientWriteRequest::fhandle, readahead_list::fhandle, XrdProto::write_list::fhandle, GenerateDescription(), XrdCl::Log::GetLevel(), XrdCl::DefaultEnv::GetLog(), ClientQueryRequest::infotype, kXR_4dirlist, kXR_async, kXR_chkpoint, kXR_chmod, kXR_ckpBegin, kXR_ckpCommit, kXR_ckpQuery, kXR_ckpRollback, kXR_ckpXeq, kXR_close, kXR_coloc, kXR_compress, kXR_delete, kXR_dirlist, kXR_fattr, kXR_fattrDel, kXR_fattrGet, kXR_fattrList, kXR_fattrSet, kXR_force, kXR_fresh, kXR_locate, kXR_mkdir, kXR_mkdirpath, kXR_mkpath, kXR_mv, kXR_new, kXR_nowait, kXR_open, kXR_open_apnd, kXR_open_read, kXR_open_updt, kXR_open_wrto, kXR_pgread, kXR_pgwrite, kXR_ping, kXR_posc, kXR_prefname, kXR_prepare, kXR_protocol, kXR_Qckscan, kXR_Qcksum, kXR_Qconfig, kXR_Qopaque, kXR_Qopaquf, kXR_Qopaqug, kXR_QPrep, kXR_Qspace, kXR_QStats, kXR_query, kXR_Qvisa, kXR_Qxattr, kXR_read, kXR_readv, kXR_refresh, kXR_replica, kXR_retstat, kXR_rm, kXR_rmdir, kXR_seqio, kXR_set, kXR_stage, kXR_stat, kXR_sync, kXR_truncate, kXR_vfs, kXR_wmode, kXR_write, kXR_writev, ClientChmodRequest::mode, ClientMkdirRequest::mode, ClientOpenRequest::mode, ClientFattrRequest::numattr, ClientPgReadRequest::offset, ClientPgWriteRequest::offset, ClientReadRequest::offset, ClientTruncateRequest::offset, ClientWriteRequest::offset, readahead_list::offset, ClientChkPointRequest::opcode, ClientFattrRequest::options, ClientLocateRequest::options, ClientMkdirRequest::options, ClientOpenRequest::options, ClientPrepareRequest::options, ClientStatRequest::options, ClientPrepareRequest::prty, ClientRequestHdr::requestid, ClientPgReadRequest::rlen, ClientReadRequest::rlen, readahead_list::rlen, ClientFattrRequest::subcode, and XrdProto::write_list::wlen.

Referenced by GenerateDescription(), and SetDescription().

Here is the call graph for this function:

Here is the caller graph for this function:

GetBindPreference()

URL XrdCl::XRootDTransport::GetBindPreference ( const URL & url, AnyObject & channelData )

virtual

Get bind preference for the next data stream.

Implements XrdCl::TransportHandler.

Definition at line 1923 of file XrdClXRootDTransport.cc.

  {
    XRootDChannelInfo *info = 0;
    channelData.Get( info );
 
    if(!info || !info->bindSelector)
      return url;
 
    return URL( info->bindSelector->Get() );
  }

References XrdCl::XRootDChannelInfo::bindSelector, and XrdCl::AnyObject::Get().

Here is the call graph for this function:

GetBody()

XRootDStatus XrdCl::XRootDTransport::GetBody ( Message & message, Socket * socket )

virtual

Read the message body from the socket, the socket is non-blocking, the method may be called multiple times - see GetHeader for details

Parameters

message	the message buffer containing the header
socket	the socket

Returns

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

Implements XrdCl::TransportHandler.

Definition at line 346 of file XrdClXRootDTransport.cc.

  {
    //--------------------------------------------------------------------------
    // Retrieve the body
    //--------------------------------------------------------------------------
    size_t   leftToBeRead = 0;
    uint32_t bodySize = 0;
    ServerResponseHeader* rsphdr = (ServerResponseHeader*)message.GetBuffer();
    bodySize = rsphdr->dlen;
 
    if( bodySize > std::numeric_limits<uint32_t>::max() - 8 )
      return XRootDStatus( stError, errInvalidMessage, 0,
                           "Response body too large." );
 
    if( message.GetSize() < bodySize + 8 )
      message.ReAllocate( bodySize + 8 );
 
    leftToBeRead = bodySize-(message.GetCursor()-8);
    while( leftToBeRead )
    {
      int bytesRead = 0;
      XRootDStatus status = socket->Read( message.GetBufferAtCursor(), leftToBeRead, bytesRead );
 
      if( !status.IsOK() || status.code == suRetry )
        return status;
 
      leftToBeRead -= bytesRead;
      message.AdvanceCursor( bytesRead );
    }
 
    return XRootDStatus( stOK, suDone );
  }

References XrdCl::Buffer::AdvanceCursor(), XrdCl::Status::code, ServerResponseHeader::dlen, XrdCl::errInvalidMessage, XrdCl::Buffer::GetBuffer(), XrdCl::Buffer::GetBufferAtCursor(), XrdCl::Buffer::GetCursor(), XrdCl::Buffer::GetSize(), XrdCl::Status::IsOK(), XrdCl::Socket::Read(), XrdCl::Buffer::ReAllocate(), XrdCl::stError, XrdCl::stOK, XrdCl::suDone, and XrdCl::suRetry.

Here is the call graph for this function:

GetHeader()

XRootDStatus XrdCl::XRootDTransport::GetHeader ( Message & message, Socket * socket )

virtual

Read a message header from the socket, the socket is non-blocking, so if there is not enough data the function should return suRetry in which case it will be called again when more data arrives, with the data previously read stored in the message buffer

Parameters

message	the message buffer
socket	the socket

Returns

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

Implements XrdCl::TransportHandler.

Definition at line 306 of file XrdClXRootDTransport.cc.

  {
    //--------------------------------------------------------------------------
    // A new message - allocate the space needed for the header
    //--------------------------------------------------------------------------
    if( message.GetCursor() == 0 && message.GetSize() < 8 )
      message.Allocate( 8 );
 
    //--------------------------------------------------------------------------
    // Read the message header
    //--------------------------------------------------------------------------
    if( message.GetCursor() < 8 )
    {
      size_t leftToBeRead = 8 - message.GetCursor();
      while( leftToBeRead )
      {
        int bytesRead = 0;
        XRootDStatus status = socket->Read( message.GetBufferAtCursor(),
                                            leftToBeRead, bytesRead );
        if( !status.IsOK() || status.code == suRetry )
          return status;
 
        leftToBeRead -= bytesRead;
        message.AdvanceCursor( bytesRead );
      }
      UnMarshallHeader( message );
 
      uint32_t bodySize = *(uint32_t*)(message.GetBuffer(4));
      Log *log = DefaultEnv::GetLog();
      log->Dump( XRootDTransportMsg, "[msg: %p] Expecting %d bytes of message "
                 "body", (void*)&message, bodySize );
 
      return XRootDStatus( stOK, suDone );
    }
    return XRootDStatus( stError, errInternal );
  }

References XrdCl::Buffer::AdvanceCursor(), XrdCl::Buffer::Allocate(), XrdCl::Status::code, XrdCl::Log::Dump(), XrdCl::errInternal, XrdCl::Buffer::GetBuffer(), XrdCl::Buffer::GetBufferAtCursor(), XrdCl::Buffer::GetCursor(), XrdCl::DefaultEnv::GetLog(), XrdCl::Buffer::GetSize(), XrdCl::Status::IsOK(), XrdCl::Socket::Read(), XrdCl::stError, XrdCl::stOK, XrdCl::suDone, XrdCl::suRetry, UnMarshallHeader(), and XrdCl::XRootDTransportMsg.

Here is the call graph for this function:

GetMore()

XRootDStatus XrdCl::XRootDTransport::GetMore ( Message & message, Socket * socket )

virtual

Read more of the message body from the socket, the socket is non-blocking the method may be called multiple times - see GetHeader for details

Parameters

message	the message buffer containing the header
socket	the socket

Returns

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

Implements XrdCl::TransportHandler.

Definition at line 382 of file XrdClXRootDTransport.cc.

  {
    ServerResponseHeader* rsphdr = (ServerResponseHeader*)message.GetBuffer();
    if( rsphdr->status != kXR_status )
      return XRootDStatus( stError, errInvalidOp );
 
    //--------------------------------------------------------------------------
    // In case of non kXR_status responses we read all the response, including
    // data. For kXR_status responses we first read only the remainder of the
    // header. The header must then be unmarshalled, and then a second call to
    // GetMore (repeated for suRetry as needed) will read the data.
    //--------------------------------------------------------------------------
 
    uint32_t bodySize = rsphdr->dlen;
    if( bodySize > std::numeric_limits<uint32_t>::max() - 8 )
      return XRootDStatus( stError, errInvalidMessage, 0,
                          "kXR_status: response body too large." );
    if( bodySize+8 < sizeof( ServerResponseStatus ) )
      return XRootDStatus( stError, errInvalidMessage, 0,
                          "kXR_status: invalid message size." );
 
    ServerResponseStatus *rspst = (ServerResponseStatus*)message.GetBuffer();
    uint32_t moreSize = static_cast<uint32_t>( rspst->bdy.dlen );
    if( moreSize > std::numeric_limits<uint32_t>::max() - 8 - bodySize )
      return XRootDStatus( stError, errInvalidMessage, 0,
                           "kXR_status: response body too large." );
    bodySize += moreSize;
 
    if( message.GetSize() < bodySize + 8 )
      message.ReAllocate( bodySize + 8 );
 
    size_t leftToBeRead = bodySize-(message.GetCursor()-8);
    while( leftToBeRead )
    {
      int bytesRead = 0;
      XRootDStatus status = socket->Read( message.GetBufferAtCursor(), leftToBeRead, bytesRead );
 
      if( !status.IsOK() || status.code == suRetry )
        return status;
 
      leftToBeRead -= bytesRead;
      message.AdvanceCursor( bytesRead );
    }
 
    // Unmarchal to message body
    Log *log = DefaultEnv::GetLog();
    XRootDStatus st = XRootDTransport::UnMarchalStatusMore( message );
    if( !st.IsOK() && st.code == errDataError )
    {
      log->Error( XRootDTransportMsg, "[msg: %p] %s", (void*)&message,
                  st.GetErrorMessage().c_str() );
      return st;
    }
 
    if( !st.IsOK() )
    {
      log->Error( XRootDTransportMsg, "[msg: %p] Failed to unmarshall status body.",
                  (void*)&message );
      return st;
    }
 
    return XRootDStatus( stOK, suDone );
  }

References XrdCl::Buffer::AdvanceCursor(), ServerResponseStatus::bdy, XrdCl::Status::code, ServerResponseBody_Status::dlen, ServerResponseHeader::dlen, XrdCl::errDataError, XrdCl::errInvalidMessage, XrdCl::errInvalidOp, XrdCl::Log::Error(), XrdCl::Buffer::GetBuffer(), XrdCl::Buffer::GetBufferAtCursor(), XrdCl::Buffer::GetCursor(), XrdCl::XRootDStatus::GetErrorMessage(), XrdCl::DefaultEnv::GetLog(), XrdCl::Buffer::GetSize(), XrdCl::Status::IsOK(), kXR_status, XrdCl::Socket::Read(), XrdCl::Buffer::ReAllocate(), ServerResponseHeader::status, XrdCl::stError, XrdCl::stOK, XrdCl::suDone, XrdCl::suRetry, UnMarchalStatusMore(), and XrdCl::XRootDTransportMsg.

Here is the call graph for this function:

GetSignature() [1/2]

Status XrdCl::XRootDTransport::GetSignature ( Message * toSign, Message *& sign, AnyObject & channelData )

virtual

Get signature for given message.

Implements XrdCl::TransportHandler.

Definition at line 1784 of file XrdClXRootDTransport.cc.

  {
    XRootDChannelInfo *info = 0;
    channelData.Get( info );
    return GetSignature( toSign, sign, info );
  }

References XrdCl::AnyObject::Get(), and GetSignature().

Referenced by GetSignature().

Here is the call graph for this function:

Here is the caller graph for this function:

GetSignature() [2/2]

Status XrdCl::XRootDTransport::GetSignature ( Message * toSign, Message *& sign, XRootDChannelInfo * info )

virtual

Get signature for given message.

Definition at line 1794 of file XrdClXRootDTransport.cc.

  {
    XrdSysRWLockHelper scope( pSecUnloadHandler->lock );
    if( pSecUnloadHandler->unloaded ) return Status( stError, errInvalidOp );
 
    ClientRequest *thereq  = reinterpret_cast<ClientRequest*>( toSign->GetBuffer() );
    if( !info ) return Status( stError, errInternal );
    if( info->protection )
    {
      SecurityRequest *newreq  = 0;
      // check if we have to secure the request in the first place
      if( !( NEED2SECURE ( info->protection )( *thereq ) ) ) return Status();
      // secure (sign/encrypt) the request
      int rc = info->protection->Secure( newreq, *thereq, 0 );
      // there was an error
      if( rc < 0 )
        return Status( stError, errInternal, -rc );
 
      sign = new Message();
      sign->Grab( reinterpret_cast<char*>( newreq ), rc );
    }
 
    return Status();
  }

References XrdCl::errInternal, XrdCl::errInvalidOp, XrdCl::Buffer::GetBuffer(), XrdCl::Buffer::Grab(), NEED2SECURE, XrdCl::XRootDChannelInfo::protection, XrdSecProtect::Secure(), and XrdCl::stError.

Here is the call graph for this function:

HandShake()

XRootDStatus XrdCl::XRootDTransport::HandShake ( HandShakeData * handShakeData, AnyObject & channelData )

virtual

HandShake.

Implements XrdCl::TransportHandler.

Definition at line 477 of file XrdClXRootDTransport.cc.

  {
    XRootDChannelInfo *info = 0;
    channelData.Get( info );
 
    if (!info)
      return XRootDStatus(stFatal, errInternal);
 
    XrdSysMutexHelper scopedLock( info->mutex );
 
    if( info->stream.size() <= handShakeData->subStreamId )
    {
      Log *log = DefaultEnv::GetLog();
      log->Error( XRootDTransportMsg,
                  "[%s] Internal error: not enough substreams",
                  handShakeData->streamName.c_str() );
      return XRootDStatus( stFatal, errInternal );
    }
 
    if( handShakeData->subStreamId == 0 )
    {
      info->streamName = handShakeData->streamName;
      return HandShakeMain( handShakeData, channelData );
    }
    return HandShakeParallel( handShakeData, channelData );
  }

References XrdCl::errInternal, XrdCl::Log::Error(), XrdCl::AnyObject::Get(), XrdCl::DefaultEnv::GetLog(), XrdCl::XRootDChannelInfo::mutex, XrdCl::stFatal, XrdCl::XRootDChannelInfo::stream, XrdCl::HandShakeData::streamName, XrdCl::XRootDChannelInfo::streamName, XrdCl::HandShakeData::subStreamId, and XrdCl::XRootDTransportMsg.

Here is the call graph for this function:

HandShakeDone()

bool XrdCl::XRootDTransport::HandShakeDone ( HandShakeData * handShakeData, AnyObject & channelData )

virtual

Implements XrdCl::TransportHandler.

Definition at line 756 of file XrdClXRootDTransport.cc.

  {
    XRootDChannelInfo *info = 0;
    channelData.Get( info );
 
    if (!info) {
      DefaultEnv::GetLog()->Error(XRootDTransportMsg,
        "[%s] Internal error: no channel info",
        handShakeData->streamName.c_str());
      return false;
    }
 
    XRootDStreamInfo &sInfo = info->stream[handShakeData->subStreamId];
    return ( sInfo.status == XRootDStreamInfo::Connected );
  }

References XrdCl::XRootDStreamInfo::Connected, XrdCl::Log::Error(), XrdCl::AnyObject::Get(), XrdCl::DefaultEnv::GetLog(), XrdCl::XRootDStreamInfo::status, XrdCl::XRootDChannelInfo::stream, XrdCl::HandShakeData::streamName, XrdCl::HandShakeData::subStreamId, and XrdCl::XRootDTransportMsg.

Here is the call graph for this function:

InitializeChannel()

void XrdCl::XRootDTransport::InitializeChannel ( const URL & url, AnyObject & channelData )

virtual

Initialize channel.

Implements XrdCl::TransportHandler.

Definition at line 449 of file XrdClXRootDTransport.cc.

  {
    XRootDChannelInfo *info = new XRootDChannelInfo( url );
    XrdSysMutexHelper scopedLock( info->mutex );
    channelData.Set( info );
 
    Env *env = DefaultEnv::GetEnv();
    int streams = DefaultSubStreamsPerChannel;
    env->GetInt( "SubStreamsPerChannel", streams );
    if( streams < 1 ) streams = 1;
    info->stream.resize( streams );
    info->strmSelector.reset( new StreamSelector( streams ) );
    info->encrypted    = url.IsSecure();
    info->istpc        = url.IsTPC();
    info->logintoken   = url.GetLoginToken();
  }

References XrdCl::DefaultSubStreamsPerChannel, XrdCl::XRootDChannelInfo::encrypted, XrdCl::DefaultEnv::GetEnv(), XrdCl::Env::GetInt(), XrdCl::URL::GetLoginToken(), XrdCl::URL::IsSecure(), XrdCl::URL::IsTPC(), XrdCl::XRootDChannelInfo::istpc, XrdCl::XRootDChannelInfo::logintoken, XrdCl::XRootDChannelInfo::mutex, XrdCl::AnyObject::Set(), XrdCl::XRootDChannelInfo::stream, and XrdCl::XRootDChannelInfo::strmSelector.

Here is the call graph for this function:

IsStreamBroken()

Status XrdCl::XRootDTransport::IsStreamBroken ( time_t inactiveTime, AnyObject & channelData )

virtual

Check the stream is broken - ie. TCP connection got broken and went undetected by the TCP stack

Implements XrdCl::TransportHandler.

Definition at line 829 of file XrdClXRootDTransport.cc.

  {
    XRootDChannelInfo *info = 0;
    channelData.Get( info );
    Env *env = DefaultEnv::GetEnv();
    Log *log = DefaultEnv::GetLog();
 
    if (!info) {
      log->Error(XRootDTransportMsg,
        "Internal error: no channel info, behaving as if stream is broken");
      return true;
    }
 
    int streamTimeout = DefaultStreamTimeout;
    env->GetInt( "StreamTimeout", streamTimeout );
 
    XrdSysMutexHelper scopedLock( info->mutex );
 
    const time_t now  = time(0);
    const bool anySID =
      info->sidManager->IsAnySIDOldAs( now - streamTimeout );
 
    log->Dump( XRootDTransportMsg, "[%s] Stream inactive since %lld seconds, "
               "stream timeout: %d, any SID: %d, wait barrier: %s",
               info->streamName.c_str(), (long long) inactiveTime, streamTimeout,
               anySID, Utils::TimeToString(info->waitBarrier).c_str() );
 
    if( inactiveTime < streamTimeout )
      return Status();
 
    if( now < info->waitBarrier )
      return Status();
 
    if( !anySID )
      return Status();
 
    return Status( stError, errSocketTimeout );
  }

References XrdCl::DefaultStreamTimeout, XrdCl::Log::Dump(), XrdCl::Log::Error(), XrdCl::errSocketTimeout, XrdCl::AnyObject::Get(), XrdCl::DefaultEnv::GetEnv(), XrdCl::Env::GetInt(), XrdCl::DefaultEnv::GetLog(), XrdCl::XRootDChannelInfo::mutex, XrdCl::XRootDChannelInfo::sidManager, XrdCl::stError, XrdCl::XRootDChannelInfo::streamName, XrdCl::Utils::TimeToString(), XrdCl::XRootDChannelInfo::waitBarrier, and XrdCl::XRootDTransportMsg.

Here is the call graph for this function:

IsStreamTTLElapsed()

bool XrdCl::XRootDTransport::IsStreamTTLElapsed ( time_t time, AnyObject & channelData )

virtual

Check if the stream should be disconnected.

Implements XrdCl::TransportHandler.

Definition at line 776 of file XrdClXRootDTransport.cc.

  {
    XRootDChannelInfo *info = 0;
    channelData.Get( info );
 
    Env *env = DefaultEnv::GetEnv();
    Log *log = DefaultEnv::GetLog();
 
    if (!info) {
      log->Error(XRootDTransportMsg,
        "Internal error: no channel info, behaving as if TTL has elapsed");
      return true;
    }
 
    //--------------------------------------------------------------------------
    // Check the TTL settings for the current server
    //--------------------------------------------------------------------------
    int ttl;
    if( info->serverFlags & kXR_isServer )
    {
      ttl = DefaultDataServerTTL;
      env->GetInt( "DataServerTTL", ttl );
    }
    else
    {
      ttl = DefaultLoadBalancerTTL;
      env->GetInt( "LoadBalancerTTL", ttl );
    }
 
    //--------------------------------------------------------------------------
    // See whether we can give a go-ahead for the disconnection
    //--------------------------------------------------------------------------
    XrdSysMutexHelper scopedLock( info->mutex );
    uint16_t allocatedSIDs = info->sidManager->GetNumberOfAllocatedSIDs();
    log->Dump( XRootDTransportMsg, "[%s] Stream inactive since %lld seconds, "
               "TTL: %d, allocated SIDs: %d, open files: %d, bound file objects: %d",
               info->streamName.c_str(), (long long) inactiveTime, ttl, allocatedSIDs,
               info->openFiles, info->finstcnt.load( std::memory_order_relaxed ) );
 
    if( info->openFiles != 0 && info->finstcnt.load( std::memory_order_relaxed ) != 0 )
      return false;
 
    if( !allocatedSIDs && inactiveTime > ttl )
      return true;
 
    return false;
  }

References XrdCl::DefaultDataServerTTL, XrdCl::DefaultLoadBalancerTTL, XrdCl::Log::Dump(), XrdCl::Log::Error(), XrdCl::XRootDChannelInfo::finstcnt, XrdCl::AnyObject::Get(), XrdCl::DefaultEnv::GetEnv(), XrdCl::Env::GetInt(), XrdCl::DefaultEnv::GetLog(), kXR_isServer, XrdCl::XRootDChannelInfo::mutex, XrdCl::XRootDChannelInfo::openFiles, XrdCl::XRootDChannelInfo::serverFlags, XrdCl::XRootDChannelInfo::sidManager, XrdCl::XRootDChannelInfo::streamName, and XrdCl::XRootDTransportMsg.

Here is the call graph for this function:

LogErrorResponse()

void XrdCl::XRootDTransport::LogErrorResponse ( const Message & msg )

static

Log server error response.

Definition at line 1517 of file XrdClXRootDTransport.cc.

  {
    Log *log = DefaultEnv::GetLog();
    ServerResponse *rsp = (ServerResponse *)msg.GetBuffer();
    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->Error( XRootDTransportMsg, "Server responded with an error [%d]: %s",
                                    rsp->body.error.errnum, errmsg );
   delete [] errmsg;
  }

References ServerResponse::body, ServerResponseHeader::dlen, XrdCl::Log::Error(), XrdCl::Buffer::GetBuffer(), XrdCl::DefaultEnv::GetLog(), ServerResponse::hdr, and XrdCl::XRootDTransportMsg.

Here is the call graph for this function:

MarshallRequest() [1/2]

XRootDStatus XrdCl::XRootDTransport::MarshallRequest ( char * msg )

static

Marshal the outgoing message.

Definition at line 1113 of file XrdClXRootDTransport.cc.

  {
    ClientRequest *req = (ClientRequest*)msg;
    switch( req->header.requestid )
    {
      //------------------------------------------------------------------------
      // kXR_protocol
      //------------------------------------------------------------------------
      case kXR_protocol:
        req->protocol.clientpv = htonl( req->protocol.clientpv );
        break;
 
      //------------------------------------------------------------------------
      // kXR_login
      //------------------------------------------------------------------------
      case kXR_login:
        req->login.pid = htonl( req->login.pid );
        break;
 
      //------------------------------------------------------------------------
      // kXR_locate
      //------------------------------------------------------------------------
      case kXR_locate:
        req->locate.options = htons( req->locate.options );
        break;
 
      //------------------------------------------------------------------------
      // kXR_query
      //------------------------------------------------------------------------
      case kXR_query:
        req->query.infotype = htons( req->query.infotype );
        break;
 
      //------------------------------------------------------------------------
      // kXR_truncate
      //------------------------------------------------------------------------
      case kXR_truncate:
        req->truncate.offset = htonll( req->truncate.offset );
        break;
 
      //------------------------------------------------------------------------
      // kXR_mkdir
      //------------------------------------------------------------------------
      case kXR_mkdir:
        req->mkdir.mode = htons( req->mkdir.mode );
        break;
 
      //------------------------------------------------------------------------
      // kXR_chmod
      //------------------------------------------------------------------------
      case kXR_chmod:
        req->chmod.mode = htons( req->chmod.mode );
        break;
 
      //------------------------------------------------------------------------
      // kXR_open
      //------------------------------------------------------------------------
      case kXR_open:
        req->open.mode    = htons( req->open.mode );
        req->open.options = htons( req->open.options );
        break;
 
      //------------------------------------------------------------------------
      // kXR_read
      //------------------------------------------------------------------------
      case kXR_read:
        req->read.offset = htonll( req->read.offset );
        req->read.rlen   = htonl( req->read.rlen );
        break;
 
      //------------------------------------------------------------------------
      // kXR_write
      //------------------------------------------------------------------------
      case kXR_write:
        req->write.offset = htonll( req->write.offset );
        break;
 
      //------------------------------------------------------------------------
      // kXR_mv
      //------------------------------------------------------------------------
      case kXR_mv:
        req->mv.arg1len = htons( req->mv.arg1len );
        break;
 
      //------------------------------------------------------------------------
      // kXR_readv
      //------------------------------------------------------------------------
      case kXR_readv:
      {
        uint16_t numChunks  = (req->readv.dlen)/16;
        readahead_list *dataChunk = (readahead_list*)( msg + 24 );
        for( size_t i = 0; i < numChunks; ++i )
        {
          dataChunk[i].rlen   = htonl( dataChunk[i].rlen );
          dataChunk[i].offset = htonll( dataChunk[i].offset );
        }
        break;
      }
 
      //------------------------------------------------------------------------
      // kXR_writev
      //------------------------------------------------------------------------
      case kXR_writev:
      {
        uint16_t numChunks  = (req->writev.dlen)/16;
        XrdProto::write_list *wrtList =
            reinterpret_cast<XrdProto::write_list*>( msg + 24 );
        for( size_t i = 0; i < numChunks; ++i )
        {
          wrtList[i].wlen   = htonl( wrtList[i].wlen );
          wrtList[i].offset = htonll( wrtList[i].offset );
        }
 
        break;
      }
 
      case kXR_pgread:
      {
        req->pgread.offset = htonll( req->pgread.offset );
        req->pgread.rlen   = htonl( req->pgread.rlen );
        break;
      }
 
      case kXR_pgwrite:
      {
        req->pgwrite.offset = htonll( req->pgwrite.offset );
        break;
      }
 
      //------------------------------------------------------------------------
      // kXR_prepare
      //------------------------------------------------------------------------
      case kXR_prepare:
      {
        req->prepare.optionX = htons( req->prepare.optionX );
        req->prepare.port    = htons( req->prepare.port );
        break;
      }
 
      case kXR_chkpoint:
      {
        if( req->chkpoint.opcode == kXR_ckpXeq )
          MarshallRequest( msg + 24 );
        break;
      }
    };
 
    req->header.requestid = htons( req->header.requestid );
    req->header.dlen      = htonl( req->header.dlen );
    return XRootDStatus();
  }

References ClientMvRequest::arg1len, ClientRequest::chkpoint, ClientRequest::chmod, ClientProtocolRequest::clientpv, ClientReadVRequest::dlen, ClientRequestHdr::dlen, ClientWriteVRequest::dlen, ClientRequest::header, ClientQueryRequest::infotype, kXR_chkpoint, kXR_chmod, kXR_ckpXeq, kXR_locate, kXR_login, kXR_mkdir, kXR_mv, kXR_open, kXR_pgread, kXR_pgwrite, kXR_prepare, kXR_protocol, kXR_query, kXR_read, kXR_readv, kXR_truncate, kXR_write, kXR_writev, ClientRequest::locate, ClientRequest::login, MarshallRequest(), ClientRequest::mkdir, ClientChmodRequest::mode, ClientMkdirRequest::mode, ClientOpenRequest::mode, ClientRequest::mv, ClientPgReadRequest::offset, ClientPgWriteRequest::offset, ClientReadRequest::offset, ClientTruncateRequest::offset, ClientWriteRequest::offset, readahead_list::offset, XrdProto::write_list::offset, ClientChkPointRequest::opcode, ClientRequest::open, ClientLocateRequest::options, ClientOpenRequest::options, ClientPrepareRequest::optionX, ClientRequest::pgread, ClientRequest::pgwrite, ClientLoginRequest::pid, ClientPrepareRequest::port, ClientRequest::prepare, ClientRequest::protocol, ClientRequest::query, ClientRequest::read, ClientRequest::readv, ClientRequestHdr::requestid, ClientPgReadRequest::rlen, ClientReadRequest::rlen, readahead_list::rlen, ClientRequest::truncate, XrdProto::write_list::wlen, ClientRequest::write, and ClientRequest::writev.

Here is the call graph for this function:

MarshallRequest() [2/2]

static XRootDStatus XrdCl::XRootDTransport::MarshallRequest ( Message * msg )

inlinestatic

Marshal the outgoing message.

Definition at line 175 of file XrdClXRootDTransport.hh.

      {
        MarshallRequest( msg->GetBuffer() );
        msg->SetIsMarshalled( true );
        return XRootDStatus();
      }

References XrdCl::Buffer::GetBuffer(), MarshallRequest(), and XrdCl::Message::SetIsMarshalled().

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

Here is the call graph for this function:

Here is the caller graph for this function:

MessageReceived()

uint32_t XrdCl::XRootDTransport::MessageReceived ( Message & msg, uint16_t subStream, AnyObject & channelData )

virtual

Check if the message invokes a stream action.

Implements XrdCl::TransportHandler.

Definition at line 1640 of file XrdClXRootDTransport.cc.

  {
    XRootDChannelInfo *info = 0;
    channelData.Get( info );
    if( !info ) return NoAction;
    XrdSysMutexHelper scopedLock( info->mutex );
    Log *log = DefaultEnv::GetLog();
 
    //--------------------------------------------------------------------------
    // Update the substream queues
    //--------------------------------------------------------------------------
    info->strmSelector->MsgReceived( subStream );
 
    //--------------------------------------------------------------------------
    // Check whether this message is a response to a request that has
    // timed out, and if so, drop it
    //--------------------------------------------------------------------------
    ServerResponse *rsp = (ServerResponse*)msg.GetBuffer();
    if( rsp->hdr.status == kXR_attn )
    {
      return NoAction;
    }
 
    if( info->sidManager->IsTimedOut( rsp->hdr.streamid ) )
    {
      log->Error( XRootDTransportMsg, "Message %p, stream [%d, %d] is a "
                  "response that we're no longer interested in (timed out)",
                  (void*)&msg, rsp->hdr.streamid[0], rsp->hdr.streamid[1] );
      //------------------------------------------------------------------------
      // If it is kXR_waitresp there will be another one,
      // so we don't release the sid yet
      //------------------------------------------------------------------------
      if( rsp->hdr.status != kXR_waitresp )
        info->sidManager->ReleaseTimedOut( rsp->hdr.streamid );
      //------------------------------------------------------------------------
      // If it is a successful response to an open request
      // that timed out, we need to send a close
      //------------------------------------------------------------------------
      uint16_t sid; memcpy( &sid, rsp->hdr.streamid, 2 );
      std::set<uint16_t>::iterator sidIt = info->sentOpens.find( sid );
      if( sidIt != info->sentOpens.end() )
      {
        info->sentOpens.erase( sidIt );
        if( rsp->hdr.status == kXR_ok ) return RequestClose;
      }
      return DigestMsg;
    }
 
    //--------------------------------------------------------------------------
    // If we have a wait or waitresp
    //--------------------------------------------------------------------------
    uint32_t seconds = 0;
    if( rsp->hdr.status == kXR_wait )
      seconds = ntohl( rsp->body.wait.seconds ) + 5; // we need extra time
                                                     // to re-send the request
    else if( rsp->hdr.status == kXR_waitresp )
    {
      seconds = ntohl( rsp->body.waitresp.seconds );
 
      log->Dump( XRootDMsg, "[%s] Got kXR_waitresp response of %u seconds, "
                 "setting up wait barrier.",
                 info->streamName.c_str(),
                 seconds );
    }
 
    time_t barrier = time(0) + seconds;
    if( info->waitBarrier < barrier )
      info->waitBarrier = barrier;
 
    //--------------------------------------------------------------------------
    // If we got a response to an open request, we may need to bump the counter
    // of open files
    //--------------------------------------------------------------------------
    uint16_t sid; memcpy( &sid, rsp->hdr.streamid, 2 );
    std::set<uint16_t>::iterator sidIt = info->sentOpens.find( sid );
    if( sidIt != info->sentOpens.end() )
    {
      if( rsp->hdr.status == kXR_waitresp )
        return NoAction;
      info->sentOpens.erase( sidIt );
      if( rsp->hdr.status == kXR_ok )
      {
        ++info->openFiles;
        info->finstcnt.fetch_add( 1, std::memory_order_relaxed ); // another file File object instance has been bound with this connection
      }
      return NoAction;
    }
 
    //--------------------------------------------------------------------------
    // If we got a response to a close, we may need to decrement the counter of
    // open files
    //--------------------------------------------------------------------------
    sidIt = info->sentCloses.find( sid );
    if( sidIt != info->sentCloses.end() )
    {
      if( rsp->hdr.status == kXR_waitresp )
        return NoAction;
      info->sentCloses.erase( sidIt );
      --info->openFiles;
      return NoAction;
    }
    return NoAction;
  }

References ServerResponse::body, XrdCl::TransportHandler::DigestMsg, XrdCl::Log::Dump(), XrdCl::Log::Error(), XrdCl::XRootDChannelInfo::finstcnt, XrdCl::AnyObject::Get(), XrdCl::Buffer::GetBuffer(), XrdCl::DefaultEnv::GetLog(), ServerResponse::hdr, kXR_attn, kXR_ok, kXR_wait, kXR_waitresp, XrdCl::XRootDChannelInfo::mutex, XrdCl::TransportHandler::NoAction, XrdCl::XRootDChannelInfo::openFiles, XrdCl::TransportHandler::RequestClose, XrdCl::XRootDChannelInfo::sentCloses, XrdCl::XRootDChannelInfo::sentOpens, XrdCl::XRootDChannelInfo::sidManager, ServerResponseHeader::status, ServerResponseHeader::streamid, XrdCl::XRootDChannelInfo::streamName, XrdCl::XRootDChannelInfo::strmSelector, XrdCl::XRootDChannelInfo::waitBarrier, XrdCl::XRootDMsg, and XrdCl::XRootDTransportMsg.

Here is the call graph for this function:

MessageSent()

void XrdCl::XRootDTransport::MessageSent ( Message * msg, uint16_t subStream, uint32_t bytesSent, AnyObject & channelData )

virtual

Notify the transport about a message having been sent.

Implements XrdCl::TransportHandler.

Definition at line 1749 of file XrdClXRootDTransport.cc.

  {
    // Called when a message has been sent. For messages that return on a
    // different pathid (and hence may use a different poller) it is possible
    // that the server has already replied and the reply will trigger
    // MessageReceived() before this method has been called. However for open
    // and close this is never the case and this method is used for tracking
    // only those.
    XRootDChannelInfo *info = 0;
    channelData.Get( info );
    if( !info ) return;
    XrdSysMutexHelper scopedLock( info->mutex );
    ClientRequest *req = (ClientRequest*)msg->GetBuffer();
    uint16_t reqid = ntohs( req->header.requestid );
 
 
    //--------------------------------------------------------------------------
    // We need to track opens to know if we can close streams due to idleness
    //--------------------------------------------------------------------------
    uint16_t sid;
    memcpy( &sid, req->header.streamid, 2 );
 
    if( reqid == kXR_open )
      info->sentOpens.insert( sid );
    else if( reqid == kXR_close )
      info->sentCloses.insert( sid );
  }

References XrdCl::AnyObject::Get(), XrdCl::Buffer::GetBuffer(), ClientRequest::header, kXR_close, kXR_open, XrdCl::XRootDChannelInfo::mutex, ClientRequestHdr::requestid, XrdCl::XRootDChannelInfo::sentCloses, XrdCl::XRootDChannelInfo::sentOpens, and ClientRequestHdr::streamid.

Here is the call graph for this function:

Multiplex()

PathID XrdCl::XRootDTransport::Multiplex ( Message * msg, AnyObject & channelData, PathID * hint = 0 )

virtual

Return the ID for the up stream this message should be sent by and the down stream which the answer should be expected at. Modify the message itself if necessary. If hint is non-zero then the message should be modified such that the answer will be returned via the hinted stream.

Implements XrdCl::TransportHandler.

Definition at line 872 of file XrdClXRootDTransport.cc.

  {
    return PathID( 0, 0 );
  }

MultiplexSubStream()

PathID XrdCl::XRootDTransport::MultiplexSubStream ( Message * msg, AnyObject & channelData, PathID * hint = 0 )

virtual

Return the ID for the up substream this message should be sent by and the down substream which the answer should be expected at. Modify the message itself if necessary. If hint is non-zero then the message should be modified such that the answer will be returned via the hinted stream.

Implements XrdCl::TransportHandler.

Definition at line 880 of file XrdClXRootDTransport.cc.

  {
    XRootDChannelInfo *info = 0;
    channelData.Get( info );
 
    if (!info) {
      DefaultEnv::GetLog()->Error(XRootDTransportMsg,
        "Internal error: no channel info, cannot multiplex");
      return PathID(0,0);
    }
 
    XrdSysMutexHelper scopedLock( info->mutex );
 
    //--------------------------------------------------------------------------
    // If we're not connected to a data server or we don't know that yet
    // we stream through 0
    //--------------------------------------------------------------------------
    if( !(info->serverFlags & kXR_isServer) || info->stream.size() == 0 )
      return PathID( 0, 0 );
 
    //--------------------------------------------------------------------------
    // Select the streams
    //--------------------------------------------------------------------------
    Log *log = DefaultEnv::GetLog();
    uint16_t upStream   = 0;
    uint16_t downStream = 0;
 
    if( hint )
    {
      upStream   = hint->up;
      downStream = hint->down;
    }
    else
    {
      upStream = 0;
      std::vector<bool> connected;
      connected.reserve( info->stream.size() - 1 );
      size_t nbConnected = 0;
      for( size_t i = 1; i < info->stream.size(); ++i )
        if( info->stream[i].status == XRootDStreamInfo::Connected )
        {
          connected.push_back( true );
          ++nbConnected;
        }
        else
          connected.push_back( false );
 
      if( nbConnected == 0 )
        downStream = 0;
      else
        downStream = info->strmSelector->Select( connected );
    }
 
    if( upStream >= info->stream.size() )
    {
      log->Debug( XRootDTransportMsg,
                  "[%s] Up link stream %d does not exist, using 0",
                  info->streamName.c_str(), upStream );
      upStream = 0;
    }
 
    if( downStream >= info->stream.size() )
    {
      log->Debug( XRootDTransportMsg,
                  "[%s] Down link stream %d does not exist, using 0",
                  info->streamName.c_str(), downStream );
      downStream = 0;
    }
 
    //--------------------------------------------------------------------------
    // Modify the message
    //--------------------------------------------------------------------------
    UnMarshallRequest( msg );
    ClientRequestHdr *hdr = (ClientRequestHdr*)msg->GetBuffer();
    switch( hdr->requestid )
    {
      //------------------------------------------------------------------------
      // Read - we update the path id to tell the server where we want to
      // get the response, but we still send the request through stream 0
      // We need to allocate space for read_args if we don't have it
      // included yet
      //------------------------------------------------------------------------
      case kXR_read:
      {
        if( msg->GetSize() < sizeof(ClientReadRequest) + 8 )
        {
          msg->ReAllocate( sizeof(ClientReadRequest) + 8 );
          void *newBuf = msg->GetBuffer(sizeof(ClientReadRequest));
          memset( newBuf, 0, 8 );
          ClientReadRequest *req = (ClientReadRequest*)msg->GetBuffer();
          req->dlen += 8;
        }
        read_args *args = (read_args*)msg->GetBuffer(sizeof(ClientReadRequest));
        args->pathid = info->stream[downStream].pathId;
        break;
      }
 
 
      //------------------------------------------------------------------------
      // PgRead - we update the path id to tell the server where we want to
      // get the response, but we still send the request through stream 0
      // We need to allocate space for ClientPgReadReqArgs if we don't have it
      // included yet
      //------------------------------------------------------------------------
      case kXR_pgread:
      {
        if( msg->GetSize() < sizeof( ClientPgReadRequest ) + sizeof( ClientPgReadReqArgs ) )
        {
          msg->ReAllocate( sizeof( ClientPgReadRequest ) + sizeof( ClientPgReadReqArgs ) );
          void *newBuf = msg->GetBuffer( sizeof( ClientPgReadRequest ) );
          memset( newBuf, 0, sizeof( ClientPgReadReqArgs ) );
          ClientPgReadRequest *req = (ClientPgReadRequest*)msg->GetBuffer();
          req->dlen += sizeof( ClientPgReadReqArgs );
        }
        ClientPgReadReqArgs *args = reinterpret_cast<ClientPgReadReqArgs*>(
            msg->GetBuffer( sizeof( ClientPgReadRequest ) ) );
        args->pathid   = info->stream[downStream].pathId;
        break;
      }
 
      //------------------------------------------------------------------------
      // ReadV - the situation is identical to read but we don't need any
      // additional structures to specify the return path
      //------------------------------------------------------------------------
      case kXR_readv:
      {
        ClientReadVRequest *req = (ClientReadVRequest*)msg->GetBuffer();
        req->pathid = info->stream[downStream].pathId;
        break;
      }
 
      //------------------------------------------------------------------------
      // Write - multiplexing writes doesn't work properly in the server
      //------------------------------------------------------------------------
      case kXR_write:
      {
//        ClientWriteRequest *req = (ClientWriteRequest*)msg->GetBuffer();
//        req->pathid = info->stream[downStream].pathId;
        break;
      }
 
      //------------------------------------------------------------------------
      // WriteV - multiplexing writes doesn't work properly in the server
      //------------------------------------------------------------------------
      case kXR_writev:
      {
//        ClientWriteVRequest *req = (ClientWriteVRequest*)msg->GetBuffer();
//        req->pathid = info->stream[downStream].pathId;
        break;
      }
 
      //------------------------------------------------------------------------
      // PgWrite - multiplexing writes doesn't work properly in the server
      //------------------------------------------------------------------------
      case kXR_pgwrite:
      {
//        ClientWriteVRequest *req = (ClientWriteVRequest*)msg->GetBuffer();
//        req->pathid = info->stream[downStream].pathId;
        break;
      }
    };
    MarshallRequest( msg );
    return PathID( upStream, downStream );
  }

References XrdCl::XRootDStreamInfo::Connected, XrdCl::Log::Debug(), ClientPgReadRequest::dlen, ClientReadRequest::dlen, XrdCl::PathID::down, XrdCl::Log::Error(), XrdCl::AnyObject::Get(), XrdCl::Buffer::GetBuffer(), XrdCl::DefaultEnv::GetLog(), XrdCl::Buffer::GetSize(), kXR_isServer, kXR_pgread, kXR_pgwrite, kXR_read, kXR_readv, kXR_write, kXR_writev, MarshallRequest(), XrdCl::XRootDChannelInfo::mutex, ClientPgReadReqArgs::pathid, ClientReadVRequest::pathid, read_args::pathid, XrdCl::Buffer::ReAllocate(), ClientRequestHdr::requestid, XrdCl::XRootDChannelInfo::serverFlags, XrdCl::XRootDChannelInfo::stream, XrdCl::XRootDChannelInfo::streamName, XrdCl::XRootDChannelInfo::strmSelector, UnMarshallRequest(), XrdCl::PathID::up, and XrdCl::XRootDTransportMsg.

Here is the call graph for this function:

NbConnectedStrm()

uint16_t XrdCl::XRootDTransport::NbConnectedStrm ( AnyObject & channelData )

static

Number of currently connected data streams.

Definition at line 1531 of file XrdClXRootDTransport.cc.

  {
    XRootDChannelInfo *info = 0;
    channelData.Get( info );
 
    if (!info) {
      DefaultEnv::GetLog()->Error(XRootDTransportMsg, "Internal error: no channel info");
      return 0;
    }
 
    XrdSysMutexHelper scopedLock( info->mutex );
 
    uint16_t nbConnected = 0;
    for( size_t i = 1; i < info->stream.size(); ++i )
      if( info->stream[i].status == XRootDStreamInfo::Connected )
        ++nbConnected;
 
    return nbConnected;
  }

References XrdCl::XRootDStreamInfo::Connected, XrdCl::Log::Error(), XrdCl::AnyObject::Get(), XrdCl::DefaultEnv::GetLog(), XrdCl::XRootDChannelInfo::mutex, XrdCl::XRootDChannelInfo::stream, and XrdCl::XRootDTransportMsg.

Referenced by XrdCl::Channel::NbConnectedStrm().

Here is the call graph for this function:

Here is the caller graph for this function:

NeedControlConnection()

virtual bool XrdCl::XRootDTransport::NeedControlConnection ( )

inlinevirtual

Return the information whether a control connection needs to be valid before establishing other connections

Definition at line 167 of file XrdClXRootDTransport.hh.

      {
        return true;
      }

NeedEncryption()

bool XrdCl::XRootDTransport::NeedEncryption ( HandShakeData * handShakeData, AnyObject & channelData )

virtual

Returns

: true if encryption should be turned on, false otherwise

Implements XrdCl::TransportHandler.

Definition at line 1844 of file XrdClXRootDTransport.cc.

  {
    XRootDChannelInfo *info = 0;
    channelData.Get( info );
 
    XrdCl::Env *env = XrdCl::DefaultEnv::GetEnv();
    int notlsok = DefaultNoTlsOK;
    env->GetInt( "NoTlsOK", notlsok );
 
 
    if( notlsok )
      return info->encrypted;
 
    // Did the server instructed us to switch to TLS right away?
    if( info->serverFlags & kXR_gotoTLS )
    {
      info->encrypted = true;
      return true ;
    }
 
    XRootDStreamInfo &sInfo = info->stream[handShakeData->subStreamId];
 
    //--------------------------------------------------------------------------
    // The control stream (sub-stream 0)  might need to switch to TLS before
    // login or after login
    //--------------------------------------------------------------------------
    if( handShakeData->subStreamId == 0 )
    {
      //------------------------------------------------------------------------
      // We are about to login and the server asked to start encrypting
      // before login
      //------------------------------------------------------------------------
      if( ( sInfo.status == XRootDStreamInfo::LoginSent ) &&
          ( info->serverFlags & kXR_tlsLogin ) )
      {
        info->encrypted = true;
        return true;
      }
 
      //--------------------------------------------------------------------
      // The hand-shake is done and the server requested to encrypt the session
      //--------------------------------------------------------------------
      if( (sInfo.status == XRootDStreamInfo::Connected ||
          //--------------------------------------------------------------------
          // we really need to turn on TLS before we sent kXR_endsess and we
          // are about to do so (1st enable encryption, then send kXR_endsess)
          //--------------------------------------------------------------------
           sInfo.status == XRootDStreamInfo::EndSessionSent ) &&
          ( info->serverFlags & kXR_tlsSess ) )
      {
        info->encrypted = true;
        return true;
      }
    }
    //--------------------------------------------------------------------------
    // A data stream (sub-stream > 0) if need be will be switched to TLS before
    // bind.
    //--------------------------------------------------------------------------
    else
    {
      //------------------------------------------------------------------------
      // We are about to bind a data stream and the server asked to start
      // encrypting before bind
      //------------------------------------------------------------------------
      if( ( sInfo.status == XRootDStreamInfo::BindSent ) &&
          ( info->serverFlags & kXR_tlsData ) )
      {
        info->encrypted = true;
        return true;
      }
    }
 
    return false;
  }

References XrdCl::XRootDStreamInfo::BindSent, XrdCl::XRootDStreamInfo::Connected, XrdCl::DefaultNoTlsOK, XrdCl::XRootDChannelInfo::encrypted, XrdCl::XRootDStreamInfo::EndSessionSent, XrdCl::AnyObject::Get(), XrdCl::DefaultEnv::GetEnv(), XrdCl::Env::GetInt(), kXR_gotoTLS, kXR_tlsData, kXR_tlsLogin, kXR_tlsSess, XrdCl::XRootDStreamInfo::LoginSent, XrdCl::XRootDChannelInfo::serverFlags, XrdCl::XRootDStreamInfo::status, XrdCl::XRootDChannelInfo::stream, and XrdCl::HandShakeData::subStreamId.

Here is the call graph for this function:

Query()

Status XrdCl::XRootDTransport::Query ( uint16_t query, AnyObject & result, AnyObject & channelData )

virtual

Query the channel.

Implements XrdCl::TransportHandler.

Definition at line 1588 of file XrdClXRootDTransport.cc.

  {
    XRootDChannelInfo *info = 0;
    channelData.Get( info );
 
    if (!info)
      return XRootDStatus(stFatal, errInternal);
 
    XrdSysMutexHelper scopedLock( info->mutex );
 
    switch( query )
    {
      //------------------------------------------------------------------------
      // Protocol name
      //------------------------------------------------------------------------
      case TransportQuery::Name:
        result.Set( (const char*)"XRootD", false );
        return Status();
 
      //------------------------------------------------------------------------
      // Authentication
      //------------------------------------------------------------------------
      case TransportQuery::Auth:
        result.Set( new std::string( info->authProtocolName ), false );
        return Status();
 
      //------------------------------------------------------------------------
      // Server flags
      //------------------------------------------------------------------------
      case XRootDQuery::ServerFlags:
        result.Set( new int( info->serverFlags ), false );
        return Status();
 
      //------------------------------------------------------------------------
      // Protocol version
      //------------------------------------------------------------------------
      case XRootDQuery::ProtocolVersion:
        result.Set( new int( info->protocolVersion ), false );
        return Status();
 
      case XRootDQuery::IsEncrypted:
        result.Set( new bool( info->encrypted ), false );
        return Status();
    };
    return Status( stError, errQueryNotSupported );
  }

References XrdCl::TransportQuery::Auth, XrdCl::XRootDChannelInfo::authProtocolName, XrdCl::XRootDChannelInfo::encrypted, XrdCl::errInternal, XrdCl::errQueryNotSupported, XrdCl::AnyObject::Get(), XrdCl::XRootDQuery::IsEncrypted, XrdCl::XRootDChannelInfo::mutex, XrdCl::TransportQuery::Name, XrdCl::XRootDQuery::ProtocolVersion, XrdCl::XRootDChannelInfo::protocolVersion, XrdCl::XRootDQuery::ServerFlags, XrdCl::XRootDChannelInfo::serverFlags, XrdCl::AnyObject::Set(), XrdCl::stError, and XrdCl::stFatal.

Here is the call graph for this function:

SetDescription()

static void XrdCl::XRootDTransport::SetDescription ( Message * msg )

inlinestatic

Get the description of a message.

Definition at line 245 of file XrdClXRootDTransport.hh.

      {
        std::ostringstream o;
        GenerateDescription( msg->GetBuffer(), o );
        msg->SetDescription( o.str() );
      }

References GenerateDescription(), XrdCl::Buffer::GetBuffer(), and XrdCl::Message::SetDescription().

Referenced by XrdCl::FileStateHandler::Checkpoint(), XrdCl::FileStateHandler::ChkptWrt(), XrdCl::FileStateHandler::ChkptWrtV(), XrdCl::FileSystem::ChMod(), XrdCl::FileStateHandler::Close(), XrdCl::FileSystem::DirList(), XrdCl::FileStateHandler::Fcntl(), XrdCl::FileSystem::Locate(), XrdCl::FileSystem::MkDir(), XrdCl::FileSystem::Mv(), XrdCl::FileStateHandler::Open(), XrdCl::FileStateHandler::PgReadImpl(), XrdCl::FileStateHandler::PgWriteImpl(), XrdCl::FileSystem::Ping(), XrdCl::FileSystem::Prepare(), XrdCl::FileSystem::Protocol(), XrdCl::FileSystem::Query(), XrdCl::FileStateHandler::Read(), XrdCl::FileStateHandler::ReadV(), XrdCl::MessageUtils::RewriteCGIAndPath(), XrdCl::FileSystem::Rm(), XrdCl::FileSystem::RmDir(), XrdCl::FileStateHandler::Stat(), XrdCl::FileSystem::Stat(), XrdCl::FileSystem::StatVFS(), XrdCl::FileStateHandler::Sync(), XrdCl::FileStateHandler::Truncate(), XrdCl::FileSystem::Truncate(), XrdCl::FileStateHandler::VectorRead(), XrdCl::FileStateHandler::VectorWrite(), XrdCl::FileStateHandler::Visa(), XrdCl::FileStateHandler::Write(), and XrdCl::FileStateHandler::WriteV().

Here is the call graph for this function:

Here is the caller graph for this function:

SubStreamNumber()

uint16_t XrdCl::XRootDTransport::SubStreamNumber ( AnyObject & channelData )

virtual

Return a number of substreams per stream that should be created.

Implements XrdCl::TransportHandler.

Definition at line 1052 of file XrdClXRootDTransport.cc.

  {
    XRootDChannelInfo *info = 0;
    channelData.Get( info );
 
    if (!info) {
      DefaultEnv::GetLog()->Error(XRootDTransportMsg, "Internal error: no channel info");
      return 1;
    }
 
    XrdSysMutexHelper scopedLock( info->mutex );
 
    //--------------------------------------------------------------------------
    // If the connection has been opened in order to orchestrate a TPC or
    // the remote server is a Manager or Metamanager we will need only one
    // (control) stream.
    //--------------------------------------------------------------------------
    if( info->istpc || !(info->serverFlags & kXR_isServer ) ) return 1;
 
    //--------------------------------------------------------------------------
    // Number of streams requested by user
    //--------------------------------------------------------------------------
    uint16_t ret = info->stream.size();
 
    XrdCl::Env *env = XrdCl::DefaultEnv::GetEnv();
    int nodata = DefaultTlsNoData;
    env->GetInt( "TlsNoData", nodata );
 
    // Does the server require the stream 0 to be encrypted?
    bool srvTlsStrm0 = ( info->serverFlags & kXR_gotoTLS )  ||
                       ( info->serverFlags & kXR_tlsLogin ) ||
                       ( info->serverFlags & kXR_tlsSess );
    // Does the server NOT require the data streams to be encrypted?
    bool srvNoTlsData = !( info->serverFlags & kXR_tlsData );
    // Does the user require the stream 0 to be encrypted?
    bool usrTlsStrm0 = info->encrypted;
    // Does the user NOT require the data streams to be encrypted?
    bool usrNoTlsData = !info->encrypted || ( info->encrypted && nodata );
 
    if( ( usrTlsStrm0 && usrNoTlsData && srvNoTlsData ) ||
        ( srvTlsStrm0 && srvNoTlsData && usrNoTlsData ) )
    {
      //------------------------------------------------------------------------
      // The server or user asked us to encrypt stream 0, but to send the data
      // (read/write) using a plain TCP connection
      //------------------------------------------------------------------------
      if( ret == 1 ) ++ret;
    }
 
    if( ret > info->stream.size() )
    {
      info->stream.resize( ret );
      info->strmSelector->AdjustQueues( ret );
    }
 
    return ret;
  }

References XrdCl::DefaultTlsNoData, XrdCl::XRootDChannelInfo::encrypted, XrdCl::Log::Error(), XrdCl::AnyObject::Get(), XrdCl::DefaultEnv::GetEnv(), XrdCl::Env::GetInt(), XrdCl::DefaultEnv::GetLog(), XrdCl::XRootDChannelInfo::istpc, kXR_gotoTLS, kXR_isServer, kXR_tlsData, kXR_tlsLogin, kXR_tlsSess, XrdCl::XRootDChannelInfo::mutex, XrdCl::XRootDChannelInfo::serverFlags, XrdCl::XRootDChannelInfo::stream, XrdCl::XRootDChannelInfo::strmSelector, and XrdCl::XRootDTransportMsg.

Here is the call graph for this function:

UnMarchalStatusMore()

XRootDStatus XrdCl::XRootDTransport::UnMarchalStatusMore ( Message & msg )

static

Unmarshall the correction-segment of the status response for pgwrite.

Definition at line 1444 of file XrdClXRootDTransport.cc.

  {
    ServerResponseV2 *rsp = (ServerResponseV2*)msg.GetBuffer();
    uint16_t reqType = rsp->status.bdy.requestid + kXR_1stRequest;
 
    switch( reqType )
    {
      case kXR_pgwrite:
      {
        //--------------------------------------------------------------------------
        // If there's no additional data there's nothing to unmarshal
        //--------------------------------------------------------------------------
        if( rsp->status.bdy.dlen == 0 ) return XRootDStatus();
        //--------------------------------------------------------------------------
        // If there's not enough data to form correction-segment report an error
        //--------------------------------------------------------------------------
        if( size_t( rsp->status.bdy.dlen ) < sizeof( ServerResponseBody_pgWrCSE ) )
          return XRootDStatus( stError, errInvalidMessage, 0,
                               "kXR_status: invalid message size." );
 
        //--------------------------------------------------------------------------
        // Calculate the crc32c for the additional data
        //--------------------------------------------------------------------------
        ServerResponseBody_pgWrCSE *cse = (ServerResponseBody_pgWrCSE*)msg.GetBuffer( sizeof( ServerResponseV2 ) );
        cse->cseCRC = ntohl( cse->cseCRC );
        size_t length = rsp->status.bdy.dlen - sizeof( uint32_t );
        void*  buffer = msg.GetBuffer( sizeof( ServerResponseV2 ) + sizeof( uint32_t ) );
        uint32_t crcval = XrdOucCRC::Calc32C( buffer, length );
 
        //--------------------------------------------------------------------------
        // Do the integrity checks
        //--------------------------------------------------------------------------
        if( crcval != cse->cseCRC )
        {
          return XRootDStatus( stError, errDataError, 0, "kXR_status response header "
                               "corrupted (crc32c integrity check failed)." );
        }
 
        cse->dlFirst = ntohs( cse->dlFirst );
        cse->dlLast  = ntohs( cse->dlLast );
 
        size_t pgcnt = ( rsp->status.bdy.dlen  - sizeof( ServerResponseBody_pgWrCSE ) ) /
                       sizeof( kXR_int64 );
        kXR_int64 *pgoffs = (kXR_int64*)msg.GetBuffer( sizeof( ServerResponseV2 ) +
                                                        sizeof( ServerResponseBody_pgWrCSE ) );
 
        for( size_t i = 0; i < pgcnt; ++i )
          pgoffs[i] = ntohll( pgoffs[i] );
 
        return XRootDStatus();
        break;
      }
 
      default:
        break;
    }
 
    return XRootDStatus( stError, errNotSupported );
  }

References ServerResponseStatus::bdy, XrdOucCRC::Calc32C(), ServerResponseBody_pgWrCSE::cseCRC, ServerResponseBody_Status::dlen, ServerResponseBody_pgWrCSE::dlFirst, ServerResponseBody_pgWrCSE::dlLast, XrdCl::errDataError, XrdCl::errInvalidMessage, XrdCl::errNotSupported, XrdCl::Buffer::GetBuffer(), kXR_1stRequest, kXR_pgwrite, ServerResponseBody_Status::requestid, ServerResponseV2::status, and XrdCl::stError.

Referenced by GetMore().

Here is the call graph for this function:

Here is the caller graph for this function:

UnMarshallBody()

XRootDStatus XrdCl::XRootDTransport::UnMarshallBody ( Message * msg, uint16_t reqType )

static

Unmarshall the body of the incoming message.

Definition at line 1290 of file XrdClXRootDTransport.cc.

  {
    ServerResponse *m = (ServerResponse *)msg->GetBuffer();
 
    //--------------------------------------------------------------------------
    // kXR_ok
    //--------------------------------------------------------------------------
    if( m->hdr.status == kXR_ok )
    {
      switch( reqType )
      {
        //----------------------------------------------------------------------
        // kXR_protocol
        //----------------------------------------------------------------------
        case kXR_protocol:
          if( m->hdr.dlen < 8 )
            return XRootDStatus( stError, errInvalidMessage, 0, "kXR_protocol: body too short." );
          m->body.protocol.pval  = ntohl( m->body.protocol.pval );
          m->body.protocol.flags = ntohl( m->body.protocol.flags );
          break;
      }
    }
    //--------------------------------------------------------------------------
    // kXR_error
    //--------------------------------------------------------------------------
    else if( m->hdr.status == kXR_error )
    {
      if( m->hdr.dlen < 4 )
        return XRootDStatus( stError, errInvalidMessage, 0, "kXR_error: body too short." );
      m->body.error.errnum = ntohl( m->body.error.errnum );
    }
 
    //--------------------------------------------------------------------------
    // kXR_wait
    //--------------------------------------------------------------------------
    else if( m->hdr.status == kXR_wait )
    {
      if( m->hdr.dlen < 4 )
        return XRootDStatus( stError, errInvalidMessage, 0, "kXR_wait: body too short." );
      m->body.wait.seconds = htonl( m->body.wait.seconds );
    }
 
    //--------------------------------------------------------------------------
    // kXR_redirect
    //--------------------------------------------------------------------------
    else if( m->hdr.status == kXR_redirect )
    {
      if( m->hdr.dlen < 4 )
        return XRootDStatus( stError, errInvalidMessage, 0, "kXR_redirect: body too short." );
      m->body.redirect.port = htonl( m->body.redirect.port );
    }
 
    //--------------------------------------------------------------------------
    // kXR_waitresp
    //--------------------------------------------------------------------------
    else if( m->hdr.status == kXR_waitresp )
    {
      if( m->hdr.dlen < 4 )
        return XRootDStatus( stError, errInvalidMessage, 0, "kXR_waitresp: body too short." );
      m->body.waitresp.seconds = htonl( m->body.waitresp.seconds );
    }
 
    //--------------------------------------------------------------------------
    // kXR_attn
    //--------------------------------------------------------------------------
    else if( m->hdr.status == kXR_attn )
    {
      if( m->hdr.dlen < 4 )
        return XRootDStatus( stError, errInvalidMessage, 0, "kXR_attn: body too short." );
      m->body.attn.actnum = htonl( m->body.attn.actnum );
    }
 
    return XRootDStatus();
  }

References ServerResponse::body, ServerResponseHeader::dlen, XrdCl::errInvalidMessage, XrdCl::Buffer::GetBuffer(), ServerResponse::hdr, kXR_attn, kXR_error, kXR_ok, kXR_protocol, kXR_redirect, kXR_wait, kXR_waitresp, ServerResponseHeader::status, and XrdCl::stError.

Referenced by XrdCl::XRootDMsgHandler::Process().

Here is the call graph for this function:

Here is the caller graph for this function:

UnMarshallHeader()

void XrdCl::XRootDTransport::UnMarshallHeader ( Message & msg )

static

Unmarshall the header incoming message.

Definition at line 1507 of file XrdClXRootDTransport.cc.

  {
    ServerResponseHeader *header = (ServerResponseHeader *)msg.GetBuffer();
    header->status = ntohs( header->status );
    header->dlen   = ntohl( header->dlen );
  }

References ServerResponseHeader::dlen, XrdCl::Buffer::GetBuffer(), and ServerResponseHeader::status.

Referenced by GetHeader().

Here is the call graph for this function:

Here is the caller graph for this function:

UnMarshallRequest()

XRootDStatus XrdCl::XRootDTransport::UnMarshallRequest ( Message * msg )

static

Unmarshall the request - sometimes the requests need to be rewritten, so we need to unmarshall them

Definition at line 1269 of file XrdClXRootDTransport.cc.

  {
    if( !msg->IsMarshalled() ) return XRootDStatus( stOK, suAlreadyDone );
    // We rely on the marshaling process to be symmetric!
    // First we unmarshall the request ID and the length because
    // MarshallRequest() relies on these, and then we need to unmarshall these
    // two again, because they get marshalled in MarshallRequest().
    // All this is pretty damn ugly and should be rewritten.
    ClientRequest *req = (ClientRequest*)msg->GetBuffer();
    req->header.requestid = htons( req->header.requestid );
    req->header.dlen      = htonl( req->header.dlen );
    XRootDStatus st = MarshallRequest( msg );
    req->header.requestid = htons( req->header.requestid );
    req->header.dlen      = htonl( req->header.dlen );
    msg->SetIsMarshalled( false );
    return st;
  }

References ClientRequestHdr::dlen, XrdCl::Buffer::GetBuffer(), ClientRequest::header, XrdCl::Message::IsMarshalled(), MarshallRequest(), ClientRequestHdr::requestid, XrdCl::Message::SetIsMarshalled(), XrdCl::stOK, and XrdCl::suAlreadyDone.

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

Here is the call graph for this function:

Here is the caller graph for this function:

UnMarshalStatusBody()

XRootDStatus XrdCl::XRootDTransport::UnMarshalStatusBody ( Message & msg, uint16_t reqType )

static

Unmarshall the body of the status response.

Definition at line 1368 of file XrdClXRootDTransport.cc.

  {
    //--------------------------------------------------------------------------
    // Calculate the crc32c before the unmarshaling the body!
    //--------------------------------------------------------------------------
    ServerResponseStatus *rspst   = (ServerResponseStatus*)msg.GetBuffer();
    char   *buffer = msg.GetBuffer( 8 + sizeof( rspst->bdy.crc32c ) );
    size_t  length = rspst->hdr.dlen - sizeof( rspst->bdy.crc32c );
    uint32_t crcval = XrdOucCRC::Calc32C( buffer, length );
 
    size_t stlen = sizeof( ServerResponseStatus );
    switch( reqType )
    {
      case kXR_pgread:
      {
        stlen += sizeof( ServerResponseBody_pgRead );
        break;
      }
 
      case kXR_pgwrite:
      {
        stlen += sizeof( ServerResponseBody_pgWrite );
        break;
      }
    }
 
    if( msg.GetSize() < stlen ) return XRootDStatus( stError, errInvalidMessage, 0,
                                                      "kXR_status: invalid message size." );
 
    rspst->bdy.crc32c = ntohl( rspst->bdy.crc32c );
    rspst->bdy.dlen   = ntohl( rspst->bdy.dlen );
 
    switch( reqType )
    {
      case kXR_pgread:
      {
        ServerResponseBody_pgRead *pgrdbdy = (ServerResponseBody_pgRead*)msg.GetBuffer( sizeof( ServerResponseStatus ) );
        pgrdbdy->offset = ntohll( pgrdbdy->offset );
        break;
      }
 
      case kXR_pgwrite:
      {
        ServerResponseBody_pgWrite *pgwrtbdy = (ServerResponseBody_pgWrite*)msg.GetBuffer( sizeof( ServerResponseStatus ) );
        pgwrtbdy->offset = ntohll( pgwrtbdy->offset );
        break;
      }
    }
 
    //--------------------------------------------------------------------------
    // Do the integrity checks
    //--------------------------------------------------------------------------
    if( crcval != rspst->bdy.crc32c )
    {
      return XRootDStatus( stError, errDataError, 0, "kXR_status response header "
                           "corrupted (crc32c integrity check failed)." );
    }
 
    if( rspst->hdr.streamid[0] != rspst->bdy.streamID[0] ||
        rspst->hdr.streamid[1] != rspst->bdy.streamID[1] )
    {
      return XRootDStatus( stError, errDataError, 0, "response header corrupted "
                  "(stream ID mismatch)." );
    }
 
 
 
    if( rspst->bdy.requestid + kXR_1stRequest != reqType )
    {
      return XRootDStatus( stError, errDataError, 0, "kXR_status response header corrupted "
                  "(request ID mismatch)." );
    }
 
    return XRootDStatus();
  }

References ServerResponseStatus::bdy, XrdOucCRC::Calc32C(), ServerResponseBody_Status::crc32c, ServerResponseBody_Status::dlen, ServerResponseHeader::dlen, XrdCl::errDataError, XrdCl::errInvalidMessage, XrdCl::Buffer::GetBuffer(), XrdCl::Buffer::GetSize(), ServerResponseStatus::hdr, kXR_1stRequest, kXR_pgread, kXR_pgwrite, ServerResponseBody_pgRead::offset, ServerResponseBody_pgWrite::offset, ServerResponseBody_Status::requestid, XrdCl::stError, ServerResponseBody_Status::streamID, and ServerResponseHeader::streamid.

Referenced by XrdCl::XRootDMsgHandler::InspectStatusRsp().

Here is the call graph for this function:

Here is the caller graph for this function:

WaitBeforeExit()

void XrdCl::XRootDTransport::WaitBeforeExit ( )

virtual

Wait until the program can safely exit.

Implements XrdCl::TransportHandler.

Definition at line 1835 of file XrdClXRootDTransport.cc.

  {
    XrdSysRWLockHelper scope( pSecUnloadHandler->lock, false ); // obtain write lock
    pSecUnloadHandler->unloaded = true;
  }

Friends And Related Symbol Documentation

PluginUnloadHandler

friend struct PluginUnloadHandler

friend

Definition at line 432 of file XrdClXRootDTransport.hh.

References PluginUnloadHandler.

Referenced by XRootDTransport(), and PluginUnloadHandler.

---

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

• XrdClXRootDTransport.hh
• XrdClXRootDTransport.cc