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