XrdCl::PollerBuiltIn Class Reference

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. More...
	~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. More...
virtual bool	Initialize ()
	Initialize the poller. More...
virtual bool	IsRegistered (Socket *socket)
	Check whether the socket is registered with the poller. More...
virtual bool	IsRunning () const
	Is the event loop running? More...
virtual bool	RemoveSocket (Socket *socket)
	Remove the socket. More...
virtual void	ShutdownEvents (Socket *socket)
virtual bool	Start ()
	Start polling. More...
virtual bool	Stop ()
	Stop polling. More...
Public Member Functions inherited from XrdCl::Poller
virtual	~Poller ()
	Destructor. More...

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.

: pNbPoller( GetNbPollerInit() ){}

~PollerBuiltIn()

XrdCl::PollerBuiltIn::~PollerBuiltIn ( )

inline

Definition at line 48 of file XrdClPollerBuiltIn.hh.

{}

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 XrdCl::Log::Dump(), XrdCl::Log::Error(), XrdCl::DefaultEnv::GetLog(), XrdCl::Socket::GetName(), XrdCl::PollerMsg, 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 XrdCl::Log::Dump(), XrdCl::Log::Error(), XrdCl::DefaultEnv::GetLog(), XrdCl::Socket::GetName(), XrdCl::PollerMsg, and XrdCl::Log::Warning().

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;
  }

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 XrdCl::Log::Debug(), 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 Create, XrdCl::Log::Debug(), XrdCl::Log::Error(), XrdCl::DefaultEnv::GetLog(), XrdCl::PollerMsg, 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 XrdCl::Log::Debug(), 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:

• XrdClPollerBuiltIn.hh
• XrdClPollerBuiltIn.cc