From: Remi Gacogne <remi.gacogne@powerdns.com>
Date: Wed, 4 Aug 2021 14:07:04 +0000 (+0200)
Subject: kqueue does not merge read and write events for the same descriptor
X-Git-Tag: dnsdist-1.7.0-alpha1~61^2~9
X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=510d368dcf16bb514a18172187a3b2737c72d9e8;p=thirdparty%2Fpdns.git

kqueue does not merge read and write events for the same descriptor

So we need to add, remove and process them as two separate events.
---

diff --git a/pdns/dnsdistdist/test-dnsdisttcp_cc.cc b/pdns/dnsdistdist/test-dnsdisttcp_cc.cc
index edcec82b84..4f506fefc1 100644
--- a/pdns/dnsdistdist/test-dnsdisttcp_cc.cc
+++ b/pdns/dnsdistdist/test-dnsdisttcp_cc.cc
@@ -349,10 +349,6 @@ public:
   {
   }
 
-  void alterFD(int fd, FDMultiplexer::EventKind kind) override
-  {
-  }
-
   string getName() const override
   {
     return "mockup";
diff --git a/pdns/epollmplexer.cc b/pdns/epollmplexer.cc
index e80f1fc066..88e58b7b78 100644
--- a/pdns/epollmplexer.cc
+++ b/pdns/epollmplexer.cc
@@ -49,7 +49,7 @@ public:
 
   void addFD(int fd, FDMultiplexer::EventKind kind) override;
   void removeFD(int fd, FDMultiplexer::EventKind kind) override;
-  void alterFD(int fd, FDMultiplexer::EventKind kind) override;
+  void alterFD(int fd, FDMultiplexer::EventKind from, FDMultiplexer::EventKind to) override;
 
   string getName() const override
   {
@@ -140,10 +140,10 @@ void EpollFDMultiplexer::removeFD(int fd, FDMultiplexer::EventKind)
   }
 }
 
-void EpollFDMultiplexer::alterFD(int fd, FDMultiplexer::EventKind kind)
+void EpollFDMultiplexer::alterFD(int fd, FDMultiplexer::EventKind, FDMultiplexer::EventKind to)
 {
   struct epoll_event eevent;
-  eevent.events = convertEventKind(kind);
+  eevent.events = convertEventKind(to);
   eevent.data.u64 = 0; // placate valgrind (I love it so much)
   eevent.data.fd = fd;
 
diff --git a/pdns/kqueuemplexer.cc b/pdns/kqueuemplexer.cc
index 34fcb8d798..447b45ea3b 100644
--- a/pdns/kqueuemplexer.cc
+++ b/pdns/kqueuemplexer.cc
@@ -95,26 +95,46 @@ static uint32_t convertEventKind(FDMultiplexer::EventKind kind)
   case FDMultiplexer::EventKind::Write:
     return EVFILT_WRITE;
   case FDMultiplexer::EventKind::Both:
-    return EVFILT_READ | EVFILT_WRITE;
+    throw std::runtime_error("Read and write events cannot be combined in one go with kqueue");
   }
 }
 
 void KqueueFDMultiplexer::addFD(int fd, FDMultiplexer::EventKind kind)
 {
-  struct kevent kqevent;
-  EV_SET(&kqevent, fd, convertEventKind(kind), EV_ADD, 0, 0, 0);
+  struct kevent kqevents[2];
+  int nevents = 0;
 
-  if (kevent(d_kqueuefd, &kqevent, 1, 0, 0, 0) < 0) {
+  if (kind == FDMultiplexer::EventKind::Both || kind == FDMultiplexer::EventKind::Read) {
+    EV_SET(&kqevents[nevents], fd, convertEventKind(FDMultiplexer::EventKind::Read), EV_ADD, 0, 0, 0);
+    nevents++;
+  }
+
+  if (kind == FDMultiplexer::EventKind::Both || kind == FDMultiplexer::EventKind::Write) {
+    EV_SET(&kqevents[nevents], fd, convertEventKind(FDMultiplexer::EventKind::Write), EV_ADD, 0, 0, 0);
+    nevents++;
+  }
+
+  if (kevent(d_kqueuefd, &kqevents, nevents, 0, 0, 0) < 0) {
     throw FDMultiplexerException("Adding fd to kqueue set: " + stringerror());
   }
 }
 
 void KqueueFDMultiplexer::removeFD(int fd, FDMultiplexer::EventKind kind)
 {
-  struct kevent kqevent;
-  EV_SET(&kqevent, fd, convertEventKind(kind), EV_DELETE, 0, 0, 0);
+  struct kevent kqevents[2];
+  int nevents = 0;
+
+  if (kind == FDMultiplexer::EventKind::Both || kind == FDMultiplexer::EventKind::Read) {
+    EV_SET(&kqevents[nevents], fd, convertEventKind(FDMultiplexer::EventKind::Read), EV_DELETE, 0, 0, 0);
+    nevents++;
+  }
+
+  if (kind == FDMultiplexer::EventKind::Both || kind == FDMultiplexer::EventKind::Write) {
+    EV_SET(&kqevents[nevents], fd, convertEventKind(FDMultiplexer::EventKind::Write), EV_DELETE, 0, 0, 0);
+    nevents++;
+  }
 
-  if (kevent(d_kqueuefd, &kqevent, 1, 0, 0, 0) < 0) {
+  if (kevent(d_kqueuefd, &kqevents, nevents, 0, 0, 0) < 0) {
     // ponder putting Callback back on the map..
     throw FDMultiplexerException("Removing fd from kqueue set: " + stringerror());
   }
@@ -132,8 +152,16 @@ void KqueueFDMultiplexer::getAvailableFDs(std::vector<int>& fds, int timeout)
     throw FDMultiplexerException("kqueue returned error: " + stringerror());
   }
 
+  // we de-duplicate here, since if a descriptor is readable AND writable
+  // we will get two events
+  std::unordered_set<int> fdSet;
+  fdSet.reserve(n);
   for (int n = 0; n < ret; ++n) {
-    fds.push_back(d_kevents[n].ident);
+    fdSet.insert(d_kevents[n].ident);
+  }
+
+  for (const auto fd : fdSet) {
+    fds.push_back(fd);
   }
 }
 
diff --git a/pdns/mplexer.hh b/pdns/mplexer.hh
index d1c573a1ef..6ea2e0ea85 100644
--- a/pdns/mplexer.hh
+++ b/pdns/mplexer.hh
@@ -97,7 +97,7 @@ public:
     bool alreadyWatched = d_writeCallbacks.count(fd) > 0;
 
     if (alreadyWatched) {
-      this->alterFD(fd, EventKind::Both);
+      this->alterFD(fd, EventKind::Write, EventKind::Both);
     }
     else {
       this->addFD(fd, EventKind::Read);
@@ -113,7 +113,7 @@ public:
     bool alreadyWatched = d_readCallbacks.count(fd) > 0;
 
     if (alreadyWatched) {
-      this->alterFD(fd, EventKind::Both);
+      this->alterFD(fd, EventKind::Read, EventKind::Both);
     }
     else {
       this->addFD(fd, EventKind::Write);
@@ -131,7 +131,7 @@ public:
     accountingRemoveFD(d_readCallbacks, fd);
 
     if (iter != d_writeCallbacks.end()) {
-      this->alterFD(fd, EventKind::Write);
+      this->alterFD(fd, EventKind::Both, EventKind::Write);
     }
     else {
       this->removeFD(fd, EventKind::Read);
@@ -146,7 +146,7 @@ public:
     accountingRemoveFD(d_writeCallbacks, fd);
 
     if (iter != d_readCallbacks.end()) {
-      this->alterFD(fd, EventKind::Read);
+      this->alterFD(fd, EventKind::Both, EventKind::Read);
     }
     else {
       this->removeFD(fd, EventKind::Write);
@@ -182,14 +182,14 @@ public:
   void alterFDToRead(int fd, callbackfunc_t toDo, const funcparam_t& parameter = funcparam_t(), const struct timeval* ttd = nullptr)
   {
     accountingRemoveFD(d_writeCallbacks, fd);
-    this->alterFD(fd, EventKind::Read);
+    this->alterFD(fd, EventKind::Write, EventKind::Read);
     accountingAddFD(d_readCallbacks, fd, toDo, parameter, ttd);
   }
 
   void alterFDToWrite(int fd, callbackfunc_t toDo, const funcparam_t& parameter = funcparam_t(), const struct timeval* ttd = nullptr)
   {
     accountingRemoveFD(d_readCallbacks, fd);
-    this->alterFD(fd, EventKind::Write);
+    this->alterFD(fd, EventKind::Read, EventKind::Write);
     accountingAddFD(d_writeCallbacks, fd, toDo, parameter, ttd);
   }
 
@@ -303,10 +303,11 @@ protected:
   virtual void addFD(int fd, EventKind kind) = 0;
   /* most implementations do not care about which event has to be removed, except for kqueue */
   virtual void removeFD(int fd, EventKind kind) = 0;
-  virtual void alterFD(int fd, EventKind kind)
+  /* most implementations do not care about which event has to be removed, except for kqueue */
+  virtual void alterFD(int fd, EventKind from, EventKind to)
   {
     /* naive implementation */
-    removeFD(fd, (kind == EventKind::Write) ? EventKind::Read : EventKind::Write);
-    addFD(fd, kind);
+    removeFD(fd, from);
+    addFD(fd, to);
   }
 };
diff --git a/pdns/portsmplexer.cc b/pdns/portsmplexer.cc
index 3ee0a37524..2f3c945912 100644
--- a/pdns/portsmplexer.cc
+++ b/pdns/portsmplexer.cc
@@ -28,7 +28,6 @@ public:
 
   void addFD(int fd, FDMultiplexer::EventKind kind) override;
   void removeFD(int fd, FDMultiplexer::EventKind kind) override;
-  void alterFD(int fd, FDMultiplexer::EventKind kind) override;
 
   string getName() const override
   {