A poller implementation using the build-in XRootD poller. More...

#include <XrdClPollerBuiltIn.hh>

Inheritance diagram for XrdCl::PollerBuiltIn:

Collaboration diagram for XrdCl::PollerBuiltIn:

Public Member Functions
	PollerBuiltIn ()
	Constructor.
	~PollerBuiltIn ()
virtual bool	AddSocket (Socket socket, SocketHandler handler)
virtual bool	EnableReadNotification (Socket *socket, bool notify, time_t timeout=60)
virtual bool	EnableWriteNotification (Socket *socket, bool notify, time_t timeout=60)
virtual bool	Finalize ()
	Finalize the poller.
virtual bool	Initialize ()
	Initialize the poller.
virtual bool	IsRegistered (Socket *socket)
	Check whether the socket is registered with the poller.
virtual bool	IsRunning () const
	Is the event loop running?
virtual bool	RemoveSocket (Socket *socket)
	Remove the socket.
virtual void	ShutdownEvents (Socket *socket)
virtual bool	Start ()
	Start polling.
virtual bool	Stop ()
	Stop polling.
Public Member Functions inherited from XrdCl::Poller
virtual	~Poller ()
	Destructor.

Detailed Description

A poller implementation using the build-in XRootD poller.

Definition at line 40 of file XrdClPollerBuiltIn.hh.

Constructor & Destructor Documentation

◆ PollerBuiltIn()

XrdCl::PollerBuiltIn::PollerBuiltIn ( )

inline

Constructor.

Definition at line 46 of file XrdClPollerBuiltIn.hh.

46: pNbPoller( GetNbPollerInit() ){}

◆ ~PollerBuiltIn()

XrdCl::PollerBuiltIn::~PollerBuiltIn ( )

inline

Definition at line 48 of file XrdClPollerBuiltIn.hh.

48{}

Member Function Documentation

◆ AddSocket()

bool XrdCl::PollerBuiltIn::AddSocket	(	Socket *	socket,
		SocketHandler *	handler )

virtual

Add socket to the polling loop

Parameters

socket	the socket
handler	object handling the events

Implements XrdCl::Poller.

Definition at line 348 of file XrdClPollerBuiltIn.cc.

  {
    Log *log = DefaultEnv::GetLog();
    XrdSysMutexHelper scopedLock( pMutex );
 
    if( !socket )
    {
      log->Error( PollerMsg, "Invalid socket, impossible to poll" );
      return false;
    }
 
    if( socket->GetStatus() != Socket::Connected &&
        socket->GetStatus() != Socket::Connecting )
    {
      log->Error( PollerMsg, "Socket is not in a state valid for polling" );
      return false;
    }
 
    log->Debug( PollerMsg, "Adding socket %p to the poller", (void*)socket );
 
    //--------------------------------------------------------------------------
    // Check if the socket is already registered
    //--------------------------------------------------------------------------
    SocketMap::const_iterator it = pSocketMap.find( socket );
    if( it != pSocketMap.end() )
    {
      log->Warning( PollerMsg, "%s Already registered with this poller",
                               socket->GetName().c_str() );
      return false;
    }
 
    //--------------------------------------------------------------------------
    // Create the socket helper
    //--------------------------------------------------------------------------
    XrdSys::IOEvents::Poller* poller = RegisterAndGetPoller( socket );
 
    if( !poller )
    {
      log->Error( PollerMsg, "No poller available, can not add socket" );
      return false;
    }
 
    PollerHelper *helper = new PollerHelper();
    helper->callBack = new ::SocketCallBack( socket, handler );
 
    if( poller )
    {
      helper->channel  = new XrdSys::IOEvents::Channel( poller,
                                                        socket->GetFD(),
                                                        helper->callBack );
    }
 
    handler->Initialize( this );
    pSocketMap[socket] = helper;
    return true;
  }

References XrdCl::Socket::Connected, XrdCl::Socket::Connecting, XrdCl::Log::Debug(), XrdCl::Log::Error(), XrdCl::Socket::GetFD(), XrdCl::DefaultEnv::GetLog(), XrdCl::Socket::GetName(), XrdCl::Socket::GetStatus(), XrdCl::SocketHandler::Initialize(), XrdCl::PollerMsg, and XrdCl::Log::Warning().

Here is the call graph for this function:

◆ EnableReadNotification()

bool XrdCl::PollerBuiltIn::EnableReadNotification	(	Socket *	socket,
		bool	notify,
		time_t	timeout = 60 )

virtual

Notify the handler about read events

Parameters

socket	the socket
notify	specify if the handler should be notified
timeout	if no read event occurred after this time a timeout event will be generated

Implements XrdCl::Poller.

Definition at line 479 of file XrdClPollerBuiltIn.cc.

  {
    using namespace XrdSys::IOEvents;
    Log *log = DefaultEnv::GetLog();
 
    if( !socket )
    {
      log->Error( PollerMsg, "Invalid socket, read events unavailable" );
      return false;
    }
 
    //--------------------------------------------------------------------------
    // Check if the socket is registered
    //--------------------------------------------------------------------------
    XrdSysMutexHelper scopedLock( pMutex );
    SocketMap::const_iterator it = pSocketMap.find( socket );
    if( it == pSocketMap.end() )
    {
      log->Warning( PollerMsg, "%s Socket is not registered",
                               socket->GetName().c_str() );
      return false;
    }
 
    PollerHelper *helper = (PollerHelper*)it->second;
    XrdSys::IOEvents::Poller *poller = GetPoller( socket );
 
    //--------------------------------------------------------------------------
    // Enable read notifications
    //--------------------------------------------------------------------------
    if( notify )
    {
      if( helper->readEnabled )
        return true;
      helper->readTimeout = timeout;
 
      log->Dump( PollerMsg, "%s Enable read notifications, timeout: %lld",
                            socket->GetName().c_str(), (long long)timeout );
 
      if( poller )
      {
        const char *errMsg;
        bool status = helper->channel->Enable( Channel::readEvents, timeout,
                                               &errMsg );
        if( !status )
        {
          log->Error( PollerMsg, "%s Unable to enable read notifications: %s",
                      socket->GetName().c_str(), errMsg );
          return false;
        }
      }
      helper->readEnabled = true;
    }
 
    //--------------------------------------------------------------------------
    // Disable read notifications
    //--------------------------------------------------------------------------
    else
    {
      if( !helper->readEnabled )
        return true;
 
      log->Dump( PollerMsg, "%s Disable read notifications",
                            socket->GetName().c_str() );
 
      if( poller )
      {
        const char *errMsg;
        bool status = helper->channel->Disable( Channel::readEvents, &errMsg );
        if( !status )
        {
          log->Error( PollerMsg, "%s Unable to disable read notifications: %s",
                      socket->GetName().c_str(), errMsg );
          return false;
        }
      }
      helper->readEnabled = false;
    }
    return true;
  }

References XrdSys::IOEvents::Channel::Disable(), XrdCl::Log::Dump(), XrdSys::IOEvents::Channel::Enable(), XrdCl::Log::Error(), XrdCl::DefaultEnv::GetLog(), XrdCl::Socket::GetName(), XrdCl::PollerMsg, XrdSys::IOEvents::Channel::readEvents, and XrdCl::Log::Warning().

Here is the call graph for this function:

◆ EnableWriteNotification()

bool XrdCl::PollerBuiltIn::EnableWriteNotification	(	Socket *	socket,
		bool	notify,
		time_t	timeout = 60 )

virtual

Notify the handler about write events

Parameters

socket	the socket
notify	specify if the handler should be notified
timeout	if no write event occurred after this time a timeout event will be generated

Implements XrdCl::Poller.

Definition at line 564 of file XrdClPollerBuiltIn.cc.

  {
    using namespace XrdSys::IOEvents;
    Log *log = DefaultEnv::GetLog();
 
    if( !socket )
    {
      log->Error( PollerMsg, "Invalid socket, write events unavailable" );
      return false;
    }
 
    //--------------------------------------------------------------------------
    // Check if the socket is registered
    //--------------------------------------------------------------------------
    XrdSysMutexHelper scopedLock( pMutex );
    SocketMap::const_iterator it = pSocketMap.find( socket );
    if( it == pSocketMap.end() )
    {
      log->Warning( PollerMsg, "%s Socket is not registered",
                               socket->GetName().c_str() );
      return false;
    }
 
    PollerHelper *helper = (PollerHelper*)it->second;
    XrdSys::IOEvents::Poller *poller = GetPoller( socket );
 
    //--------------------------------------------------------------------------
    // Enable write notifications
    //--------------------------------------------------------------------------
    if( notify )
    {
      if( helper->writeEnabled )
        return true;
 
      helper->writeTimeout = timeout;
 
      log->Dump( PollerMsg, "%s Enable write notifications, timeout: %lld",
                            socket->GetName().c_str(), (long long)timeout );
 
      if( poller )
      {
        const char *errMsg;
        bool status = helper->channel->Enable( Channel::writeEvents, timeout,
                                               &errMsg );
        if( !status )
        {
          log->Error( PollerMsg, "%s Unable to enable write notifications: %s",
                      socket->GetName().c_str(), errMsg );
          return false;
        }
      }
      helper->writeEnabled = true;
    }
 
    //--------------------------------------------------------------------------
    // Disable read notifications
    //--------------------------------------------------------------------------
    else
    {
      if( !helper->writeEnabled )
        return true;
 
      log->Dump( PollerMsg, "%s Disable write notifications",
                            socket->GetName().c_str() );
      if( poller )
      {
        const char *errMsg;
        bool status = helper->channel->Disable( Channel::writeEvents, &errMsg );
        if( !status )
        {
          log->Error( PollerMsg, "%s Unable to disable write notifications: %s",
                      socket->GetName().c_str(), errMsg );
          return false;
        }
      }
      helper->writeEnabled = false;
    }
    return true;
  }

References XrdSys::IOEvents::Channel::Disable(), XrdCl::Log::Dump(), XrdSys::IOEvents::Channel::Enable(), XrdCl::Log::Error(), XrdCl::DefaultEnv::GetLog(), XrdCl::Socket::GetName(), XrdCl::PollerMsg, XrdCl::Log::Warning(), and XrdSys::IOEvents::Channel::writeEvents.

Here is the call graph for this function:

◆ Finalize()

bool XrdCl::PollerBuiltIn::Finalize ( )

virtual

Finalize the poller.

Implements XrdCl::Poller.

Definition at line 190 of file XrdClPollerBuiltIn.cc.

  {
    //--------------------------------------------------------------------------
    // Clean up the channels
    //--------------------------------------------------------------------------
    SocketMap::iterator it;
    for( it = pSocketMap.begin(); it != pSocketMap.end(); ++it )
    {
      PollerHelper *helper = (PollerHelper*)it->second;
      if( helper->channel ) helper->channel->Delete();
      delete helper->callBack;
      delete helper;
    }
    pSocketMap.clear();
 
    return true;
  }

References XrdSys::IOEvents::Channel::Delete().

Here is the call graph for this function:

◆ Initialize()

bool XrdCl::PollerBuiltIn::Initialize ( )

virtual

Initialize the poller.

Implements XrdCl::Poller.

Definition at line 182 of file XrdClPollerBuiltIn.cc.

  {
    return true;
  }

◆ IsRegistered()

bool XrdCl::PollerBuiltIn::IsRegistered ( Socket * socket )

virtual

Check whether the socket is registered with the poller.

Implements XrdCl::Poller.

Definition at line 649 of file XrdClPollerBuiltIn.cc.

  {
    XrdSysMutexHelper scopedLock( pMutex );
    SocketMap::iterator it = pSocketMap.find( socket );
    return it != pSocketMap.end();
  }

◆ IsRunning()

virtual bool XrdCl::PollerBuiltIn::IsRunning ( ) const

inlinevirtual

Is the event loop running?

Implements XrdCl::Poller.

Definition at line 123 of file XrdClPollerBuiltIn.hh.

      {
        return !pPollerPool.empty();
      }

◆ RemoveSocket()

bool XrdCl::PollerBuiltIn::RemoveSocket ( Socket * socket )

virtual

Remove the socket.

Implements XrdCl::Poller.

Definition at line 433 of file XrdClPollerBuiltIn.cc.

  {
    using namespace XrdSys::IOEvents;
    Log *log = DefaultEnv::GetLog();
 
    //--------------------------------------------------------------------------
    // Find the right socket
    //--------------------------------------------------------------------------
    XrdSysMutexHelper scopedLock( pMutex );
    SocketMap::iterator it = pSocketMap.find( socket );
    if( it == pSocketMap.end() )
      return true;
 
    log->Debug( PollerMsg, "%s Removing socket from the poller",
                           socket->GetName().c_str() );
 
    // unregister from the poller it's currently associated with
    UnregisterFromPoller( socket );
 
    //--------------------------------------------------------------------------
    // Remove the socket
    //--------------------------------------------------------------------------
    PollerHelper *helper = (PollerHelper*)it->second;
    pSocketMap.erase( it );
    scopedLock.UnLock();
 
    if( helper->channel )
    {
      const char *errMsg;
      bool status = helper->channel->Disable( Channel::allEvents, &errMsg );
      if( !status )
      {
        log->Error( PollerMsg, "%s Unable to disable write notifications: %s",
                    socket->GetName().c_str(), errMsg );
        return false;
      }
      helper->channel->Delete();
    }
    delete helper->callBack;
    delete helper;
    return true;
  }

References XrdSys::IOEvents::Channel::allEvents, XrdCl::Log::Debug(), XrdSys::IOEvents::Channel::Delete(), XrdSys::IOEvents::Channel::Disable(), XrdCl::Log::Error(), XrdCl::DefaultEnv::GetLog(), XrdCl::Socket::GetName(), XrdCl::PollerMsg, and XrdSysMutexHelper::UnLock().

Here is the call graph for this function:

◆ ShutdownEvents()

void XrdCl::PollerBuiltIn::ShutdownEvents ( Socket * socket )

virtual

Disables further callbacks to the socket's event handler. If callback is currently running wait for it, unless it is the current thread.

Implements XrdCl::Poller.

Definition at line 412 of file XrdClPollerBuiltIn.cc.

  {
    XrdSysMutexHelper scopedLock( pMutex );
    SocketMap::iterator it = pSocketMap.find( socket );
    if( it == pSocketMap.end() )
      return;
 
    PollerHelper *helper = (PollerHelper*)it->second;
    if( !helper ) return;
    XrdSys::IOEvents::CallBack *cb = helper->callBack;
    if( !cb ) return;
    SocketCallBack *scb = dynamic_cast<SocketCallBack*>( cb );
    if( !scb ) return;
    auto dc = scb->GetControl();
    scopedLock.UnLock();
    dc->DisableCallBack();
  }

References XrdSysMutexHelper::UnLock().

Here is the call graph for this function:

◆ Start()

bool XrdCl::PollerBuiltIn::Start ( )

virtual

Start polling.

Implements XrdCl::Poller.

Definition at line 211 of file XrdClPollerBuiltIn.cc.

  {
    //--------------------------------------------------------------------------
    // Start the poller
    //--------------------------------------------------------------------------
    using namespace XrdSys;
 
    Log *log = DefaultEnv::GetLog();
    log->Debug( PollerMsg, "Creating and starting the built-in poller..." );
    XrdSysMutexHelper scopedLock( pMutex );
    int         errNum = 0;
    const char *errMsg = 0;
 
    for( int i = 0; i < pNbPoller; ++i )
    {
      XrdSys::IOEvents::Poller* poller = IOEvents::Poller::Create( errNum, &errMsg );
      if( !poller )
      {
        log->Error( PollerMsg, "Unable to create the internal poller object: "
                               "%s (%s)", XrdSysE2T( errno ), errMsg );
        return false;
      }
      pPollerPool.push_back( poller );
    }
 
    pNext = pPollerPool.begin();
 
    log->Debug( PollerMsg, "Using %d poller threads", pNbPoller );
 
    //--------------------------------------------------------------------------
    // Check if we have any descriptors to reinsert from the last time we
    // were started
    //--------------------------------------------------------------------------
    SocketMap::iterator it;
    for( it = pSocketMap.begin(); it != pSocketMap.end(); ++it )
    {
      PollerHelper *helper = (PollerHelper*)it->second;
      Socket       *socket = it->first;
 
      // static cast for the downcast as we're sure it's a SocketCallBack
      auto *scb = static_cast<SocketCallBack*>( helper->callBack );
      if( scb )
      {
        auto dc = scb->GetControl();
        if( dc ) dc->Restart();
      }
 
      helper->channel = new IOEvents::Channel( RegisterAndGetPoller( socket ), socket->GetFD(),
                                               helper->callBack );
      if( helper->readEnabled )
      {
        bool status = helper->channel->Enable( IOEvents::Channel::readEvents,
                                               helper->readTimeout, &errMsg );
        if( !status )
        {
          log->Error( PollerMsg, "Unable to enable read notifications "
                      "while re-starting %s (%s)", XrdSysE2T( errno ), errMsg );
 
          return false;
        }
      }
 
      if( helper->writeEnabled )
      {
        bool status = helper->channel->Enable( IOEvents::Channel::writeEvents,
                                               helper->writeTimeout, &errMsg );
        if( !status )
        {
          log->Error( PollerMsg, "Unable to enable write notifications "
                      "while re-starting %s (%s)", XrdSysE2T( errno ), errMsg );
 
          return false;
        }
      }
    }
    return true;
  }

References XrdSys::IOEvents::Poller::Create(), XrdCl::Log::Debug(), XrdSys::IOEvents::Channel::Enable(), XrdCl::Log::Error(), XrdCl::Socket::GetFD(), XrdCl::DefaultEnv::GetLog(), XrdCl::PollerMsg, XrdSys::IOEvents::Channel::readEvents, XrdSys::IOEvents::Channel::writeEvents, and XrdSysE2T().

Here is the call graph for this function:

◆ Stop()

bool XrdCl::PollerBuiltIn::Stop ( )

virtual

Stop polling.

Implements XrdCl::Poller.

Definition at line 292 of file XrdClPollerBuiltIn.cc.

  {
    using namespace XrdSys::IOEvents;
 
    Log *log = DefaultEnv::GetLog();
    log->Debug( PollerMsg, "Stopping the poller..." );
 
    XrdSysMutexHelper scopedLock( pMutex );
 
    if( pPollerPool.empty() )
    {
      log->Debug( PollerMsg, "Stopping a poller that has not been started" );
      return true;
    }
 
    while( !pPollerPool.empty() )
    {
      XrdSys::IOEvents::Poller *poller = pPollerPool.back();
      if( *pNext == poller )
        pNext = pPollerPool.begin();
      pPollerPool.pop_back();
 
      if( !poller ) continue;
 
      scopedLock.UnLock();
      poller->Stop();
      delete poller;
      scopedLock.Lock( &pMutex );
    }
    pNext = pPollerPool.end();
    pPollerMap.clear();
 
    SocketMap::iterator  it;
    const char          *errMsg = 0;
 
    for( it = pSocketMap.begin(); it != pSocketMap.end(); ++it )
    {
      PollerHelper *helper = (PollerHelper*)it->second;
      if( !helper->channel ) continue;
      bool status = helper->channel->Disable( Channel::allEvents, &errMsg );
      if( !status )
      {
        Socket *socket = it->first;
        log->Error( PollerMsg, "%s Unable to disable write notifications: %s",
                    socket->GetName().c_str(), errMsg );
      }
      helper->channel->Delete();
      helper->channel = 0;
    }
 
    return true;
  }

References XrdSys::IOEvents::Channel::allEvents, XrdCl::Log::Debug(), XrdSys::IOEvents::Channel::Delete(), XrdSys::IOEvents::Channel::Disable(), XrdCl::Log::Error(), XrdCl::DefaultEnv::GetLog(), XrdCl::Socket::GetName(), XrdSysMutexHelper::Lock(), XrdCl::PollerMsg, XrdSys::IOEvents::Poller::Stop(), and XrdSysMutexHelper::UnLock().

Here is the call graph for this function:

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

Public Member Functions

Detailed Description

Constructor & Destructor Documentation

◆ PollerBuiltIn()

◆ ~PollerBuiltIn()

Member Function Documentation

◆ AddSocket()

◆ EnableReadNotification()

◆ EnableWriteNotification()

◆ Finalize()

◆ Initialize()

◆ IsRegistered()

◆ IsRunning()

◆ RemoveSocket()

◆ ShutdownEvents()

◆ Start()

◆ Stop()