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__) || defined(__OpenBSD__) || defined(__APPLE__)
  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(unsigned int maxEventsHint);
  42   ~KqueueFDMultiplexer()
  43   {
  44     if (d_kqueuefd >= 0) {
  45       close(d_kqueuefd);
  46     }
  47   }
  48
  49   int run(struct timeval* tv, int timeout = 500) override;
  50   void getAvailableFDs(std::vector<int>& fds, int timeout) override;
  51
  52   void addFD(int fd, FDMultiplexer::EventKind kind) override;
  53   void removeFD(int fd, FDMultiplexer::EventKind kind) override;
  54
  55   string getName() const override
  56   {
  57     return "kqueue";
  58   }
  59
  60 private:
  61   int d_kqueuefd;
  62   std::vector<struct kevent> d_kevents;
  63 };
  64
  65 static FDMultiplexer* make(unsigned int maxEventsHint)
  66 {
  67   return new KqueueFDMultiplexer(maxEventsHint);
  68 }
  69
  70 static struct KqueueRegisterOurselves
  71 {
  72   KqueueRegisterOurselves()
  73   {
  74     FDMultiplexer::getMultiplexerMap().emplace(0, &make); // priority 0!
  75   }
  76 } kQueueDoIt;
  77
  78 KqueueFDMultiplexer::KqueueFDMultiplexer(unsigned int maxEventsHint) :
  79   d_kevents(maxEventsHint)
  80 {
  81   d_kqueuefd = kqueue();
  82   if (d_kqueuefd < 0) {
  83     throw FDMultiplexerException("Setting up kqueue: " + stringerror());
  84   }
  85 }
  86
  87 static uint32_t convertEventKind(FDMultiplexer::EventKind kind)
  88 {
  89   switch (kind) {
  90   case FDMultiplexer::EventKind::Read:
  91     return EVFILT_READ;
  92   case FDMultiplexer::EventKind::Write:
  93     return EVFILT_WRITE;
  94   case FDMultiplexer::EventKind::Both:
  95     throw std::runtime_error("Read and write events cannot be combined in one go with kqueue");
  96   }
  97
  98   throw std::runtime_error("Unhandled event kind in the kqueue multiplexer");
  99 }
 100
 101 void KqueueFDMultiplexer::addFD(int fd, FDMultiplexer::EventKind kind)
 102 {
 103   struct kevent kqevents[2];
 104   int nevents = 0;
 105
 106   if (kind == FDMultiplexer::EventKind::Both || kind == FDMultiplexer::EventKind::Read) {
 107     EV_SET(&kqevents[nevents], fd, convertEventKind(FDMultiplexer::EventKind::Read), EV_ADD, 0, 0, 0);
 108     nevents++;
 109   }
 110
 111   if (kind == FDMultiplexer::EventKind::Both || kind == FDMultiplexer::EventKind::Write) {
 112     EV_SET(&kqevents[nevents], fd, convertEventKind(FDMultiplexer::EventKind::Write), EV_ADD, 0, 0, 0);
 113     nevents++;
 114   }
 115
 116   if (kevent(d_kqueuefd, kqevents, nevents, 0, 0, 0) < 0) {
 117     throw FDMultiplexerException("Adding fd to kqueue set: " + stringerror());
 118   }
 119 }
 120
 121 void KqueueFDMultiplexer::removeFD(int fd, FDMultiplexer::EventKind kind)
 122 {
 123   struct kevent kqevents[2];
 124   int nevents = 0;
 125
 126   if (kind == FDMultiplexer::EventKind::Both || kind == FDMultiplexer::EventKind::Read) {
 127     EV_SET(&kqevents[nevents], fd, convertEventKind(FDMultiplexer::EventKind::Read), EV_DELETE, 0, 0, 0);
 128     nevents++;
 129   }
 130
 131   if (kind == FDMultiplexer::EventKind::Both || kind == FDMultiplexer::EventKind::Write) {
 132     EV_SET(&kqevents[nevents], fd, convertEventKind(FDMultiplexer::EventKind::Write), EV_DELETE, 0, 0, 0);
 133     nevents++;
 134   }
 135
 136   if (kevent(d_kqueuefd, kqevents, nevents, 0, 0, 0) < 0) {
 137     // ponder putting Callback back on the map..
 138     throw FDMultiplexerException("Removing fd from kqueue set: " + stringerror());
 139   }
 140 }
 141
 142 void KqueueFDMultiplexer::getAvailableFDs(std::vector<int>& fds, int timeout)
 143 {
 144   struct timespec ts;
 145   ts.tv_sec = timeout / 1000;
 146   ts.tv_nsec = (timeout % 1000) * 1000000;
 147
 148   int ret = kevent(d_kqueuefd, 0, 0, d_kevents.data(), d_kevents.size(), timeout != -1 ? &ts : nullptr);
 149
 150   if (ret < 0 && errno != EINTR) {
 151     throw FDMultiplexerException("kqueue returned error: " + stringerror());
 152   }
 153
 154   // we de-duplicate here, since if a descriptor is readable AND writable
 155   // we will get two events
 156   std::unordered_set<int> fdSet;
 157   fdSet.reserve(ret);
 158   for (int n = 0; n < ret; ++n) {
 159     fdSet.insert(d_kevents[n].ident);
 160   }
 161
 162   for (const auto fd : fdSet) {
 163     fds.push_back(fd);
 164   }
 165 }
 166
 167 int KqueueFDMultiplexer::run(struct timeval* now, int timeout)
 168 {
 169   if (d_inrun) {
 170     throw FDMultiplexerException("FDMultiplexer::run() is not reentrant!\n");
 171   }
 172
 173   struct timespec ts;
 174   ts.tv_sec = timeout / 1000;
 175   ts.tv_nsec = (timeout % 1000) * 1000000;
 176
 177   int ret = kevent(d_kqueuefd, 0, 0, d_kevents.data(), d_kevents.size(), timeout != -1 ? &ts : nullptr);
 178   gettimeofday(now, nullptr); // MANDATORY!
 179
 180   if (ret < 0 && errno != EINTR) {
 181     throw FDMultiplexerException("kqueue returned error: " + stringerror());
 182   }
 183
 184   if (ret < 0) {
 185     // nothing - thanks AB!
 186     return 0;
 187   }
 188
 189   d_inrun = true;
 190
 191   for (int n = 0; n < ret; ++n) {
 192     if (d_kevents[n].filter == EVFILT_READ) {
 193       const auto& iter = d_readCallbacks.find(d_kevents[n].ident);
 194       if (iter != d_readCallbacks.end()) {
 195         iter->d_callback(iter->d_fd, iter->d_parameter);
 196       }
 197     }
 198
 199     if (d_kevents[n].filter == EVFILT_WRITE) {
 200       const auto& iter = d_writeCallbacks.find(d_kevents[n].ident);
 201       if (iter != d_writeCallbacks.end()) {
 202         iter->d_callback(iter->d_fd, iter->d_parameter);
 203       }
 204     }
 205   }
 206
 207   d_inrun = false;
 208   return ret;
 209 }
 210
 211 #if 0
 212 void acceptData(int fd, boost::any& parameter)
 213 {
 214   cout<<"Have data on fd "<<fd<<endl;
 215   Socket* sock=boost::any_cast<Socket*>(parameter);
 216   string packet;
 217   IPEndpoint rem;
 218   sock->recvFrom(packet, rem);
 219   cout<<"Received "<<packet.size()<<" bytes!\n";
 220 }
 221
 222 int main()
 223 {
 224   Socket s(AF_INET, SOCK_DGRAM);
 225
 226   IPEndpoint loc("0.0.0.0", 2000);
 227   s.bind(loc);
 228
 229   KqueueFDMultiplexer sfm;
 230
 231   sfm.addReadFD(s.getHandle(), &acceptData, &s);
 232
 233   for(int n=0; n < 100 ; ++n) {
 234     sfm.run();
 235   }
 236   sfm.removeReadFD(s.getHandle());
 237   sfm.removeReadFD(s.getHandle());
 238 }
 239 #endif