pdns/pollmplexer.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 <poll.h>
   8 #include "misc.hh"
   9 #include "namespaces.hh"
  10
  11 FDMultiplexer* FDMultiplexer::getMultiplexerSilent()
  12 {
  13   FDMultiplexer* ret = nullptr;
  14   for(const auto& i : FDMultiplexer::getMultiplexerMap()) {
  15     try {
  16       ret = i.second();
  17       return ret;
  18     }
  19     catch(const FDMultiplexerException& fe) {
  20     }
  21     catch(...) {
  22     }
  23   }
  24   return ret;
  25 }
  26
  27
  28 class PollFDMultiplexer : public FDMultiplexer
  29 {
  30 public:
  31   PollFDMultiplexer()
  32   {}
  33   virtual ~PollFDMultiplexer()
  34   {
  35   }
  36
  37   virtual int run(struct timeval* tv, int timeout=500);
  38
  39   virtual void addFD(callbackmap_t& cbmap, int fd, callbackfunc_t toDo, const funcparam_t& parameter);
  40   virtual void removeFD(callbackmap_t& cbmap, int fd);
  41   string getName()
  42   {
  43     return "poll";
  44   }
  45 private:
  46 };
  47
  48 static FDMultiplexer* make()
  49 {
  50   return new PollFDMultiplexer();
  51 }
  52
  53 static struct RegisterOurselves
  54 {
  55   RegisterOurselves() {
  56     FDMultiplexer::getMultiplexerMap().insert(make_pair(1, &make));
  57   }
  58 } doIt;
  59
  60 void PollFDMultiplexer::addFD(callbackmap_t& cbmap, int fd, callbackfunc_t toDo, const boost::any& parameter)
  61 {
  62   Callback cb;
  63   cb.d_callback=toDo;
  64   cb.d_parameter=parameter;
  65   memset(&cb.d_ttd, 0, sizeof(cb.d_ttd));
  66   if(cbmap.count(fd))
  67     throw FDMultiplexerException("Tried to add fd "+std::to_string(fd)+ " to multiplexer twice");
  68   cbmap[fd]=cb;
  69 }
  70
  71 void PollFDMultiplexer::removeFD(callbackmap_t& cbmap, int fd)
  72 {
  73   if(d_inrun && d_iter->first==fd)  // trying to remove us!
  74     d_iter++;
  75
  76   if(!cbmap.erase(fd))
  77     throw FDMultiplexerException("Tried to remove unlisted fd "+std::to_string(fd)+ " from multiplexer");
  78 }
  79
  80 bool pollfdcomp(const struct pollfd& a, const struct pollfd& b)
  81 {
  82   return a.fd < b.fd;
  83 }
  84
  85 int PollFDMultiplexer::run(struct timeval* now, int timeout)
  86 {
  87   if(d_inrun) {
  88     throw FDMultiplexerException("FDMultiplexer::run() is not reentrant!\n");
  89   }
  90
  91   vector<struct pollfd> pollfds;
  92
  93   struct pollfd pollfd;
  94   for(callbackmap_t::const_iterator i=d_readCallbacks.begin(); i != d_readCallbacks.end(); ++i) {
  95     pollfd.fd = i->first;
  96     pollfd.events = POLLIN;
  97     pollfds.push_back(pollfd);
  98   }
  99
 100   for(callbackmap_t::const_iterator i=d_writeCallbacks.begin(); i != d_writeCallbacks.end(); ++i) {
 101     pollfd.fd = i->first;
 102     pollfd.events = POLLOUT;
 103     pollfds.push_back(pollfd);
 104   }
 105
 106   int ret=poll(&pollfds[0], pollfds.size(), timeout);
 107   gettimeofday(now, 0); // MANDATORY!
 108
 109   if(ret < 0 && errno!=EINTR)
 110     throw FDMultiplexerException("poll returned error: "+stringerror());
 111
 112   d_iter=d_readCallbacks.end();
 113   d_inrun=true;
 114
 115   for(unsigned int n = 0; n < pollfds.size(); ++n) {
 116     if(pollfds[n].revents == POLLIN) {
 117       d_iter=d_readCallbacks.find(pollfds[n].fd);
 118
 119       if(d_iter != d_readCallbacks.end()) {
 120         d_iter->second.d_callback(d_iter->first, d_iter->second.d_parameter);
 121         continue; // so we don't refind ourselves as writable!
 122       }
 123     }
 124     else if(pollfds[n].revents == POLLOUT) {
 125       d_iter=d_writeCallbacks.find(pollfds[n].fd);
 126
 127       if(d_iter != d_writeCallbacks.end()) {
 128         d_iter->second.d_callback(d_iter->first, d_iter->second.d_parameter);
 129       }
 130     }
 131   }
 132   d_inrun=false;
 133   return ret;
 134 }
 135
 136 #if 0
 137
 138 void acceptData(int fd, boost::any& parameter)
 139 {
 140   cout<<"Have data on fd "<<fd<<endl;
 141   Socket* sock=boost::any_cast<Socket*>(parameter);
 142   string packet;
 143   IPEndpoint rem;
 144   sock->recvFrom(packet, rem);
 145   cout<<"Received "<<packet.size()<<" bytes!\n";
 146 }
 147
 148
 149 int main()
 150 {
 151   Socket s(AF_INET, SOCK_DGRAM);
 152
 153   IPEndpoint loc("0.0.0.0", 2000);
 154   s.bind(loc);
 155
 156   PollFDMultiplexer sfm;
 157
 158   sfm.addReadFD(s.getHandle(), &acceptData, &s);
 159
 160   for(int n=0; n < 100 ; ++n) {
 161     sfm.run();
 162   }
 163   sfm.removeReadFD(s.getHandle());
 164   sfm.removeReadFD(s.getHandle());
 165 }
 166 #endif
 167