XrdSys::IOEvents::Poller Class Reference

#include <XrdSysIOEvents.hh>

Inheritance diagram for XrdSys::IOEvents::Poller:

Collaboration diagram for XrdSys::IOEvents::Poller:

Classes
struct	PipeData

Public Types
enum	CreateOpts { optTOM }

Public Member Functions
	Poller (int cFD, int rFD)
virtual	~Poller ()
	Destructor. Stop() is effecively called when this object is deleted. More...
void	Stop ()

Static Public Member Functions
static Poller *	Create (int &eNum, const char **eTxt=0, int crOpts=0)

Protected Member Functions
virtual void	Begin (XrdSysSemaphore *syncp, int &rc, const char **eTxt)=0
void	CbkTMO ()
bool	CbkXeq (Channel *cP, int events, int eNum, const char *eTxt)
	CPP_ATOMIC_TYPE (bool) wakePend
virtual void	Exclude (Channel *cP, bool &isLocked, bool dover=1)=0
int	GetFault (Channel *cP)
int	GetPollEnt (Channel *cP)
int	GetRequest ()
virtual bool	Include (Channel *cP, int &eNum, const char **eTxt, bool &isLocked)=0
bool	Init (Channel *cP, int &eNum, const char **eTxt, bool &isLockd)
void	LockChannel (Channel *cP)
virtual bool	Modify (Channel *cP, int &eNum, const char **eTxt, bool &isLocked)=0
int	Poll2Enum (short events)
int	SendCmd (PipeData &cmd)
void	SetPollEnt (Channel *cP, int ptEnt)
virtual void	Shutdown ()=0
bool	TmoAdd (Channel *cP, int tmoSet)
void	TmoDel (Channel *cP)
int	TmoGet ()
void	UnLockChannel (Channel *cP)

Protected Attributes
Channel *	attBase
bool	chDead
int	cmdFD
int	pipeBlen
char *	pipeBuff
struct pollfd	pipePoll
pthread_t	pollTid
PipeData	reqBuff
int	reqFD
Channel *	tmoBase
unsigned char	tmoMask

Static Protected Attributes
static time_t	maxTime = (sizeof(time_t) == 8 ? 0x7fffffffffffffffLL : 0x7fffffff)
static pid_t	parentPID = getpid()

Friends
class	BootStrap
class	Channel

Detailed Description

Define a poller object interface. A poller fields and dispatches event callbacks. An actual instance of a poller object is obtained by using the Create() method. You cannot simply create an instance of this object using new or in-place declaration since it is abstract. Any number of these objects may created. Each creation spawns a polling thread.

Definition at line 371 of file XrdSysIOEvents.hh.

Member Enumeration Documentation

CreateOpts

enum XrdSys::IOEvents::Poller::CreateOpts

Create a specialized instance of a poller object, initialize it, and start the polling process. You must call Create() to obtain a specialized poller.

Parameters

eNum	Place where errno is placed upon failure.
eTxt	Place where a pointer to the description of the failing operation is to be set. If null, no description is returned.
crOpts	Poller options (see static const optxxx): optTOM - Timeout resumption after a timeout event must be manually reenabled. By default, event timeouts are automatically renabled after successful callbacks.

Returns

!0 Poller successfully created and started. eNum contains zero. eTxt if not null contains a null string. The returned value is a pointer to the Poller object. 0 Poller could not be created. eNum contains the associated errno value. eTxt if not null contains the failing operation.

Enumerator
optTOM

Definition at line 398 of file XrdSysIOEvents.hh.

{optTOM};

Constructor & Destructor Documentation

Poller()

XrdSys::IOEvents::Poller::Poller	(	int	cFD,
		int	rFD
	)

Constructor

Parameters

cFD	The file descriptor to send commands to the poll thread.
rFD	The file descriptor to recv commands in the poll thread.

Definition at line 572 of file XrdSysIOEvents.cc.

{
 
// Now initialize local class members
//
   attBase         = 0;
   tmoBase         = 0;
   cmdFD           = cFD;
   reqFD           = rFD;
   wakePend        = false;
   pipeBuff        = 0;
   pipeBlen        = 0;
   pipePoll.fd     = rFD;
   pipePoll.events = POLLIN | POLLRDNORM;
   tmoMask         = 255;
}

~Poller()

virtual XrdSys::IOEvents::Poller::~Poller ( )

inlinevirtual

Destructor. Stop() is effecively called when this object is deleted.

Definition at line 430 of file XrdSysIOEvents.hh.

{}

Member Function Documentation

Begin()

virtual void XrdSys::IOEvents::Poller::Begin ( XrdSysSemaphore *  syncp, int &  rc, const char **  eTxt  )

protectedpure virtual

Start the polling event loop. An implementation must be supplied. Begin() is called via the internal BootStrap class from a new thread.

Implemented in XrdSys::IOEvents::PollerWait, XrdSys::IOEvents::PollerInit, XrdSys::IOEvents::PollerErr1, XrdSys::IOEvents::PollPort, XrdSys::IOEvents::PollPoll, XrdSys::IOEvents::PollKQ, and XrdSys::IOEvents::PollE.

Referenced by XrdSys::IOEvents::BootStrap::Start().

Here is the caller graph for this function:

CbkTMO()

void XrdSys::IOEvents::Poller::CbkTMO ( )

protected

Definition at line 615 of file XrdSysIOEvents.cc.

{
   Channel *cP;
 
// Process each element in the timeout queue, calling the callback function
// if the timeout has passed. As this method can be called with a lock on the
// channel mutex, we need to drop it prior to calling the callback.
//
   toMutex.Lock();
   while((cP = tmoBase) && cP->deadLine <= time(0))
        {int dlType = cP->dlType;
         toMutex.UnLock();
         CbkXeq(cP, dlType, 0, 0);
         toMutex.Lock();
        }
   toMutex.UnLock();
}

CbkXeq()

bool XrdSys::IOEvents::Poller::CbkXeq ( Channel *  cP, int  events, int  eNum, const char *  eTxt  )

protected

Definition at line 637 of file XrdSysIOEvents.cc.

{
   XrdSysMutexHelper cbkMHelp(cP->chMutex);
   char oldEvents;
   bool cbok, retval, isRead, isWrite, isLocked = true;
 
// Perform any required tracing
//
   if (TRACING)
      {const char *cbtype = (cP->chPoller == cP->chPollXQ ? "norm" :
                            (cP->chPoller == &pollInit    ? "init" :
                            (cP->chPoller == &pollWait    ? "wait" : "err")));
       DO_TRACE(CbkXeq,cP->chFD,"callback events=" <<events
            <<" chev=" <<static_cast<int>(cP->chEvents)
            <<" toq=" <<(cP->inTOQ != 0) <<" erc=" <<eNum
            <<" callback " <<(cP->chCB ? "present" : "missing")
            <<" poller=" <<cbtype);
      }
 
// Remove this from the timeout queue if there and reset the deadlines based
// on the event we are reflecting. This separates read and write deadlines
//
   if (cP->inTOQ)
      {TmoDel(cP);
       cP->dlType |= (events & CallBack::ValidEvents) << 4;
       isRead = events & (CallBack::ReadyToRead | CallBack:: ReadTimeOut);
       if (isRead)  cP->rdDL = maxTime;
       isWrite= events & (CallBack::ReadyToWrite | CallBack::WriteTimeOut);
       if (isWrite) cP->wrDL = maxTime;
      } else {
       cP->dlType &= CallBack::ValidEvents;
       isRead = isWrite = false;
      }
 
// Verify that there is a callback here and the channel is ready. If not,
// disable this channel for the events being refelcted unless the event is a
// fatal error. In this case we need to abandon the channel since error events
// may continue to be generated as we can't always disable them.
//
   if (!(cP->chCB) || cP->chPoller != cP->chPollXQ)
      {if (eNum)
          {cP->chPoller = &pollErr1; cP->chFault = eNum;
           cP->inPSet   = 0;
           return false;
          }
       oldEvents = cP->chEvents;
       cP->chEvents = 0;
       retval = cP->chPoller->Modify(cP, eNum, 0, isLocked);
       TRACE_MOD(CbkXeq,cP->chFD,0);
       if (!isLocked) cP->chMutex.Lock();
       cP->chEvents = oldEvents;
       return true;
      }
 
// Resolve the problem where we get an error event but the channel wants them
// presented as a read or write event. If neither is possible then defer the
// error until the channel is enabled again.
//
   if (eNum)
      {if (cP->chEvents & Channel::errorEvents)
          {cP->chPoller = &pollErr1; cP->chFault = eNum;
           cP->chStat   = Channel::isCBMode;
           chDead       = false;
           cbkMHelp.UnLock();
           cP->chCB->Fatal(cP,cP->chCBA, eNum, eTxt);
           if (chDead) return true;
           cbkMHelp.Lock(&(cP->chMutex));
           cP->inPSet   = 0;
           return false;
          }
            if (REVENTS(cP->chEvents)) events = CallBack::ReadyToRead;
       else if (WEVENTS(cP->chEvents)) events = CallBack::ReadyToWrite;
       else    {cP->chPoller = &pollErr1; cP->chFault = eNum; cP->inPSet = 0;
                return false;
               }
      }
 
// Indicate that we are in callback mode then drop the channel lock and effect
// the callback. This allows the callback to freely manage locks.
//
   cP->chStat = Channel::isCBMode;
   chDead     = false;
   // Detach() may be called after unlocking the channel and would zero the
   // callback pointer and argument. So keep a copy.
   CallBack *cb = cP->chCB;
   void *cba = cP->chCBA;
   cbkMHelp.UnLock();
   IF_TRACE(CbkXeq,cP->chFD,"invoking callback; events=" <<events);
   cbok = cb->Event(cP,cba, events);
   IF_TRACE(CbkXeq,cP->chFD,"callback returned " <<BOOLNAME(cbok));
 
// If channel destroyed by the callback, bail really fast. Otherwise, regain
// the channel lock.
//
   if (chDead) return true;
   cbkMHelp.Lock(&(cP->chMutex));
 
// If the channel is being destroyed; then another thread must have done so.
// Tell it the callback has finished and just return.
//
   if (cP->chStat != Channel::isCBMode)
      {if (cP->chStat == Channel::isDead)
          {XrdSysSemaphore *theSem = (XrdSysSemaphore *)cP->chCBA;
           // channel will be destroyed shortly after post, unlock mutex before
           cbkMHelp.UnLock();
           theSem->Post();
          }
       return true;
      }
   cP->chStat = Channel::isClear;
 
// Handle enable or disable here. If we keep the channel enabled then reset
// the timeout if it hasn't been handled via a call from the callback.
//
        if (!cbok) Detach(cP,isLocked,false);
   else if ((isRead || isWrite) && !(cP->inTOQ) && (cP->chRTO || cP->chWTO))
           TmoAdd(cP, 0);
 
// All done. While the mutex should not have been unlocked, we relock it if
// it has to keep the mutex helper from croaking.
//
   if (!isLocked) cP->chMutex.Lock();
   return true;
}

References BOOLNAME, DO_TRACE, XrdSys::IOEvents::Channel::errorEvents, XrdSys::IOEvents::CallBack::Event(), XrdSys::IOEvents::CallBack::Fatal(), IF_TRACE, XrdSysMutex::Lock(), XrdSysMutexHelper::Lock(), Modify(), XrdSys::IOEvents::pollErr1, XrdSys::IOEvents::pollInit, XrdSys::IOEvents::pollWait, XrdSysSemaphore::Post(), XrdSys::IOEvents::CallBack::ReadTimeOut, XrdSys::IOEvents::CallBack::ReadyToRead, XrdSys::IOEvents::CallBack::ReadyToWrite, REVENTS, TRACE_MOD, TRACING, XrdSysMutexHelper::UnLock(), XrdSys::IOEvents::CallBack::ValidEvents, WEVENTS, and XrdSys::IOEvents::CallBack::WriteTimeOut.

Here is the call graph for this function:

CPP_ATOMIC_TYPE()

XrdSys::IOEvents::Poller::CPP_ATOMIC_TYPE ( bool  )

protected

Create()

XrdSys::IOEvents::Poller * XrdSys::IOEvents::Poller::Create ( int &  eNum, const char **  eTxt = 0, int  crOpts = 0  )

static

Definition at line 767 of file XrdSysIOEvents.cc.

{
   int fildes[2];
   struct pollArg pArg;
   pthread_t tid;
 
// Create a pipe used to break the poll wait loop
//
   if (XrdSysFD_Pipe(fildes))
      {eNum = errno;
       if (eTxt) *eTxt = "creating poll pipe";
       return 0;
      }
 
// Create an actual implementation of a poller
//
   if (!(pArg.pollP = newPoller(fildes, eNum, eTxt)))
      {close(fildes[0]);
       close(fildes[1]);
       return 0;
      }
 
// Now start a thread to handle this poller object
//
   if ((eNum = XrdSysThread::Run(&tid, XrdSys::IOEvents::BootStrap::Start,
                                 (void *)&pArg, XRDSYSTHREAD_BIND, "Poller")))
      {if (eTxt) *eTxt = "creating poller thread"; return 0;}
 
// Now wait for the thread to finish initializing before we allow use
// Note that the bootstrap takes ownership of the semaphore and will delete it
// once the thread positing the semaphore actually ends. This is to avoid
// semaphore bugs present in certain (e.g. Linux) kernels.
//
   pArg.pollSync->Wait();
 
// Check if all went well
//
   if (pArg.retCode)
      {if (eTxt) *eTxt = (pArg.retMsg ? pArg.retMsg : "starting poller");
       eNum = pArg.retCode;
       delete pArg.pollP;
       return 0;
      }
 
// Set creation options in the new poller
//
   if (crOpts & optTOM)
      pArg.pollP->tmoMask = ~(CallBack::ReadTimeOut|CallBack::WriteTimeOut);
 
// All done
//
   eNum = 0;
   if (eTxt) *eTxt = "";
   return pArg.pollP;
}

References close, XrdSys::IOEvents::pollArg::pollP, XrdSys::IOEvents::pollArg::pollSync, XrdSys::IOEvents::CallBack::ReadTimeOut, XrdSys::IOEvents::pollArg::retCode, XrdSys::IOEvents::pollArg::retMsg, XrdSysThread::Run(), XrdSys::IOEvents::BootStrap::Start(), tmoMask, XrdSysSemaphore::Wait(), XrdSys::IOEvents::CallBack::WriteTimeOut, and XRDSYSTHREAD_BIND.

Here is the call graph for this function:

Exclude()

virtual void XrdSys::IOEvents::Poller::Exclude ( Channel *  cP, bool &  isLocked, bool  dover = 1  )

protectedpure virtual

Remove a channel to the poll set. An implementation must be supplied. The channel is locked when this method is called but must be unlocked by the method if a command is sent to the poller thread and isLocked set to false.

Implemented in XrdSys::IOEvents::PollPort, XrdSys::IOEvents::PollPoll, XrdSys::IOEvents::PollKQ, XrdSys::IOEvents::PollE, XrdSys::IOEvents::PollerWait, XrdSys::IOEvents::PollerInit, and XrdSys::IOEvents::PollerErr1.

GetFault()

int XrdSys::IOEvents::Poller::GetFault ( Channel *  cP )

inlineprotected

Definition at line 437 of file XrdSysIOEvents.hh.

{return cP->chFault;}

Referenced by XrdSys::IOEvents::PollerErr1::Include(), and XrdSys::IOEvents::PollerErr1::Modify().

Here is the caller graph for this function:

GetPollEnt()

int XrdSys::IOEvents::Poller::GetPollEnt ( Channel *  cP )

inlineprotected

Definition at line 438 of file XrdSysIOEvents.hh.

{return cP->pollEnt;}

GetRequest()

int XrdSys::IOEvents::Poller::GetRequest ( )

protected

Definition at line 875 of file XrdSysIOEvents.cc.

{
   ssize_t rlen;
   int rc;
 
// See if we are to resume a read or start a fresh one
//
   if (!pipeBlen) 
      {pipeBuff = (char *)&reqBuff; pipeBlen = sizeof(reqBuff);}
 
// Wait for the next request. Some OS's (like Linux) don't support non-blocking
// pipes. So, we must front the read with a poll.
//
   do {rc = poll(&pipePoll, 1, 0);}
      while(rc < 0 && (errno == EAGAIN || errno == EINTR));
   if (rc < 1) return 0;
 
// Now we can put up a read without a delay. Normally a full command will be
// present. Under some heavy conditions, this may not be the case.
//
   do {rlen = read(reqFD, pipeBuff, pipeBlen);} 
      while(rlen < 0 && errno == EINTR);
   if (rlen <= 0)
      {std::cerr <<"Poll: "<<XrdSysE2T(errno)<<" reading from request pipe\n"<< std::flush;
       return 0;
      }
 
// Check if all the data has arrived. If not all the data is present, defer
// this request until more data arrives.
//
   if (!(pipeBlen -= rlen)) return 1;
   pipeBuff += rlen;
   return 0;
}

References read(), and XrdSysE2T().

Here is the call graph for this function:

Include()

virtual bool XrdSys::IOEvents::Poller::Include ( Channel *  cP, int &  eNum, const char **  eTxt, bool &  isLocked  )

protectedpure virtual

Add a channel to the poll set. An implementation must be supplied. The channel is locked when this method is called but must be unlocked by the method if a command is sent to the poller thread and isLocked set to false.

Implemented in XrdSys::IOEvents::PollPort, XrdSys::IOEvents::PollPoll, XrdSys::IOEvents::PollKQ, XrdSys::IOEvents::PollE, XrdSys::IOEvents::PollerWait, XrdSys::IOEvents::PollerInit, and XrdSys::IOEvents::PollerErr1.

Referenced by Init().

Here is the caller graph for this function:

Init()

bool XrdSys::IOEvents::Poller::Init ( Channel *  cP, int &  eNum, const char **  eTxt, bool &  isLockd  )

protected

Definition at line 914 of file XrdSysIOEvents.cc.

{
// The channel must be locked upon entry!
//
   bool retval;
 
 
// If we are already in progress then simply update the shadow events and
// resuppress all current events.
//
   if (cP->chPoller == &pollWait)
      {cP->reMod    = cP->chEvents;
       cP->chEvents = 0;
       IF_TRACE(Init,cP->chFD,"defer events=" <<cP->reMod);
       return true;
      }
 
// Trace this entry
//
   IF_TRACE(Init,cP->chFD,"begin events=" <<int(cP->chEvents));
 
// If no events are enabled at this point, just return
//
   if (!(cP->chEvents)) return true;
 
// Refuse to enable a channel without a callback function
//
   if (!(cP->chCB))
      {eNum = EDESTADDRREQ;
       if (eTxt) *eTxt = "enabling without a callback";
       return false;
      }
 
// So, now we can include the channel in the poll set. We will include it
// with no events enabled to prevent callbacks prior to completion here.
//
   cP->chPoller = &pollWait; cP->reMod = cP->chEvents; cP->chEvents = 0;
   retval = cP->chPollXQ->Include(cP, eNum, eTxt, isLocked);
   IF_TRACE(Init,cP->chFD,"Include() returned " <<BOOLNAME(retval) <<TRACE_LOK);
   if (!isLocked) {cP->chMutex.Lock(); isLocked = true;}
 
// Determine what future poller to use. If we can use the regular poller then
// set the correct event mask for the channel. Note that we could have lost
// control but the correct events will be reflected in the "reMod" member.
//
   if (!retval) {cP->chPoller = &pollErr1; cP->chFault = eNum;}
      else {cP->chPoller = cP->chPollXQ;
            cP->inPSet   = 1;
            if (cP->reMod)
               {cP->chEvents =  cP->reMod;
                retval = cP->chPoller->Modify(cP, eNum, eTxt, isLocked);
                TRACE_MOD(Init,cP->chFD,int(cP->reMod));
                if (!isLocked) {cP->chMutex.Lock(); isLocked = true;}
               } else {
                TRACE_NOD(Init,cP->chFD,0);
               }
           }
 
// All done
//
   cP->reMod = 0;
   return retval;
}

References BOOLNAME, IF_TRACE, Include(), XrdSysMutex::Lock(), Modify(), XrdSys::IOEvents::pollErr1, XrdSys::IOEvents::pollWait, TRACE_LOK, TRACE_MOD, and TRACE_NOD.

Referenced by XrdSys::IOEvents::PollerInit::Modify(), and XrdSys::IOEvents::PollerWait::Modify().

Here is the call graph for this function:

Here is the caller graph for this function:

LockChannel()

void XrdSys::IOEvents::Poller::LockChannel ( Channel *  cP )

inlineprotected

Definition at line 441 of file XrdSysIOEvents.hh.

{cP->chMutex.Lock();}

References XrdSysMutex::Lock().

Here is the call graph for this function:

Modify()

virtual bool XrdSys::IOEvents::Poller::Modify ( Channel *  cP, int &  eNum, const char **  eTxt, bool &  isLocked  )

protectedpure virtual

Modify the event status of a channel. An implementation must be supplied. The channel is locked when this method is called but must be unlocked by the method if a command is sent to the poller thread and isLocked set to false.

Implemented in XrdSys::IOEvents::PollPort, XrdSys::IOEvents::PollPoll, XrdSys::IOEvents::PollKQ, XrdSys::IOEvents::PollE, XrdSys::IOEvents::PollerWait, XrdSys::IOEvents::PollerInit, and XrdSys::IOEvents::PollerErr1.

Referenced by CbkXeq(), and Init().

Here is the caller graph for this function:

Poll2Enum()

int XrdSys::IOEvents::Poller::Poll2Enum ( short  events )

protected

Definition at line 983 of file XrdSysIOEvents.cc.

{
   if (events & POLLERR)  return EPIPE;
 
   if (events & POLLHUP)  return ECONNRESET;
 
   if (events & POLLNVAL) return EBADF;
 
   return EOPNOTSUPP;
}

SendCmd()

int XrdSys::IOEvents::Poller::SendCmd ( PipeData &  cmd )

protected

Definition at line 998 of file XrdSysIOEvents.cc.

{
   int wlen;
 
// Pipe writes are atomic so we don't need locks. Some commands require
// confirmation. We handle that here based on the command. Note that pipes
// gaurantee that all of the data will be written or we will block.
//
   if (cmd.req >= PipeData::Post)
      {XrdSysSemaphore mySem(0);
       cmd.theSem = &mySem;
       do {wlen = write(cmdFD, (char *)&cmd, sizeof(PipeData));}
          while (wlen < 0 && errno == EINTR);
       if (wlen > 0) mySem.Wait();
      } else {
       do {wlen = write(cmdFD, (char *)&cmd, sizeof(PipeData));}
          while (wlen < 0 && errno == EINTR);
      }
 
// All done
//
   return (wlen >= 0 ? 0 : errno);
}

References XrdSys::IOEvents::Poller::PipeData::req, XrdSys::IOEvents::Poller::PipeData::theSem, XrdSysSemaphore::Wait(), and write.

Here is the call graph for this function:

SetPollEnt()

void XrdSys::IOEvents::Poller::SetPollEnt ( Channel *  cP, int  ptEnt  )

protected

Definition at line 1026 of file XrdSysIOEvents.cc.

{
   cP->pollEnt = pe;
}

Shutdown()

virtual void XrdSys::IOEvents::Poller::Shutdown ( )

protectedpure virtual

Shutdown the poller. An implementation must be supplied. The shutdown method must release any allocated storage and close private file descriptors. The polling thread will have already been terminated and x-thread pipe closed. Warning: the derived destructor must call Stop() and do nothing else!

Implemented in XrdSys::IOEvents::PollPort, XrdSys::IOEvents::PollPoll, XrdSys::IOEvents::PollKQ, XrdSys::IOEvents::PollE, XrdSys::IOEvents::PollerWait, XrdSys::IOEvents::PollerInit, and XrdSys::IOEvents::PollerErr1.

Stop()

void XrdSys::IOEvents::Poller::Stop

(

)

Stop a poller object. Active callbacks are completed. Pending callbacks are discarded. After which the poller event thread exits. Subsequently, each associated channel is disabled and removed from the poller object. If the channel is enabled for a StopEvent, the stop callback is invoked. However, any attempt to use the channel methods that require an active poller will return an error.

Since a stopped poller cannot be restarted; the only thing left is to delete it. This also applies to all the associated channels since they no longer have an active poller.

Definition at line 1035 of file XrdSysIOEvents.cc.

{
   PipeData  cmdbuff;
   CallBack *theCB;
   Channel  *cP;
   void     *cbArg;
   int       doCB;
 
// Initialize the pipdata structure
//
   memset(static_cast<void*>( &cmdbuff ), 0, sizeof(cmdbuff));
   cmdbuff.req = PipeData::Stop;
 
// Lock all of this
//
   adMutex.Lock();
 
// If we are already shutdown then we are done
//
   if (cmdFD == -1) {adMutex.UnLock(); return;}
 
// First we must stop the poller thread in an orderly fashion.
//
   adMutex.UnLock();
   SendCmd(cmdbuff);
   adMutex.Lock();
 
// Close the pipe communication mechanism
//
   close(cmdFD); cmdFD = -1;
   close(reqFD); reqFD = -1;
 
// Run through cleaning up the channels. While there should not be any other
// operations happening on this poller, we take the conservative approach.
//
   while((cP = attBase))
        {REMOVE(attBase, attList, cP);
         adMutex.UnLock();
         cP->chMutex.Lock();
         doCB = cP->chCB != 0 && (cP->chEvents & Channel::stopEvent);
         if (cP->inTOQ) TmoDel(cP);
         cP->Reset(&pollErr1, cP->chFD, EIDRM);
         cP->chPollXQ = &pollErr1;
         if (doCB)
            {cP->chStat = Channel::isClear;
             theCB = cP->chCB; cbArg = cP->chCBA;
             cP->chMutex.UnLock();
             theCB->Stop(cP, cbArg);
            } else cP->chMutex.UnLock();
         adMutex.Lock();
        }
 
// Now invoke the poller specific shutdown
//
   Shutdown();
   adMutex.UnLock();
}

References close, XrdSysMutex::Lock(), XrdSys::IOEvents::pollErr1, REMOVE, XrdSys::IOEvents::Poller::PipeData::req, XrdSys::IOEvents::CallBack::Stop(), XrdSys::IOEvents::Channel::stopEvent, and XrdSysMutex::UnLock().

Referenced by XrdSys::IOEvents::PollE::~PollE(), XrdSys::IOEvents::PollKQ::~PollKQ(), XrdSys::IOEvents::PollPoll::~PollPoll(), XrdSys::IOEvents::PollPort::~PollPort(), and XrdCl::PollerBuiltIn::Stop().

Here is the call graph for this function:

Here is the caller graph for this function:

TmoAdd()

bool XrdSys::IOEvents::Poller::TmoAdd ( Channel *  cP, int  tmoSet  )

protected

Definition at line 1097 of file XrdSysIOEvents.cc.

{
   XrdSysMutexHelper mHelper(toMutex);
   time_t tNow;
   Channel *ncP;
   bool setRTO, setWTO;
 
// Do some tracing
//
   IF_TRACE(TmoAdd,cP->chFD,"chan="<< std::hex<<(void*)cP<< std::dec
            <<" inTOQ="<<BOOLNAME(cP->inTOQ)<<" status="<<STATUSOF(cP));
 
// Remove element from timeout queue if it is there
//
   if (cP->inTOQ)
      {REMOVE(tmoBase, tmoList, cP);
       cP->inTOQ = 0;
      }
 
// Determine which timeouts need to be reset
//
   tmoSet|= cP->dlType >> 4;
   setRTO = (tmoSet&tmoMask) & (CallBack::ReadyToRead |CallBack:: ReadTimeOut);
   setWTO = (tmoSet&tmoMask) & (CallBack::ReadyToWrite|CallBack::WriteTimeOut);
 
// Reset the required deadlines
//
   tNow = time(0);
   if (setRTO && REVENTS(cP->chEvents) && cP->chRTO)
      cP->rdDL = cP->chRTO + tNow;
   if (setWTO && WEVENTS(cP->chEvents) && cP->chWTO)
      cP->wrDL = cP->chWTO + tNow;
 
// Calculate the closest enabled deadline
//
   if (cP->rdDL < cP->wrDL)
      {cP->deadLine  = cP->rdDL; cP->dlType  = CallBack:: ReadTimeOut;
      } else {
       cP->deadLine  = cP->wrDL; cP->dlType  = CallBack::WriteTimeOut;
       if (cP->rdDL == cP->wrDL) cP->dlType |= CallBack:: ReadTimeOut;
      }
   IF_TRACE(TmoAdd, cP->chFD, "t=" <<tNow <<" rdDL=" <<setRTO <<' ' <<cP->rdDL
                                          <<" wrDL=" <<setWTO <<' ' <<cP->wrDL);
 
// If no timeout really applies, we are done
//
   if (cP->deadLine == maxTime) return false;
 
// Add the channel to the timeout queue in correct deadline position.
//
   if ((ncP = tmoBase))
      {do {if (cP->deadLine < ncP->deadLine) break;
           ncP = ncP->tmoList.next;
          } while(ncP != tmoBase);
       INSERT(tmoList, ncP, cP);
       if (cP->deadLine < tmoBase->deadLine) tmoBase = cP;
      } else tmoBase = cP;
   cP->inTOQ = 1;
 
// Indicate to the caller whether or not a wakeup is required
//
   return (tmoBase == cP);
}

References BOOLNAME, Xrd::dec, Xrd::hex, IF_TRACE, INSERT, XrdSys::IOEvents::CallBack::ReadTimeOut, XrdSys::IOEvents::CallBack::ReadyToRead, XrdSys::IOEvents::CallBack::ReadyToWrite, REMOVE, REVENTS, STATUSOF, WEVENTS, and XrdSys::IOEvents::CallBack::WriteTimeOut.

TmoDel()

void XrdSys::IOEvents::Poller::TmoDel ( Channel *  cP )

protected

Definition at line 1165 of file XrdSysIOEvents.cc.

{
 
// Do some tracing
//
   IF_TRACE(TmoDel,cP->chFD,"chan="<< std::hex<<(void*)cP<< std::dec
            <<" inTOQ="<<BOOLNAME(cP->inTOQ)<<" status="<<STATUSOF(cP));
 
// Get the timeout queue lock and remove the channel from the queue
//
   toMutex.Lock();
   REMOVE(tmoBase, tmoList, cP);
   cP->inTOQ = 0;
   toMutex.UnLock();
}

References BOOLNAME, Xrd::dec, Xrd::hex, IF_TRACE, REMOVE, and STATUSOF.

TmoGet()

int XrdSys::IOEvents::Poller::TmoGet ( )

protected

Definition at line 1185 of file XrdSysIOEvents.cc.

{
   int wtval;
 
// Lock the timeout queue
//
   toMutex.Lock();
 
// Calculate wait time. If the deadline passed, invoke the timeout callback.
// we will need to drop the timeout lock as we don't have the channel lock.
//
   do {if (!tmoBase) {wtval = -1; break;}
       wtval = (tmoBase->deadLine - time(0)) * 1000;
       if (wtval > 0) break;
       toMutex.UnLock();
       CbkTMO();
       toMutex.Lock();
      } while(1);
 
// Return the value
//
   CPP_ATOMIC_STORE(wakePend, false, std::memory_order_release);
   toMutex.UnLock();
   return wtval;
}

References CPP_ATOMIC_STORE.

Referenced by XrdSys::IOEvents::PollPort::BegTO().

Here is the caller graph for this function:

UnLockChannel()

void XrdSys::IOEvents::Poller::UnLockChannel ( Channel *  cP )

inlineprotected

Definition at line 448 of file XrdSysIOEvents.hh.

{cP->chMutex.UnLock();}

References XrdSysMutex::UnLock().

Here is the call graph for this function:

Friends And Related Function Documentation

BootStrap

friend class BootStrap

friend

Definition at line 373 of file XrdSysIOEvents.hh.

Channel

friend class Channel

friend

Definition at line 374 of file XrdSysIOEvents.hh.

Member Data Documentation

attBase

Channel* XrdSys::IOEvents::Poller::attBase

protected

Definition at line 488 of file XrdSysIOEvents.hh.

chDead

bool XrdSys::IOEvents::Poller::chDead

protected

Definition at line 511 of file XrdSysIOEvents.hh.

Referenced by XrdSys::IOEvents::Channel::Delete().

cmdFD

int XrdSys::IOEvents::Poller::cmdFD

protected

Definition at line 494 of file XrdSysIOEvents.hh.

maxTime

time_t XrdSys::IOEvents::Poller::maxTime = (sizeof(time_t) == 8 ? 0x7fffffffffffffffLL : 0x7fffffff)

staticprotected

Definition at line 513 of file XrdSysIOEvents.hh.

Referenced by XrdSys::IOEvents::Channel::Enable().

parentPID

pid_t XrdSys::IOEvents::Poller::parentPID = getpid()

staticprotected

Definition at line 515 of file XrdSysIOEvents.hh.

pipeBlen

int XrdSys::IOEvents::Poller::pipeBlen

protected

Definition at line 508 of file XrdSysIOEvents.hh.

pipeBuff

char* XrdSys::IOEvents::Poller::pipeBuff

protected

Definition at line 507 of file XrdSysIOEvents.hh.

pipePoll

struct pollfd XrdSys::IOEvents::Poller::pipePoll

protected

Definition at line 491 of file XrdSysIOEvents.hh.

pollTid

pthread_t XrdSys::IOEvents::Poller::pollTid

protected

Definition at line 491 of file XrdSysIOEvents.hh.

Referenced by XrdSys::IOEvents::Channel::Delete(), and XrdSys::IOEvents::BootStrap::Start().

reqBuff

PipeData XrdSys::IOEvents::Poller::reqBuff

protected

Definition at line 506 of file XrdSysIOEvents.hh.

reqFD

int XrdSys::IOEvents::Poller::reqFD

protected

Definition at line 495 of file XrdSysIOEvents.hh.

Referenced by XrdSys::IOEvents::PollKQ::PollKQ().

tmoBase

Channel* XrdSys::IOEvents::Poller::tmoBase

protected

Definition at line 489 of file XrdSysIOEvents.hh.

tmoMask

unsigned char XrdSys::IOEvents::Poller::tmoMask

protected

Definition at line 509 of file XrdSysIOEvents.hh.

Referenced by Create().

---

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

• XrdSysIOEvents.hh
• XrdSysIOEvents.cc
• XrdSysIOEventsPollE.icc
• XrdSysIOEventsPollKQ.icc
• XrdSysIOEventsPollPoll.icc
• XrdSysIOEventsPollPort.icc