Loading...
Searching...
No Matches
XrdClStream.hh
Go to the documentation of this file.
//------------------------------------------------------------------------------
// Copyright (c) 2011-2012 by European Organization for Nuclear Research (CERN)
// Author: Lukasz Janyst <ljanyst@cern.ch>
//------------------------------------------------------------------------------
// XRootD is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// XRootD is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with XRootD. If not, see <http://www.gnu.org/licenses/>.
//------------------------------------------------------------------------------
#ifndef __XRD_CL_STREAM_HH__
#define __XRD_CL_STREAM_HH__
#include "XrdCl/XrdClPoller.hh"
#include "XrdCl/XrdClStatus.hh"
#include "XrdCl/XrdClURL.hh"
#include "XrdCl/XrdClPostMasterInterfaces.hh"
#include "XrdCl/XrdClChannelHandlerList.hh"
#include "XrdCl/XrdClJobManager.hh"
#include "XrdCl/XrdClInQueue.hh"
#include "XrdCl/XrdClUtils.hh"
#include "XrdSys/XrdSysPthread.hh"
#include "XrdSys/XrdSysRAtomic.hh"
#include "XrdNet/XrdNetAddr.hh"
#include "XrdOuc/XrdOucCompiler.hh"
#include <list>
#include <vector>
#include <functional>
#include <memory>
namespace XrdCl
{
class Message;
class Channel;
class TransportHandler;
class TaskManager;
struct SubStreamData;
//----------------------------------------------------------------------------
//----------------------------------------------------------------------------
class Stream
{
public:
//------------------------------------------------------------------------
//------------------------------------------------------------------------
enum StreamStatus
{
Disconnected = 0,
Connected = 1,
Connecting = 2,
Error = 3
};
//------------------------------------------------------------------------
//------------------------------------------------------------------------
Stream( const URL *url, const URL &prefer = URL() );
//------------------------------------------------------------------------
//------------------------------------------------------------------------
~Stream();
//------------------------------------------------------------------------
//------------------------------------------------------------------------
XRootDStatus Initialize();
//------------------------------------------------------------------------
//------------------------------------------------------------------------
XRootDStatus Send( Message *msg,
MsgHandler *handler,
bool stateful,
time_t expires );
//------------------------------------------------------------------------
//------------------------------------------------------------------------
void SetTransport( TransportHandler *transport )
{
pTransport = transport;
}
//------------------------------------------------------------------------
//------------------------------------------------------------------------
void SetPoller( Poller *poller )
{
pPoller = poller;
}
//------------------------------------------------------------------------
//------------------------------------------------------------------------
void SetIncomingQueue( InQueue *incomingQueue )
{
pIncomingQueue = incomingQueue;
}
//------------------------------------------------------------------------
//------------------------------------------------------------------------
void SetChannelData( AnyObject *channelData )
{
pChannelData = channelData;
}
//------------------------------------------------------------------------
//------------------------------------------------------------------------
void SetTaskManager( TaskManager *taskManager )
{
pTaskManager = taskManager;
}
//------------------------------------------------------------------------
//------------------------------------------------------------------------
void SetJobManager( JobManager *jobManager )
{
pJobManager = jobManager;
}
//------------------------------------------------------------------------
//------------------------------------------------------------------------
XRootDStatus EnableLink( PathID &path );
//------------------------------------------------------------------------
//------------------------------------------------------------------------
void Disconnect( bool force = false );
//------------------------------------------------------------------------
//------------------------------------------------------------------------
void Tick( time_t now );
//------------------------------------------------------------------------
//------------------------------------------------------------------------
const URL *GetURL() const
{
return pUrl;
}
//------------------------------------------------------------------------
//------------------------------------------------------------------------
void ForceConnect();
//------------------------------------------------------------------------
//------------------------------------------------------------------------
const std::string &GetName() const
{
return pStreamName;
}
//------------------------------------------------------------------------
//------------------------------------------------------------------------
void DisableIfEmpty( uint16_t subStream );
//------------------------------------------------------------------------
//------------------------------------------------------------------------
void OnIncoming( uint16_t subStream,
std::shared_ptr<Message> msg,
uint32_t bytesReceived );
//------------------------------------------------------------------------
// Call when one of the sockets is ready to accept a new message
//------------------------------------------------------------------------
std::pair<Message *, MsgHandler *>
OnReadyToWrite( uint16_t subStream );
//------------------------------------------------------------------------
// Call when a message is written to the socket
//------------------------------------------------------------------------
void OnMessageSent( uint16_t subStream,
Message *msg,
uint32_t bytesSent );
//------------------------------------------------------------------------
//------------------------------------------------------------------------
void OnConnect( uint16_t subStream );
//------------------------------------------------------------------------
//------------------------------------------------------------------------
void OnConnectError( uint16_t subStream, XRootDStatus status );
//------------------------------------------------------------------------
//------------------------------------------------------------------------
void OnError( uint16_t subStream, XRootDStatus status );
//------------------------------------------------------------------------
//------------------------------------------------------------------------
void ForceError( XRootDStatus status, bool hush=false );
//------------------------------------------------------------------------
//------------------------------------------------------------------------
bool OnReadTimeout( uint16_t subStream ) XRD_WARN_UNUSED_RESULT;
//------------------------------------------------------------------------
//------------------------------------------------------------------------
bool OnWriteTimeout( uint16_t subStream ) XRD_WARN_UNUSED_RESULT;
//------------------------------------------------------------------------
//------------------------------------------------------------------------
void RegisterEventHandler( ChannelEventHandler *handler );
//------------------------------------------------------------------------
//------------------------------------------------------------------------
void RemoveEventHandler( ChannelEventHandler *handler );
//------------------------------------------------------------------------
//------------------------------------------------------------------------
MsgHandler*
InstallIncHandler( std::shared_ptr<Message> &msg, uint16_t stream );
//------------------------------------------------------------------------
//------------------------------------------------------------------------
uint16_t InspectStatusRsp( uint16_t stream, MsgHandler *&incHandler );
//------------------------------------------------------------------------
//------------------------------------------------------------------------
void SetOnDataConnectHandler( std::shared_ptr<Job> &onConnJob )
{
XrdSysMutexHelper scopedLock( pMutex );
pOnDataConnJob = onConnJob;
}
//------------------------------------------------------------------------
//------------------------------------------------------------------------
bool CanCollapse( const URL &url );
//------------------------------------------------------------------------
//------------------------------------------------------------------------
Status Query( uint16_t query, AnyObject &result );
private:
//------------------------------------------------------------------------
//------------------------------------------------------------------------
static bool IsPartial( Message &msg );
//------------------------------------------------------------------------
//------------------------------------------------------------------------
inline static bool HasNetAddr( const XrdNetAddr &addr,
std::vector<XrdNetAddr> &addresses )
{
auto itr = addresses.begin();
for( ; itr != addresses.end() ; ++itr )
{
if( itr->Same( &addr ) ) return true;
}
return false;
}
//------------------------------------------------------------------------
// Job handling the incoming messages
//------------------------------------------------------------------------
class HandleIncMsgJob: public Job
{
public:
HandleIncMsgJob( MsgHandler *handler ): pHandler( handler ) {};
virtual ~HandleIncMsgJob() {};
virtual void Run( void* )
{
pHandler->Process();
delete this;
}
private:
MsgHandler *pHandler;
};
//------------------------------------------------------------------------
//------------------------------------------------------------------------
void OnFatalError( uint16_t subStream,
XRootDStatus status,
XrdSysMutexHelper &lock );
//------------------------------------------------------------------------
//------------------------------------------------------------------------
void MonitorDisconnection( XRootDStatus status );
//------------------------------------------------------------------------
//------------------------------------------------------------------------
XRootDStatus RequestClose( Message &resp );
typedef std::vector<SubStreamData*> SubStreamList;
//------------------------------------------------------------------------
// Data members
//------------------------------------------------------------------------
const URL *pUrl;
const URL pPrefer;
std::string pStreamName;
TransportHandler *pTransport;
Poller *pPoller;
TaskManager *pTaskManager;
JobManager *pJobManager;
XrdSysRecMutex pMutex;
InQueue *pIncomingQueue;
AnyObject *pChannelData;
uint32_t pLastStreamError;
XRootDStatus pLastFatalError;
uint16_t pStreamErrorWindow;
uint16_t pConnectionCount;
uint16_t pConnectionRetry;
time_t pConnectionInitTime;
uint16_t pConnectionWindow;
SubStreamList pSubStreams;
std::vector<XrdNetAddr> pAddresses;
Utils::AddressType pAddressType;
ChannelHandlerList pChannelEvHandlers;
uint64_t pSessionId;
//------------------------------------------------------------------------
// Monitoring info
//------------------------------------------------------------------------
timeval pConnectionStarted;
timeval pConnectionDone;
std::atomic<uint64_t> pBytesSent;
std::atomic<uint64_t> pBytesReceived;
//------------------------------------------------------------------------
// Data stream on-connect handler
//------------------------------------------------------------------------
std::shared_ptr<Job> pOnDataConnJob;
//------------------------------------------------------------------------
// Track last assigned Id across all Streams, to ensure unique sessionId
//------------------------------------------------------------------------
static RAtomic_uint64_t sSessCntGen;
};
}
#endif // __XRD_CL_STREAM_HH__
XrdClChannelHandlerList.hh
XrdClInQueue.hh
XrdClJobManager.hh
XrdClPoller.hh
XrdClPostMasterInterfaces.hh
XrdClStatus.hh
XrdClURL.hh
XrdClUtils.hh
XrdNetAddr.hh
XrdOucCompiler.hh
XRD_WARN_UNUSED_RESULT
#define XRD_WARN_UNUSED_RESULT
Definition XrdOucCompiler.hh:31
XrdSysPthread.hh
XrdSysRAtomic.hh
RAtomic_uint64_t
XrdSys::RAtomic< uint64_t > RAtomic_uint64_t
Definition XrdSysRAtomic.hh:339
XrdCl::AnyObject
Definition XrdClAnyObject.hh:33
XrdCl::ChannelEventHandler
Channel event handler.
Definition XrdClPostMasterInterfaces.hh:210
XrdCl::Channel
A communication channel between the client and the server.
Definition XrdClChannel.hh:49
XrdCl::InQueue
A synchronize queue for incoming data.
Definition XrdClInQueue.hh:37
XrdCl::JobManager
A synchronized queue.
Definition XrdClJobManager.hh:51
XrdCl::Message
The message representation used throughout the system.
Definition XrdClMessage.hh:32
XrdCl::MsgHandler
Message handler.
Definition XrdClPostMasterInterfaces.hh:51
XrdCl::Poller
Interface for socket pollers.
Definition XrdClPoller.hh:87
XrdCl::Stream::Send
XRootDStatus Send(Message *msg, MsgHandler *handler, bool stateful, time_t expires)
Queue the message for sending.
Definition XrdClStream.cc:297
XrdCl::Stream::SetTransport
void SetTransport(TransportHandler *transport)
Set the transport.
Definition XrdClStream.hh:91
XrdCl::Stream::StreamStatus
StreamStatus
Status of the stream.
Definition XrdClStream.hh:58
XrdCl::Stream::Disconnected
@ Disconnected
Not connected.
Definition XrdClStream.hh:59
XrdCl::Stream::Error
@ Error
Broken.
Definition XrdClStream.hh:62
XrdCl::Stream::Connected
@ Connected
Connected.
Definition XrdClStream.hh:60
XrdCl::Stream::Connecting
@ Connecting
In the process of being connected.
Definition XrdClStream.hh:61
XrdCl::Stream::SetIncomingQueue
void SetIncomingQueue(InQueue *incomingQueue)
Set the incoming queue.
Definition XrdClStream.hh:107
XrdCl::Stream::OnReadTimeout
bool OnReadTimeout(uint16_t subStream) XRD_WARN_UNUSED_RESULT
On read timeout.
Definition XrdClStream.cc:1050
XrdCl::Stream::CanCollapse
bool CanCollapse(const URL &url)
Definition XrdClStream.cc:1197
XrdCl::Stream::SetPoller
void SetPoller(Poller *poller)
Set the poller.
Definition XrdClStream.hh:99
XrdCl::Stream::ForceConnect
void ForceConnect()
Force connection.
Definition XrdClStream.cc:347
XrdCl::Stream::ForceError
void ForceError(XRootDStatus status, bool hush=false)
Force error.
Definition XrdClStream.cc:933
XrdCl::Stream::SetTaskManager
void SetTaskManager(TaskManager *taskManager)
Set task manager.
Definition XrdClStream.hh:123
XrdCl::Stream::SetOnDataConnectHandler
void SetOnDataConnectHandler(std::shared_ptr< Job > &onConnJob)
Set the on-connect handler for data streams.
Definition XrdClStream.hh:263
XrdCl::Stream::SetJobManager
void SetJobManager(JobManager *jobManager)
Set job manager.
Definition XrdClStream.hh:131
XrdCl::Stream::Disconnect
void Disconnect(bool force=false)
Disconnect the stream.
Definition XrdClStream.cc:363
XrdCl::Stream::EnableLink
XRootDStatus EnableLink(PathID &path)
Definition XrdClStream.cc:187
XrdCl::Stream::Stream
Stream(const URL *url, const URL &prefer=URL())
Constructor.
Definition XrdClStream.cc:96
XrdCl::Stream::OnConnect
void OnConnect(uint16_t subStream)
Call back when a message has been reconstructed.
Definition XrdClStream.cc:623
XrdCl::Stream::GetName
const std::string & GetName() const
Return stream name.
Definition XrdClStream.hh:170
XrdCl::Stream::Tick
void Tick(time_t now)
Definition XrdClStream.cc:377
XrdCl::Stream::OnConnectError
void OnConnectError(uint16_t subStream, XRootDStatus status)
On connect error.
Definition XrdClStream.cc:728
XrdCl::Stream::~Stream
~Stream()
Destructor.
Definition XrdClStream.cc:153
XrdCl::Stream::OnWriteTimeout
bool OnWriteTimeout(uint16_t subStream) XRD_WARN_UNUSED_RESULT
On write timeout.
Definition XrdClStream.cc:1120
XrdCl::Stream::DisableIfEmpty
void DisableIfEmpty(uint16_t subStream)
Disables respective uplink if empty.
Definition XrdClStream.cc:585
XrdCl::Stream::RemoveEventHandler
void RemoveEventHandler(ChannelEventHandler *handler)
Remove a channel event handler.
Definition XrdClStream.cc:1136
XrdCl::Stream::OnMessageSent
void OnMessageSent(uint16_t subStream, Message *msg, uint32_t bytesSent)
Definition XrdClStream.cc:601
XrdCl::Stream::InstallIncHandler
MsgHandler * InstallIncHandler(std::shared_ptr< Message > &msg, uint16_t stream)
Definition XrdClStream.cc:1145
XrdCl::Stream::OnIncoming
void OnIncoming(uint16_t subStream, std::shared_ptr< Message > msg, uint32_t bytesReceived)
Call back when a message has been reconstructed.
Definition XrdClStream.cc:471
XrdCl::Stream::OnError
void OnError(uint16_t subStream, XRootDStatus status)
On error.
Definition XrdClStream.cc:828
XrdCl::Stream::GetURL
const URL * GetURL() const
Get the URL.
Definition XrdClStream.hh:157
XrdCl::Stream::InspectStatusRsp
uint16_t InspectStatusRsp(uint16_t stream, MsgHandler *&incHandler)
Definition XrdClStream.cc:1166
XrdCl::Stream::SetChannelData
void SetChannelData(AnyObject *channelData)
Set the channel data.
Definition XrdClStream.hh:115
XrdCl::Stream::OnReadyToWrite
std::pair< Message *, MsgHandler * > OnReadyToWrite(uint16_t subStream)
Definition XrdClStream.cc:545
XrdCl::Stream::Initialize
XRootDStatus Initialize()
Initializer.
Definition XrdClStream.cc:171
XrdCl::Stream::RegisterEventHandler
void RegisterEventHandler(ChannelEventHandler *handler)
Register channel event handler.
Definition XrdClStream.cc:1128
XrdCl::TaskManager
Definition XrdClTaskManager.hh:76
XrdCl::TransportHandler
Perform the handshake and the authentication for each physical stream.
Definition XrdClPostMasterInterfaces.hh:310
XrdCl::URL
URL representation.
Definition XrdClURL.hh:31
XrdCl::Utils::AddressType
AddressType
Address type.
Definition XrdClUtils.hh:87
XrdCl::XRootDStatus
Request status.
Definition XrdClXRootDResponses.hh:219
XrdNetAddr
Definition XrdNetAddr.hh:42
XrdSysMutexHelper
Definition XrdSysPthread.hh:263
XrdCl
Definition XrdClAction.hh:34
XrdCl::Query
QueryImpl< false > Query
Definition XrdClFileSystemOperations.hh:280
XrdCl::PathID
Definition XrdClPostMasterInterfaces.hh:269
XrdCl::Status
Procedure execution status.
Definition XrdClStatus.hh:115
XrdCl::SubStreamData
Definition XrdClStream.cc:76