From ee271fc486ba93bd8bee65197b54f7eeaa73fe14 Mon Sep 17 00:00:00 2001
From: Remi Gacogne <remi.gacogne@powerdns.com>
Date: Wed, 13 Mar 2019 11:35:42 +0100
Subject: [PATCH] rec: Add a distribution-pipe-buffer-size setting

---
 pdns/misc.cc                        | 33 +++++++++++++++++++++++++++++
 pdns/misc.hh                        |  5 ++++-
 pdns/pdns_recursor.cc               | 16 ++++++++++++++
 pdns/recursordist/docs/settings.rst | 15 +++++++++++++
 4 files changed, 68 insertions(+), 1 deletion(-)

diff --git a/pdns/misc.cc b/pdns/misc.cc
index 9b6eea36d5..1f90bbcbd1 100644
--- a/pdns/misc.cc
+++ b/pdns/misc.cc
@@ -1485,3 +1485,36 @@ std::vector<ComboAddress> getResolvers(const std::string& resolvConfPath)
 
   return results;
 }
+
+size_t getPipeBufferSize(int fd)
+{
+#ifdef F_GETPIPE_SZ
+  int res = fcntl(fd, F_GETPIPE_SZ);
+  if (res == -1) {
+    return 0;
+  }
+  return res;
+#else
+  errno = ENOSYS;
+  return 0;
+#endif /* F_GETPIPE_SZ */
+}
+
+bool setPipeBufferSize(int fd, size_t size)
+{
+#ifdef F_SETPIPE_SZ
+  if (size > std::numeric_limits<int>::max()) {
+    errno = EINVAL;
+    return false;
+  }
+  int newSize = static_cast<int>(size);
+  int res = fcntl(fd, F_SETPIPE_SZ, newSize);
+  if (res == -1) {
+    return false;
+  }
+  return true;
+#else
+  errno = ENOSYS;
+  return false;
+#endif /* F_SETPIPE_SZ */
+}
diff --git a/pdns/misc.hh b/pdns/misc.hh
index 9adfe35dc4..ee255e9b90 100644
--- a/pdns/misc.hh
+++ b/pdns/misc.hh
@@ -537,8 +537,11 @@ bool isNonBlocking(int sock);
 bool setReceiveSocketErrors(int sock, int af);
 int closesocket(int fd);
 bool setCloseOnExec(int sock);
-uint64_t udpErrorStats(const std::string& str);
 
+size_t getPipeBufferSize(int fd);
+bool setPipeBufferSize(int fd, size_t size);
+
+uint64_t udpErrorStats(const std::string& str);
 uint64_t getRealMemoryUsage(const std::string&);
 uint64_t getSpecialMemoryUsage(const std::string&);
 uint64_t getOpenFileDescriptors(const std::string&);
diff --git a/pdns/pdns_recursor.cc b/pdns/pdns_recursor.cc
index d0b8b121c2..1e171080c0 100644
--- a/pdns/pdns_recursor.cc
+++ b/pdns/pdns_recursor.cc
@@ -2781,6 +2781,11 @@ static void houseKeeping(void *)
 
 static void makeThreadPipes()
 {
+  auto pipeBufferSize = ::arg().asNum("distribution-pipe-buffer-size");
+  if (pipeBufferSize > 0) {
+    g_log<<Logger::Info<<"Resizing the buffer of the distribution pipe to "<<pipeBufferSize<<endl;
+  }
+
   /* thread 0 is the handler / SNMP, we start at 1 */
   for(unsigned int n = 1; n <= (g_numWorkerThreads + g_numDistributorThreads); ++n) {
     auto& threadInfos = s_threadInfos.at(n);
@@ -2804,6 +2809,16 @@ static void makeThreadPipes()
     threadInfos.pipes.readQueriesToThread = fd[0];
     threadInfos.pipes.writeQueriesToThread = fd[1];
 
+    if (pipeBufferSize > 0) {
+      if (!setPipeBufferSize(threadInfos.pipes.writeQueriesToThread, pipeBufferSize)) {
+        g_log<<Logger::Warning<<"Error resizing the buffer of the distribution pipe for thread "<<n<<" to "<<pipeBufferSize<<": "<<strerror(errno)<<endl;
+        auto existingSize = getPipeBufferSize(threadInfos.pipes.writeQueriesToThread);
+        if (existingSize > 0) {
+          g_log<<Logger::Warning<<"The current size of the distribution pipe's buffer for thread "<<n<<" is "<<existingSize<<endl;
+        }
+      }
+    }
+
     if (!setNonBlocking(threadInfos.pipes.writeQueriesToThread)) {
       unixDie("Making pipe for inter-thread communications non-blocking");
     }
@@ -4402,6 +4417,7 @@ int main(int argc, char **argv)
     ::arg().set("max-recursion-depth", "Maximum number of internal recursion calls per query, 0 for unlimited")="40";
     ::arg().set("max-udp-queries-per-round", "Maximum number of UDP queries processed per recvmsg() round, before returning back to normal processing")="10000";
     ::arg().set("protobuf-use-kernel-timestamp", "Compute the latency of queries in protobuf messages by using the timestamp set by the kernel when the query was received (when available)")="";
+    ::arg().set("distribution-pipe-buffer-size", "Size in bytes of the internal buffer of the pipe used by the distributor to pass incoming queries to a worker thread")="0";
 
     ::arg().set("include-dir","Include *.conf files from this directory")="";
     ::arg().set("security-poll-suffix","Domain name from which to query security update notifications")="secpoll.powerdns.com.";
diff --git a/pdns/recursordist/docs/settings.rst b/pdns/recursordist/docs/settings.rst
index 5bb081fa83..5405bb3a09 100644
--- a/pdns/recursordist/docs/settings.rst
+++ b/pdns/recursordist/docs/settings.rst
@@ -331,6 +331,21 @@ average load. This helps making sure that all the workers have roughly the same
 share of queries, even if the incoming traffic is very skewed, with a larger
 number of requests asking for the same qname.
 
+.. _setting-distribution-pipe-buffer-size:
+
+``distribution-pipe-buffer-size``
+---------------------------------
+.. versionadded:: 4.2.0
+
+-  Integer
+-  Default: 0
+
+Size in bytes of the internal buffer of the pipe used by the distributor to pass incoming queries to a worker thread.
+Requires support for `F_SETPIPE_SZ` which is present in Linux since 2.6.35. The actual size might be rounded up to
+a multiple of a page size. 0 means that the OS default size is used.
+A large buffer might allow the recursor to deal with very short-lived load spikes during which a worker thread gets
+overloaded, but it will be at the cost of an increased latency.
+
 .. _setting-distributor-threads:
 
 ``distributor-threads``
-- 
2.47.2