XrdCl::PostMaster Class Reference

A hub for dispatching and receiving messages. More...

#include <XrdClPostMaster.hh>

Collaboration diagram for XrdCl::PostMaster:

Public Member Functions
	PostMaster ()
	Constructor. More...
virtual	~PostMaster ()
	Destructor. More...
void	CollapseRedirect (const URL &oldurl, const URL &newURL)
	Collapse channel URL - replace the URL of the channel. More...
void	DecFileInstCnt (const URL &url)
	Decrement file object instance count bound to this channel. More...
bool	Finalize ()
	Finalizer. More...
Status	ForceDisconnect (const URL &url)
	Shut down a channel. More...
Status	ForceDisconnect (const URL &url, bool hush)
	Shut down a channel. More...
Status	ForceDisconnect (std::shared_ptr< Channel > channel, const uint64_t sess)
	Shut down a channel. This version is used by the channel itself. More...
Status	ForceReconnect (const URL &url)
	Reconnect the channel. More...
JobManager *	GetJobManager ()
	Get the job manager object user by the post master. More...
TaskManager *	GetTaskManager ()
	Get the task manager object user by the post master. More...
bool	Initialize ()
	Initializer. More...
bool	IsRunning ()
uint16_t	NbConnectedStrm (const URL &url)
	Get the number of connected data streams. More...
void	NotifyConnectHandler (const URL &url)
	Notify the global on-connect handler. More...
void	NotifyConnErrHandler (const URL &url, const XRootDStatus &status)
	Notify the global error connection handler. More...
Status	QueryTransport (const URL &url, uint16_t query, AnyObject &result)
Status	Redirect (const URL &url, Message *msg, MsgHandler *handler)
Status	RegisterEventHandler (const URL &url, ChannelEventHandler *handler)
	Register channel event handler. More...
bool	Reinitialize ()
	Reinitialize after fork. More...
Status	RemoveEventHandler (const URL &url, ChannelEventHandler *handler)
	Remove a channel event handler. More...
XRootDStatus	Send (const URL &url, Message *msg, MsgHandler *handler, bool stateful, time_t expires)
void	SetConnectionErrorHandler (std::function< void(const URL &, const XRootDStatus &)> handler)
	Set the global on-error on-connect handler for control streams. More...
void	SetOnConnectHandler (std::unique_ptr< Job > onConnJob)
	Set the global connection error handler. More...
void	SetOnDataConnectHandler (const URL &url, std::shared_ptr< Job > onConnJob)
	Set the on-connect handler for data streams. More...
bool	Start ()
	Start the post master. More...
bool	Stop ()
	Stop the postmaster. More...

Detailed Description

A hub for dispatching and receiving messages.

Definition at line 47 of file XrdClPostMaster.hh.

Constructor & Destructor Documentation

PostMaster()

XrdCl::PostMaster::PostMaster

(

)

Constructor.

Definition at line 127 of file XrdClPostMaster.cc.

                        : pImpl( new PostMasterImpl() )
  {
  }

~PostMaster()

XrdCl::PostMaster::~PostMaster ( )

virtual

Destructor.

Definition at line 134 of file XrdClPostMaster.cc.

  {
  }

Member Function Documentation

CollapseRedirect()

void XrdCl::PostMaster::CollapseRedirect	(	const URL &	oldurl,
		const URL &	newURL
	)

Collapse channel URL - replace the URL of the channel.

Definition at line 478 of file XrdClPostMaster.cc.

  {
    XrdSysMutexHelper scopedLock( pImpl->pChannelMapMutex );
    //--------------------------------------------------------------------------
    // Get the passive channel
    //--------------------------------------------------------------------------
    std::shared_ptr<Channel> passive;
    PostMasterImpl::ChannelMap::iterator it =
        pImpl->pChannelMap.find( alias.GetChannelId() );
    if( it != pImpl->pChannelMap.end() )
      passive = it->second;
 
    //--------------------------------------------------------------------------
    // If the channel does not exist there's nothing to do
    //--------------------------------------------------------------------------
    if( !passive ) return;
 
    //--------------------------------------------------------------------------
    // Check if this URL is eligible for collapsing. To avoid depencencies
    // we don't call CanCollapse while holding the channel map mutex. So we
    // reverify the content of the map afterwards.
    //--------------------------------------------------------------------------
    scopedLock.UnLock();
    if( !passive->CanCollapse( url ) ) return;
 
    scopedLock.Lock( &pImpl->pChannelMapMutex );
    it = pImpl->pChannelMap.find( alias.GetChannelId() );
    if( it == pImpl->pChannelMap.end() || it->second != passive )
    {
      // something changed. Retry.
      scopedLock.UnLock();
      CollapseRedirect( alias, url );
      return;
    }
 
    //--------------------------------------------------------------------------
    // Create the active channel
    //--------------------------------------------------------------------------
    TransportManager *trManager = DefaultEnv::GetTransportManager();
    TransportHandler *trHandler = trManager->GetHandler( url.GetProtocol() );
 
    if( !trHandler )
    {
      Log *log = DefaultEnv::GetLog();
      log->Error( PostMasterMsg, "Unable to get transport handler for %s "
                  "protocol", url.GetProtocol().c_str() );
      return;
    }
 
    Log *log = DefaultEnv::GetLog();
    log->Info( PostMasterMsg, "Label channel %s with alias %s.",
               url.GetHostId().c_str(), alias.GetHostId().c_str() );
 
    std::shared_ptr<Channel> active(new Channel{ alias,
      pImpl->pPoller, trHandler, pImpl->pTaskManager, pImpl->pJobManager, url },
      [this](Channel *ch) { this->pImpl->removeFinalize( ch ); delete ch; });
    pImpl->addFinalize( active.get() );
    active->SetSelf( active );
 
    pImpl->pChannelMap[alias.GetChannelId()] = active;
    //--------------------------------------------------------------------------
    // The passive channel will be deallocated by TTL
    //--------------------------------------------------------------------------
  }

References XrdCl::Log::Error(), XrdCl::URL::GetChannelId(), XrdCl::TransportManager::GetHandler(), XrdCl::URL::GetHostId(), XrdCl::DefaultEnv::GetLog(), XrdCl::URL::GetProtocol(), XrdCl::DefaultEnv::GetTransportManager(), XrdCl::Log::Info(), XrdSysMutexHelper::Lock(), XrdCl::PostMasterMsg, and XrdSysMutexHelper::UnLock().

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

Here is the call graph for this function:

Here is the caller graph for this function:

DecFileInstCnt()

void XrdCl::PostMaster::DecFileInstCnt

(

const URL &

url

)

Decrement file object instance count bound to this channel.

Definition at line 546 of file XrdClPostMaster.cc.

  {
    auto channel = pImpl->GetChannel( url );
 
    if( !channel ) return;
 
    return channel->DecFileInstCnt();
  }

Referenced by XrdCl::FileStateHandler::~FileStateHandler().

Here is the caller graph for this function:

Finalize()

bool XrdCl::PostMaster::Finalize

(

)

Finalizer.

Definition at line 168 of file XrdClPostMaster.cc.

  {
    //--------------------------------------------------------------------------
    // Clean up the channels
    //--------------------------------------------------------------------------
    if( !pImpl->pInitialized )
      return true;
 
    pImpl->pInitialized = false;
    pImpl->pJobManager->Finalize();
 
    //--------------------------------------------------------------------------
    // Finalize may cause some of the channels to remove themselves from the
    // finalize set. So make a copy. Should be no concurrency as poller and
    // jobmanager are stopped, so no lock.
    //--------------------------------------------------------------------------
    auto finSet = pImpl->pFinalizeSet;
    for( auto ch: finSet ) ch->Finalize();
 
    pImpl->pChannelMap.clear();
    return pImpl->pPoller->Finalize();
  }

Referenced by XrdCl::ForkHandler::Child(), and XrdCl::DefaultEnv::GetPostMaster().

Here is the caller graph for this function:

ForceDisconnect() [1/3]

Status XrdCl::PostMaster::ForceDisconnect

(

const URL &

url

)

Shut down a channel.

Definition at line 345 of file XrdClPostMaster.cc.

  {
    return ForceDisconnect(url, false);
  }

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

Here is the caller graph for this function:

ForceDisconnect() [2/3]

Status XrdCl::PostMaster::ForceDisconnect	(	const URL &	url,
		bool	hush
	)

Shut down a channel.

Definition at line 353 of file XrdClPostMaster.cc.

  {
    std::shared_ptr<Channel> channel;
    {
      XrdSysMutexHelper scopedLock( pImpl->pChannelMapMutex );
      PostMasterImpl::ChannelMap::iterator it =
          pImpl->pChannelMap.find( url.GetChannelId() );
 
      if( it == pImpl->pChannelMap.end() )
        return Status( stError, errInvalidOp );
      channel = it->second;
      pImpl->pChannelMap.erase( it );
    }
 
    channel->ForceDisconnect( hush );
    return Status();
  }

References XrdCl::errInvalidOp, XrdCl::URL::GetChannelId(), and XrdCl::stError.

Here is the call graph for this function:

ForceDisconnect() [3/3]

Status XrdCl::PostMaster::ForceDisconnect	(	std::shared_ptr< Channel >	channel,
		const uint64_t	sess
	)

Shut down a channel. This version is used by the channel itself.

Definition at line 374 of file XrdClPostMaster.cc.

  {
    if( !channel )
      return Status( stError, errNotSupported );
 
    {
      XrdSysMutexHelper scopedLock( pImpl->pChannelMapMutex );
      PostMasterImpl::ChannelMap::iterator it =
          pImpl->pChannelMap.find( channel->GetURL().GetChannelId() );
 
      if( it != pImpl->pChannelMap.end() && it->second == channel )
        pImpl->pChannelMap.erase( it );
    }
 
    channel->ForceDisconnect( channel, sess );
    return Status();
  }

References XrdCl::errNotSupported, and XrdCl::stError.

ForceReconnect()

Status XrdCl::PostMaster::ForceReconnect

(

const URL &

url

)

Reconnect the channel.

Definition at line 393 of file XrdClPostMaster.cc.

  {
    std::shared_ptr<Channel> channel;
    {
      XrdSysMutexHelper scopedLock( pImpl->pChannelMapMutex );
      PostMasterImpl::ChannelMap::iterator it =
          pImpl->pChannelMap.find( url.GetChannelId() );
 
      if( it == pImpl->pChannelMap.end() )
        return Status( stError, errInvalidOp );
      channel = it->second;
    }
 
    channel->ForceReconnect();
    return Status();
  }

References XrdCl::errInvalidOp, XrdCl::URL::GetChannelId(), and XrdCl::stError.

Here is the call graph for this function:

GetJobManager()

JobManager * XrdCl::PostMaster::GetJobManager

(

)

Get the job manager object user by the post master.

Definition at line 337 of file XrdClPostMaster.cc.

  {
    return pImpl->pJobManager;
  }

Referenced by XrdCl::FileStateHandler::Close(), XrdCl::Operation< HasHndl >::Run(), XrdEc::ScheduleHandler(), and XrdCl::FileStateHandler::TimeOutRequests().

Here is the caller graph for this function:

GetTaskManager()

TaskManager * XrdCl::PostMaster::GetTaskManager

(

)

Get the task manager object user by the post master.

Definition at line 329 of file XrdClPostMaster.cc.

  {
    return pImpl->pTaskManager;
  }

Referenced by XrdCl::ForkHandler::Child(), XrdCl::DefaultEnv::GetPostMaster(), and XrdCl::XRootDMsgHandler::Process().

Here is the caller graph for this function:

Initialize()

bool XrdCl::PostMaster::Initialize

(

)

Initializer.

Definition at line 141 of file XrdClPostMaster.cc.

  {
    Env *env = DefaultEnv::GetEnv();
    std::string pollerPref = DefaultPollerPreference;
    env->GetString( "PollerPreference", pollerPref );
 
    pImpl->pPoller = PollerFactory::CreatePoller( pollerPref );
 
    if( !pImpl->pPoller )
      return false;
 
    bool st = pImpl->pPoller->Initialize();
 
    if( !st )
    {
      delete pImpl->pPoller;
      return false;
    }
 
    pImpl->pJobManager->Initialize();
    pImpl->pInitialized = true;
    return true;
  }

References XrdCl::PollerFactory::CreatePoller(), XrdCl::DefaultPollerPreference, XrdCl::DefaultEnv::GetEnv(), and XrdCl::Env::GetString().

Referenced by XrdCl::ForkHandler::Child(), and XrdCl::DefaultEnv::GetPostMaster().

Here is the call graph for this function:

Here is the caller graph for this function:

IsRunning()

bool XrdCl::PostMaster::IsRunning

(

)

Returns

: true if underlying threads are running, false otherwise

Definition at line 558 of file XrdClPostMaster.cc.

  {
    return pImpl->pRunning;
  }

NbConnectedStrm()

uint16_t XrdCl::PostMaster::NbConnectedStrm

(

const URL &

url

)

Get the number of connected data streams.

Definition at line 413 of file XrdClPostMaster.cc.

  {
    auto channel = pImpl->GetChannel( url );
    if( !channel ) return 0;
    return channel->NbConnectedStrm();
  }

NotifyConnectHandler()

void XrdCl::PostMaster::NotifyConnectHandler

(

const URL &

url

)

Notify the global on-connect handler.

Definition at line 452 of file XrdClPostMaster.cc.

  {
    XrdSysMutexHelper lck( pImpl->pMtx );
    if( pImpl->pOnConnJob )
    {
      URL *ptr = new URL( url );
      pImpl->pJobManager->QueueJob( pImpl->pOnConnJob.get(), ptr );
    }
  }

Referenced by XrdCl::Stream::OnConnect().

Here is the caller graph for this function:

NotifyConnErrHandler()

void XrdCl::PostMaster::NotifyConnErrHandler	(	const URL &	url,
		const XRootDStatus &	status
	)

Notify the global error connection handler.

Definition at line 465 of file XrdClPostMaster.cc.

  {
    XrdSysMutexHelper lck( pImpl->pMtx );
    if( pImpl->pOnConnErrCB )
    {
      ConnErrJob *job = new ConnErrJob( url, status, pImpl->pOnConnErrCB );
      pImpl->pJobManager->QueueJob( job, nullptr );
    }
  }

Referenced by XrdCl::Stream::OnConnectError().

Here is the caller graph for this function:

QueryTransport()

Status XrdCl::PostMaster::QueryTransport	(	const URL &	url,
		uint16_t	query,
		AnyObject &	result
	)

Query the transport handler for a given URL

Parameters

url	the channel to be queried
query	the query as defined in the TransportQuery struct or others that may be recognized by the protocol transport
result	the result of the query

Returns

status of the query

Definition at line 276 of file XrdClPostMaster.cc.

  {
    std::shared_ptr<Channel> channel;
    {
      XrdSysMutexHelper scopedLock( pImpl->pChannelMapMutex );
      PostMasterImpl::ChannelMap::iterator it =
          pImpl->pChannelMap.find( url.GetChannelId() );
      if( it == pImpl->pChannelMap.end() )
        return Status( stError, errInvalidOp );
      channel = it->second;
    }
 
    if( !channel )
      return Status( stError, errNotSupported );
 
    return channel->QueryTransport( query, result );
  }

References XrdCl::errInvalidOp, XrdCl::errNotSupported, XrdCl::URL::GetChannelId(), and XrdCl::stError.

Referenced by XrdCl::FileSystem::DirList(), XrdCl::FileStateHandler::PgRead(), and XrdCl::XRootDMsgHandler::Process().

Here is the call graph for this function:

Here is the caller graph for this function:

Redirect()

Status XrdCl::PostMaster::Redirect	(	const URL &	url,
		Message *	msg,
		MsgHandler *	handler
	)

Definition at line 262 of file XrdClPostMaster.cc.

  {
    RedirectorRegistry &registry  = RedirectorRegistry::Instance();
    VirtualRedirector *redirector = registry.Get( url );
    if( !redirector )
      return Status( stError, errInvalidOp );
    return redirector->HandleRequest( msg, inHandler );
  }

References XrdCl::errInvalidOp, XrdCl::RedirectorRegistry::Get(), XrdCl::VirtualRedirector::HandleRequest(), XrdCl::RedirectorRegistry::Instance(), and XrdCl::stError.

Referenced by XrdCl::MessageUtils::RedirectMessage().

Here is the call graph for this function:

Here is the caller graph for this function:

RegisterEventHandler()

Status XrdCl::PostMaster::RegisterEventHandler	(	const URL &	url,
		ChannelEventHandler *	handler
	)

Register channel event handler.

Definition at line 299 of file XrdClPostMaster.cc.

  {
    auto channel = pImpl->GetChannel( url );
 
    if( !channel )
      return Status( stError, errNotSupported );
 
    channel->RegisterEventHandler( handler );
    return Status();
  }

References XrdCl::errNotSupported, and XrdCl::stError.

Reinitialize()

bool XrdCl::PostMaster::Reinitialize

(

)

Reinitialize after fork.

Definition at line 240 of file XrdClPostMaster.cc.

  {
    return true;
  }

RemoveEventHandler()

Status XrdCl::PostMaster::RemoveEventHandler	(	const URL &	url,
		ChannelEventHandler *	handler
	)

Remove a channel event handler.

Definition at line 314 of file XrdClPostMaster.cc.

  {
    auto channel = pImpl->GetChannel( url );
 
    if( !channel )
      return Status( stError, errNotSupported );
 
    channel->RemoveEventHandler( handler );
    return Status();
  }

References XrdCl::errNotSupported, and XrdCl::stError.

Send()

XRootDStatus XrdCl::PostMaster::Send	(	const URL &	url,
		Message *	msg,
		MsgHandler *	handler,
		bool	stateful,
		time_t	expires
	)

Send the message asynchronously - the message is inserted into the send queue and a listener is called when the message is succesfsully pushed through the wire or when the timeout elapses

DEADLOCK WARNING: no lock should be taken while calling this method that are used in the callback as well.

Parameters

url	recipient of the message
msg	message to be sent
expires	unix timestamp after which a failure is reported to the handler
handler	handler will be notified about the status
stateful	physical stream disconnection causes an error

Returns

success if the message was successfully inserted into the send queues, failure otherwise

Definition at line 248 of file XrdClPostMaster.cc.

  {
    auto channel = pImpl->GetChannel( url );
 
    if( !channel )
      return XRootDStatus( stError, errNotSupported );
 
    return channel->Send( msg, handler, stateful, expires );
  }

References XrdCl::errNotSupported, and XrdCl::stError.

Referenced by XrdCl::MessageUtils::SendMessage().

Here is the caller graph for this function:

SetConnectionErrorHandler()

void XrdCl::PostMaster::SetConnectionErrorHandler

(

std::function< void(const URL &, const XRootDStatus &)>

handler

)

Set the global on-error on-connect handler for control streams.

Definition at line 443 of file XrdClPostMaster.cc.

  {
    XrdSysMutexHelper lck( pImpl->pMtx );
    pImpl->pOnConnErrCB = std::move( handler );
  }

SetOnConnectHandler()

void XrdCl::PostMaster::SetOnConnectHandler

(

std::unique_ptr< Job >

onConnJob

)

Set the global connection error handler.

Set the global on-connect handler for control streams.

Definition at line 434 of file XrdClPostMaster.cc.

  {
    XrdSysMutexHelper lck( pImpl->pMtx );
    pImpl->pOnConnJob = std::move( onConnJob );
  }

Referenced by XrdPosixConfig::conTracker().

Here is the caller graph for this function:

SetOnDataConnectHandler()

void XrdCl::PostMaster::SetOnDataConnectHandler	(	const URL &	url,
		std::shared_ptr< Job >	onConnJob
	)

Set the on-connect handler for data streams.

Definition at line 423 of file XrdClPostMaster.cc.

  {
    auto channel = pImpl->GetChannel( url );
    if( !channel ) return;
    channel->SetOnDataConnectHandler( onConnJob );
  }

Start()

bool XrdCl::PostMaster::Start

(

)

Start the post master.

Definition at line 194 of file XrdClPostMaster.cc.

  {
    if( !pImpl->pInitialized )
      return false;
 
    if( !pImpl->pPoller->Start() )
      return false;
 
    if( !pImpl->pTaskManager->Start() )
    {
      pImpl->pPoller->Stop();
      return false;
    }
 
    if( !pImpl->pJobManager->Start() )
    {
      pImpl->pPoller->Stop();
      pImpl->pTaskManager->Stop();
      return false;
    }
 
    pImpl->pRunning = true;
    return true;
  }

Referenced by XrdCl::ForkHandler::Child(), XrdCl::DefaultEnv::GetPostMaster(), and XrdCl::ForkHandler::Parent().

Here is the caller graph for this function:

Stop()

bool XrdCl::PostMaster::Stop

(

)

Stop the postmaster.

Definition at line 222 of file XrdClPostMaster.cc.

  {
    if( !pImpl->pInitialized || !pImpl->pRunning )
      return true;
 
    if( !pImpl->pJobManager->Stop() )
      return false;
    if( !pImpl->pPoller->Stop() )
      return false;
    if( !pImpl->pTaskManager->Stop() )
      return false;
    pImpl->pRunning = false;
    return true;
  }

Referenced by main(), and XrdCl::ForkHandler::Prepare().

Here is the caller graph for this function:

---

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

• XrdClPostMaster.hh
• XrdClPostMaster.cc