Loading...
Searching...
No Matches
XrdClChannel.cc
Go to the documentation of this file.
//------------------------------------------------------------------------------
// Copyright (c) 2011-2014 by European Organization for Nuclear Research (CERN)
// Author: Lukasz Janyst <ljanyst@cern.ch>
//------------------------------------------------------------------------------
// This file is part of the XRootD software suite.
//
// XRootD is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// XRootD is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with XRootD. If not, see <http://www.gnu.org/licenses/>
//
// In applying this licence, CERN does not waive the privileges and immunities
// granted to it by virtue of its status as an Intergovernmental Organization
// or submit itself to any jurisdiction.
//------------------------------------------------------------------------------
#include "XrdCl/XrdClChannel.hh"
#include "XrdCl/XrdClDefaultEnv.hh"
#include "XrdCl/XrdClStream.hh"
#include "XrdCl/XrdClSocket.hh"
#include "XrdCl/XrdClConstants.hh"
#include "XrdCl/XrdClLog.hh"
#include "XrdCl/XrdClRedirectorRegistry.hh"
#include "XrdCl/XrdClXRootDTransport.hh"
#include "XrdSys/XrdSysPthread.hh"
#include <ctime>
namespace XrdCl
{
class TickGeneratorTask: public XrdCl::Task
{
public:
//------------------------------------------------------------------------
// Constructor
//------------------------------------------------------------------------
TickGeneratorTask( XrdCl::Channel *channel, const std::string &hostId ):
pChannel( channel )
{
std::string name = "TickGeneratorTask for: ";
name += hostId;
SetName( name );
}
//------------------------------------------------------------------------
// Run the task
//------------------------------------------------------------------------
time_t Run( time_t now )
{
XrdSysMutexHelper lck( pMtx );
if( !pChannel ) return 0;
using namespace XrdCl;
pChannel->Tick( now );
Env *env = DefaultEnv::GetEnv();
int timeoutResolution = DefaultTimeoutResolution;
env->GetInt( "TimeoutResolution", timeoutResolution );
return now+timeoutResolution;
}
void Invalidate()
{
XrdSysMutexHelper lck( pMtx );
pChannel = 0;
}
private:
XrdCl::Channel *pChannel;
XrdSysMutex pMtx;
};
//----------------------------------------------------------------------------
// Constructor
//----------------------------------------------------------------------------
Channel::Channel( const URL &url,
Poller *poller,
TransportHandler *transport,
TaskManager *taskManager,
JobManager *jobManager,
const URL &prefurl ):
pUrl( url.GetHostId() ),
pPoller( poller ),
pTransport( transport ),
pTaskManager( taskManager ),
pTickGenerator( 0 ),
pJobManager( jobManager )
{
Env *env = DefaultEnv::GetEnv();
Log *log = DefaultEnv::GetLog();
int timeoutResolution = DefaultTimeoutResolution;
env->GetInt( "TimeoutResolution", timeoutResolution );
pTransport->InitializeChannel( url, pChannelData );
log->Debug( PostMasterMsg, "Creating new channel to: %s",
url.GetChannelId().c_str() );
pUrl.SetParams( url.GetParams() );
pUrl.SetProtocol( url.GetProtocol() );
//--------------------------------------------------------------------------
// Create the stream
//--------------------------------------------------------------------------
pStream = new Stream( &pUrl, prefurl );
pStream->SetTransport( transport );
pStream->SetPoller( poller );
pStream->SetIncomingQueue( &pIncoming );
pStream->SetTaskManager( taskManager );
pStream->SetJobManager( jobManager );
pStream->SetChannelData( &pChannelData );
pStream->Initialize();
//--------------------------------------------------------------------------
// Register the task generating timeout events
//--------------------------------------------------------------------------
pTickGenerator = new TickGeneratorTask( this, pUrl.GetChannelId() );
pTaskManager->RegisterTask( pTickGenerator, ::time(0)+timeoutResolution );
}
//----------------------------------------------------------------------------
// Destructor
//----------------------------------------------------------------------------
Channel::~Channel()
{
pTickGenerator->Invalidate();
delete pStream;
pTransport->FinalizeChannel( pChannelData );
}
//----------------------------------------------------------------------------
// Send the message asynchronously
//----------------------------------------------------------------------------
XRootDStatus Channel::Send( Message *msg,
MsgHandler *handler,
bool stateful,
time_t expires )
{
return pStream->Send( msg, handler, stateful, expires );
}
//----------------------------------------------------------------------------
// Handle a time event
//----------------------------------------------------------------------------
void Channel::Tick( time_t now )
{
pStream->Tick( now );
}
//----------------------------------------------------------------------------
// Force disconnect of all streams
//----------------------------------------------------------------------------
Status Channel::ForceDisconnect()
{
return ForceDisconnect(false);
}
//----------------------------------------------------------------------------
// Force disconnect of all streams
//----------------------------------------------------------------------------
Status Channel::ForceDisconnect( bool hush )
{
//--------------------------------------------------------------------------
// Disconnect and recreate the streams
//--------------------------------------------------------------------------
pStream->ForceError( Status( stError, errOperationInterrupted ), hush );
return Status();
}
//----------------------------------------------------------------------------
// Force reconnect
//----------------------------------------------------------------------------
Status Channel::ForceReconnect()
{
pStream->ForceConnect();
return Status();
}
//------------------------------------------------------------------------
// Get the number of connected data streams
//------------------------------------------------------------------------
uint16_t Channel::NbConnectedStrm()
{
return XRootDTransport::NbConnectedStrm( pChannelData );
}
//------------------------------------------------------------------------
// Set the on-connect handler for data streams
//------------------------------------------------------------------------
void Channel::SetOnDataConnectHandler( std::shared_ptr<Job> &onConnJob )
{
pStream->SetOnDataConnectHandler( onConnJob );
}
//------------------------------------------------------------------------
// Check if channel can be collapsed using given URL
//------------------------------------------------------------------------
bool Channel::CanCollapse( const URL &url )
{
return pStream->CanCollapse( url );
}
//------------------------------------------------------------------------
// Decrement file object instance count bound to this channel
//------------------------------------------------------------------------
void Channel::DecFileInstCnt()
{
pTransport->DecFileInstCnt( pChannelData );
}
//----------------------------------------------------------------------------
// Query the transport handler
//----------------------------------------------------------------------------
Status Channel::QueryTransport( uint16_t query, AnyObject &result )
{
if( query < 2000 )
return pTransport->Query( query, result, pChannelData );
return pStream->Query( query, result );
}
//----------------------------------------------------------------------------
// Register channel event handler
//----------------------------------------------------------------------------
void Channel::RegisterEventHandler( ChannelEventHandler *handler )
{
pStream->RegisterEventHandler( handler );
}
//------------------------------------------------------------------------
// Remove a channel event handler
//------------------------------------------------------------------------
void Channel::RemoveEventHandler( ChannelEventHandler *handler )
{
pStream->RemoveEventHandler( handler );
}
}
XrdClChannel.hh
XrdClConstants.hh
XrdClDefaultEnv.hh
XrdClLog.hh
XrdClRedirectorRegistry.hh
XrdClSocket.hh
XrdClStream.hh
XrdClXRootDTransport.hh
XrdSysPthread.hh
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::Channel::NbConnectedStrm
uint16_t NbConnectedStrm()
Get the number of connected data streams.
Definition XrdClChannel.cc:192
XrdCl::Channel::ForceReconnect
Status ForceReconnect()
Force reconnect.
Definition XrdClChannel.cc:183
XrdCl::Channel::RemoveEventHandler
void RemoveEventHandler(ChannelEventHandler *handler)
Remove a channel event handler.
Definition XrdClChannel.cc:242
XrdCl::Channel::SetOnDataConnectHandler
void SetOnDataConnectHandler(std::shared_ptr< Job > &onConnJob)
Set the on-connect handler for data streams.
Definition XrdClChannel.cc:200
XrdCl::Channel::~Channel
~Channel()
Destructor.
Definition XrdClChannel.cc:132
XrdCl::Channel::RegisterEventHandler
void RegisterEventHandler(ChannelEventHandler *handler)
Register channel event handler.
Definition XrdClChannel.cc:234
XrdCl::Channel::DecFileInstCnt
void DecFileInstCnt()
Decrement file object instance count bound to this channel.
Definition XrdClChannel.cc:216
XrdCl::Channel::ForceDisconnect
Status ForceDisconnect()
Force disconnect of all streams.
Definition XrdClChannel.cc:162
XrdCl::Channel::Send
XRootDStatus Send(Message *msg, MsgHandler *handler, bool stateful, time_t expires)
Definition XrdClChannel.cc:142
XrdCl::Channel::Tick
void Tick(time_t now)
Handle a time event.
Definition XrdClChannel.cc:154
XrdCl::Channel::CanCollapse
bool CanCollapse(const URL &url)
Definition XrdClChannel.cc:208
XrdCl::Channel::QueryTransport
Status QueryTransport(uint16_t query, AnyObject &result)
Definition XrdClChannel.cc:224
XrdCl::Channel::Channel
Channel(const URL &url, Poller *poller, TransportHandler *transport, TaskManager *taskManager, JobManager *jobManager, const URL &prefurl=URL())
Definition XrdClChannel.cc:84
XrdCl::DefaultEnv::GetLog
static Log * GetLog()
Get default log.
Definition XrdClDefaultEnv.cc:493
XrdCl::DefaultEnv::GetEnv
static Env * GetEnv()
Get default client environment.
Definition XrdClDefaultEnv.cc:445
XrdCl::Env
Definition XrdClEnv.hh:38
XrdCl::Env::GetInt
bool GetInt(const std::string &key, int &value)
Definition XrdClEnv.cc:89
XrdCl::JobManager
A synchronized queue.
Definition XrdClJobManager.hh:51
XrdCl::Log
Handle diagnostics.
Definition XrdClLog.hh:101
XrdCl::Log::Debug
void Debug(uint64_t topic, const char *format,...)
Print a debug message.
Definition XrdClLog.cc:282
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
Stream.
Definition XrdClStream.hh:52
XrdCl::TaskManager
Definition XrdClTaskManager.hh:76
XrdCl::Task
Interface for a task to be run by the TaskManager.
Definition XrdClTaskManager.hh:36
XrdCl::Task::SetName
void SetName(const std::string &name)
Set name of the task.
Definition XrdClTaskManager.hh:60
XrdCl::TickGeneratorTask
Definition XrdClChannel.cc:40
XrdCl::TickGeneratorTask::Invalidate
void Invalidate()
Definition XrdClChannel.cc:70
XrdCl::TickGeneratorTask::Run
time_t Run(time_t now)
Definition XrdClChannel.cc:56
XrdCl::TickGeneratorTask::TickGeneratorTask
TickGeneratorTask(XrdCl::Channel *channel, const std::string &hostId)
Definition XrdClChannel.cc:45
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::URL::GetChannelId
std::string GetChannelId() const
Definition XrdClURL.cc:512
XrdCl::URL::GetParams
const ParamsMap & GetParams() const
Get the URL params.
Definition XrdClURL.hh:244
XrdCl::URL::GetProtocol
const std::string & GetProtocol() const
Get the protocol.
Definition XrdClURL.hh:118
XrdCl::XRootDStatus
Request status.
Definition XrdClXRootDResponses.hh:219
XrdCl::XRootDTransport::NbConnectedStrm
static uint16_t NbConnectedStrm(AnyObject &channelData)
Number of currently connected data streams.
Definition XrdClXRootDTransport.cc:1521
XrdSysMutexHelper
Definition XrdSysPthread.hh:263
XrdSysMutex
Definition XrdSysPthread.hh:165
XrdCl
Definition XrdClAction.hh:34
XrdCl::stError
const uint16_t stError
An error occurred that could potentially be retried.
Definition XrdClStatus.hh:32
XrdCl::PostMasterMsg
const uint64_t PostMasterMsg
Definition XrdClConstants.hh:36
XrdCl::DefaultTimeoutResolution
const int DefaultTimeoutResolution
Definition XrdClConstants.hh:56
XrdCl::errOperationInterrupted
const uint16_t errOperationInterrupted
Definition XrdClStatus.hh:91
XrdCl::Status
Procedure execution status.
Definition XrdClStatus.hh:115