pdns/epollmplexer.cc

   1 #include "mplexer.hh"
   2 #include "sstuff.hh"
   3 #include <iostream>
   4 #include <unistd.h>
   5 #include "misc.hh"
   6 #include <boost/lexical_cast.hpp>
   7 #include "syncres.hh"
   8 #ifdef __linux__
   9 #include <sys/epoll.h>
  10 #endif
  11
  12 #include "namespaces.hh"
  13 #include "namespaces.hh"
  14
  15 class EpollFDMultiplexer : public FDMultiplexer
  16 {
  17 public:
  18   EpollFDMultiplexer();
  19   virtual ~EpollFDMultiplexer()
  20   {
  21     close(d_epollfd);
  22   }
  23
  24   virtual int run(struct timeval* tv);
  25
  26   virtual void addFD(callbackmap_t& cbmap, int fd, callbackfunc_t toDo, const funcparam_t& parameter);
  27   virtual void removeFD(callbackmap_t& cbmap, int fd);
  28   string getName()
  29   {
  30     return "epoll";
  31   }
  32 private:
  33   int d_epollfd;
  34   boost::shared_array<epoll_event> d_eevents;
  35   static int s_maxevents; // not a hard maximum
  36 };
  37
  38
  39 static FDMultiplexer* makeEpoll()
  40 {
  41   return new EpollFDMultiplexer();
  42 }
  43
  44 static struct EpollRegisterOurselves
  45 {
  46   EpollRegisterOurselves() {
  47     FDMultiplexer::getMultiplexerMap().insert(make_pair(0, &makeEpoll)); // priority 0!
  48   }
  49 } doItEpoll;
  50
  51
  52 int EpollFDMultiplexer::s_maxevents=1024;
  53 EpollFDMultiplexer::EpollFDMultiplexer() : d_eevents(new epoll_event[s_maxevents])
  54 {
  55   d_epollfd=epoll_create(s_maxevents); // not hard max
  56   if(d_epollfd < 0)
  57     throw FDMultiplexerException("Setting up epoll: "+stringerror());
  58   int fd=socket(AF_INET, SOCK_DGRAM, 0); // for self-test
  59   if(fd < 0)
  60     return;
  61   try {
  62     addReadFD(fd, 0);
  63     removeReadFD(fd);
  64     close(fd);
  65     return;
  66   }
  67   catch(FDMultiplexerException &fe) {
  68     close(fd);
  69     close(d_epollfd);
  70     throw FDMultiplexerException("epoll multiplexer failed self-test: "+string(fe.what()));
  71   }
  72
  73 }
  74
  75 void EpollFDMultiplexer::addFD(callbackmap_t& cbmap, int fd, callbackfunc_t toDo, const funcparam_t& parameter)
  76 {
  77   accountingAddFD(cbmap, fd, toDo, parameter);
  78
  79   struct epoll_event eevent;
  80
  81   eevent.events = (&cbmap == &d_readCallbacks) ? EPOLLIN : EPOLLOUT;
  82
  83   eevent.data.u64=0; // placate valgrind (I love it so much)
  84   eevent.data.fd=fd;
  85
  86   if(epoll_ctl(d_epollfd, EPOLL_CTL_ADD, fd, &eevent) < 0) {
  87     cbmap.erase(fd);
  88     throw FDMultiplexerException("Adding fd to epoll set: "+stringerror());
  89   }
  90 }
  91
  92 void EpollFDMultiplexer::removeFD(callbackmap_t& cbmap, int fd)
  93 {
  94   if(!cbmap.erase(fd))
  95     throw FDMultiplexerException("Tried to remove unlisted fd "+lexical_cast<string>(fd)+ " from multiplexer");
  96
  97   struct epoll_event dummy;
  98   dummy.events = 0;
  99   dummy.data.u64 = 0;
 100
 101   if(epoll_ctl(d_epollfd, EPOLL_CTL_DEL, fd, &dummy) < 0)
 102     throw FDMultiplexerException("Removing fd from epoll set: "+stringerror());
 103 }
 104
 105 int EpollFDMultiplexer::run(struct timeval* now)
 106 {
 107   if(d_inrun) {
 108     throw FDMultiplexerException("FDMultiplexer::run() is not reentrant!\n");
 109   }
 110
 111   int ret=epoll_wait(d_epollfd, d_eevents.get(), s_maxevents, 500);
 112   gettimeofday(now,0); // MANDATORY
 113
 114   if(ret < 0 && errno!=EINTR)
 115     throw FDMultiplexerException("epoll returned error: "+stringerror());
 116
 117   if(ret < 1) // thanks AB!
 118     return 0;
 119
 120   d_inrun=true;
 121   for(int n=0; n < ret; ++n) {
 122     d_iter=d_readCallbacks.find(d_eevents[n].data.fd);
 123
 124     if(d_iter != d_readCallbacks.end()) {
 125       d_iter->second.d_callback(d_iter->first, d_iter->second.d_parameter);
 126       continue; // so we don't refind ourselves as writable!
 127     }
 128     d_iter=d_writeCallbacks.find(d_eevents[n].data.fd);
 129
 130     if(d_iter != d_writeCallbacks.end()) {
 131       d_iter->second.d_callback(d_iter->first, d_iter->second.d_parameter);
 132     }
 133   }
 134   d_inrun=false;
 135   return 0;
 136 }
 137
 138 #if 0
 139 void acceptData(int fd, funcparam_t& parameter)
 140 {
 141   cout<<"Have data on fd "<<fd<<endl;
 142   Socket* sock=funcparam_t_cast<Socket*>(parameter);
 143   string packet;
 144   IPEndpoint rem;
 145   sock->recvFrom(packet, rem);
 146   cout<<"Received "<<packet.size()<<" bytes!\n";
 147 }
 148
 149
 150 int main()
 151 {
 152   Socket s(AF_INET, SOCK_DGRAM);
 153
 154   IPEndpoint loc("0.0.0.0", 2000);
 155   s.bind(loc);
 156
 157   EpollFDMultiplexer sfm;
 158
 159   sfm.addReadFD(s.getHandle(), &acceptData, &s);
 160
 161   for(int n=0; n < 100 ; ++n) {
 162     sfm.run();
 163   }
 164   sfm.removeReadFD(s.getHandle());
 165   sfm.removeReadFD(s.getHandle());
 166 }
 167 #endif
 168
 169