Loading...
Searching...
No Matches
XrdSysIOEvents.cc
Go to the documentation of this file.
/******************************************************************************/
/* */
/* X r d S y s I O E v e n t s . c c */
/* */
/* (c) 2012 by the Board of Trustees of the Leland Stanford, Jr., University */
/* All Rights Reserved */
/* Produced by Andrew Hanushevsky for Stanford University under contract */
/* DE-AC02-76-SFO0515 with the Department of Energy */
/* */
/* 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 Lesser General Public */
/* License for more details. */
/* */
/* You should have received a copy of the GNU Lesser General Public License */
/* along with XRootD in a file called COPYING.LESSER (LGPL license) and file */
/* COPYING (GPL license). If not, see <http://www.gnu.org/licenses/>. */
/* */
/* The copyright holder's institutional names and contributor's names may not */
/* be used to endorse or promote products derived from this software without */
/* specific prior written permission of the institution or contributor. */
/******************************************************************************/
#include <cstdio>
#include <cstdlib>
#include "XrdSys/XrdSysE2T.hh"
#include "XrdSys/XrdSysFD.hh"
#include "XrdSys/XrdSysIOEvents.hh"
#include "XrdSys/XrdSysHeaders.hh"
#include "XrdSys/XrdSysPlatform.hh"
#include "XrdSys/XrdSysPthread.hh"
/******************************************************************************/
/* L o c a l D a t a */
/******************************************************************************/
namespace
{
// Status code to name array corresponding to:
// enum Status {isClear = 0, isCBMode, isDead};
//
const char *statName[] = {"isClear", "isCBMode", "isDead"};
}
/******************************************************************************/
/* L o c a l D e f i n e s */
/******************************************************************************/
#define STATUS statName[(int)chStat]
#define STATUSOF(x) statName[(int)(x->chStat)]
#define SINGLETON(dlvar, theitem)\
theitem ->dlvar .next == theitem
#define INSERT(dlvar, curitem, newitem) \
newitem ->dlvar .next = curitem; \
newitem ->dlvar .prev = curitem ->dlvar .prev; \
curitem ->dlvar .prev-> dlvar .next = newitem; \
curitem ->dlvar .prev = newitem
#define REMOVE(dlbase, dlvar, curitem) \
if (dlbase == curitem) dlbase = (SINGLETON(dlvar,curitem) \
? 0 : curitem ->dlvar .next);\
curitem ->dlvar .prev-> dlvar .next = curitem ->dlvar .next;\
curitem ->dlvar .next-> dlvar .prev = curitem ->dlvar .prev;\
curitem ->dlvar .next = curitem;\
curitem ->dlvar .prev = curitem
#define REVENTS(x) x & Channel:: readEvents
#define WEVENTS(x) x & Channel::writeEvents
#define ISPOLLER XrdSysThread::Same(XrdSysThread::ID(),pollTid)
#define BOOLNAME(x) (x ? "true" : "false")
#define DO_TRACE(x,fd,y) \
{PollerInit::traceMTX.Lock(); \
std::cerr <<"IOE fd "<<fd<<' '<<#x <<": "<<y<<'\n'<< std::flush; \
PollerInit::traceMTX.UnLock();}
#define TRACING PollerInit::doTrace
#define IF_TRACE(x,fd,y) if (TRACING) DO_TRACE(x,fd,y)
#define TRACE_LOK " channel now " <<(isLocked ? "locked" : "unlocked")
#define TRACE_MOD(x,fd,y) \
IF_TRACE(x,fd,"Modify(" <<y <<") == " \
<<BOOLNAME(retval) <<TRACE_LOK)
#define TRACE_NOD(x,fd,y) \
IF_TRACE(x,fd,"Modify(" <<y <<") skipped; no events changed")
/******************************************************************************/
/* G l o b a l D a t a */
/******************************************************************************/
time_t XrdSys::IOEvents::Poller::maxTime
= (sizeof(time_t) == 8 ? 0x7fffffffffffffffLL : 0x7fffffff);
pid_t XrdSys::IOEvents::Poller::parentPID = getpid();
/******************************************************************************/
/* L o c a l C l a s s e s */
/******************************************************************************/
/******************************************************************************/
/* T h r e a d S t a r t u p I n t e r f a c e */
/******************************************************************************/
namespace XrdSys
{
namespace IOEvents
{
struct pollArg
{Poller *pollP;
const char *retMsg;
int retCode;
XrdSysSemaphore *pollSync;
pollArg() : retMsg(0), retCode(0)
{pollSync = new XrdSysSemaphore(0, "poll sync");}
~pollArg() {}
};
class BootStrap
{
public:
static void *Start(void *parg);
};
void *BootStrap::Start(void *parg)
{
struct pollArg *pollArg = (struct pollArg *)parg;
Poller *thePoller = pollArg->pollP;
XrdSysSemaphore *theSem = pollArg->pollSync;
thePoller->pollTid = XrdSysThread::ID();
thePoller->Begin(theSem, pollArg->retCode, &(pollArg->retMsg));
delete theSem;
return (void *)0;
}
/******************************************************************************/
/* P o l l e r E r r 1 */
/******************************************************************************/
// This class is set in the channel when an error occurs but cannot be reflected
// immediately because either there is no callback function or all events are
// disabled. We need to do this because error events can be physically presented
// by the kernel even when logical events are disabled. Note that the error
// number and text will have been set and remain set as the channel was actually
// disabled preventing any new operation on the channel.
//
class PollerErr1 : public Poller
{
public:
PollerErr1() : Poller(-1, -1) {}
~PollerErr1() {}
protected:
void Begin(XrdSysSemaphore *syncp, int &rc, const char **eTxt)
{(void)syncp; (void)rc; (void)eTxt;}
void Exclude(Channel *cP, bool &isLocked, bool dover=1)
{(void)cP; (void)isLocked; (void)dover;}
bool Include(Channel *cP, int &eNum, const char **eTxt, bool &isLocked)
{(void)isLocked;
if (!(eNum = GetFault(cP))) eNum = EPROTO;
if (eTxt) *eTxt = "initializing channel";
return false;
}
bool Modify (Channel *cP, int &eNum, const char **eTxt, bool &isLocked)
{(void)isLocked;
if (!(eNum = GetFault(cP))) eNum = EPROTO;
if (eTxt) *eTxt = "modifying channel";
return false;
}
void Shutdown() {}
};
/******************************************************************************/
/* P o l l e r I n i t */
/******************************************************************************/
// This class is used as the initial poller on a channel. It is responsible
// for adding the file descriptor to the poll set upon the initial enable. This
// avoids enabling a channel prior to it receiving a call back function.
//
class PollerInit : public Poller
{
public:
PollerInit() : Poller(-1, -1) {}
~PollerInit() {}
static XrdSysMutex traceMTX;
static bool doTrace;
protected:
void Begin(XrdSysSemaphore *syncp, int &rc, const char **eTxt) {}
void Exclude(Channel *cP, bool &isLocked, bool dover=1) {}
bool Include(Channel *cP, int &eNum, const char **eTxt, bool &isLocked)
{eNum = EPROTO;
if (eTxt) *eTxt = "initializing channel";
return false;
}
bool Modify (Channel *cP, int &eNum, const char **eTxt, bool &isLocked)
{bool rc = Init(cP, eNum, eTxt, isLocked);
IF_TRACE(Modify,cP->GetFD(), "Init() returned " <<BOOLNAME(rc));
return rc;
}
void Shutdown() {}
};
bool PollerInit::doTrace = (getenv("XrdSysIOE_TRACE") != 0);
XrdSysMutex PollerInit::traceMTX;
/******************************************************************************/
/* P o l l e r W a i t */
/******************************************************************************/
// This class is set in the channel when we need to serialize aces to the
// channel. Channel methods (as some others) check for this to see if they need
// to defer the current operation. We need to do his because some poller
// implementations must release the channel lock to avoid a deadlock.
//
class PollerWait : public Poller
{
public:
PollerWait() : Poller(-1, -1) {}
~PollerWait() {}
protected:
void Begin(XrdSysSemaphore *syncp, int &rc, const char **eTxt) {}
void Exclude(Channel *cP, bool &isLocked, bool dover=1) {}
bool Include(Channel *cP, int &eNum, const char **eTxt, bool &isLocked)
{eNum = EIDRM;
if (eTxt) *eTxt = "initializing channel";
return false;
}
bool Modify (Channel *cP, int &eNum, const char **eTxt, bool &isLocked)
{return Init(cP, eNum, eTxt, isLocked);}
void Shutdown() {}
};
PollerErr1 pollErr1;
PollerInit pollInit;
PollerWait pollWait;
};
};
/******************************************************************************/
/* C l a s s C h a n n e l M e t h o d s */
/******************************************************************************/
/******************************************************************************/
/* C o n s t r u c t o r */
/******************************************************************************/
XrdSys::IOEvents::Channel::Channel(Poller *pollP, int fd,
CallBack *cbP, void *cbArg)
: chPollXQ(pollP), chCB(cbP), chCBA(cbArg)
{
attList.next = attList.prev = this;
tmoList.next = tmoList.prev = this;
inTOQ = 0;
pollEnt = 0;
chStat = isClear;
Reset(&pollInit, fd);
pollP->Attach(this);
}
/******************************************************************************/
/* D e l e t e */
/******************************************************************************/
void XrdSys::IOEvents::Channel::Delete()
{
Poller *myPoller;
bool isLocked = true;
// Do some tracing
//
IF_TRACE(Delete,chFD,"status="<<STATUS);
// Lock ourselves during the delete process. If the channel is disassociated
// or the real poller is set to the error poller then this channel is clean
// and can be deleted (i.e. the channel ran through Detach()).
//
chMutex.Lock();
if (!chPollXQ || chPollXQ == &pollErr1)
{chMutex.UnLock();
delete this;
return;
}
// Disable and remove ourselves from all queues
//
myPoller = chPollXQ;
chPollXQ->Detach(this,isLocked,false);
if (!isLocked) chMutex.Lock();
// If we are in callback mode then we will need to delay the destruction until
// after the callback completes unless this is the poller thread. In that case,
// we need to tell the poller that we have been destroyed in a shelf-stable way.
//
if (chStat)
{if (XrdSysThread::Same(XrdSysThread::ID(),myPoller->pollTid))
{myPoller->chDead = true;
chMutex.UnLock();
} else {
XrdSysSemaphore cbDone(0);
IF_TRACE(Delete,chFD,"waiting for callback");
chStat = isDead;
chCBA = (void *)&cbDone;
chMutex.UnLock();
cbDone.Wait();
}
}
// It is now safe to release the storage
//
IF_TRACE(Delete,chFD,"chan="<< std::hex<<(void *)this<< std::dec);
delete this;
}
/******************************************************************************/
/* D i s a b l e */
/******************************************************************************/
bool XrdSys::IOEvents::Channel::Disable(int events, const char **eText)
{
int eNum = 0, newev, curev;
bool retval = true, isLocked = true;
// Lock this channel
//
chMutex.Lock();
// Get correct current events; depending on the state of the channel
//
if (chPoller == &pollWait) curev = static_cast<int>(reMod);
else curev = static_cast<int>(chEvents);
// Trace this entry
//
IF_TRACE(Disable,chFD,"->Disable(" <<events <<") chev=" <<curev);
// Calculate new event mask
//
events &= allEvents;
newev = curev & ~events;
// If something has changed, then modify the event mask in the poller. The
// poller may or may not unlock this channel during the process.
//
if (newev != curev)
{chEvents = newev;
retval = chPoller->Modify(this, eNum, eText, isLocked);
TRACE_MOD(Disable,chFD,newev);
} else {
TRACE_NOD(Disable,chFD,newev);
}
if (isLocked) chMutex.UnLock();
// All done
//
if (!retval) errno = eNum;
return retval;
}
/******************************************************************************/
/* E n a b l e */
/******************************************************************************/
bool XrdSys::IOEvents::Channel::Enable(int events, int timeout,
const char **eText)
{
int eNum = 0, newev, curev, tmoSet = 0;
bool retval, setTO, isLocked = true;
// Lock ourselves against any changes (this is a recursive mutex)
//
chMutex.Lock();
// Get correct current events; depending on the state of the channel
//
if (chPoller == &pollWait) curev = static_cast<int>(reMod);
else curev = static_cast<int>(chEvents);
// Trace this entry
//
IF_TRACE(Enable,chFD,"->Enable("<<events<<','<<timeout<<") chev="<<curev);
// Establish events that should be enabled
//
events &= allEvents;
newev = (curev ^ events) & events;
chEvents = curev | events;
// Handle timeout changes now
//
if (REVENTS(events))
{ if (timeout > 0) chRTO = timeout;
else if (timeout < 0) chRTO = 0;
if (rdDL != Poller::maxTime || chRTO) tmoSet |= CallBack::ReadyToRead;
}
if (WEVENTS(events))
{ if (timeout > 0) chWTO = timeout;
else if (timeout < 0) chWTO = 0;
if (wrDL != Poller::maxTime || chWTO) tmoSet |= CallBack::ReadyToWrite;
}
// Check if we have to reset the timeout. We need to hold the channel lock here.
//
if (tmoSet && chPoller != &pollErr1)
setTO = chPollXQ->TmoAdd(this, tmoSet);
else setTO = false;
// Check if any modifcations needed here. If so, invoke the modifier. Note that
// the modify will unlock the channel if the operation causes a wait. So,
// we cannot depend on the channel being locked upon return. The reason we do
// not unlock here is because we must ensure the channel doesn't change while
// we call modify. We let modify determine what to do.
//
if (newev)
{retval = chPoller->Modify(this, eNum, eText, isLocked);
TRACE_MOD(Enable,chFD,(curev | events));
} else {
retval = true;
TRACE_NOD(Enable,chFD,(curev | events));
}
// We need to notify the poller thread if the added deadline is the first in the
// queue and the poller is waiting. We also optimize for the case where the
// poller thread is always woken up to perform an action in which case it
// doesn't need a separate wakeup. We only do this if the enable succeeed. Note
// that we cannot hold the channel mutex for this call because it may wait.
//
if (isLocked) chMutex.UnLock();
bool isWakePend = CPP_ATOMIC_LOAD(chPollXQ->wakePend, std::memory_order_consume);
if (retval && !isWakePend && setTO && isLocked) chPollXQ->WakeUp();
// All done
//
if (!retval) errno = eNum;
return retval;
}
/******************************************************************************/
/* G e t C a l l B a c k */
/******************************************************************************/
void XrdSys::IOEvents::Channel::GetCallBack(CallBack **cbP, void **cbArg)
{
chMutex.Lock();
*cbP = chCB;
*cbArg = chCBA;
chMutex.UnLock();
}
/******************************************************************************/
/* Private: R e s e t */
/******************************************************************************/
void XrdSys::IOEvents::Channel::Reset(XrdSys::IOEvents::Poller *thePoller,
int fd, int eNum)
{
chPoller = thePoller;
chFD = fd;
chFault = eNum;
chRTO = 0;
chWTO = 0;
chEvents = 0;
dlType = 0;
inPSet = 0;
reMod = 0;
rdDL = Poller::maxTime;
wrDL = Poller::maxTime;
deadLine = Poller::maxTime;
}
/******************************************************************************/
/* S e t C a l l B a c k */
/******************************************************************************/
void XrdSys::IOEvents::Channel::SetCallBack(CallBack *cbP, void *cbArg)
{
// We only need to have the channel lock to set the callback. If the object
// is in the process of being destroyed, we do nothing.
//
chMutex.Lock();
if (chStat != isDead)
{chCB = cbP;
chCBA = cbArg;
}
chMutex.UnLock();
}
/******************************************************************************/
/* S e t F D */
/******************************************************************************/
void XrdSys::IOEvents::Channel::SetFD(int fd)
{
bool isLocked = true;
// Obtain the channel lock. If the object is in callback mode we have some
// extra work to do. If normal callback then indicate the channel transitioned
// to prevent it being automatically re-enabled. If it's being destroyed, then
// do nothing. Otherwise, this is a stupid double setFD call.
//
chMutex.Lock();
if (chStat == isDead)
{chMutex.UnLock();
return;
}
// This is a tricky deal here because we need to protect ourselves from other
// threads as well as the poller trying to do a callback. We first, set the
// poller target. This means the channel is no longer ready and callbacks will
// be skipped. We then remove the current file descriptor. This may unlock the
// channel but at this point that's ok.
//
if (inPSet)
{chPoller = &pollWait;
chPollXQ->Detach(this, isLocked, true);
if (!isLocked) chMutex.Lock();
}
// Indicate channel needs to be re-enabled then unlock the channel
//
Reset(&pollInit, fd);
chMutex.UnLock();
}
/******************************************************************************/
/* C l a s s P o l l e r */
/******************************************************************************/
/******************************************************************************/
/* C o n s t r u c t o r */
/******************************************************************************/
XrdSys::IOEvents::Poller::Poller(int rFD, int cFD)
{
// 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;
}
/******************************************************************************/
/* A t t a c h */
/******************************************************************************/
void XrdSys::IOEvents::Poller::Attach(XrdSys::IOEvents::Channel *cP)
{
Channel *pcP;
// We allow only one attach at a time to simplify the processing
//
adMutex.Lock();
// Chain this channel into the list of attached channels
//
if ((pcP = attBase)) {INSERT(attList, pcP, cP);}
else attBase = cP;
// All done
//
adMutex.UnLock();
}
/******************************************************************************/
/* C b k T M O */
/******************************************************************************/
void XrdSys::IOEvents::Poller::CbkTMO()
{
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();
}
/******************************************************************************/
/* C b k X e q */
/******************************************************************************/
bool XrdSys::IOEvents::Poller::CbkXeq(XrdSys::IOEvents::Channel *cP, int events,
int eNum, const char *eTxt)
{
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 *)cP->chCBA)->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;
}
/******************************************************************************/
/* Static: C r e a t e */
/******************************************************************************/
XrdSys::IOEvents::Poller *XrdSys::IOEvents::Poller::Create(int &eNum,
const char **eTxt,
int crOpts)
{
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;
}
/******************************************************************************/
/* D e t a c h */
/******************************************************************************/
void XrdSys::IOEvents::Poller::Detach(XrdSys::IOEvents::Channel *cP,
bool &isLocked, bool keep)
{
// The caller must hold the channel lock!
//
bool detFD = (cP->inPSet != 0);
// First remove the channel from the timeout queue
//
if (cP->inTOQ)
{toMutex.Lock();
REMOVE(tmoBase, tmoList, cP);
toMutex.UnLock();
}
// Allow only one detach at a time
//
adMutex.Lock();
// Preset channel to prohibit callback if we are not keeping this channel
//
if (!keep)
{cP->Reset(&pollErr1, cP->chFD);
cP->chPollXQ = &pollErr1;
cP->chCB = 0;
cP->chCBA = 0;
if (cP->attList.next != cP) {REMOVE(attBase, attList, cP);}
else if (attBase == cP) attBase = 0;
}
// Exclude this channel from the associated poll set, don't hold the ad lock
//
adMutex.UnLock();
if (detFD)
{cP->inPSet = 0;
if (cmdFD >= 0) Exclude(cP, isLocked, !ISPOLLER);
}
}
/******************************************************************************/
/* Protected: G e t R e q u e s t */
/******************************************************************************/
// Warning: This method runs unlocked. The caller must have exclusive use of
// the reqBuff otherwise unpredictable results will occur.
int XrdSys::IOEvents::Poller::GetRequest()
{
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;
}
/******************************************************************************/
/* Protected: I n i t */
/******************************************************************************/
bool XrdSys::IOEvents::Poller::Init(XrdSys::IOEvents::Channel *cP, int &eNum,
const char **eTxt, bool &isLocked)
{
// 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;
}
/******************************************************************************/
/* P o l l 2 E n u m */
/******************************************************************************/
int XrdSys::IOEvents::Poller::Poll2Enum(short events)
{
if (events & POLLERR) return EPIPE;
if (events & POLLHUP) return ECONNRESET;
if (events & POLLNVAL) return EBADF;
return EOPNOTSUPP;
}
/******************************************************************************/
/* S e n d C m d */
/******************************************************************************/
int XrdSys::IOEvents::Poller::SendCmd(PipeData &cmd)
{
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);
}
/******************************************************************************/
/* Protected: S e t P o l l E n t */
/******************************************************************************/
void XrdSys::IOEvents::Poller::SetPollEnt(XrdSys::IOEvents::Channel *cP, int pe)
{
cP->pollEnt = pe;
}
/******************************************************************************/
/* S t o p */
/******************************************************************************/
void XrdSys::IOEvents::Poller::Stop()
{
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();
}
/******************************************************************************/
/* T m o A d d */
/******************************************************************************/
bool XrdSys::IOEvents::Poller::TmoAdd(XrdSys::IOEvents::Channel *cP, int tmoSet)
{
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);
}
/******************************************************************************/
/* T m o D e l */
/******************************************************************************/
void XrdSys::IOEvents::Poller::TmoDel(XrdSys::IOEvents::Channel *cP)
{
// 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();
}
/******************************************************************************/
/* T m o G e t */
/******************************************************************************/
int XrdSys::IOEvents::Poller::TmoGet()
{
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;
}
/******************************************************************************/
/* W a k e U p */
/******************************************************************************/
void XrdSys::IOEvents::Poller::WakeUp()
{
static PipeData cmdbuff(PipeData::NoOp);
// Send it off to wakeup the poller thread, but only if here is no wakeup in
// progress.
//
// We use a mutex here because we want to produce a synchronization point - all
// threads that might be interested timeouts and wakeups are going to incur a
// cache bounce for the page where wakePend resides; they will see a consistent
// view of the wakePend flag. For those threads, this is equivalent to
// an atomic with memory_order std::memory_order_seq_cst (the strongest ordering).
// However, the threads that are not interested in timeouts will not get a flush
// for their copy of the wakePend page. They will still have the weaker memory
// ordering of consume/release (which is guaranteed anyway on all current architectures
// except for DEC Alpha).
toMutex.Lock();
bool isWakePend = CPP_ATOMIC_LOAD(wakePend, std::memory_order_consume);
if (isWakePend) {toMutex.UnLock();}
else {CPP_ATOMIC_STORE(wakePend, true, std::memory_order_release);
toMutex.UnLock();
SendCmd(cmdbuff);
}
}
/******************************************************************************/
/* I m p l e m e n t a t i o n S p e c i f i c s */
/******************************************************************************/
#if defined( __solaris__ )
#include "XrdSys/XrdSysIOEventsPollPort.icc"
#elif defined( __linux__ )
#include "XrdSys/XrdSysIOEventsPollE.icc"
#elif defined(__APPLE__)
#include "XrdSys/XrdSysIOEventsPollKQ.icc"
#else
#include "XrdSys/XrdSysIOEventsPollPoll.icc"
#endif
close
#define close(a)
Definition XrdPosix.hh:48
write
#define write(a, b, c)
Definition XrdPosix.hh:115
read
#define read(a, b, c)
Definition XrdPosix.hh:82
REMOVE
#define REMOVE(dlbase, dlvar, curitem)
Definition XrdSsiSessReal.cc:69
INSERT
#define INSERT(dlvar, curitem, newitem)
Definition XrdSsiSessReal.cc:63
CPP_ATOMIC_LOAD
#define CPP_ATOMIC_LOAD(x, order)
Definition XrdSysAtomics.hh:95
CPP_ATOMIC_STORE
#define CPP_ATOMIC_STORE(x, val, order)
Definition XrdSysAtomics.hh:96
XrdSysE2T
const char * XrdSysE2T(int errcode)
Definition XrdSysE2T.cc:104
XrdSysE2T.hh
XrdSysFD.hh
XrdSysHeaders.hh
XrdSysIOEventsPollE.icc
XrdSysIOEventsPollKQ.icc
XrdSysIOEventsPollPoll.icc
XrdSysIOEventsPollPort.icc
IF_TRACE
#define IF_TRACE(x, fd, y)
Definition XrdSysIOEvents.cc:93
TRACE_LOK
#define TRACE_LOK
Definition XrdSysIOEvents.cc:95
STATUS
#define STATUS
Definition XrdSysIOEvents.cc:57
TRACE_NOD
#define TRACE_NOD(x, fd, y)
Definition XrdSysIOEvents.cc:101
STATUSOF
#define STATUSOF(x)
Definition XrdSysIOEvents.cc:59
DO_TRACE
#define DO_TRACE(x, fd, y)
Definition XrdSysIOEvents.cc:86
REVENTS
#define REVENTS(x)
Definition XrdSysIOEvents.cc:78
BOOLNAME
#define BOOLNAME(x)
Definition XrdSysIOEvents.cc:84
TRACE_MOD
#define TRACE_MOD(x, fd, y)
Definition XrdSysIOEvents.cc:97
ISPOLLER
#define ISPOLLER
Definition XrdSysIOEvents.cc:82
WEVENTS
#define WEVENTS(x)
Definition XrdSysIOEvents.cc:80
XrdSysIOEvents.hh
XrdSysPlatform.hh
XrdSysPthread.hh
XRDSYSTHREAD_BIND
#define XRDSYSTHREAD_BIND
Definition XrdSysPthread.hh:543
TRACING
#define TRACING(x)
Definition XrdTrace.hh:70
XrdSysMutexHelper
Definition XrdSysPthread.hh:263
XrdSysMutexHelper::UnLock
void UnLock()
Definition XrdSysPthread.hh:274
XrdSysMutexHelper::Lock
void Lock(XrdSysMutex *Mutex)
Definition XrdSysPthread.hh:266
XrdSysMutex
Definition XrdSysPthread.hh:165
XrdSysMutex::Lock
void Lock()
Definition XrdSysPthread.hh:222
XrdSysMutex::UnLock
void UnLock()
Definition XrdSysPthread.hh:224
XrdSysSemaphore
Definition XrdSysPthread.hh:494
XrdSysSemaphore::Wait
void Wait()
Definition XrdSysPthread.hh:509
XrdSysThread::Same
static int Same(pthread_t t1, pthread_t t2)
Definition XrdSysPthread.hh:593
XrdSysThread::Run
static int Run(pthread_t *, void *(*proc)(void *), void *arg, int opts=0, const char *desc=0)
Definition XrdSysPthread.cc:307
XrdSysThread::ID
static pthread_t ID(void)
Definition XrdSysPthread.hh:584
XrdSys::IOEvents::BootStrap
Definition XrdSysIOEvents.cc:136
XrdSys::IOEvents::BootStrap::Start
static void * Start(void *parg)
Definition XrdSysIOEvents.cc:142
XrdSys::IOEvents::CallBack
Definition XrdSysIOEvents.hh:79
XrdSys::IOEvents::CallBack::Fatal
virtual void Fatal(Channel *chP, void *cbArg, int eNum, const char *eTxt)
Definition XrdSysIOEvents.hh:131
XrdSys::IOEvents::CallBack::Event
virtual bool Event(Channel *chP, void *cbArg, int evFlags)=0
XrdSys::IOEvents::CallBack::Stop
virtual void Stop(Channel *chP, void *cbArg)
Definition XrdSysIOEvents.hh:146
XrdSys::IOEvents::CallBack::ReadyToWrite
@ ReadyToWrite
Writing won't block.
Definition XrdSysIOEvents.hh:90
XrdSys::IOEvents::CallBack::ReadyToRead
@ ReadyToRead
New data has arrived.
Definition XrdSysIOEvents.hh:88
XrdSys::IOEvents::CallBack::ReadTimeOut
@ ReadTimeOut
Read timeout.
Definition XrdSysIOEvents.hh:89
XrdSys::IOEvents::CallBack::WriteTimeOut
@ WriteTimeOut
Write timeout.
Definition XrdSysIOEvents.hh:91
XrdSys::IOEvents::CallBack::ValidEvents
@ ValidEvents
Mask to test for valid events.
Definition XrdSysIOEvents.hh:92
XrdSys::IOEvents::Channel
Definition XrdSysIOEvents.hh:174
XrdSys::IOEvents::Channel::SetCallBack
void SetCallBack(CallBack *cbP, void *cbArg=0)
Definition XrdSysIOEvents.cc:513
XrdSys::IOEvents::Channel::GetCallBack
void GetCallBack(CallBack **cbP, void **cbArg)
Definition XrdSysIOEvents.cc:480
XrdSys::IOEvents::Channel::allEvents
@ allEvents
All of the above.
Definition XrdSysIOEvents.hh:197
XrdSys::IOEvents::Channel::errorEvents
@ errorEvents
Error event non-r/w specific.
Definition XrdSysIOEvents.hh:195
XrdSys::IOEvents::Channel::stopEvent
@ stopEvent
Poller stop event.
Definition XrdSysIOEvents.hh:196
XrdSys::IOEvents::Channel::Enable
bool Enable(int events, int timeout=0, const char **eText=0)
Definition XrdSysIOEvents.cc:401
XrdSys::IOEvents::Channel::SetFD
void SetFD(int fd)
Definition XrdSysIOEvents.cc:531
XrdSys::IOEvents::Channel::Delete
void Delete()
Definition XrdSysIOEvents.cc:303
XrdSys::IOEvents::Channel::Poller
friend class Poller
Definition XrdSysIOEvents.hh:175
XrdSys::IOEvents::Channel::GetFD
int GetFD()
Definition XrdSysIOEvents.hh:276
XrdSys::IOEvents::Channel::Channel
Channel(Poller *pollP, int fd, CallBack *cbP=0, void *cbArg=0)
Definition XrdSysIOEvents.cc:285
XrdSys::IOEvents::Channel::Disable
bool Disable(int events, const char **eText=0)
Definition XrdSysIOEvents.cc:356
XrdSys::IOEvents::PollerErr1
Definition XrdSysIOEvents.cc:167
XrdSys::IOEvents::PollerErr1::~PollerErr1
~PollerErr1()
Definition XrdSysIOEvents.cc:171
XrdSys::IOEvents::PollerErr1::Exclude
void Exclude(Channel *cP, bool &isLocked, bool dover=1)
Definition XrdSysIOEvents.cc:177
XrdSys::IOEvents::PollerErr1::Modify
bool Modify(Channel *cP, int &eNum, const char **eTxt, bool &isLocked)
Definition XrdSysIOEvents.cc:187
XrdSys::IOEvents::PollerErr1::Begin
void Begin(XrdSysSemaphore *syncp, int &rc, const char **eTxt)
Definition XrdSysIOEvents.cc:174
XrdSys::IOEvents::PollerErr1::Shutdown
void Shutdown()
Definition XrdSysIOEvents.cc:194
XrdSys::IOEvents::PollerErr1::PollerErr1
PollerErr1()
Definition XrdSysIOEvents.cc:170
XrdSys::IOEvents::PollerErr1::Include
bool Include(Channel *cP, int &eNum, const char **eTxt, bool &isLocked)
Definition XrdSysIOEvents.cc:180
XrdSys::IOEvents::PollerInit
Definition XrdSysIOEvents.cc:206
XrdSys::IOEvents::PollerInit::Begin
void Begin(XrdSysSemaphore *syncp, int &rc, const char **eTxt)
Definition XrdSysIOEvents.cc:217
XrdSys::IOEvents::PollerInit::Shutdown
void Shutdown()
Definition XrdSysIOEvents.cc:233
XrdSys::IOEvents::PollerInit::Exclude
void Exclude(Channel *cP, bool &isLocked, bool dover=1)
Definition XrdSysIOEvents.cc:219
XrdSys::IOEvents::PollerInit::PollerInit
PollerInit()
Definition XrdSysIOEvents.cc:209
XrdSys::IOEvents::PollerInit::Modify
bool Modify(Channel *cP, int &eNum, const char **eTxt, bool &isLocked)
Definition XrdSysIOEvents.cc:227
XrdSys::IOEvents::PollerInit::doTrace
static bool doTrace
Definition XrdSysIOEvents.cc:213
XrdSys::IOEvents::PollerInit::traceMTX
static XrdSysMutex traceMTX
Definition XrdSysIOEvents.cc:212
XrdSys::IOEvents::PollerInit::Include
bool Include(Channel *cP, int &eNum, const char **eTxt, bool &isLocked)
Definition XrdSysIOEvents.cc:221
XrdSys::IOEvents::PollerInit::~PollerInit
~PollerInit()
Definition XrdSysIOEvents.cc:210
XrdSys::IOEvents::PollerWait
Definition XrdSysIOEvents.cc:249
XrdSys::IOEvents::PollerWait::Shutdown
void Shutdown()
Definition XrdSysIOEvents.cc:269
XrdSys::IOEvents::PollerWait::PollerWait
PollerWait()
Definition XrdSysIOEvents.cc:252
XrdSys::IOEvents::PollerWait::Include
bool Include(Channel *cP, int &eNum, const char **eTxt, bool &isLocked)
Definition XrdSysIOEvents.cc:260
XrdSys::IOEvents::PollerWait::Modify
bool Modify(Channel *cP, int &eNum, const char **eTxt, bool &isLocked)
Definition XrdSysIOEvents.cc:266
XrdSys::IOEvents::PollerWait::Exclude
void Exclude(Channel *cP, bool &isLocked, bool dover=1)
Definition XrdSysIOEvents.cc:258
XrdSys::IOEvents::PollerWait::~PollerWait
~PollerWait()
Definition XrdSysIOEvents.cc:253
XrdSys::IOEvents::PollerWait::Begin
void Begin(XrdSysSemaphore *syncp, int &rc, const char **eTxt)
Definition XrdSysIOEvents.cc:256
XrdSys::IOEvents::Poller
Definition XrdSysIOEvents.hh:372
XrdSys::IOEvents::Poller::Modify
virtual bool Modify(Channel *cP, int &eNum, const char **eTxt, bool &isLocked)=0
XrdSys::IOEvents::Poller::pipeBuff
char * pipeBuff
Definition XrdSysIOEvents.hh:507
XrdSys::IOEvents::Poller::Include
virtual bool Include(Channel *cP, int &eNum, const char **eTxt, bool &isLocked)=0
XrdSys::IOEvents::Poller::GetFault
int GetFault(Channel *cP)
Definition XrdSysIOEvents.hh:437
XrdSys::IOEvents::Poller::attBase
Channel * attBase
Definition XrdSysIOEvents.hh:488
XrdSys::IOEvents::Poller::tmoBase
Channel * tmoBase
Definition XrdSysIOEvents.hh:489
XrdSys::IOEvents::Poller::reqFD
int reqFD
Definition XrdSysIOEvents.hh:495
XrdSys::IOEvents::Poller::pollTid
pthread_t pollTid
Definition XrdSysIOEvents.hh:491
XrdSys::IOEvents::Poller::Poller
Poller(int cFD, int rFD)
Definition XrdSysIOEvents.cc:571
XrdSys::IOEvents::Poller::Create
static Poller * Create(int &eNum, const char **eTxt=0, int crOpts=0)
Definition XrdSysIOEvents.cc:762
XrdSys::IOEvents::Poller::TmoDel
void TmoDel(Channel *cP)
Definition XrdSysIOEvents.cc:1160
XrdSys::IOEvents::Poller::pipePoll
struct pollfd pipePoll
Definition XrdSysIOEvents.hh:493
XrdSys::IOEvents::Poller::Begin
virtual void Begin(XrdSysSemaphore *syncp, int &rc, const char **eTxt)=0
XrdSys::IOEvents::Poller::CbkXeq
bool CbkXeq(Channel *cP, int events, int eNum, const char *eTxt)
Definition XrdSysIOEvents.cc:636
XrdSys::IOEvents::Poller::TmoGet
int TmoGet()
Definition XrdSysIOEvents.cc:1180
XrdSys::IOEvents::Poller::cmdFD
int cmdFD
Definition XrdSysIOEvents.hh:494
XrdSys::IOEvents::Poller::GetRequest
int GetRequest()
Definition XrdSysIOEvents.cc:870
XrdSys::IOEvents::Poller::maxTime
static time_t maxTime
Definition XrdSysIOEvents.hh:513
XrdSys::IOEvents::Poller::SendCmd
int SendCmd(PipeData &cmd)
Definition XrdSysIOEvents.cc:993
XrdSys::IOEvents::Poller::tmoMask
unsigned char tmoMask
Definition XrdSysIOEvents.hh:509
XrdSys::IOEvents::Poller::Shutdown
virtual void Shutdown()=0
XrdSys::IOEvents::Poller::pipeBlen
int pipeBlen
Definition XrdSysIOEvents.hh:508
XrdSys::IOEvents::Poller::CbkTMO
void CbkTMO()
Definition XrdSysIOEvents.cc:614
XrdSys::IOEvents::Poller::Poll2Enum
int Poll2Enum(short events)
Definition XrdSysIOEvents.cc:978
XrdSys::IOEvents::Poller::Channel
friend class Channel
Definition XrdSysIOEvents.hh:374
XrdSys::IOEvents::Poller::reqBuff
PipeData reqBuff
Definition XrdSysIOEvents.hh:506
XrdSys::IOEvents::Poller::TmoAdd
bool TmoAdd(Channel *cP, int tmoSet)
Definition XrdSysIOEvents.cc:1092
XrdSys::IOEvents::Poller::chDead
bool chDead
Definition XrdSysIOEvents.hh:511
XrdSys::IOEvents::Poller::optTOM
@ optTOM
Definition XrdSysIOEvents.hh:398
XrdSys::IOEvents::Poller::Stop
void Stop()
Definition XrdSysIOEvents.cc:1030
XrdSys::IOEvents::Poller::SetPollEnt
void SetPollEnt(Channel *cP, int ptEnt)
Definition XrdSysIOEvents.cc:1021
XrdSys::IOEvents::Poller::parentPID
static pid_t parentPID
Definition XrdSysIOEvents.hh:515
XrdSys::IOEvents::Poller::Init
bool Init(Channel *cP, int &eNum, const char **eTxt, bool &isLockd)
Definition XrdSysIOEvents.cc:909
XrdSys::IOEvents::pollInit
PollerInit pollInit
Definition XrdSysIOEvents.cc:273
XrdSys::IOEvents::pollErr1
PollerErr1 pollErr1
Definition XrdSysIOEvents.cc:272
XrdSys::IOEvents::pollWait
PollerWait pollWait
Definition XrdSysIOEvents.cc:274
XrdSys
Definition XrdClPollerBuiltIn.hh:28
XrdSys::IOEvents::Poller::PipeData
Definition XrdSysIOEvents.hh:496
XrdSys::IOEvents::Poller::PipeData::theSem
XrdSysSemaphore * theSem
Definition XrdSysIOEvents.hh:497
XrdSys::IOEvents::Poller::PipeData::req
char req
Definition XrdSysIOEvents.hh:496
XrdSys::IOEvents::Poller::PipeData::Post
@ Post
Definition XrdSysIOEvents.hh:498
XrdSys::IOEvents::Poller::PipeData::Stop
@ Stop
Definition XrdSysIOEvents.hh:499
XrdSys::IOEvents::pollArg
Definition XrdSysIOEvents.cc:125
XrdSys::IOEvents::pollArg::retCode
int retCode
Definition XrdSysIOEvents.cc:127
XrdSys::IOEvents::pollArg::pollArg
pollArg()
Definition XrdSysIOEvents.cc:130
XrdSys::IOEvents::pollArg::pollP
Poller * pollP
Definition XrdSysIOEvents.cc:125
XrdSys::IOEvents::pollArg::retMsg
const char * retMsg
Definition XrdSysIOEvents.cc:126
XrdSys::IOEvents::pollArg::pollSync
XrdSysSemaphore * pollSync
Definition XrdSysIOEvents.cc:128
XrdSys::IOEvents::pollArg::~pollArg
~pollArg()
Definition XrdSysIOEvents.cc:132