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 edcec82b84335e56e17feda31a2481f726450f17..4f506fefc1ca517edb1d939dbbc7e84b31f467c5 100644 (file)
--- 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 e80f1fc06632f28de9c1268b95361e489748f9b9..88e58b7b784c2f5ced8b11feaa4bd2881b47c35b 100644 (file)
--- 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 34fcb8d798eb453735db3d1885ddbc5be3a1989d..447b45ea3bfa467aeafaadb20f1f9651a31207bc 100644 (file)
--- 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 d1c573a1efcc812a0a901875e81fa12ba58da2dc..6ea2e0ea85b7940eb8b3936ad0a8c776de19974b 100644 (file)
--- 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 3ee0a37524c709127139b6b2c22aa0468bd400fe..2f3c945912eb60d6e50322526eb5fa865078d096 100644 (file)
--- 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
   {