XrdCl::Stream Class Reference

Stream. More...

#include <XrdClStream.hh>

Collaboration diagram for XrdCl::Stream:

Public Types
enum	StreamStatus {   Disconnected = 0 ,   Connected = 1 ,   Connecting = 2 ,   Error = 3 }
	Status of the stream. More...

Public Member Functions
	Stream (std::shared_ptr< URL > url, const URL &prefer=URL())
	Constructor. More...
	~Stream ()
	Destructor. More...
bool	CanCollapse (const URL &url)
void	DisableIfEmpty (uint16_t subStream)
	Disables respective uplink if empty. More...
void	Disconnect (bool force=false)
	Disconnect the stream. More...
XRootDStatus	EnableLink (PathID &path)
void	ForceConnect ()
	Force connection. More...
void	ForceError (XRootDStatus status, bool hush=false)
	Force error. More...
const std::string &	GetName () const
	Return stream name. More...
const URL *	GetURL () const
	Get the URL. More...
XRootDStatus	Initialize ()
	Initializer. More...
uint16_t	InspectStatusRsp (uint16_t stream, MsgHandler *&incHandler)
MsgHandler *	InstallIncHandler (std::shared_ptr< Message > &msg, uint16_t stream)
void	OnConnect (uint16_t subStream)
	Call back when a message has been reconstructed. More...
void	OnConnectError (uint16_t subStream, XRootDStatus status)
	On connect error. More...
void	OnError (uint16_t subStream, XRootDStatus status)
	On error. More...
void	OnIncoming (uint16_t subStream, std::shared_ptr< Message > msg, uint32_t bytesReceived)
	Call back when a message has been reconstructed. More...
void	OnMessageSent (uint16_t subStream, Message *msg, uint32_t bytesSent)
bool	OnReadTimeout (uint16_t subStream) XRD_WARN_UNUSED_RESULT
	On read timeout. More...
std::pair< Message *, MsgHandler * >	OnReadyToWrite (uint16_t subStream)
bool	OnWriteTimeout (uint16_t subStream) XRD_WARN_UNUSED_RESULT
	On write timeout. More...
Status	Query (uint16_t query, AnyObject &result)
	Query the stream. More...
void	RegisterEventHandler (ChannelEventHandler *handler)
	Register channel event handler. More...
void	RemoveEventHandler (ChannelEventHandler *handler)
	Remove a channel event handler. More...
XRootDStatus	Send (Message *msg, MsgHandler *handler, bool stateful, time_t expires)
	Queue the message for sending. More...
void	SetChannelData (AnyObject *channelData)
	Set the channel data. More...
void	SetIncomingQueue (InQueue *incomingQueue)
	Set the incoming queue. More...
void	SetJobManager (JobManager *jobManager)
	Set job manager. More...
void	SetOnDataConnectHandler (std::shared_ptr< Job > &onConnJob)
	Set the on-connect handler for data streams. More...
void	SetPoller (Poller *poller)
	Set the poller. More...
void	SetTaskManager (TaskManager *taskManager)
	Set task manager. More...
void	SetTransport (TransportHandler *transport)
	Set the transport. More...
void	Tick (time_t now)

Detailed Description

Stream.

Definition at line 51 of file XrdClStream.hh.

Member Enumeration Documentation

StreamStatus

enum XrdCl::Stream::StreamStatus

Status of the stream.

Enumerator
Disconnected	Not connected.
Connected	Connected.
Connecting	In the process of being connected.
Error	Broken.

Definition at line 57 of file XrdClStream.hh.

      {
        Disconnected    = 0,    
        Connected       = 1,    
        Connecting      = 2,    
        Error           = 3     
      };

Constructor & Destructor Documentation

Stream()

XrdCl::Stream::Stream	(	std::shared_ptr< URL >	url,
		const URL &	prefer = URL()
	)

Constructor.

Definition at line 96 of file XrdClStream.cc.

                                                           :
    pUrl( url ),
    pPrefer( prefer ),
    pTransport( 0 ),
    pPoller( 0 ),
    pTaskManager( 0 ),
    pJobManager( 0 ),
    pIncomingQueue( 0 ),
    pChannelData( 0 ),
    pLastStreamError( 0 ),
    pConnectionCount( 0 ),
    pConnectionInitTime( 0 ),
    pAddressType( Utils::IPAll ),
    pSessionId( 0 ),
    pTTLDiscJob( nullptr ),
    pSubsWaitingClose( 0 ),
    pDiscCV( 0 ),
    pDiscAllCnt( 0 ),
    pBytesSent( 0 ),
    pBytesReceived( 0 )
  {
    pConnectionStarted.tv_sec = 0; pConnectionStarted.tv_usec = 0;
    pConnectionDone.tv_sec = 0;    pConnectionDone.tv_usec = 0;
 
    std::ostringstream o;
    o << pUrl->GetHostId();
    pStreamName = o.str();
 
    pConnectionWindow  = Utils::GetIntParameter( *url, "ConnectionWindow",
                                                 DefaultConnectionWindow );
    pConnectionRetry   = Utils::GetIntParameter( *url, "ConnectionRetry",
                                                 DefaultConnectionRetry );
    pStreamErrorWindow = Utils::GetIntParameter( *url, "StreamErrorWindow",
                                                 DefaultStreamErrorWindow );
 
    std::string netStack = Utils::GetStringParameter( *url, "NetworkStack",
                                                      DefaultNetworkStack );
 
    pAddressType = Utils::String2AddressType( netStack );
    if( pAddressType == Utils::AddressType::IPAuto )
    {
      XrdNetUtils::NetProt stacks = XrdNetUtils::NetConfig( XrdNetUtils::NetType::qryINIF );
      if( !( stacks & XrdNetUtils::hasIP64 ) )
      {
        if( stacks & XrdNetUtils::hasIPv4 )
          pAddressType = Utils::AddressType::IPv4;
        else if( stacks & XrdNetUtils::hasIPv6 )
          pAddressType = Utils::AddressType::IPv6;
      }
    }
 
    Log *log = DefaultEnv::GetLog();
    log->Debug( PostMasterMsg, "[%s] Stream parameters: Network Stack: %s, "
                "Connection Window: %d, ConnectionRetry: %d, Stream Error "
                "Window: %d", pStreamName.c_str(), netStack.c_str(),
                pConnectionWindow, pConnectionRetry, pStreamErrorWindow );
  }

References XrdCl::Log::Debug(), XrdCl::DefaultConnectionRetry, XrdCl::DefaultConnectionWindow, XrdCl::DefaultNetworkStack, XrdCl::DefaultStreamErrorWindow, XrdCl::Utils::GetIntParameter(), XrdCl::DefaultEnv::GetLog(), XrdCl::Utils::GetStringParameter(), XrdNetUtils::hasIP64, XrdNetUtils::hasIPv4, XrdNetUtils::hasIPv6, XrdNetUtils::NetConfig(), XrdCl::PostMasterMsg, and XrdCl::Utils::String2AddressType().

Here is the call graph for this function:

~Stream()

XrdCl::Stream::~Stream

(

)

Destructor.

Definition at line 157 of file XrdClStream.cc.

  {
    Disconnect( true );
 
    Log *log = DefaultEnv::GetLog();
    log->Debug( PostMasterMsg, "[%s] Destroying stream",
                pStreamName.c_str() );
 
    MonitorDisconnection( XRootDStatus() );
 
    SubStreamList::iterator it;
    for( it = pSubStreams.begin(); it != pSubStreams.end(); ++it )
      delete *it;
  }

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

Here is the call graph for this function:

Member Function Documentation

CanCollapse()

bool XrdCl::Stream::CanCollapse

(

const URL &

url

)

Returns

: true is this channel can be collapsed using this URL, false otherwise

Definition at line 1353 of file XrdClStream.cc.

  {
    Log *log = DefaultEnv::GetLog();
 
    //--------------------------------------------------------------------------
    // Resolve all the addresses of the host we're supposed to connect to
    //--------------------------------------------------------------------------
    std::vector<XrdNetAddr> prefaddrs;
    XRootDStatus st = Utils::GetHostAddresses( prefaddrs, url, pAddressType );
    if( !st.IsOK() )
    {
      log->Error( PostMasterMsg, "[%s] Unable to resolve IP address for %s."
                  , pStreamName.c_str(), url.GetHostName().c_str() );
      return false;
    }
 
    //--------------------------------------------------------------------------
    // Resolve all the addresses of the alias
    //--------------------------------------------------------------------------
    std::vector<XrdNetAddr> aliasaddrs;
    st = Utils::GetHostAddresses( aliasaddrs, *pUrl, pAddressType );
    if( !st.IsOK() )
    {
      log->Error( PostMasterMsg, "[%s] Unable to resolve IP address for %s."
                  , pStreamName.c_str(), pUrl->GetHostName().c_str() );
      return false;
    }
 
    //--------------------------------------------------------------------------
    // Now check if the preferred host is part of the alias
    //--------------------------------------------------------------------------
    auto itr = prefaddrs.begin();
    for( ; itr != prefaddrs.end() ; ++itr )
    {
      auto itr2 = aliasaddrs.begin();
      for( ; itr2 != aliasaddrs.end() ; ++itr2 )
        if( itr->Same( &*itr2 ) ) return true;
    }
 
    return false;
  }

References XrdCl::Log::Error(), XrdCl::Utils::GetHostAddresses(), XrdCl::URL::GetHostName(), XrdCl::DefaultEnv::GetLog(), XrdCl::Status::IsOK(), and XrdCl::PostMasterMsg.

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

Here is the call graph for this function:

Here is the caller graph for this function:

DisableIfEmpty()

void XrdCl::Stream::DisableIfEmpty

(

uint16_t

subStream

)

Disables respective uplink if empty.

Definition at line 607 of file XrdClStream.cc.

  {
    XrdSysMutexHelper scopedLock( pMutex );
    Log *log = DefaultEnv::GetLog();
 
    if( pSubStreams[subStream]->outQueue->IsEmpty() )
    {
      log->Dump( PostMasterMsg, "[%s] All messages consumed, disable uplink",
                 pSubStreams[subStream]->socket->GetStreamName().c_str() );
      pSubStreams[subStream]->socket->DisableUplink();
    }
  }

References XrdCl::Log::Dump(), XrdCl::DefaultEnv::GetLog(), and XrdCl::PostMasterMsg.

Referenced by XrdCl::AsyncSocketHandler::OnWrite().

Here is the call graph for this function:

Here is the caller graph for this function:

Disconnect()

void XrdCl::Stream::Disconnect

(

bool

force = false

)

Disconnect the stream.

Definition at line 367 of file XrdClStream.cc.

  {
    //--------------------------------------------------------------------------
    // See comment about deadlocks in ForceError() method. We don't expect
    // to be called from a callback thread.
    //--------------------------------------------------------------------------
    XrdSysCondVarHelper discLock( pDiscCV );
    while ( pDiscAllCnt )
    {
      pDiscCV.Wait();
    }
    ++pDiscAllCnt;
    discLock.UnLock();
 
    XrdSysMutexHelper scopedLock( pMutex );
    SubStreamList::iterator it;
    for( it = pSubStreams.begin(); it != pSubStreams.end(); ++it )
    {
      (*it)->socket->Close();
      (*it)->status = Socket::Disconnected;
    }
    pSubsWaitingClose = 0;
 
    scopedLock.UnLock();
    discLock.Lock( &pDiscCV );
    --pDiscAllCnt;
    pDiscCV.Signal();
  }

References XrdCl::Socket::Disconnected, XrdSysCondVarHelper::Lock(), XrdSysCondVar::Signal(), XrdSysCondVarHelper::UnLock(), XrdSysMutexHelper::UnLock(), and XrdSysCondVar::Wait().

Referenced by ~Stream().

Here is the call graph for this function:

Here is the caller graph for this function:

EnableLink()

XRootDStatus XrdCl::Stream::EnableLink

(

PathID &

path

)

Connect if needed, otherwise make sure that the underlying socket handler gets write readiness events, it will update the path with what it has actually enabled

Definition at line 191 of file XrdClStream.cc.

  {
    XrdSysMutexHelper scopedLock( pMutex );
 
    //--------------------------------------------------------------------------
    // We are in the process of connecting the main stream, so we do nothing
    // because when the main stream connection is established it will connect
    // all the other streams
    //--------------------------------------------------------------------------
    if( pSubStreams[0]->status == Socket::Connecting )
      return XRootDStatus();
 
    //--------------------------------------------------------------------------
    // The main stream is connected, so we can verify whether we have
    // the up and the down stream connected and ready to handle data.
    // If anything is not right we fall back to stream 0.
    //--------------------------------------------------------------------------
    if( pSubStreams[0]->status == Socket::Connected )
    {
      if( pSubStreams[path.down]->status != Socket::Connected )
        path.down = 0;
 
      if( pSubStreams[path.up]->status == Socket::Disconnected )
      {
        path.up = 0;
        return pSubStreams[0]->socket->EnableUplink();
      }
 
      if( pSubStreams[path.up]->status == Socket::Connected )
        return pSubStreams[path.up]->socket->EnableUplink();
 
      return XRootDStatus();
    }
 
    //--------------------------------------------------------------------------
    // The main stream is not connected, we need to check whether enough time
    // has passed since we last encountered an error (if any) so that we could
    // re-attempt the connection
    //--------------------------------------------------------------------------
    Log *log = DefaultEnv::GetLog();
    time_t now = ::time(0);
 
    if( now-pLastStreamError < pStreamErrorWindow )
      return pLastFatalError;
 
    gettimeofday( &pConnectionStarted, 0 );
    ++pConnectionCount;
 
    //--------------------------------------------------------------------------
    // Resolve all the addresses of the host we're supposed to connect to
    //--------------------------------------------------------------------------
    XRootDStatus st = Utils::GetHostAddresses( pAddresses, *pUrl, pAddressType );
    if( !st.IsOK() )
    {
      log->Error( PostMasterMsg, "[%s] Unable to resolve IP address for "
                  "the host", pStreamName.c_str() );
      pLastStreamError = now;
      st.status        = stFatal;
      pLastFatalError  = st;
      return st;
    }
 
    if( pPrefer.IsValid() )
    {
      std::vector<XrdNetAddr> addrresses;
      XRootDStatus st = Utils::GetHostAddresses( addrresses, pPrefer, pAddressType );
      if( !st.IsOK() )
      {
        log->Error( PostMasterMsg, "[%s] Unable to resolve IP address for %s",
                    pStreamName.c_str(), pPrefer.GetHostName().c_str() );
      }
      else
      {
        std::vector<XrdNetAddr> tmp;
        tmp.reserve( pAddresses.size() );
        // first add all remaining addresses
        auto itr = pAddresses.begin();
        for( ; itr != pAddresses.end() ; ++itr )
        {
          if( !HasNetAddr( *itr, addrresses ) )
            tmp.push_back( *itr );
        }
        // then copy all 'preferred' addresses
        std::copy( addrresses.begin(), addrresses.end(), std::back_inserter( tmp ) );
        // and keep the result
        pAddresses.swap( tmp );
      }
    }
 
    Utils::LogHostAddresses( log, PostMasterMsg, pUrl->GetHostId(),
                             pAddresses );
 
    while( !pAddresses.empty() )
    {
      pSubStreams[0]->socket->SetAddress( pAddresses.back() );
      pAddresses.pop_back();
      pConnectionInitTime = ::time( 0 );
      st = pSubStreams[0]->socket->Connect( pConnectionWindow );
      if( st.IsOK() )
      {
        pSubStreams[0]->status = Socket::Connecting;
        break;
      }
    }
    return st;
  }

References XrdCl::Socket::Connected, XrdCl::Socket::Connecting, XrdCl::Socket::Disconnected, XrdCl::PathID::down, XrdCl::Log::Error(), XrdCl::Utils::GetHostAddresses(), XrdCl::URL::GetHostName(), XrdCl::DefaultEnv::GetLog(), XrdCl::Status::IsOK(), XrdCl::URL::IsValid(), XrdCl::Utils::LogHostAddresses(), XrdCl::PostMasterMsg, XrdCl::Status::status, XrdCl::stFatal, and XrdCl::PathID::up.

Referenced by ForceConnect(), OnConnectError(), OnError(), and Send().

Here is the call graph for this function:

Here is the caller graph for this function:

ForceConnect()

void XrdCl::Stream::ForceConnect

(

)

Force connection.

Definition at line 351 of file XrdClStream.cc.

  {
    XrdSysMutexHelper scopedLock( pMutex );
    if( pSubStreams[0]->status == Socket::Connecting )
    {
      pSubStreams[0]->status = Socket::Disconnected;
      XrdCl::PathID path( 0, 0 );
      XrdCl::XRootDStatus st = EnableLink( path );
      if( !st.IsOK() )
        OnConnectError( 0, st );
    }
  }

References XrdCl::Socket::Connecting, XrdCl::Socket::Disconnected, EnableLink(), XrdCl::Status::IsOK(), and OnConnectError().

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

Here is the call graph for this function:

Here is the caller graph for this function:

ForceError()

void XrdCl::Stream::ForceError	(	XRootDStatus	status,
		bool	hush = false
	)

Force error.

Definition at line 1042 of file XrdClStream.cc.

  {
    //----------------------------------------------------------------------
    // We can be called in two ways: first is by by a non-poller thread,
    // with errOperationInterrupted as error as part of ForceDisconnect. In
    // which case the Stream will be destoryed shortly after we return. The
    // second way is call by a poller thread with another type of error.
    // Further, when we call socket handler Close() for a socket handled a
    // callback running we wait for that to complete (unless it is
    // ourselves). This raises the possibility of a deadlock. We avoid this
    // by returning quickly if we detect we're in a callback thread and
    // there's already a disconnect affecting multiple streams in progress.
    //----------------------------------------------------------------------
    XrdSysCondVarHelper discLock( pDiscCV );
    if( pDiscAllCnt &&
      !( status.IsError() && status.code == errOperationInterrupted ) )
    {
      return;
    }
    while( pDiscAllCnt )
    {
      pDiscCV.Wait();
    }
    ++pDiscAllCnt;
    discLock.UnLock();
 
    XrdSysMutexHelper scopedLock( pMutex );
    Log    *log = DefaultEnv::GetLog();
    for( size_t substream = 0; substream < pSubStreams.size(); ++substream )
    {
      if( pSubStreams[substream]->status != Socket::Connected ) continue;
      pSubStreams[substream]->socket->Close();
      pSubStreams[substream]->status = Socket::Disconnected;
 
      if( !hush )
        log->Debug( PostMasterMsg, "[%s] Forcing error on disconnect: %s.",
                    pStreamName.c_str(), status.ToString().c_str() );
 
      //--------------------------------------------------------------------
      // Reinsert the stuff that we have failed to sent
      //--------------------------------------------------------------------
      if( pSubStreams[substream]->outMsgHelper.msg )
      {
        OutQueue::MsgHelper &h = pSubStreams[substream]->outMsgHelper;
        pSubStreams[substream]->outQueue->PushFront( h.msg, h.handler, h.expires,
                                                     h.stateful );
        pIncomingQueue->RemoveMessageHandler(h.handler);
        pSubStreams[substream]->outMsgHelper.Reset();
      }
 
      //--------------------------------------------------------------------
      // Reinsert the receiving handler and reset any partially read partial
      //--------------------------------------------------------------------
      if( pSubStreams[substream]->inMsgHelper.handler )
      {
        InMessageHelper &h = pSubStreams[substream]->inMsgHelper;
        pIncomingQueue->ReAddMessageHandler( h.handler, h.expires );
        XRootDMsgHandler *xrdHandler = dynamic_cast<XRootDMsgHandler*>( h.handler );
        if( xrdHandler ) xrdHandler->PartialReceived();
        h.Reset();
      }
    }
 
    pConnectionCount = 0;
    pSubsWaitingClose = 0;
 
    //------------------------------------------------------------------------
    // We're done here, unlock the stream mutex to avoid deadlocks and
    // report the disconnection event to the handlers
    //------------------------------------------------------------------------
    log->Debug( PostMasterMsg, "[%s] Reporting disconnection to queued "
                "message handlers.", pStreamName.c_str() );
 
    SubStreamList::iterator it;
    OutQueue q;
    for( it = pSubStreams.begin(); it != pSubStreams.end(); ++it )
      q.GrabItems( *(*it)->outQueue );
    scopedLock.UnLock();
 
    q.Report( status );
 
    pIncomingQueue->ReportStreamEvent( MsgHandler::Broken, status );
    pChannelEvHandlers.ReportEvent( ChannelEventHandler::StreamBroken, status );
 
    discLock.Lock( &pDiscCV );
    --pDiscAllCnt;
    pDiscCV.Signal();
  }

References XrdCl::MsgHandler::Broken, XrdCl::Status::code, XrdCl::Socket::Connected, XrdCl::Log::Debug(), XrdCl::Socket::Disconnected, XrdCl::errOperationInterrupted, XrdCl::OutQueue::MsgHelper::expires, XrdCl::InMessageHelper::expires, XrdCl::DefaultEnv::GetLog(), XrdCl::OutQueue::GrabItems(), XrdCl::OutQueue::MsgHelper::handler, XrdCl::InMessageHelper::handler, XrdCl::Status::IsError(), XrdSysCondVarHelper::Lock(), XrdCl::OutQueue::MsgHelper::msg, XrdCl::XRootDMsgHandler::PartialReceived(), XrdCl::PostMasterMsg, XrdCl::InQueue::ReAddMessageHandler(), XrdCl::InQueue::RemoveMessageHandler(), XrdCl::OutQueue::Report(), XrdCl::ChannelHandlerList::ReportEvent(), XrdCl::InQueue::ReportStreamEvent(), XrdCl::InMessageHelper::Reset(), XrdSysCondVar::Signal(), XrdCl::OutQueue::MsgHelper::stateful, XrdCl::ChannelEventHandler::StreamBroken, XrdCl::Status::ToString(), XrdSysCondVarHelper::UnLock(), XrdSysMutexHelper::UnLock(), and XrdSysCondVar::Wait().

Referenced by XrdCl::Channel::ForceDisconnect(), and XrdCl::AsyncSocketHandler::OnHeaderCorruption().

Here is the call graph for this function:

Here is the caller graph for this function:

GetName()

const std::string& XrdCl::Stream::GetName ( ) const

inline

Return stream name.

Definition at line 170 of file XrdClStream.hh.

      {
        return pStreamName;
      }

GetURL()

const URL* XrdCl::Stream::GetURL ( ) const

inline

Get the URL.

Definition at line 157 of file XrdClStream.hh.

      {
        return pUrl.get();
      }

Referenced by XrdCl::AsyncSocketHandler::OnConnectionReturn().

Here is the caller graph for this function:

Initialize()

XRootDStatus XrdCl::Stream::Initialize

(

)

Initializer.

Definition at line 175 of file XrdClStream.cc.

  {
    if( !pTransport || !pPoller || !pChannelData )
      return XRootDStatus( stError, errUninitialized );
 
    AsyncSocketHandler *s = new AsyncSocketHandler( *pUrl, pPoller, pTransport,
                                                    pChannelData, 0, this );
    pSubStreams.push_back( new SubStreamData() );
    pSubStreams[0]->socket = s;
    return XRootDStatus();
  }

References XrdCl::errUninitialized, and XrdCl::stError.

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

Here is the caller graph for this function:

InspectStatusRsp()

uint16_t XrdCl::Stream::InspectStatusRsp	(	uint16_t	stream,
		MsgHandler *&	incHandler
	)

In case the message is a kXR_status response it needs further attention

Returns

: a MsgHandler in case we need to read out raw data

Definition at line 1322 of file XrdClStream.cc.

  {
    InMessageHelper &mh = pSubStreams[stream]->inMsgHelper;
    if( !mh.handler )
      return MsgHandler::RemoveHandler;
 
    uint16_t action = mh.handler->InspectStatusRsp();
    mh.action |= action;
 
    if( action & MsgHandler::RemoveHandler )
      pIncomingQueue->RemoveMessageHandler( mh.handler );
 
    if( action & MsgHandler::Raw )
    {
      incHandler = mh.handler;
      return MsgHandler::Raw;
    }
 
    if( action & MsgHandler::Corrupted )
      return MsgHandler::Corrupted;
 
    if( action & MsgHandler::More )
      return MsgHandler::More;
 
    return MsgHandler::None;
  }

References XrdCl::InMessageHelper::action, XrdCl::MsgHandler::Corrupted, XrdCl::InMessageHelper::handler, XrdCl::MsgHandler::InspectStatusRsp(), XrdCl::MsgHandler::More, XrdCl::MsgHandler::None, XrdCl::MsgHandler::Raw, XrdCl::MsgHandler::RemoveHandler, and XrdCl::InQueue::RemoveMessageHandler().

Referenced by XrdCl::AsyncMsgReader::Read().

Here is the call graph for this function:

Here is the caller graph for this function:

InstallIncHandler()

MsgHandler * XrdCl::Stream::InstallIncHandler	(	std::shared_ptr< Message > &	msg,
		uint16_t	stream
	)

Install a message handler for the given message if there is one available, if the handler want's to be called in the raw mode it will be returned, the message ownership flag is returned in any case

Parameters

msg	message header
stream	stream concerned

Returns

a pair containing the handler and ownership flag

Definition at line 1301 of file XrdClStream.cc.

  {
    InMessageHelper &mh = pSubStreams[stream]->inMsgHelper;
    if( !mh.handler )
      mh.handler = pIncomingQueue->GetHandlerForMessage( msg,
                                                         mh.expires,
                                                         mh.action );
 
    if( !mh.handler )
      return nullptr;
 
    if( mh.action & MsgHandler::Raw )
      return mh.handler;
    return nullptr;
  }

References XrdCl::InMessageHelper::action, XrdCl::InMessageHelper::expires, XrdCl::InQueue::GetHandlerForMessage(), XrdCl::InMessageHelper::handler, and XrdCl::MsgHandler::Raw.

Referenced by XrdCl::AsyncMsgReader::Read().

Here is the call graph for this function:

Here is the caller graph for this function:

OnConnect()

void XrdCl::Stream::OnConnect

(

uint16_t

subStream

)

Call back when a message has been reconstructed.

Definition at line 645 of file XrdClStream.cc.

  {
    XrdSysMutexHelper scopedLock( pMutex );
    if( subStream == 0 )
    {
      int nsubconn = 0;
      if( pSubStreams.size() > 1 )
      {
        for( size_t i = 1; i < pSubStreams.size(); ++i )
          if( pSubStreams[i]->status != Socket::Disconnected ) nsubconn++;
      }
      if( nsubconn )
      {
        pSubsWaitingClose = nsubconn;
        pSubStreams[0]->socket->DisableUplink();
        return;
      }
      else
        pSubStreams[0]->socket->EnableUplink();
    }
    else
    {
      if( pSubsWaitingClose > 0 )
      {
        pSubStreams[subStream]->socket->Close();
        pSubStreams[subStream]->status = Socket::Disconnected;
        if( --pSubsWaitingClose == 0 )
        {
          scopedLock.UnLock();
          OnConnect( 0 );
        }
        return;
      }
    }
 
    pSubStreams[subStream]->status = Socket::Connected;
 
    std::string ipstack( pSubStreams[0]->socket->GetIpStack() );
    Log *log = DefaultEnv::GetLog();
    log->Debug( PostMasterMsg, "[%s] Stream %d connected (%s).", pStreamName.c_str(),
                subStream, ipstack.c_str() );
 
    if( subStream == 0 )
    {
      pLastStreamError = 0;
      pLastFatalError  = XRootDStatus();
      pConnectionCount = 0;
      uint16_t numSub = pTransport->SubStreamNumber( *pChannelData );
      pSessionId = ++sSessCntGen;
 
      //------------------------------------------------------------------------
      // Create the streams if they don't exist yet
      //------------------------------------------------------------------------
      if( pSubStreams.size() == 1 && numSub > 1 )
      {
        for( uint16_t i = 1; i < numSub; ++i )
        {
          URL url = pTransport->GetBindPreference( *pUrl, *pChannelData );
          AsyncSocketHandler *s = new AsyncSocketHandler( url, pPoller, pTransport,
                                                          pChannelData, i, this );
          pSubStreams.push_back( new SubStreamData() );
          pSubStreams[i]->socket = s;
        }
      }
 
      //------------------------------------------------------------------------
      // Connect the extra streams, if we fail we move all the outgoing items
      // to stream 0, we don't need to enable the uplink here, because it
      // should be already enabled after the handshaking process is completed.
      //------------------------------------------------------------------------
      if( pSubStreams.size() > 1 )
      {
        log->Debug( PostMasterMsg, "[%s] Attempting to connect %zu additional streams.",
                    pStreamName.c_str(), pSubStreams.size() - 1 );
        for( size_t i = 1; i < pSubStreams.size(); ++i )
        {
          pSubStreams[i]->socket->SetAddress( pSubStreams[0]->socket->GetAddress() );
          XRootDStatus st = pSubStreams[i]->socket->Connect( pConnectionWindow );
          if( !st.IsOK() )
          {
            pSubStreams[0]->outQueue->GrabItems( *pSubStreams[i]->outQueue );
            // mark as disconnected. We don't try to actively Close here as
            // we're in a poller callback thread and the i'th substream here
            // may be hadled by a different poller callback thread, raising
            // the possibility of deadlock.
            pSubStreams[i]->status = Socket::Disconnected;
          }
          else
          {
            pSubStreams[i]->status = Socket::Connecting;
          }
        }
      }
 
      //------------------------------------------------------------------------
      // Inform monitoring
      //------------------------------------------------------------------------
      pBytesSent     = 0;
      pBytesReceived = 0;
      gettimeofday( &pConnectionDone, 0 );
      Monitor *mon = DefaultEnv::GetMonitor();
      if( mon )
      {
        Monitor::ConnectInfo i;
        i.server  = pUrl->GetHostId();
        i.sTOD    = pConnectionStarted;
        i.eTOD    = pConnectionDone;
        i.streams = pSubStreams.size();
 
        AnyObject    qryResult;
        std::string *qryResponse = nullptr;
        pTransport->Query( TransportQuery::Auth, qryResult, *pChannelData );
        qryResult.Get( qryResponse );
 
        if (qryResponse) {
          i.auth = *qryResponse;
          delete qryResponse;
        } else {
          i.auth = "";
        }
 
        mon->Event( Monitor::EvConnect, &i );
      }
 
      //------------------------------------------------------------------------
      // For every connected control-stream call the global on-connect handler
      //------------------------------------------------------------------------
      XrdCl::DefaultEnv::GetPostMaster()->NotifyConnectHandler( *pUrl );
    }
    else if( pOnDataConnJob )
    {
      //------------------------------------------------------------------------
      // For every connected data-stream call the on-connect handler
      //------------------------------------------------------------------------
      pJobManager->QueueJob( pOnDataConnJob.get(), 0 );
    }
  }

References XrdCl::Monitor::ConnectInfo::auth, XrdCl::TransportQuery::Auth, XrdCl::Socket::Connected, XrdCl::Socket::Connecting, XrdCl::Log::Debug(), XrdCl::Socket::Disconnected, XrdCl::Monitor::ConnectInfo::eTOD, XrdCl::Monitor::EvConnect, XrdCl::Monitor::Event(), XrdCl::AnyObject::Get(), XrdCl::TransportHandler::GetBindPreference(), XrdCl::DefaultEnv::GetLog(), XrdCl::DefaultEnv::GetMonitor(), XrdCl::DefaultEnv::GetPostMaster(), XrdCl::Status::IsOK(), XrdCl::PostMaster::NotifyConnectHandler(), XrdCl::PostMasterMsg, XrdCl::TransportHandler::Query(), XrdCl::JobManager::QueueJob(), XrdCl::Monitor::ConnectInfo::server, XrdCl::Monitor::ConnectInfo::sTOD, XrdCl::Monitor::ConnectInfo::streams, XrdCl::TransportHandler::SubStreamNumber(), and XrdSysMutexHelper::UnLock().

Referenced by XrdCl::AsyncSocketHandler::HandShakeNextStep(), OnConnectError(), OnError(), and OnReadTimeout().

Here is the call graph for this function:

Here is the caller graph for this function:

OnConnectError()

void XrdCl::Stream::OnConnectError	(	uint16_t	subStream,
		XRootDStatus	status
	)

On connect error.

Definition at line 786 of file XrdClStream.cc.

  {
    XrdSysMutexHelper scopedLock( pMutex );
    Log *log = DefaultEnv::GetLog();
    pSubStreams[subStream]->socket->Close();
    time_t now = ::time(0);
 
    //--------------------------------------------------------------------------
    // For every connection error call the global connection error handler
    //--------------------------------------------------------------------------
    XrdCl::DefaultEnv::GetPostMaster()->NotifyConnErrHandler( *pUrl, status );
 
    //--------------------------------------------------------------------------
    // If we connected subStream == 0 and cannot connect >0 then we just give
    // up and move the outgoing messages to another queue
    //--------------------------------------------------------------------------
    if( subStream > 0 )
    {
      const Socket::SocketStatus oldstate = pSubStreams[subStream]->status;
      pSubStreams[subStream]->status = Socket::Disconnected;
 
      if( pSubsWaitingClose > 0 && oldstate != Socket::Disconnected )
      {
        if( --pSubsWaitingClose == 0 )
        {
          scopedLock.UnLock();
          OnConnect( 0 );
        }
        return;
      }
 
      pSubStreams[0]->outQueue->GrabItems( *pSubStreams[subStream]->outQueue );
      if( pSubStreams[0]->status == Socket::Connected )
      {
        XRootDStatus st = pSubStreams[0]->socket->EnableUplink();
        if( !st.IsOK() )
          OnFatalError( 0, st, scopedLock );
        return;
      }
 
      if( pSubStreams[0]->status == Socket::Connecting )
        return;
 
      OnFatalError( subStream, status, scopedLock );
      return;
    }
 
    //--------------------------------------------------------------------------
    // Check if we still have time to try and do something in the current window
    //--------------------------------------------------------------------------
    time_t elapsed = now-pConnectionInitTime;
    log->Error( PostMasterMsg, "[%s] elapsed = %lld, pConnectionWindow = %d seconds.",
                pStreamName.c_str(), (long long) elapsed, pConnectionWindow );
 
    //------------------------------------------------------------------------
    // If we have some IP addresses left we try them
    //------------------------------------------------------------------------
    if( !pAddresses.empty() )
    {
      XRootDStatus st;
      do
      {
        pSubStreams[0]->socket->SetAddress( pAddresses.back() );
        pAddresses.pop_back();
        pConnectionInitTime = ::time( 0 );
        st = pSubStreams[0]->socket->Connect( pConnectionWindow );
      }
      while( !pAddresses.empty() && !st.IsOK() );
 
      if( !st.IsOK() )
        OnFatalError( subStream, st, scopedLock );
 
      return;
    }
    //------------------------------------------------------------------------
    // If we still can retry with the same host name, we sleep until the end
    // of the connection window and try
    //------------------------------------------------------------------------
    else if( elapsed < pConnectionWindow && pConnectionCount < pConnectionRetry
             && !status.IsFatal() )
    {
      log->Info( PostMasterMsg, "[%s] Attempting reconnection in %lld seconds.",
                 pStreamName.c_str(), (long long) (pConnectionWindow - elapsed) );
 
      Task *task = new ::StreamConnectorTask( *pUrl, pStreamName );
      pTaskManager->RegisterTask( task, pConnectionInitTime+pConnectionWindow );
      return;
    }
    //--------------------------------------------------------------------------
    // We are out of the connection window, the only thing we can do here
    // is re-resolving the host name and retrying if we still can
    //--------------------------------------------------------------------------
    else if( pConnectionCount < pConnectionRetry && !status.IsFatal() )
    {
      pAddresses.clear();
      pSubStreams[0]->status = Socket::Disconnected;
      PathID path( 0, 0 );
      XRootDStatus st = EnableLink( path );
      if( !st.IsOK() )
        OnFatalError( subStream, st, scopedLock );
      return;
    }
 
    //--------------------------------------------------------------------------
    // Else, we fail
    //--------------------------------------------------------------------------
    OnFatalError( subStream, status, scopedLock );
  }

References XrdCl::Socket::Connected, XrdCl::Socket::Connecting, XrdCl::Socket::Disconnected, EnableLink(), XrdCl::Log::Error(), XrdCl::DefaultEnv::GetLog(), XrdCl::DefaultEnv::GetPostMaster(), XrdCl::Log::Info(), XrdCl::Status::IsFatal(), XrdCl::Status::IsOK(), XrdCl::PostMaster::NotifyConnErrHandler(), OnConnect(), XrdCl::PostMasterMsg, XrdCl::TaskManager::RegisterTask(), and XrdSysMutexHelper::UnLock().

Referenced by ForceConnect(), XrdCl::AsyncSocketHandler::OnConnectionReturn(), and XrdCl::AsyncSocketHandler::OnFaultWhileHandshaking().

Here is the call graph for this function:

Here is the caller graph for this function:

OnError()

void XrdCl::Stream::OnError	(	uint16_t	subStream,
		XRootDStatus	status
	)

On error.

Definition at line 898 of file XrdClStream.cc.

  {
    //--------------------------------------------------------------------------
    // See comment about deadlocks in ForceError() method. We expect to be
    // called form a callback thread. However we take care to only potentially
    // disconnect the socket for our own subStream. We require no ongoing
    // disconnect of all substreams and ensure that remains true throughout
    // our execution by releasing discLock only after acquiring pMutex.
    //--------------------------------------------------------------------------
 
    XrdSysCondVarHelper discLock( pDiscCV );
    if( pDiscAllCnt ) return;
 
    XrdSysMutexHelper scopedLock( pMutex );
    discLock.UnLock();
    Log *log = DefaultEnv::GetLog();
    const Socket::SocketStatus oldstate = pSubStreams[subStream]->status;
    pSubStreams[subStream]->socket->Close();
    pSubStreams[subStream]->status = Socket::Disconnected;
 
    log->Debug( PostMasterMsg, "[%s] Recovering error for stream #%d: %s.",
                pStreamName.c_str(), subStream, status.ToString().c_str() );
 
    //--------------------------------------------------------------------------
    // Reinsert the stuff that we have failed to sent
    //--------------------------------------------------------------------------
    if( pSubStreams[subStream]->outMsgHelper.msg )
    {
      OutQueue::MsgHelper &h = pSubStreams[subStream]->outMsgHelper;
      pSubStreams[subStream]->outQueue->PushFront( h.msg, h.handler, h.expires,
                                                   h.stateful );
      pIncomingQueue->RemoveMessageHandler(h.handler);
      pSubStreams[subStream]->outMsgHelper.Reset();
    }
 
    //--------------------------------------------------------------------------
    // Reinsert the receiving handler and reset any partially read partial
    //--------------------------------------------------------------------------
    if( pSubStreams[subStream]->inMsgHelper.handler )
    {
      InMessageHelper &h = pSubStreams[subStream]->inMsgHelper;
      pIncomingQueue->ReAddMessageHandler( h.handler, h.expires );
      XRootDMsgHandler *xrdHandler = dynamic_cast<XRootDMsgHandler*>( h.handler );
      if( xrdHandler ) xrdHandler->PartialReceived();
      h.Reset();
    }
 
    //--------------------------------------------------------------------------
    // We are dealing with an error of a peripheral stream. If we don't have
    // anything to send don't bother recovering. Otherwise move the requests
    // to stream 0 if possible.
    //--------------------------------------------------------------------------
    if( subStream > 0 )
    {
      if( pSubsWaitingClose > 0 && oldstate != Socket::Disconnected )
      {
        if( --pSubsWaitingClose == 0 )
        {
          scopedLock.UnLock();
          OnConnect( 0 );
        }
        return;
      }
 
      if( pSubStreams[0]->status != Socket::Disconnected )
      {
        pSubStreams[subStream]->socket->SetAddress( pSubStreams[0]->socket->GetAddress() );
        XRootDStatus st = pSubStreams[subStream]->socket->Connect( pConnectionWindow );
        if( !st.IsOK() )
        {
          pSubStreams[subStream]->socket->Close();
          if( pSubStreams[subStream]->outQueue->IsEmpty() )
            return;
          pSubStreams[0]->outQueue->GrabItems( *pSubStreams[subStream]->outQueue );
          if( pSubStreams[0]->status == Socket::Connected )
          {
            XRootDStatus st = pSubStreams[0]->socket->EnableUplink();
            if( !st.IsOK() )
              OnFatalError( 0, st, scopedLock );
            return;
          }
          if( pSubStreams[0]->status == Socket::Connecting )
            return;
        }
        else
        {
          pSubStreams[subStream]->status = Socket::Connecting;
          return;
        }
        OnFatalError( subStream, status, scopedLock );
        return;
      }
      if( pSubStreams[subStream]->outQueue->IsEmpty() )
        return;
      OnFatalError( subStream, status, scopedLock );
      return;
    }
 
    //--------------------------------------------------------------------------
    // If we lost the stream 0 we have lost the session, we re-enable the
    // stream if we still have things in one of the outgoing queues, otherwise
    // there is not point to recover at this point.
    //--------------------------------------------------------------------------
    if( subStream == 0 )
    {
      MonitorDisconnection( status );
 
      SubStreamList::iterator it;
      size_t outstanding = 0;
      for( it = pSubStreams.begin(); it != pSubStreams.end(); ++it )
        outstanding += (*it)->outQueue->GetSizeStateless();
 
      if( outstanding )
      {
        PathID path( 0, 0 );
        XRootDStatus st = EnableLink( path );
        if( !st.IsOK() )
        {
          OnFatalError( 0, st, scopedLock );
          return;
        }
      }
 
      //------------------------------------------------------------------------
      // We're done here, unlock the stream mutex to avoid deadlocks and
      // report the disconnection event to the handlers
      //------------------------------------------------------------------------
      log->Debug( PostMasterMsg, "[%s] Reporting disconnection to queued "
                  "message handlers.", pStreamName.c_str() );
      OutQueue q;
      for( it = pSubStreams.begin(); it != pSubStreams.end(); ++it )
        q.GrabStateful( *(*it)->outQueue );
      scopedLock.UnLock();
 
      q.Report( status );
      pIncomingQueue->ReportStreamEvent( MsgHandler::Broken, status );
      pChannelEvHandlers.ReportEvent( ChannelEventHandler::StreamBroken, status );
      return;
    }
  }

References XrdCl::MsgHandler::Broken, XrdCl::Socket::Connected, XrdCl::Socket::Connecting, XrdCl::Log::Debug(), XrdCl::Socket::Disconnected, EnableLink(), XrdCl::OutQueue::MsgHelper::expires, XrdCl::InMessageHelper::expires, XrdCl::DefaultEnv::GetLog(), XrdCl::OutQueue::GrabStateful(), XrdCl::OutQueue::MsgHelper::handler, XrdCl::InMessageHelper::handler, XrdCl::Status::IsOK(), XrdCl::OutQueue::MsgHelper::msg, OnConnect(), XrdCl::XRootDMsgHandler::PartialReceived(), XrdCl::PostMasterMsg, XrdCl::InQueue::ReAddMessageHandler(), XrdCl::InQueue::RemoveMessageHandler(), XrdCl::OutQueue::Report(), XrdCl::ChannelHandlerList::ReportEvent(), XrdCl::InQueue::ReportStreamEvent(), XrdCl::InMessageHelper::Reset(), XrdCl::OutQueue::MsgHelper::stateful, XrdCl::ChannelEventHandler::StreamBroken, XrdCl::Status::ToString(), XrdSysCondVarHelper::UnLock(), and XrdSysMutexHelper::UnLock().

Referenced by XrdCl::AsyncSocketHandler::OnFault(), and OnReadTimeout().

Here is the call graph for this function:

Here is the caller graph for this function:

OnIncoming()

void XrdCl::Stream::OnIncoming	(	uint16_t	subStream,
		std::shared_ptr< Message >	msg,
		uint32_t	bytesReceived
	)

Call back when a message has been reconstructed.

Definition at line 493 of file XrdClStream.cc.

  {
    msg->SetSessionId( pSessionId );
    pBytesReceived += bytesReceived;
 
    MsgHandler *handler = nullptr;
    uint16_t action = 0;
    {
      InMessageHelper &mh = pSubStreams[subStream]->inMsgHelper;
      handler = mh.handler;
      action = mh.action;
      mh.Reset();
    }
 
    if( !IsPartial( *msg ) )
    {
      uint32_t streamAction = pTransport->MessageReceived( *msg, subStream,
                                                           *pChannelData );
      if( streamAction & TransportHandler::DigestMsg )
        return;
 
      if( streamAction & TransportHandler::RequestClose )
      {
        RequestClose( *msg );
        return;
      }
    }
 
    Log *log = DefaultEnv::GetLog();
 
    //--------------------------------------------------------------------------
    // No handler, we discard the message ...
    //--------------------------------------------------------------------------
    if( !handler )
    {
      ServerResponse *rsp = (ServerResponse*)msg->GetBuffer();
      log->Warning( PostMasterMsg, "[%s] Discarding received message: %p "
                    "(status=%d, SID=[%d,%d]), no MsgHandler found.",
                    pStreamName.c_str(), (void*)msg.get(), rsp->hdr.status,
                    rsp->hdr.streamid[0], rsp->hdr.streamid[1] );
      return;
    }
 
    //--------------------------------------------------------------------------
    // We have a handler, so we call the callback
    //--------------------------------------------------------------------------
    log->Dump( PostMasterMsg, "[%s] Handling received message: %p.",
               pStreamName.c_str(), (void*)msg.get() );
 
    if( action & (MsgHandler::NoProcess|MsgHandler::Ignore) )
    {
      log->Dump( PostMasterMsg, "[%s] Ignoring the processing handler for: %s.",
                 pStreamName.c_str(), msg->GetObfuscatedDescription().c_str() );
 
      // if we are handling partial response we have to take down the timeout fence
      if( IsPartial( *msg ) )
      {
        XRootDMsgHandler *xrdHandler = dynamic_cast<XRootDMsgHandler*>( handler );
        if( xrdHandler ) xrdHandler->PartialReceived();
      }
 
      return;
    }
 
    Job *job = new HandleIncMsgJob( handler );
    pJobManager->QueueJob( job );
  }

References XrdCl::InMessageHelper::action, XrdCl::TransportHandler::DigestMsg, XrdCl::Log::Dump(), XrdCl::DefaultEnv::GetLog(), XrdCl::InMessageHelper::handler, ServerResponse::hdr, XrdCl::MsgHandler::Ignore, XrdCl::TransportHandler::MessageReceived(), XrdCl::MsgHandler::NoProcess, XrdCl::XRootDMsgHandler::PartialReceived(), XrdCl::PostMasterMsg, XrdCl::JobManager::QueueJob(), XrdCl::TransportHandler::RequestClose, XrdCl::InMessageHelper::Reset(), ServerResponseHeader::status, ServerResponseHeader::streamid, and XrdCl::Log::Warning().

Referenced by XrdCl::AsyncMsgReader::Read().

Here is the call graph for this function:

Here is the caller graph for this function:

OnMessageSent()

void XrdCl::Stream::OnMessageSent	(	uint16_t	subStream,
		Message *	msg,
		uint32_t	bytesSent
	)

Definition at line 623 of file XrdClStream.cc.

  {
    pTransport->MessageSent( msg, subStream, bytesSent,
                             *pChannelData );
    OutQueue::MsgHelper &h = pSubStreams[subStream]->outMsgHelper;
    pBytesSent += bytesSent;
    if( h.handler )
    {
      // ensure expiration time is assigned if still in queue
      pIncomingQueue->AssignTimeout( h.handler );
      // OnStatusReady may cause the handler to delete itself, in
      // which case the handler or the user callback may also delete msg
      h.handler->OnStatusReady( msg, XRootDStatus() );
    }
    pSubStreams[subStream]->outMsgHelper.Reset();
  }

References XrdCl::InQueue::AssignTimeout(), XrdCl::OutQueue::MsgHelper::handler, XrdCl::TransportHandler::MessageSent(), and XrdCl::MsgHandler::OnStatusReady().

Referenced by XrdCl::AsyncMsgWriter::Write().

Here is the call graph for this function:

Here is the caller graph for this function:

OnReadTimeout()

bool XrdCl::Stream::OnReadTimeout

(

uint16_t

subStream

)

On read timeout.

Definition at line 1188 of file XrdClStream.cc.

  {
    //--------------------------------------------------------------------------
    // We only take the main stream into account
    //--------------------------------------------------------------------------
    if( substream != 0 )
    {
      if( pSubsWaitingClose )
      {
        XrdSysMutexHelper scopedLock( pMutex );
        if( !pSubsWaitingClose ) return true;
        if( pSubStreams[substream]->status == Socket::Disconnected ) return true;
        pSubStreams[substream]->socket->Close();
        pSubStreams[substream]->status = Socket::Disconnected;
        if( --pSubsWaitingClose == 0 )
        {
          scopedLock.UnLock();
          OnConnect( 0 );
        }
      }
      return true;
    }
 
    //--------------------------------------------------------------------------
    // Check if there is no outgoing messages and if the stream TTL is elapesed.
    // It is assumed that the underlying transport makes sure that there is no
    // pending requests that are not answered, ie. all possible virtual streams
    // are de-allocated
    //--------------------------------------------------------------------------
    Log *log = DefaultEnv::GetLog();
    SubStreamList::iterator it;
    time_t                  now = time(0);
 
    XrdSysMutexHelper scopedLock( pMutex );
    uint32_t outgoingMessages = 0;
    time_t   lastActivity     = 0;
    for( it = pSubStreams.begin(); it != pSubStreams.end(); ++it )
    {
      outgoingMessages += (*it)->outQueue->GetSize();
      time_t sockLastActivity = (*it)->socket->GetLastActivity();
      if( lastActivity < sockLastActivity )
        lastActivity = sockLastActivity;
    }
 
    if( !outgoingMessages )
    {
      bool disconnect = pTransport->IsStreamTTLElapsed( now-lastActivity,
                                                        *pChannelData );
      if( disconnect )
      {
        log->Debug( PostMasterMsg, "[%s] Stream TTL elapsed, disconnecting...",
                    pStreamName.c_str() );
        //----------------------------------------------------------------------
        // Important note!
        //
        // This job destroys the Stream object itself, the underlying
        // AsyncSocketHandler object (that called this method) and the Channel
        // object that aggregates this Stream.
        //
        // Additionally &(*pUrl) is used by ForceDisconnect to check if we are
        // in a Channel that was previously collapsed in a redirect.
        //----------------------------------------------------------------------
        if( !pTTLDiscJob )
        {
          pTTLDiscJob = new ForceDisconnectJob( pUrl );
          pJobManager->QueueJob( pTTLDiscJob );
        }
        return false;
      }
    }
 
    //--------------------------------------------------------------------------
    // Check if the stream is broken
    //--------------------------------------------------------------------------
    XRootDStatus st = pTransport->IsStreamBroken( now-lastActivity,
                                            *pChannelData );
    if( !st.IsOK() )
    {
      scopedLock.UnLock();
      OnError( substream, st );
      return false;
    }
    return true;
  }

References XrdCl::Log::Debug(), XrdCl::Socket::Disconnected, XrdCl::DefaultEnv::GetLog(), XrdCl::Status::IsOK(), XrdCl::TransportHandler::IsStreamBroken(), XrdCl::TransportHandler::IsStreamTTLElapsed(), OnConnect(), OnError(), XrdCl::PostMasterMsg, XrdCl::JobManager::QueueJob(), and XrdSysMutexHelper::UnLock().

Referenced by XrdCl::AsyncSocketHandler::OnReadTimeout().

Here is the call graph for this function:

Here is the caller graph for this function:

OnReadyToWrite()

std::pair< Message *, MsgHandler * > XrdCl::Stream::OnReadyToWrite

(

uint16_t

subStream

)

Definition at line 567 of file XrdClStream.cc.

  {
    XrdSysMutexHelper scopedLock( pMutex );
    Log *log = DefaultEnv::GetLog();
    if( pSubStreams[subStream]->outQueue->IsEmpty() )
    {
      log->Dump( PostMasterMsg, "[%s] Nothing to write, disable uplink",
                 pSubStreams[subStream]->socket->GetStreamName().c_str() );
 
      pSubStreams[subStream]->socket->DisableUplink();
      return std::make_pair( (Message *)0, (MsgHandler *)0 );
    }
 
    OutQueue::MsgHelper &h = pSubStreams[subStream]->outMsgHelper;
    h.msg = pSubStreams[subStream]->outQueue->PopMessage( h.handler,
                                                          h.expires,
                                                          h.stateful );
 
    log->Debug( PostMasterMsg, "[%s] Duplicating MsgHandler: %p (message: %s) "
                "from out-queue to in-queue, starting to send outgoing.",
                pUrl->GetHostId().c_str(), (void*)h.handler,
                h.msg->GetObfuscatedDescription().c_str() );
 
    scopedLock.UnLock();
 
    if( h.handler )
    {
      bool rmMsg = false;
      pIncomingQueue->AddMessageHandler( h.handler, rmMsg );
      if( rmMsg )
      {
        Log *log = DefaultEnv::GetLog();
        log->Warning( PostMasterMsg, "[%s] Removed a leftover msg from the in-queue.",
                      pStreamName.c_str() );
      }
      h.handler->OnReadyToSend( h.msg );
    }
    return std::make_pair( h.msg, h.handler );
  }

References XrdCl::InQueue::AddMessageHandler(), XrdCl::Log::Debug(), XrdCl::Log::Dump(), XrdCl::OutQueue::MsgHelper::expires, XrdCl::DefaultEnv::GetLog(), XrdCl::Message::GetObfuscatedDescription(), XrdCl::OutQueue::MsgHelper::handler, XrdCl::OutQueue::MsgHelper::msg, XrdCl::MsgHandler::OnReadyToSend(), XrdCl::PostMasterMsg, XrdCl::OutQueue::MsgHelper::stateful, XrdSysMutexHelper::UnLock(), and XrdCl::Log::Warning().

Referenced by XrdCl::AsyncMsgWriter::Write().

Here is the call graph for this function:

Here is the caller graph for this function:

OnWriteTimeout()

bool XrdCl::Stream::OnWriteTimeout

(

uint16_t

subStream

)

On write timeout.

Definition at line 1276 of file XrdClStream.cc.

  {
    return true;
  }

Referenced by XrdCl::AsyncSocketHandler::OnWriteTimeout().

Here is the caller graph for this function:

Query()

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

Query the stream.

Definition at line 1398 of file XrdClStream.cc.

  {
    switch( query )
    {
      case StreamQuery::IpAddr:
      {
        result.Set( new std::string( pSubStreams[0]->socket->GetIpAddr() ), false );
        return Status();
      }
 
      case StreamQuery::IpStack:
      {
        result.Set( new std::string( pSubStreams[0]->socket->GetIpStack() ), false );
        return Status();
      }
 
      case StreamQuery::HostName:
      {
        result.Set( new std::string( pSubStreams[0]->socket->GetHostName() ), false );
        return Status();
      }
 
      default:
        return Status( stError, errQueryNotSupported );
    }
  }

References XrdCl::errQueryNotSupported, XrdCl::StreamQuery::HostName, XrdCl::StreamQuery::IpAddr, XrdCl::StreamQuery::IpStack, XrdCl::AnyObject::Set(), and XrdCl::stError.

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

Here is the call graph for this function:

Here is the caller graph for this function:

RegisterEventHandler()

void XrdCl::Stream::RegisterEventHandler

(

ChannelEventHandler *

handler

)

Register channel event handler.

Definition at line 1284 of file XrdClStream.cc.

  {
    pChannelEvHandlers.AddHandler( handler );
  }

References XrdCl::ChannelHandlerList::AddHandler().

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

Here is the call graph for this function:

Here is the caller graph for this function:

RemoveEventHandler()

void XrdCl::Stream::RemoveEventHandler

(

ChannelEventHandler *

handler

)

Remove a channel event handler.

Definition at line 1292 of file XrdClStream.cc.

  {
    pChannelEvHandlers.RemoveHandler( handler );
  }

References XrdCl::ChannelHandlerList::RemoveHandler().

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

Here is the call graph for this function:

Here is the caller graph for this function:

Send()

XRootDStatus XrdCl::Stream::Send	(	Message *	msg,
		MsgHandler *	handler,
		bool	stateful,
		time_t	expires
	)

Queue the message for sending.

Definition at line 301 of file XrdClStream.cc.

  {
    XrdSysMutexHelper scopedLock( pMutex );
    Log *log = DefaultEnv::GetLog();
 
    //--------------------------------------------------------------------------
    // Check the session ID and bounce if needed
    //--------------------------------------------------------------------------
    if( msg->GetSessionId() &&
        (pSubStreams[0]->status != Socket::Connected ||
        pSessionId != msg->GetSessionId()) )
      return XRootDStatus( stError, errInvalidSession );
 
    //--------------------------------------------------------------------------
    // Decide on the path to send the message
    //--------------------------------------------------------------------------
    PathID path = pTransport->MultiplexSubStream( msg, *pChannelData );
    if( pSubStreams.size() <= path.up )
    {
      log->Warning( PostMasterMsg, "[%s] Unable to send message %s through "
                    "substream %d, using 0 instead", pStreamName.c_str(),
                    msg->GetObfuscatedDescription().c_str(), path.up );
      path.up = 0;
    }
 
    log->Dump( PostMasterMsg, "[%s] Sending message %s (%p) through "
               "substream %d expecting answer at %d", pStreamName.c_str(),
               msg->GetObfuscatedDescription().c_str(), (void*)msg, path.up, path.down );
 
    //--------------------------------------------------------------------------
    // Enable *a* path and insert the message to the right queue
    //--------------------------------------------------------------------------
    XRootDStatus st = EnableLink( path );
    if( st.IsOK() )
    {
      pTransport->MultiplexSubStream( msg, *pChannelData, &path );
      pSubStreams[path.up]->outQueue->PushBack( msg, handler,
                                                expires, stateful );
    }
    else
      st.status = stFatal;
    return st;
  }

References XrdCl::Socket::Connected, XrdCl::PathID::down, XrdCl::Log::Dump(), EnableLink(), XrdCl::errInvalidSession, XrdCl::DefaultEnv::GetLog(), XrdCl::Message::GetObfuscatedDescription(), XrdCl::Message::GetSessionId(), XrdCl::Status::IsOK(), XrdCl::TransportHandler::MultiplexSubStream(), XrdCl::PostMasterMsg, XrdCl::Status::status, XrdCl::stError, XrdCl::stFatal, XrdCl::PathID::up, and XrdCl::Log::Warning().

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

Here is the call graph for this function:

Here is the caller graph for this function:

SetChannelData()

void XrdCl::Stream::SetChannelData ( AnyObject *  channelData )

inline

Set the channel data.

Definition at line 115 of file XrdClStream.hh.

      {
        pChannelData = channelData;
      }

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

Here is the caller graph for this function:

SetIncomingQueue()

void XrdCl::Stream::SetIncomingQueue ( InQueue *  incomingQueue )

inline

Set the incoming queue.

Definition at line 107 of file XrdClStream.hh.

      {
        pIncomingQueue = incomingQueue;
      }

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

Here is the caller graph for this function:

SetJobManager()

void XrdCl::Stream::SetJobManager ( JobManager *  jobManager )

inline

Set job manager.

Definition at line 131 of file XrdClStream.hh.

      {
        pJobManager = jobManager;
      }

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

Here is the caller graph for this function:

SetOnDataConnectHandler()

void XrdCl::Stream::SetOnDataConnectHandler ( std::shared_ptr< Job > &  onConnJob )

inline

Set the on-connect handler for data streams.

Definition at line 263 of file XrdClStream.hh.

      {
        XrdSysMutexHelper scopedLock( pMutex );
        pOnDataConnJob = onConnJob;
      }

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

Here is the caller graph for this function:

SetPoller()

void XrdCl::Stream::SetPoller ( Poller *  poller )

inline

Set the poller.

Definition at line 99 of file XrdClStream.hh.

      {
        pPoller = poller;
      }

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

Here is the caller graph for this function:

SetTaskManager()

void XrdCl::Stream::SetTaskManager ( TaskManager *  taskManager )

inline

Set task manager.

Definition at line 123 of file XrdClStream.hh.

      {
        pTaskManager = taskManager;
      }

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

Here is the caller graph for this function:

SetTransport()

void XrdCl::Stream::SetTransport ( TransportHandler *  transport )

inline

Set the transport.

Definition at line 91 of file XrdClStream.hh.

      {
        pTransport = transport;
      }

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

Here is the caller graph for this function:

Tick()

void XrdCl::Stream::Tick

(

time_t

now

)

Handle a clock event generated either by socket timeout, or by the task manager event

Definition at line 399 of file XrdClStream.cc.

  {
    //--------------------------------------------------------------------------
    // Check for timed-out requests and incoming handlers
    //--------------------------------------------------------------------------
    pMutex.Lock();
    OutQueue q;
    SubStreamList::iterator it;
    for( it = pSubStreams.begin(); it != pSubStreams.end(); ++it )
      q.GrabExpired( *(*it)->outQueue, now );
    pMutex.UnLock();
 
    q.Report( XRootDStatus( stError, errOperationExpired ) );
    pIncomingQueue->ReportTimeout( now );
  }

References XrdCl::errOperationExpired, XrdCl::OutQueue::GrabExpired(), XrdSysMutex::Lock(), XrdCl::OutQueue::Report(), XrdCl::InQueue::ReportTimeout(), XrdCl::stError, and XrdSysMutex::UnLock().

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

Here is the call graph for this function:

Here is the caller graph for this function:

---

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

• XrdClStream.hh
• XrdClStream.cc