From: Remi Gacogne <remi.gacogne@powerdns.com>
Date: Thu, 2 Jan 2020 16:09:44 +0000 (+0100)
Subject: dnsdist: Implement fast-{c,w}hashed FFI policies
X-Git-Tag: auth-4.3.0-beta2~1^2~10
X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=cc958b79aaa65137e0c9a90f74f3af562556acb8;p=thirdparty%2Fpdns.git

dnsdist: Implement fast-{c,w}hashed FFI policies
---

diff --git a/pdns/dnsdist-lbpolicies.hh b/pdns/dnsdist-lbpolicies.hh
index 90c208da37..587c22d956 100644
--- a/pdns/dnsdist-lbpolicies.hh
+++ b/pdns/dnsdist-lbpolicies.hh
@@ -71,6 +71,8 @@ std::shared_ptr<DownstreamState> firstAvailable(const ServerPolicy::NumberedServ
 std::shared_ptr<DownstreamState> leastOutstanding(const ServerPolicy::NumberedServerVector& servers, const DNSQuestion* dq);
 std::shared_ptr<DownstreamState> wrandom(const ServerPolicy::NumberedServerVector& servers, const DNSQuestion* dq);
 std::shared_ptr<DownstreamState> whashed(const ServerPolicy::NumberedServerVector& servers, const DNSQuestion* dq);
+std::shared_ptr<DownstreamState> whashedFromHash(const ServerPolicy::NumberedServerVector& servers, size_t hash);
 std::shared_ptr<DownstreamState> chashed(const ServerPolicy::NumberedServerVector& servers, const DNSQuestion* dq);
+std::shared_ptr<DownstreamState> chashedFromHash(const ServerPolicy::NumberedServerVector& servers, size_t hash);
 std::shared_ptr<DownstreamState> roundrobin(const ServerPolicy::NumberedServerVector& servers, const DNSQuestion* dq);
 std::shared_ptr<DownstreamState> getSelectedBackendFromPolicy(const ServerPolicy& policy, const ServerPolicy::NumberedServerVector& servers, DNSQuestion& dq);
diff --git a/pdns/dnsdistdist/dnsdist-lbpolicies.cc b/pdns/dnsdistdist/dnsdist-lbpolicies.cc
index 4b707dcaa2..095ad9620f 100644
--- a/pdns/dnsdistdist/dnsdist-lbpolicies.cc
+++ b/pdns/dnsdistdist/dnsdist-lbpolicies.cc
@@ -59,9 +59,10 @@ shared_ptr<DownstreamState> firstAvailable(const ServerPolicy::NumberedServerVec
   return leastOutstanding(servers, dq);
 }
 
-static shared_ptr<DownstreamState> valrandom(unsigned int val, const ServerPolicy::NumberedServerVector& servers, const DNSQuestion* dq)
+static shared_ptr<DownstreamState> valrandom(unsigned int val, const ServerPolicy::NumberedServerVector& servers)
 {
-  vector<pair<int, shared_ptr<DownstreamState>>> poss;
+  vector<pair<int, size_t>> poss;
+  poss.reserve(servers.size());
   int sum = 0;
   int max = std::numeric_limits<int>::max();
 
@@ -74,36 +75,44 @@ static shared_ptr<DownstreamState> valrandom(unsigned int val, const ServerPolic
         sum += d.second->weight;
       }
 
-      poss.push_back({sum, d.second});
+      poss.emplace_back(sum, d.first);
     }
   }
 
   // Catch poss & sum are empty to avoid SIGFPE
-  if(poss.empty())
+  if (poss.empty()) {
     return shared_ptr<DownstreamState>();
+  }
 
   int r = val % sum;
   auto p = upper_bound(poss.begin(), poss.end(),r, [](int r_, const decltype(poss)::value_type& a) { return  r_ < a.first;});
-  if(p==poss.end())
+  if (p == poss.end()) {
     return shared_ptr<DownstreamState>();
-  return p->second;
+  }
+
+  return servers.at(p->second - 1).second;
 }
 
 shared_ptr<DownstreamState> wrandom(const ServerPolicy::NumberedServerVector& servers, const DNSQuestion* dq)
 {
-  return valrandom(random(), servers, dq);
+  return valrandom(random(), servers);
 }
 
 uint32_t g_hashperturb;
 double g_consistentHashBalancingFactor = 0;
+
+shared_ptr<DownstreamState> whashedFromHash(const ServerPolicy::NumberedServerVector& servers, size_t hash)
+{
+  return valrandom(hash, servers);
+}
+
 shared_ptr<DownstreamState> whashed(const ServerPolicy::NumberedServerVector& servers, const DNSQuestion* dq)
 {
-  return valrandom(dq->qname->hash(g_hashperturb), servers, dq);
+  return whashedFromHash(servers, dq->qname->hash(g_hashperturb));
 }
 
-shared_ptr<DownstreamState> chashed(const ServerPolicy::NumberedServerVector& servers, const DNSQuestion* dq)
+shared_ptr<DownstreamState> chashedFromHash(const ServerPolicy::NumberedServerVector& servers, size_t qhash)
 {
-  unsigned int qhash = dq->qname->hash(g_hashperturb);
   unsigned int sel = std::numeric_limits<unsigned int>::max();
   unsigned int min = std::numeric_limits<unsigned int>::max();
   shared_ptr<DownstreamState> ret = nullptr, first = nullptr;
@@ -152,6 +161,11 @@ shared_ptr<DownstreamState> chashed(const ServerPolicy::NumberedServerVector& se
   return shared_ptr<DownstreamState>();
 }
 
+shared_ptr<DownstreamState> chashed(const ServerPolicy::NumberedServerVector& servers, const DNSQuestion* dq)
+{
+  return chashedFromHash(servers, dq->qname->hash(g_hashperturb));
+}
+
 shared_ptr<DownstreamState> roundrobin(const ServerPolicy::NumberedServerVector& servers, const DNSQuestion* dq)
 {
   ServerPolicy::NumberedServerVector poss;
diff --git a/pdns/dnsdistdist/dnsdist-lua-ffi-interface.h b/pdns/dnsdistdist/dnsdist-lua-ffi-interface.h
index 2650af7ce5..327da5b498 100644
--- a/pdns/dnsdistdist/dnsdist-lua-ffi-interface.h
+++ b/pdns/dnsdistdist/dnsdist-lua-ffi-interface.h
@@ -97,6 +97,8 @@ typedef struct dnsdist_ffi_server_t dnsdist_ffi_server_t;
 
 size_t dnsdist_ffi_servers_list_get_count(const dnsdist_ffi_servers_list_t* list) __attribute__ ((visibility ("default")));
 void dnsdist_ffi_servers_list_get_server(const dnsdist_ffi_servers_list_t* list, size_t idx, const dnsdist_ffi_server_t** out) __attribute__ ((visibility ("default")));
+size_t dnsdist_ffi_servers_list_chashed(const dnsdist_ffi_servers_list_t* list, const dnsdist_ffi_dnsquestion_t* dq, size_t hash) __attribute__ ((visibility ("default")));
+size_t dnsdist_ffi_servers_list_whashed(const dnsdist_ffi_servers_list_t* list, const dnsdist_ffi_dnsquestion_t* dq, size_t hash) __attribute__ ((visibility ("default")));
 
 uint64_t dnsdist_ffi_server_get_outstanding(const dnsdist_ffi_server_t* server) __attribute__ ((visibility ("default")));
 bool dnsdist_ffi_server_is_up(const dnsdist_ffi_server_t* server) __attribute__ ((visibility ("default")));
diff --git a/pdns/dnsdistdist/dnsdist-lua-ffi.cc b/pdns/dnsdistdist/dnsdist-lua-ffi.cc
index 181232ede2..548357785b 100644
--- a/pdns/dnsdistdist/dnsdist-lua-ffi.cc
+++ b/pdns/dnsdistdist/dnsdist-lua-ffi.cc
@@ -425,6 +425,28 @@ void dnsdist_ffi_servers_list_get_server(const dnsdist_ffi_servers_list_t* list,
   *out = &list->ffiServers.at(idx);
 }
 
+static size_t dnsdist_ffi_servers_get_index_from_server(const ServerPolicy::NumberedServerVector& servers, const std::shared_ptr<DownstreamState>& server)
+{
+  for (const auto& pair : servers) {
+    if (pair.second == server) {
+      return pair.first - 1;
+    }
+  }
+  throw std::runtime_error("Unable to find servers in server list");
+}
+
+size_t dnsdist_ffi_servers_list_chashed(const dnsdist_ffi_servers_list_t* list, const dnsdist_ffi_dnsquestion_t* dq, size_t hash)
+{
+  auto server = chashedFromHash(list->servers, hash);
+  return dnsdist_ffi_servers_get_index_from_server(list->servers, server);
+}
+
+size_t dnsdist_ffi_servers_list_whashed(const dnsdist_ffi_servers_list_t* list, const dnsdist_ffi_dnsquestion_t* dq, size_t hash)
+{
+  auto server = whashedFromHash(list->servers, hash);
+  return dnsdist_ffi_servers_get_index_from_server(list->servers, server);
+}
+
 uint64_t dnsdist_ffi_server_get_outstanding(const dnsdist_ffi_server_t* server)
 {
   return server->server->outstanding;
diff --git a/pdns/dnsdistdist/dnsdist-lua-ffi.hh b/pdns/dnsdistdist/dnsdist-lua-ffi.hh
index 811a7f777e..118fd0ce8d 100644
--- a/pdns/dnsdistdist/dnsdist-lua-ffi.hh
+++ b/pdns/dnsdistdist/dnsdist-lua-ffi.hh
@@ -93,7 +93,7 @@ struct LuaContext::Pusher<dnsdist_ffi_servers_list_t*> {
 
 struct dnsdist_ffi_servers_list_t
 {
-  dnsdist_ffi_servers_list_t(const ServerPolicy::NumberedServerVector& servers)
+  dnsdist_ffi_servers_list_t(const ServerPolicy::NumberedServerVector& servers_): servers(servers_)
   {
     ffiServers.reserve(servers.size());
     for (const auto& server: servers) {
@@ -102,6 +102,7 @@ struct dnsdist_ffi_servers_list_t
   }
 
   std::vector<dnsdist_ffi_server_t> ffiServers;
+  const ServerPolicy::NumberedServerVector& servers;
 };
 
 const std::string& getLuaFFIWrappers();
diff --git a/pdns/dnsdistdist/test-dnsdistlbpolicies_cc.cc b/pdns/dnsdistdist/test-dnsdistlbpolicies_cc.cc
index be75da1929..e04168b2e3 100644
--- a/pdns/dnsdistdist/test-dnsdistlbpolicies_cc.cc
+++ b/pdns/dnsdistdist/test-dnsdistlbpolicies_cc.cc
@@ -112,20 +112,34 @@ static void benchPolicy(const ServerPolicy& pol)
   for (size_t idx = 1; idx <= 10; idx++) {
     servers.push_back({ idx, std::make_shared<DownstreamState>(ComboAddress("192.0.2." + std::to_string(idx) + ":53")) });
     servers.at(idx - 1).second->setUp();
+    /* we need to have a weight of at least 1000 to get an optimal repartition with the consistent hashing algo */
+    servers.at(idx - 1).second->setWeight(1000);
+    /* make sure that the hashes have been computed */
+    servers.at(idx - 1).second->hash();
   }
 
   StopWatch sw;
   sw.start();
+  for (size_t idx = 0; idx < 1000; idx++) {
   for (const auto& name : names) {
     auto dq = getDQ(&name);
     auto server = getSelectedBackendFromPolicy(pol, servers, dq);
   }
+  }
   cerr<<pol.name<<" took "<<std::to_string(sw.udiff())<<" us for "<<names.size()<<endl;
 
   g_verbose = existingVerboseValue;
 #endif /* BENCH_POLICIES */
 }
 
+static void resetLuaContext()
+{
+  /* we need to reset this before cleaning the Lua state because the server policy might holds
+     a reference to a Lua function (Lua policies) */
+  g_policy.setState(ServerPolicy("leastOutstanding", leastOutstanding, false));
+  g_lua = LuaContext();
+}
+
 BOOST_AUTO_TEST_SUITE(dnsdistlbpolicies)
 
 BOOST_AUTO_TEST_CASE(test_firstAvailable) {
@@ -352,7 +366,6 @@ BOOST_AUTO_TEST_CASE(test_whashed) {
   BOOST_CHECK_LT(got, expected * 2);
 }
 
-
 BOOST_AUTO_TEST_CASE(test_chashed) {
   bool existingVerboseValue = g_verbose;
   g_verbose = false;
@@ -456,6 +469,7 @@ BOOST_AUTO_TEST_CASE(test_lua) {
 
     setServerPolicyLua("luaroundrobin", luaroundrobin)
   )foo";
+  resetLuaContext();
   g_lua.writeFunction("setServerPolicyLua", [](string name, ServerPolicy::policyfunc_t policy) {
       g_policy.setState(ServerPolicy{name, policy, true});
     });
@@ -491,7 +505,7 @@ BOOST_AUTO_TEST_CASE(test_lua) {
 }
 
 #ifdef LUAJIT_VERSION
-BOOST_AUTO_TEST_CASE(test_lua_ffi) {
+BOOST_AUTO_TEST_CASE(test_lua_ffi_rr) {
   std::vector<DNSName> names;
   names.reserve(1000);
   for (size_t idx = 0; idx < 1000; idx++) {
@@ -505,16 +519,69 @@ BOOST_AUTO_TEST_CASE(test_lua_ffi) {
     function ffilb(servers_list, dq)
       local serversCount = tonumber(C.dnsdist_ffi_servers_list_get_count(servers_list))
       counter = counter + 1
+      return counter % serversCount
+    end
+
+    setServerPolicyLuaFFI("FFI round-robin", ffilb)
+  )foo";
+  resetLuaContext();
+  g_lua.executeCode(getLuaFFIWrappers());
+  g_lua.writeFunction("setServerPolicyLuaFFI", [](string name, ServerPolicy::ffipolicyfunc_t policy) {
+      g_policy.setState(ServerPolicy(name, policy));
+    });
+  g_lua.executeCode(policySetupStr);
+
+  ServerPolicy pol = g_policy.getCopy();
+  ServerPolicy::NumberedServerVector servers;
+  std::map<std::shared_ptr<DownstreamState>, uint64_t> serversMap;
+  for (size_t idx = 1; idx <= 10; idx++) {
+    servers.push_back({ idx, std::make_shared<DownstreamState>(ComboAddress("192.0.2." + std::to_string(idx) + ":53")) });
+    serversMap[servers.at(idx - 1).second] = 0;
+    servers.at(idx - 1).second->setUp();
+  }
+  BOOST_REQUIRE_EQUAL(servers.size(), 10);
+
+  for (const auto& name : names) {
+    auto dq = getDQ(&name);
+    auto server = getSelectedBackendFromPolicy(pol, servers, dq);
+    BOOST_REQUIRE(serversMap.count(server) == 1);
+    ++serversMap[server];
+  }
+
+  uint64_t total = 0;
+  for (const auto& entry : serversMap) {
+    BOOST_CHECK_GT(entry.second, 0);
+    BOOST_CHECK_GT(entry.second, (names.size() / servers.size() / 2));
+    BOOST_CHECK_LT(entry.second, (names.size() / servers.size() * 2));
+    total += entry.second;
+  }
+  BOOST_CHECK_EQUAL(total, names.size());
+
+  benchPolicy(pol);
+}
+
+BOOST_AUTO_TEST_CASE(test_lua_ffi_hashed) {
+  std::vector<DNSName> names;
+  names.reserve(1000);
+  for (size_t idx = 0; idx < 1000; idx++) {
+    names.push_back(DNSName("powerdns-" + std::to_string(idx) + ".com."));
+  }
+
+  static const std::string policySetupStr = R"foo(
+    local ffi = require("ffi")
+    local C = ffi.C
+    function ffilb(servers_list, dq)
+      local serversCount = tonumber(C.dnsdist_ffi_servers_list_get_count(servers_list))
       local hash = tonumber(C.dnsdist_ffi_dnsquestion_get_qname_hash(dq, 0))
       return hash % serversCount
     end
 
-    setServerPolicyLuaFFI("FFI", ffilb)
+    setServerPolicyLuaFFI("FFI hashed", ffilb)
   )foo";
+  resetLuaContext();
   g_lua.executeCode(getLuaFFIWrappers());
   g_lua.writeFunction("setServerPolicyLuaFFI", [](string name, ServerPolicy::ffipolicyfunc_t policy) {
-      auto pol = ServerPolicy(name, policy);
-      g_policy.setState(std::move(pol));
+      g_policy.setState(ServerPolicy(name, policy));
     });
   g_lua.executeCode(policySetupStr);
 
@@ -546,6 +613,119 @@ BOOST_AUTO_TEST_CASE(test_lua_ffi) {
 
   benchPolicy(pol);
 }
+
+BOOST_AUTO_TEST_CASE(test_lua_ffi_whashed) {
+  std::vector<DNSName> names;
+  names.reserve(1000);
+  for (size_t idx = 0; idx < 1000; idx++) {
+    names.push_back(DNSName("powerdns-" + std::to_string(idx) + ".com."));
+  }
+
+  static const std::string policySetupStr = R"foo(
+    local ffi = require("ffi")
+    local C = ffi.C
+    function ffilb(servers_list, dq)
+      return tonumber(C.dnsdist_ffi_servers_list_whashed(servers_list, dq, C.dnsdist_ffi_dnsquestion_get_qname_hash(dq, 0)))
+    end
+
+    setServerPolicyLuaFFI("FFI whashed", ffilb)
+  )foo";
+  resetLuaContext();
+  g_lua.executeCode(getLuaFFIWrappers());
+  g_lua.writeFunction("setServerPolicyLuaFFI", [](string name, ServerPolicy::ffipolicyfunc_t policy) {
+      g_policy.setState(ServerPolicy(name, policy));
+    });
+  g_lua.executeCode(policySetupStr);
+
+  ServerPolicy pol = g_policy.getCopy();
+  ServerPolicy::NumberedServerVector servers;
+  std::map<std::shared_ptr<DownstreamState>, uint64_t> serversMap;
+  for (size_t idx = 1; idx <= 10; idx++) {
+    servers.push_back({ idx, std::make_shared<DownstreamState>(ComboAddress("192.0.2." + std::to_string(idx) + ":53")) });
+    serversMap[servers.at(idx - 1).second] = 0;
+    servers.at(idx - 1).second->setUp();
+  }
+  BOOST_REQUIRE_EQUAL(servers.size(), 10);
+
+  for (const auto& name : names) {
+    auto dq = getDQ(&name);
+    auto server = getSelectedBackendFromPolicy(pol, servers, dq);
+    BOOST_REQUIRE(serversMap.count(server) == 1);
+    ++serversMap[server];
+  }
+
+  uint64_t total = 0;
+  for (const auto& entry : serversMap) {
+    BOOST_CHECK_GT(entry.second, 0);
+    BOOST_CHECK_GT(entry.second, (names.size() / servers.size() / 2));
+    BOOST_CHECK_LT(entry.second, (names.size() / servers.size() * 2));
+    total += entry.second;
+  }
+  BOOST_CHECK_EQUAL(total, names.size());
+
+  benchPolicy(pol);
+}
+
+BOOST_AUTO_TEST_CASE(test_lua_ffi_chashed) {
+  bool existingVerboseValue = g_verbose;
+  g_verbose = false;
+
+  std::vector<DNSName> names;
+  names.reserve(1000);
+  for (size_t idx = 0; idx < 1000; idx++) {
+    names.push_back(DNSName("powerdns-" + std::to_string(idx) + ".com."));
+  }
+
+  static const std::string policySetupStr = R"foo(
+    local ffi = require("ffi")
+    local C = ffi.C
+    function ffilb(servers_list, dq)
+      return tonumber(C.dnsdist_ffi_servers_list_chashed(servers_list, dq, C.dnsdist_ffi_dnsquestion_get_qname_hash(dq, 0)))
+    end
+
+    setServerPolicyLuaFFI("FFI chashed", ffilb)
+  )foo";
+  resetLuaContext();
+  g_lua.executeCode(getLuaFFIWrappers());
+  g_lua.writeFunction("setServerPolicyLuaFFI", [](string name, ServerPolicy::ffipolicyfunc_t policy) {
+      g_policy.setState(ServerPolicy(name, policy));
+    });
+  g_lua.executeCode(policySetupStr);
+
+  ServerPolicy pol = g_policy.getCopy();
+  ServerPolicy::NumberedServerVector servers;
+  std::map<std::shared_ptr<DownstreamState>, uint64_t> serversMap;
+  for (size_t idx = 1; idx <= 10; idx++) {
+    servers.push_back({ idx, std::make_shared<DownstreamState>(ComboAddress("192.0.2." + std::to_string(idx) + ":53")) });
+    serversMap[servers.at(idx - 1).second] = 0;
+    servers.at(idx - 1).second->setUp();
+    /* we need to have a weight of at least 1000 to get an optimal repartition with the consistent hashing algo */
+    servers.at(idx - 1).second->setWeight(1000);
+    /* make sure that the hashes have been computed */
+    servers.at(idx - 1).second->hash();
+  }
+  BOOST_REQUIRE_EQUAL(servers.size(), 10);
+
+  for (const auto& name : names) {
+    auto dq = getDQ(&name);
+    auto server = getSelectedBackendFromPolicy(pol, servers, dq);
+    BOOST_REQUIRE(serversMap.count(server) == 1);
+    ++serversMap[server];
+  }
+
+  uint64_t total = 0;
+  for (const auto& entry : serversMap) {
+    BOOST_CHECK_GT(entry.second, 0);
+    BOOST_CHECK_GT(entry.second, (names.size() / servers.size() / 2));
+    BOOST_CHECK_LT(entry.second, (names.size() / servers.size() * 2));
+    total += entry.second;
+  }
+  BOOST_CHECK_EQUAL(total, names.size());
+
+  benchPolicy(pol);
+
+  g_verbose = existingVerboseValue;
+}
 #endif /* LUAJIT_VERSION */
 
 BOOST_AUTO_TEST_SUITE_END()