Loading...
Searching...
No Matches
XrdPollE.icc
Go to the documentation of this file.
/******************************************************************************/
/* */
/* X r d P o l l E . i c c */
/* */
/* (c) 2008 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 <fcntl.h>
#include <cstdlib>
#include <sys/types.h>
#include <sys/stat.h>
#include <sys/epoll.h>
#include <sys/eventfd.h>
#include "Xrd/XrdPollE.hh"
#include "Xrd/XrdScheduler.hh"
/******************************************************************************/
/* n e w P o l l e r */
/******************************************************************************/
XrdPoll *XrdPoll::newPoller(int pollid, int maxfd)
{
int pfd, wfd, bytes, alignment, pagsz = getpagesize();
struct epoll_event *pp;
// Open the /dev/poll driver
//
#ifndef EPOLL_CLOEXEC
if ((pfd = epoll_create(maxfd)) >= 0) fcntl(pfd, F_SETFD, FD_CLOEXEC);
else
#else
if ((pfd = epoll_create1(EPOLL_CLOEXEC)) < 0)
#endif
{Log.Emsg("Poll", errno, "create epoll device"); return 0;}
if ((wfd = eventfd(0, EFD_CLOEXEC)) < 0)
{Log.Emsg("Poll", errno,
"create an eventfd as the wait-poller descriptor");
close(pfd);
return 0;
}
// Calculate the size of the poll table and allocate it
//
bytes = maxfd * sizeof(struct epoll_event);
alignment = (bytes < pagsz ? 1024 : pagsz);
if (posix_memalign((void **)&pp, alignment, bytes))
{Log.Emsg("Poll", ENOMEM, "create poll table");
close(wfd);
close(pfd);
return 0;
}
// Create new poll object
//
memset((void *)pp, 0, bytes);
return (XrdPoll *)new XrdPollE(pp, maxfd, pfd, wfd);
}
/******************************************************************************/
/* D e s t r u c t o r */
/******************************************************************************/
XrdPollE::~XrdPollE()
{
if (PollTab) free(PollTab);
if (WaitFd >= 0) close(WaitFd);
if (PollDfd >= 0) close(PollDfd);
}
/******************************************************************************/
/* A d d W a i t F d */
/******************************************************************************/
int XrdPollE::AddWaitFd()
{
const unsigned int myPollEvts = EPOLLIN;
struct epoll_event myEvent = {myPollEvts, {(void *)&WaitFd}};
// Add the waitfd to the poll set
//
if (epoll_ctl(PollDfd, EPOLL_CTL_ADD, WaitFd, &myEvent) < 0)
{int rc = errno;
Log.Emsg("Poll", rc, "include the wait FD in the poll set");
return rc;
}
return 0;
}
/******************************************************************************/
/* D i s a b l e */
/******************************************************************************/
void XrdPollE::Disable(XrdPollInfo &pInfo, const char *etxt)
{
// Simply return if the link is already disabled
//
if (!pInfo.isEnabled) return;
// If Linux 2.6.9 we use EPOLLONESHOT to automatically disable a polled fd.
// So, the Disable() method need not do anything. Prior kernels did not have
// this mechanism so we need to do this manually.
//
#ifndef EPOLLONESHOT
struct epoll_event myEvents = {0, (void *)&pInfo};
// Disable this fd. Unlike solaris, epoll_ctl() does not block when the pollfd
// is being waited upon by another thread.
//
if (epoll_ctl(PollDfd, EPOLL_CTL_MOD, pInfo.FD, &myEvents))
{Log.Emsg("Poll", errno, "disable link", lp->ID); return;}
#endif
// Trace this event
//
pInfo.isEnabled = false;
TRACEI(POLL, "Poller " <<PID <<" async disabling link " <<pInfo.FD);
// Check if this link needs to be rescheduled. If so, the caller better have
// the link opMutex lock held for this to work!
//
if (etxt && Finish(pInfo, etxt)) Sched.Schedule((XrdJob *)&pInfo.Link);
}
/******************************************************************************/
/* E n a b l e */
/******************************************************************************/
int XrdPollE::Enable(XrdPollInfo &pInfo)
{
struct epoll_event myEvents = {ePollEvents, {(void *)&pInfo}};
// Simply return if the link is already enabled
//
if (pInfo.isEnabled) return 1;
// Enable this fd. Unlike solaris, epoll_ctl() does not block when the pollfd
// is being waited upon by another thread.
//
pInfo.isEnabled = true;
if (epoll_ctl(PollDfd, EPOLL_CTL_MOD, pInfo.FD, &myEvents))
{Log.Emsg("Poll", errno, "enable link", pInfo.Link.ID);
pInfo.isEnabled = false;
return 0;
}
// Do final processing
//
TRACE(POLL, "Poller " <<PID <<" enabled " <<pInfo.Link.ID);
numEnabled++;
return 1;
}
/******************************************************************************/
/* E x c l u d e */
/******************************************************************************/
void XrdPollE::Exclude(XrdPollInfo &pInfo)
{
// Make sure this link is not enabled
//
if (pInfo.isEnabled)
{Log.Emsg("Poll", "Detach of enabled link", pInfo.Link.ID);
Disable(pInfo);
}
// Forcibly remove the link from the poll set
//
remFD(pInfo, 0);
// Wait to make sure the poll thread has completed handling the last set of
// events which may have included this Link. The handing includes examining the
// Link's PollInfo before deciding if to schedule the Link. After we return,
// the PollInfo may be reset and the Link could be subsequently reused.
//
Wait4Poller();
}
/******************************************************************************/
/* H a n d l e W a i t F d */
/******************************************************************************/
void XrdPollE::HandleWaitFd(const unsigned int events)
{
// Used by the polling thread to signal waiters that a polling loop has been
// completed.
//
eventfd_t ic, cnt;
// We don't expect anything but EPOLLIN. But if we get an error (errors
// can be reported, despite not being selected events) abort rather than
// potentially keeping the poll thread busy repeatedly looping.
//
if (!(events & EPOLLIN) || (events & (EPOLLERR | EPOLLHUP)))
{char eBuff[64];
Log.Emsg("Poll", "wait-poller handler:", x2Text(events, eBuff));
if (events & (EPOLLERR | EPOLLHUP))
abort();
return;
}
if (eventfd_read(WaitFd, &cnt) < 0)
{Log.Emsg("Poll", errno, "read from the wait-poller descriptor");
return;
}
for (ic=0;ic<cnt;++ic) WaitFdSem.Post();
}
/******************************************************************************/
/* I n c l u d e */
/******************************************************************************/
int XrdPollE::Include(XrdPollInfo &pInfo)
{
struct epoll_event myEvent = {0, {(void *)&pInfo}};
int rc;
// Add this fd to the poll set
//
if ((rc = epoll_ctl(PollDfd, EPOLL_CTL_ADD, pInfo.FD, &myEvent)) < 0)
Log.Emsg("Poll", errno, "include link", pInfo.Link.ID);
// All done
//
return rc == 0;
}
/******************************************************************************/
/* r e m F D */
/******************************************************************************/
void XrdPollE::remFD(XrdPollInfo &pInfo, unsigned int events)
{
struct epoll_event myEvents = {0, {(void *)&pInfo}};
// It works out that ONESHOT mode or even CTL_MOD requests do not necessarily
// prevent epoll_wait() from returning on an error event. So, we must manually
// remove the fd from the set and assume the logic was actually correct. If it
// wasn't then the client will eventually timeout and retry the request. This
// may cause a double remove; so we don't consider these an arror.
//
if (pInfo.FD > 0)
{TRACEI(POLL, "Poller " <<PID <<" removing FD " <<pInfo.FD);
if (epoll_ctl(PollDfd, EPOLL_CTL_DEL, pInfo.FD, &myEvents)
&& (errno != ENOENT || events != 0))
{const char *why;
char buff[96];
if (events & (EPOLLHUP | EPOLLRDHUP)) why = "sever";
else if (events & EPOLLERR) why = "error";
else why = "disc";
snprintf(buff, sizeof(buff), "exclude fd %d during %s (%x) event; link",
pInfo.FD, why, events);
Log.Emsg("Poll", errno, buff, pInfo.Link.ID);
}
}
}
/******************************************************************************/
/* S t a r t */
/******************************************************************************/
void XrdPollE::Start(XrdSysSemaphore *syncsem, int &retcode)
{
char eBuff[64];
int rc, i, numpolled, num2sched;
unsigned int waitFdEvents;
bool haveWaiters;
XrdJob *jfirst, *jlast;
const short pollOK = EPOLLIN | EPOLLPRI;
XrdLink *lp;
XrdPollInfo *pInfo;
if ((rc = AddWaitFd()))
{retcode = rc;
syncsem->Post();
return;
}
// Indicate to the starting thread that all went well
//
retcode = 0;
syncsem->Post();
// Now start dispatching links that are ready
//
do {do {numpolled = epoll_wait(PollDfd, PollTab, PollMax, -1);}
while (numpolled < 0 && errno == EINTR);
if (numpolled == 0) continue;
if (numpolled < 0)
{Log.Emsg("Poll", errno, "poll for events");
abort();
}
numEvents += numpolled;
// Checkout which links must be dispatched (no need to lock)
//
jfirst = jlast = 0; num2sched = 0;
haveWaiters = false; waitFdEvents = 0;
for (i = 0; i < numpolled; i++)
{if (PollTab[i].data.ptr == &WaitFd)
{haveWaiters = true; waitFdEvents = PollTab[i].events;}
else if ((pInfo = (XrdPollInfo *)PollTab[i].data.ptr))
{if (!(pInfo->isEnabled) && pInfo->FD >= 0)
remFD(*pInfo, PollTab[i].events);
else {pInfo->isEnabled = 0;
if (!(PollTab[i].events & pollOK)
|| (PollTab[i].events & POLLRDHUP))
Finish(*pInfo, x2Text(PollTab[i].events, eBuff));
lp = &(pInfo->Link);
lp->NextJob = jfirst; jfirst = (XrdJob *)lp;
if (!jlast) jlast=(XrdJob *)lp;
num2sched++;
#ifndef EPOLLONESHOT
PollTab[i].events = 0;
if (epoll_ctl(PollDfd,EPOLL_CTL_MOD,pInfo.FD,&PollTab[i]))
Log.Emsg("Poll",errno,"disable link",pInfo.Link.ID);
#endif
}
} else Log.Emsg("Poll", "null link event!!!!");
}
// Schedule the polled links
//
if (num2sched == 1) Sched.Schedule(jfirst);
else if (num2sched) Sched.Schedule(num2sched, jfirst, jlast);
if (haveWaiters) HandleWaitFd(waitFdEvents);
} while(1);
}
/******************************************************************************/
/* W a i t 4 P o l l e r */
/******************************************************************************/
void XrdPollE::Wait4Poller()
{
// Makes the caller wait for the polling thread to complete an event processing
// loop.
//
if (eventfd_write(WaitFd, 1) < 0)
{Log.Emsg("Poll", errno, "write to the wait-poller descriptor");
return;
}
WaitFdSem.Wait();
}
/******************************************************************************/
/* x 2 T e x t */
/******************************************************************************/
const char *XrdPollE::x2Text(unsigned int events, char *buff)
{
if (events & EPOLLERR) return "socket error";
if (events & (EPOLLHUP | EPOLLRDHUP)) return "hangup";
sprintf(buff, "unusual event (%.4x)", events);
return buff;
}
XrdPollE.hh
EPOLLRDHUP
#define EPOLLRDHUP
Definition XrdPollE.hh:37
close
#define close(a)
Definition XrdPosix.hh:48
XrdScheduler.hh
TRACE
#define TRACE(act, x)
Definition XrdTrace.hh:63
TRACEI
#define TRACEI(act, x)
Definition XrdTrace.hh:66
XrdJob
Definition XrdJob.hh:43
XrdJob::NextJob
XrdJob * NextJob
Definition XrdJob.hh:46
XrdLink
Definition XrdLink.hh:52
XrdLink::ID
char * ID
Pointer to the client's link identity.
Definition XrdLink.hh:210
XrdPollE
Definition XrdPollE.hh:41
XrdPollE::x2Text
const char * x2Text(unsigned int evf, char *buff)
Definition XrdPollE.icc:380
XrdPollE::Exclude
void Exclude(XrdPollInfo &pInfo)
Definition XrdPollE.icc:184
XrdPollE::Disable
void Disable(XrdPollInfo &pInfo, const char *etxt=0)
Definition XrdPollE.icc:119
XrdPollE::Start
void Start(XrdSysSemaphore *syncp, int &rc)
Definition XrdPollE.icc:290
XrdPollE::Enable
int Enable(XrdPollInfo &pInfo)
Definition XrdPollE.icc:155
XrdPollE::Include
int Include(XrdPollInfo &pInfo)
Definition XrdPollE.icc:241
XrdPollE::~XrdPollE
~XrdPollE()
Definition XrdPollE.icc:88
XrdPollInfo
Definition XrdPollInfo.hh:37
XrdPollInfo::FD
int FD
Definition XrdPollInfo.hh:44
XrdPollInfo::Link
XrdLink & Link
Definition XrdPollInfo.hh:41
XrdPollInfo::isEnabled
bool isEnabled
Definition XrdPollInfo.hh:46
XrdPoll::XrdPoll
XrdPoll()
Definition XrdPoll.cc:118
XrdPoll::PID
int PID
Definition XrdPoll.hh:82
XrdPoll::numEvents
int numEvents
Definition XrdPoll.hh:133
XrdPoll::newPoller
static XrdPoll * newPoller(int pollid, int numfd)
Definition XrdPollE.icc:45
XrdPoll::Finish
static int Finish(XrdPollInfo &pInfo, const char *etxt=0)
Definition XrdPoll.cc:204
XrdPoll::numEnabled
int numEnabled
Definition XrdPoll.hh:132
XrdScheduler::Schedule
void Schedule(XrdJob *jp)
Definition XrdScheduler.cc:373
XrdSysError::Emsg
int Emsg(const char *esfx, int ecode, const char *text1, const char *text2=0)
Definition XrdSysError.cc:95
XrdSysSemaphore
Definition XrdSysPthread.hh:494
XrdSysSemaphore::Post
void Post()
Definition XrdSysPthread.hh:505
XrdGlobal::Log
XrdSysError Log
Definition XrdConfig.cc:113