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) override;
  38   virtual void getAvailableFDs(std::vector<int>& fds, int timeout) override;
  39
  40   virtual void addFD(callbackmap_t& cbmap, int fd, callbackfunc_t toDo, const funcparam_t& parameter, const struct timeval* ttd=nullptr) override;
  41   virtual void removeFD(callbackmap_t& cbmap, int fd) override;
  42
  43   string getName() const override
  44   {
  45     return "poll";
  46   }
  47 private:
  48   vector<struct pollfd> preparePollFD() const;
  49 };
  50
  51 static FDMultiplexer* make()
  52 {
  53   return new PollFDMultiplexer();
  54 }
  55
  56 static struct RegisterOurselves
  57 {
  58   RegisterOurselves() {
  59     FDMultiplexer::getMultiplexerMap().insert(make_pair(1, &make));
  60   }
  61 } doIt;
  62
  63 void PollFDMultiplexer::addFD(callbackmap_t& cbmap, int fd, callbackfunc_t toDo, const boost::any& parameter, const struct timeval* ttd)
  64 {
  65   accountingAddFD(cbmap, fd, toDo, parameter, ttd);
  66 }
  67
  68 void PollFDMultiplexer::removeFD(callbackmap_t& cbmap, int fd)
  69 {
  70   if(d_inrun && d_iter->d_fd==fd)  // trying to remove us!
  71     ++d_iter;
  72
  73   if(!cbmap.erase(fd))
  74     throw FDMultiplexerException("Tried to remove unlisted fd "+std::to_string(fd)+ " from multiplexer");
  75 }
  76
  77 vector<struct pollfd> PollFDMultiplexer::preparePollFD() const
  78 {
  79   vector<struct pollfd> pollfds;
  80   pollfds.reserve(d_readCallbacks.size() + d_writeCallbacks.size());
  81
  82   struct pollfd pollfd;
  83   for(const auto& cb : d_readCallbacks) {
  84     pollfd.fd = cb.d_fd;
  85     pollfd.events = POLLIN;
  86     pollfds.push_back(pollfd);
  87   }
  88
  89   for(const auto& cb : d_writeCallbacks) {
  90     pollfd.fd = cb.d_fd;
  91     pollfd.events = POLLOUT;
  92     pollfds.push_back(pollfd);
  93   }
  94
  95   return pollfds;
  96 }
  97
  98 void PollFDMultiplexer::getAvailableFDs(std::vector<int>& fds, int timeout)
  99 {
 100   auto pollfds = preparePollFD();
 101   int ret = poll(&pollfds[0], pollfds.size(), timeout);
 102
 103   if (ret < 0 && errno != EINTR)
 104     throw FDMultiplexerException("poll returned error: " + stringerror());
 105
 106   for(const auto& pollfd : pollfds) {
 107     if (pollfd.revents & POLLIN || pollfd.revents & POLLOUT) {
 108       fds.push_back(pollfd.fd);
 109     }
 110   }
 111 }
 112
 113 int PollFDMultiplexer::run(struct timeval* now, int timeout)
 114 {
 115   if(d_inrun) {
 116     throw FDMultiplexerException("FDMultiplexer::run() is not reentrant!\n");
 117   }
 118
 119   auto pollfds = preparePollFD();
 120
 121   int ret=poll(&pollfds[0], pollfds.size(), timeout);
 122   gettimeofday(now, 0); // MANDATORY!
 123
 124   if(ret < 0 && errno!=EINTR)
 125     throw FDMultiplexerException("poll returned error: "+stringerror());
 126
 127   d_iter=d_readCallbacks.end();
 128   d_inrun=true;
 129
 130   for(const auto& pollfd : pollfds) {
 131     if(pollfd.revents & POLLIN) {
 132       d_iter=d_readCallbacks.find(pollfd.fd);
 133
 134       if(d_iter != d_readCallbacks.end()) {
 135         d_iter->d_callback(d_iter->d_fd, d_iter->d_parameter);
 136         continue; // so we don't refind ourselves as writable!
 137       }
 138     }
 139     else if(pollfd.revents & POLLOUT) {
 140       d_iter=d_writeCallbacks.find(pollfd.fd);
 141
 142       if(d_iter != d_writeCallbacks.end()) {
 143         d_iter->d_callback(d_iter->d_fd, d_iter->d_parameter);
 144       }
 145     }
 146   }
 147   d_inrun=false;
 148   return ret;
 149 }
 150
 151 #if 0
 152
 153 void acceptData(int fd, boost::any& parameter)
 154 {
 155   cout<<"Have data on fd "<<fd<<endl;
 156   Socket* sock=boost::any_cast<Socket*>(parameter);
 157   string packet;
 158   IPEndpoint rem;
 159   sock->recvFrom(packet, rem);
 160   cout<<"Received "<<packet.size()<<" bytes!\n";
 161 }
 162
 163
 164 int main()
 165 {
 166   Socket s(AF_INET, SOCK_DGRAM);
 167
 168   IPEndpoint loc("0.0.0.0", 2000);
 169   s.bind(loc);
 170
 171   PollFDMultiplexer sfm;
 172
 173   sfm.addReadFD(s.getHandle(), &acceptData, &s);
 174
 175   for(int n=0; n < 100 ; ++n) {
 176     sfm.run();
 177   }
 178   sfm.removeReadFD(s.getHandle());
 179   sfm.removeReadFD(s.getHandle());
 180 }
 181 #endif
 182