pdns/epollmplexer.cc

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