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