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