XrdClPollerBuiltIn.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/XrdClPollerBuiltIn.hh"
#include "XrdCl/XrdClLog.hh"
#include "XrdCl/XrdClDefaultEnv.hh"
#include "XrdCl/XrdClConstants.hh"
#include "XrdCl/XrdClSocket.hh"
#include "XrdCl/XrdClOptimizers.hh"
#include "XrdSys/XrdSysE2T.hh"
#include "XrdSys/XrdSysIOEvents.hh"
 
namespace
{
  //----------------------------------------------------------------------------
  // A helper struct passed to the callback as a custom arg
  //----------------------------------------------------------------------------
  struct PollerHelper
  {
    PollerHelper():
      channel(0), callBack(0), readEnabled(false), writeEnabled(false),
      readTimeout(0), writeTimeout(0)
    {}
    XrdSys::IOEvents::Channel  *channel;
    XrdSys::IOEvents::CallBack *callBack;
    bool                        readEnabled;
    bool                        writeEnabled;
    time_t                      readTimeout;
    time_t                      writeTimeout;
  };
 
  //----------------------------------------------------------------------------
  // Call back implementation
  //----------------------------------------------------------------------------
  class SocketCallBack: public XrdSys::IOEvents::CallBack
  {
    public:
      enum CallBackFlags
      {
        kRunningCallBack = 1,
        kIdSet           = 2,
        kWantDisable     = 4,
        kDisabled        = 8
      };
 
      struct DisableControl
      {
         DisableControl() : pFlags( 0 ), pCnd( 0 ) { }
         ~DisableControl() { }
 
        //------------------------------------------------------------------------
        // Want to disable further callbacks. If callback is currently running
        // in a different thread from our caller, wait until callback finishes.
        //------------------------------------------------------------------------
        void DisableCallBack()
        {
          const int flags = pFlags.fetch_or( kWantDisable );
          if( !(flags & kIdSet) ) return;
          if( !(flags & kRunningCallBack) ) return;
          XrdSysCondVarHelper lck( pCnd );
          if( XrdSysThread::Same( XrdSysThread::ID(), pSelfId ) ) return;
          while( !(pFlags.load() & kDisabled) ) pCnd.Wait();
        }
 
         std::atomic<int>      pFlags;
         XrdSysCondVar         pCnd;
         pthread_t             pSelfId;
      };
 
      SocketCallBack( XrdCl::Socket *sock, XrdCl::SocketHandler *sh ):
        pSocket( sock ), pHandler( sh )
      {
        pControl = std::make_shared<DisableControl>();
      }
 
      virtual ~SocketCallBack() {};
 
      virtual bool Event( XrdSys::IOEvents::Channel *chP,
                          void                      *cbArg,
                          int                        evFlags )
      {
        using namespace XrdCl;
        uint8_t ev      = 0;
 
        if( evFlags & ReadyToRead )  ev |= SocketHandler::ReadyToRead;
        if( evFlags & ReadTimeOut )  ev |= SocketHandler::ReadTimeOut;
        if( evFlags & ReadyToWrite ) ev |= SocketHandler::ReadyToWrite;
        if( evFlags & WriteTimeOut ) ev |= SocketHandler::WriteTimeOut;
 
        Log *log = DefaultEnv::GetLog();
        if( unlikely(log->GetLevel() >= Log::DumpMsg) )
        {
          log->Dump( PollerMsg, "%s Got an event: %s",
                                pSocket->GetName().c_str(),
                                SocketHandler::EventTypeToString( ev ).c_str() );
        }
 
        int flags = pControl->pFlags.fetch_or( kRunningCallBack );
        if( !( flags & kIdSet ) )
        {
          XrdSysCondVarHelper lck( pControl->pCnd );
          pControl->pSelfId = XrdSysThread::ID();
          flags = pControl->pFlags.fetch_or( kIdSet );
        }
        if( flags & kWantDisable )
        {
          XrdSysCondVarHelper lck( pControl->pCnd );
          pControl->pFlags &= ~kRunningCallBack;
          pControl->pFlags |= kDisabled;
          pControl->pCnd.Broadcast();
          return false;
        }
 
        //----------------------------------------------------------------------
        // If the event handler calls RemoveSocket for itself this object may
        // be destroyed during the Event call.
        //----------------------------------------------------------------------
        auto control = pControl;
        pHandler->Event( ev, pSocket );
 
        flags = control->pFlags.fetch_and( ~kRunningCallBack );
        if( flags & kWantDisable )
        {
          XrdSysCondVarHelper lck( control->pCnd );
          control->pFlags |= kDisabled;
          control->pCnd.Broadcast();
          return false;
        }
        return true;
      }
 
      std::shared_ptr<DisableControl> GetControl()
      {
        return pControl;
      }
 
    private:
      XrdCl::Socket        *pSocket;
      XrdCl::SocketHandler *pHandler;
      std::shared_ptr<DisableControl> pControl;
  };
}
 
 
namespace XrdCl
{
  //----------------------------------------------------------------------------
  // Initialize the poller
  //----------------------------------------------------------------------------
  bool PollerBuiltIn::Initialize()
  {
    return true;
  }
 
  //----------------------------------------------------------------------------
  // Finalize the poller
  //----------------------------------------------------------------------------
  bool PollerBuiltIn::Finalize()
  {
    //--------------------------------------------------------------------------
    // 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;
  }
 
  //------------------------------------------------------------------------
  // Start polling
  //------------------------------------------------------------------------
  bool PollerBuiltIn::Start()
  {
    //--------------------------------------------------------------------------
    // 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;
      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;
  }
 
  //------------------------------------------------------------------------
  // Stop polling
  //------------------------------------------------------------------------
  bool PollerBuiltIn::Stop()
  {
    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;
  }
 
  //------------------------------------------------------------------------
  // Add socket to the polling queue
  //------------------------------------------------------------------------
  bool PollerBuiltIn::AddSocket( Socket        *socket,
                                 SocketHandler *handler )
  {
    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;
  }
 
  //----------------------------------------------------------------------------
  // This disables further callbacks to the socket's handler from the poller
  // thread. Will wait until any current callback completes (unless our caller
  // is the same thread). We do not yet remove the socket from the poller.
  // Removal may still block until the relevant poller thread can run.
  //----------------------------------------------------------------------------
  void PollerBuiltIn::ShutdownEvents( Socket *socket )
  {
    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();
  }
 
  //------------------------------------------------------------------------
  // Remove the socket
  //------------------------------------------------------------------------
  bool PollerBuiltIn::RemoveSocket( Socket *socket )
  {
    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;
  }
 
  //----------------------------------------------------------------------------
  // Notify the handler about read events
  //----------------------------------------------------------------------------
  bool PollerBuiltIn::EnableReadNotification( Socket  *socket,
                                              bool     notify,
                                              time_t   timeout )
  {
    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;
  }
 
  //----------------------------------------------------------------------------
  // Notify the handler about write events
  //----------------------------------------------------------------------------
  bool PollerBuiltIn::EnableWriteNotification( Socket  *socket,
                                               bool     notify,
                                               time_t   timeout )
  {
    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;
  }
 
  //----------------------------------------------------------------------------
  // Check whether the socket is registered with the poller
  //----------------------------------------------------------------------------
  bool PollerBuiltIn::IsRegistered( Socket *socket )
  {
    XrdSysMutexHelper scopedLock( pMutex );
    SocketMap::iterator it = pSocketMap.find( socket );
    return it != pSocketMap.end();
  }
 
  //----------------------------------------------------------------------------
  // Return poller threads in round-robin fashion
  //----------------------------------------------------------------------------
  XrdSys::IOEvents::Poller* PollerBuiltIn::GetNextPoller()
  {
    if( pPollerPool.empty() ) return 0;
 
    PollerPool::iterator ret = pNext;
    ++pNext;
    if( pNext == pPollerPool.end() )
      pNext = pPollerPool.begin();
    return *ret;
  }
 
  //----------------------------------------------------------------------------
  // Return the poller associated with the respective channel
  //----------------------------------------------------------------------------
  XrdSys::IOEvents::Poller* PollerBuiltIn::RegisterAndGetPoller(const Socket * socket)
  {
    PollerMap::iterator itr = pPollerMap.find( socket->GetFD() );
 
    if( itr == pPollerMap.end() )
    {
      XrdSys::IOEvents::Poller* poller = GetNextPoller();
      if( poller )
        pPollerMap[socket->GetFD()] = poller;
      return poller;
    }
 
    return itr->second;
  }
 
  void PollerBuiltIn::UnregisterFromPoller( const Socket *socket )
  {
    PollerMap::iterator itr = pPollerMap.find( socket->GetFD() );
    if( itr == pPollerMap.end() ) return;
    pPollerMap.erase( itr );
  }
 
  XrdSys::IOEvents::Poller* PollerBuiltIn::GetPoller(const Socket * socket)
  {
    PollerMap::iterator itr = pPollerMap.find( socket->GetFD() );
    if( itr == pPollerMap.end() ) return 0;
    return itr->second;
  }
 
  //----------------------------------------------------------------------------
  // Get the initial value for pNbPoller
  //----------------------------------------------------------------------------
  int PollerBuiltIn::GetNbPollerInit()
  {
    Env * env = DefaultEnv::GetEnv();
    int ret = XrdCl::DefaultParallelEvtLoop;
    env->GetInt("ParallelEvtLoop", ret);
    return ret;
  }
}