pdns/kqueuemplexer.cc

   1 /*
   2  * This file is part of PowerDNS or dnsdist.
   3  * Copyright -- PowerDNS.COM B.V. and its contributors
   4  *
   5  * This program is free software; you can redistribute it and/or modify
   6  * it under the terms of version 2 of the GNU General Public License as
   7  * published by the Free Software Foundation.
   8  *
   9  * In addition, for the avoidance of any doubt, permission is granted to
  10  * link this program with OpenSSL and to (re)distribute the binaries
  11  * produced as the result of such linking.
  12  *
  13  * This program is distributed in the hope that it will be useful,
  14  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  15  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  16  * GNU General Public License for more details.
  17  *
  18  * You should have received a copy of the GNU General Public License
  19  * along with this program; if not, write to the Free Software
  20  * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  21  */
  22 #ifdef HAVE_CONFIG_H
  23 #include "config.h"
  24 #endif
  25 #include "mplexer.hh"
  26 #include "sstuff.hh"
  27 #include <iostream>
  28 #include <unistd.h>
  29 #include "misc.hh"
  30 #include <sys/types.h>
  31 #if defined(__FreeBSD__) || defined(__FreeBSD_kernel__) || defined(__NetBSD__)
  32 #include <sys/event.h>
  33 #endif
  34 #include <sys/time.h>
  35
  36 #include "namespaces.hh"
  37
  38 class KqueueFDMultiplexer : public FDMultiplexer
  39 {
  40 public:
  41   KqueueFDMultiplexer();
  42   virtual ~KqueueFDMultiplexer()
  43   {
  44     close(d_kqueuefd);
  45   }
  46
  47   virtual int run(struct timeval* tv, int timeout=500) override;
  48   virtual void getAvailableFDs(std::vector<int>& fds, int timeout) override;
  49
  50   virtual void addFD(callbackmap_t& cbmap, int fd, callbackfunc_t toDo, const boost::any& parameter, const struct timeval* ttd=nullptr) override;
  51   virtual void removeFD(callbackmap_t& cbmap, int fd) override;
  52   string getName() const override
  53   {
  54     return "kqueue";
  55   }
  56 private:
  57   int d_kqueuefd;
  58   boost::shared_array<struct kevent> d_kevents;
  59   static unsigned int s_maxevents; // not a hard maximum
  60 };
  61
  62 unsigned int KqueueFDMultiplexer::s_maxevents=1024;
  63
  64 static FDMultiplexer* make()
  65 {
  66   return new KqueueFDMultiplexer();
  67 }
  68
  69 static struct KqueueRegisterOurselves
  70 {
  71   KqueueRegisterOurselves() {
  72     FDMultiplexer::getMultiplexerMap().insert(make_pair(0, &make)); // priority 0!
  73   }
  74 } kQueuedoIt;
  75
  76 KqueueFDMultiplexer::KqueueFDMultiplexer() : d_kevents(new struct kevent[s_maxevents])
  77 {
  78   d_kqueuefd=kqueue();
  79   if(d_kqueuefd < 0)
  80     throw FDMultiplexerException("Setting up kqueue: "+stringerror());
  81 }
  82
  83 void KqueueFDMultiplexer::addFD(callbackmap_t& cbmap, int fd, callbackfunc_t toDo, const boost::any& parameter, const struct timeval* ttd)
  84 {
  85   accountingAddFD(cbmap, fd, toDo, parameter, ttd);
  86
  87   struct kevent kqevent;
  88   EV_SET(&kqevent, fd, (&cbmap == &d_readCallbacks) ? EVFILT_READ : EVFILT_WRITE, EV_ADD, 0,0,0);
  89
  90   if(kevent(d_kqueuefd, &kqevent, 1, 0, 0, 0) < 0) {
  91     cbmap.erase(fd);
  92     throw FDMultiplexerException("Adding fd to kqueue set: "+stringerror());
  93   }
  94 }
  95
  96 void KqueueFDMultiplexer::removeFD(callbackmap_t& cbmap, int fd)
  97 {
  98   accountingRemoveFD(cbmap, fd);
  99
 100   struct kevent kqevent;
 101   EV_SET(&kqevent, fd, (&cbmap == &d_readCallbacks) ? EVFILT_READ : EVFILT_WRITE, EV_DELETE, 0,0,0);
 102
 103   if(kevent(d_kqueuefd, &kqevent, 1, 0, 0, 0) < 0) // ponder putting Callback back on the map..
 104     throw FDMultiplexerException("Removing fd from kqueue set: "+stringerror());
 105 }
 106
 107 void KqueueFDMultiplexer::getAvailableFDs(std::vector<int>& fds, int timeout)
 108 {
 109   struct timespec ts;
 110   ts.tv_sec=timeout/1000;
 111   ts.tv_nsec=(timeout % 1000) * 1000000;
 112
 113   int ret = kevent(d_kqueuefd, 0, 0, d_kevents.get(), s_maxevents, &ts);
 114
 115   if(ret < 0 && errno != EINTR)
 116     throw FDMultiplexerException("kqueue returned error: "+stringerror());
 117
 118   for(int n=0; n < ret; ++n) {
 119     fds.push_back(d_kevents[n].ident);
 120   }
 121 }
 122
 123 int KqueueFDMultiplexer::run(struct timeval* now, int timeout)
 124 {
 125   if(d_inrun) {
 126     throw FDMultiplexerException("FDMultiplexer::run() is not reentrant!\n");
 127   }
 128
 129   struct timespec ts;
 130   ts.tv_sec=timeout/1000;
 131   ts.tv_nsec=(timeout % 1000) * 1000000;
 132
 133   int ret=kevent(d_kqueuefd, 0, 0, d_kevents.get(), s_maxevents, &ts);
 134   gettimeofday(now,0); // MANDATORY!
 135
 136   if(ret < 0 && errno!=EINTR)
 137     throw FDMultiplexerException("kqueue returned error: "+stringerror());
 138
 139   if(ret < 0) // nothing - thanks AB!
 140     return 0;
 141
 142   d_inrun=true;
 143
 144   for(int n=0; n < ret; ++n) {
 145     d_iter=d_readCallbacks.find(d_kevents[n].ident);
 146     if(d_iter != d_readCallbacks.end()) {
 147       d_iter->d_callback(d_iter->d_fd, d_iter->d_parameter);
 148       continue; // so we don't find ourselves as writable again
 149     }
 150
 151     d_iter=d_writeCallbacks.find(d_kevents[n].ident);
 152
 153     if(d_iter != d_writeCallbacks.end()) {
 154       d_iter->d_callback(d_iter->d_fd, d_iter->d_parameter);
 155     }
 156   }
 157
 158   d_inrun=false;
 159   return ret;
 160 }
 161
 162 #if 0
 163 void acceptData(int fd, boost::any& parameter)
 164 {
 165   cout<<"Have data on fd "<<fd<<endl;
 166   Socket* sock=boost::any_cast<Socket*>(parameter);
 167   string packet;
 168   IPEndpoint rem;
 169   sock->recvFrom(packet, rem);
 170   cout<<"Received "<<packet.size()<<" bytes!\n";
 171 }
 172
 173 int main()
 174 {
 175   Socket s(AF_INET, SOCK_DGRAM);
 176
 177   IPEndpoint loc("0.0.0.0", 2000);
 178   s.bind(loc);
 179
 180   KqueueFDMultiplexer sfm;
 181
 182   sfm.addReadFD(s.getHandle(), &acceptData, &s);
 183
 184   for(int n=0; n < 100 ; ++n) {
 185     sfm.run();
 186   }
 187   sfm.removeReadFD(s.getHandle());
 188   sfm.removeReadFD(s.getHandle());
 189 }
 190 #endif
 191
 192
 193