dnsdist: Stop the related responder thread when a backend is removed

(cherry picked from commit be55a20ce9bb7140071279d70bcb460f1f2b7b7d)

diff --git a/pdns/dnsdist-lua.cc b/pdns/dnsdist-lua.cc
index 8c4995e01b1321f80c3a845a8778b9b4b2164b9d..33260e69ad99cb64e18df138cbb36dc2fa41e158 100644 (file)
--- a/pdns/dnsdist-lua.cc
+++ b/pdns/dnsdist-lua.cc
@@ -554,6 +554,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 9346c386e1aa24e23e8c928803872db3720860ed..ecf659952b5fd9e8606963dca9d8be08e46a3e12 100644 (file)
--- 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 f0c98a94cfdbb09fcbe975cffe57f49907a663f0..9188eb48ff996ce91761d75f406dace051772fff 100644 (file)
--- 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;
@@ -882,6 +875,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)
   {
@@ -891,6 +889,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 2f7a7dbff822b531cbfd43495f434561d75a0c64..38398e28eb11406b3128e5d2d57481d994300943 100644 (file)
--- 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();
+  }
 }