From be55a20ce9bb7140071279d70bcb460f1f2b7b7d Mon Sep 17 00:00:00 2001
From: Remi Gacogne <remi.gacogne@powerdns.com>
Date: Mon, 10 Aug 2020 15:51:18 +0200
Subject: [PATCH] dnsdist: Stop the related responder thread when a backend is
 removed

---
 pdns/dnsdist-lua.cc                 |  1 +
 pdns/dnsdist.cc                     |  9 +++++++--
 pdns/dnsdist.hh                     | 17 ++++++++--------
 pdns/dnsdistdist/dnsdist-backend.cc | 30 +++++++++++++++++++++++++++++
 4 files changed, 46 insertions(+), 11 deletions(-)

diff --git a/pdns/dnsdist-lua.cc b/pdns/dnsdist-lua.cc
index c6477f17f8..b46a04ca46 100644
--- a/pdns/dnsdist-lua.cc
+++ b/pdns/dnsdist-lua.cc
@@ -558,6 +558,7 @@ static void setupLuaConfig(bool client, bool configCheck)
                         g_pools.setState(localPools);
                         states.erase(remove(states.begin(), states.end(), server), states.end());
                         g_dstates.setState(states);
+                        server->stop();
                       } );
 
   g_lua.writeFunction("truncateTC", [](bool tc) { setLuaSideEffect(); g_truncateTC=tc; });
diff --git a/pdns/dnsdist.cc b/pdns/dnsdist.cc
index 9346c386e1..ecf659952b 100644
--- a/pdns/dnsdist.cc
+++ b/pdns/dnsdist.cc
@@ -552,7 +552,7 @@ try {
   std::vector<int> sockets;
   sockets.reserve(dss->sockets.size());
 
-  for(;;) {
+  for(; !dss->isStopped(); ) {
     dnsheader* dh = reinterpret_cast<struct dnsheader*>(packet);
     try {
       pickBackendSocketsReadyForReceiving(dss, sockets);
@@ -561,8 +561,13 @@ try {
         char * response = packet;
         size_t responseSize = sizeof(packet);
 
-        if (got < 0 || static_cast<size_t>(got) < sizeof(dnsheader))
+        if (got == 0 && dss->isStopped()) {
+          break;
+        }
+
+        if (got < 0 || static_cast<size_t>(got) < sizeof(dnsheader)) {
           continue;
+        }
 
         uint16_t responseLen = static_cast<uint16_t>(got);
         queryId = dh->id;
diff --git a/pdns/dnsdist.hh b/pdns/dnsdist.hh
index 04e280cca9..9c6dcd16c0 100644
--- a/pdns/dnsdist.hh
+++ b/pdns/dnsdist.hh
@@ -765,15 +765,8 @@ struct DownstreamState
 
   DownstreamState(const ComboAddress& remote_, const ComboAddress& sourceAddr_, unsigned int sourceItf, const std::string& sourceItfName, size_t numberOfSockets, bool connect);
   DownstreamState(const ComboAddress& remote_): DownstreamState(remote_, ComboAddress(), 0, std::string(), 1, true) {}
-  ~DownstreamState()
-  {
-    for (auto& fd : sockets) {
-      if (fd >= 0) {
-        close(fd);
-        fd = -1;
-      }
-    }
-  }
+  ~DownstreamState();
+
   boost::uuids::uuid id;
   std::vector<unsigned int> hashes;
   mutable ReadWriteLock d_lock;
@@ -883,6 +876,11 @@ struct DownstreamState
   void hash();
   void setId(const boost::uuids::uuid& newId);
   void setWeight(int newWeight);
+  void stop();
+  bool isStopped() const
+  {
+    return d_stopped;
+  }
 
   void updateTCPMetrics(size_t nbQueries, uint64_t durationMs)
   {
@@ -892,6 +890,7 @@ struct DownstreamState
 private:
   std::string name;
   std::string nameWithAddr;
+  bool d_stopped{false};
 };
 using servers_t =vector<std::shared_ptr<DownstreamState>>;
 
diff --git a/pdns/dnsdistdist/dnsdist-backend.cc b/pdns/dnsdistdist/dnsdist-backend.cc
index 2f7a7dbff8..38398e28eb 100644
--- a/pdns/dnsdistdist/dnsdist-backend.cc
+++ b/pdns/dnsdistdist/dnsdist-backend.cc
@@ -98,6 +98,21 @@ bool DownstreamState::reconnect()
 
   return connected;
 }
+
+void DownstreamState::stop()
+{
+  std::unique_lock<std::mutex> tl(connectLock);
+  std::lock_guard<std::mutex> slock(socketsLock);
+  d_stopped = true;
+
+  for (auto& fd : sockets) {
+    if (fd != -1) {
+      /* shutdown() is needed to wake up recv() in the responderThread */
+      shutdown(fd, SHUT_RDWR);
+    }
+  }
+}
+
 void DownstreamState::hash()
 {
   vinfolog("Computing hashes for id=%s and weight=%d", id, weight);
@@ -152,5 +167,20 @@ DownstreamState::DownstreamState(const ComboAddress& remote_, const ComboAddress
     idStates.resize(g_maxOutstanding);
     sw.start();
   }
+}
+
+DownstreamState::~DownstreamState()
+{
+  for (auto& fd : sockets) {
+    if (fd >= 0) {
+      close(fd);
+      fd = -1;
+    }
+  }
 
+  // we need to either detach or join the thread before it
+  // is destroyed
+  if (threadStarted.test_and_set()) {
+    tid.detach();
+  }
 }
-- 
2.47.2