dnsdist: remove pipeline functionality from RedisClient

Signed-off-by: Ensar Sarajčić <dev@ensarsarajcic.com>

Author: esensar

pdns/dnsdistdist/dnsdist-configuration-yaml.cc

@@ -2095,7 +2095,7 @@ void registerRedisClientObjects([[maybe_unused]] const ::rust::Vec<RedisClientCo
 {
 #ifdef HAVE_REDIS
   for (const auto& redisClient : redisClients) {
-    dnsdist::configuration::yaml::registerType<RedisClient>(std::make_shared<RedisClient>(std::string(redisClient.url), redisClient.pipeline_enabled, redisClient.pipeline_interval), redisClient.name);
+    dnsdist::configuration::yaml::registerType<RedisClient>(std::make_shared<RedisClient>(std::string(redisClient.url)), redisClient.name);
   }
 #endif
 }

pdns/dnsdistdist/dnsdist-lua-bindings-redis.cc

@@ -26,18 +26,12 @@
 void setupLuaBindingsRedis([[maybe_unused]] LuaContext& luaCtx, [[maybe_unused]] bool client)
 {
 #ifdef HAVE_REDIS
-  luaCtx.writeFunction("newRedisClient", [client](const std::string& url, std::optional<LuaAssociativeTable<boost::variant<bool, std::string>>> vars) {
+  luaCtx.writeFunction("newRedisClient", [client](const std::string& url) {
     if (client) {
       return std::shared_ptr<RedisClient>(nullptr);
     }
 
-    bool pipelineEnabled{true};
-    int pipelineInterval{10};
-    getOptionalValue<bool>(vars, "pipelineEnabled", pipelineEnabled);
-    getOptionalIntegerValue<int>("newRedisClient", vars, "pipelineInterval", pipelineInterval);
-    checkAllParametersConsumed("newRedisClient", vars);
-
-    return std::make_shared<RedisClient>(url, pipelineEnabled, pipelineInterval);
+    return std::make_shared<RedisClient>(url);
   });
 
   luaCtx.registerFunction<std::string (std::shared_ptr<RedisClient>::*)(const std::string&)>("get", [](std::shared_ptr<RedisClient>& rc, const std::string& key) {

pdns/dnsdistdist/dnsdist-rust-lib/rust-post-in.rs

@@ -96,6 +96,7 @@ fn get_global_configuration_from_serde(
     dnsdistsettings::registerKVSObjects(&config.key_value_stores);
     // this needs to be done before the rules so that they can refer to the NMG objects
     dnsdistsettings::registerNMGObjects(&config.netmask_groups);
+    dnsdistsettings::registerRedisClientObjects(&config.redis_clients);
     // this needs to be done before the rules so that they can refer to the TimeIPSet objects
     dnsdistsettings::registerTimedIPSetObjects(&config.timed_ip_sets);
     // this needs to be done BEFORE the rules so that they can refer to the selectors

pdns/dnsdistdist/dnsdist-settings-definitions.yml

@@ -201,14 +201,6 @@ redis_client:
     - name: "url"
       type: "String"
       description: "URL of the Redis instance to connect to"
-    - name: "pipeline_enabled"
-      type: "bool"
-      default: "true"
-      description: "Whether command pipelining should be enabled for this client"
-    - name: "pipeline_interval"
-      type: "u32"
-      default: "10"
-      description: "Interval to send commands at when pipelining is enabled"
 
 remote_logging:
   description: "Queries and/or responses remote logging settings"

pdns/dnsdistdist/redis.cc

@@ -83,7 +83,11 @@ redisReply* RedisClient::executeCommand(const char* format, ...) const
 {
   va_list ap;
   va_start(ap, format);
-  auto result = d_executor->executeCommand(format, ap);
+  auto connection = d_connection.getConnection();
+  auto result = static_cast<redisReply*>(redisvCommand(connection->get(), format, ap));
+  if (connection->get()->err != 0) {
+    vinfolog("Redis connection error %s", connection->get()->errstr);
+  }
   va_end(ap);
   return result;
 };
@@ -96,143 +100,14 @@ redisReply* RedisClient::executeCommandArgv(std::vector<std::string> args) const
     argv.push_back(args[i].data());
     argvlen.push_back(args[i].length());
   }
-  return d_executor->executeCommandArgv(args.size(), argv.data(), argvlen.data());
-};
-
-redisReply* RedisClient::DirectExecutor::executeCommand(const char* format, va_list ap) const
-{
   auto connection = d_connection.getConnection();
-  auto result = static_cast<redisReply*>(redisvCommand(connection->get(), format, ap));
+  auto result = static_cast<redisReply*>(redisCommandArgv(connection->get(), args.size(), argv.data(), argvlen.data()));
   if (connection->get()->err != 0) {
     vinfolog("Redis connection error %s", connection->get()->errstr);
   }
   return result;
-}
-
-redisReply* RedisClient::DirectExecutor::executeCommandArgv(int argc, const char** argv, const size_t* argvlen) const
-{
-  auto connection = d_connection.getConnection();
-  auto result = static_cast<redisReply*>(redisCommandArgv(connection->get(), argc, argv, argvlen));
-  if (connection->get()->err != 0) {
-    vinfolog("Redis connection error %s", connection->get()->errstr);
-  }
-  return result;
-}
-
-RedisClient::PipelineExecutor::PipelineExecutor(const std::string& url, uint32_t pipelineInterval) :
-  d_connection(url), d_interval(pipelineInterval)
-{
-  auto [sender, receiver] = pdns::channel::createObjectQueue<PipelineCommand>();
-  d_pipelineSender = std::move(sender);
-  d_pipelineReceiver = std::move(receiver);
-
-  d_thread = std::thread(&RedisClient::PipelineExecutor::maintenanceThread, this);
-}
-
-redisReply* RedisClient::PipelineExecutor::executeCommandArgv(int argc, const char** argv, const size_t* argvlen) const
-{
-  char* command;
-  auto len = redisFormatCommandArgv(&command, argc, argv, argvlen);
-  if (len < 0) {
-    // TODO: handle formatting errors?
-    vinfolog("Redis command formatting error");
-    return nullptr;
-  }
-
-  return pipelineCommand(command, len);
-}
-
-redisReply* RedisClient::PipelineExecutor::executeCommand(const char* format, va_list ap) const
-{
-  char* command;
-  auto len = redisvFormatCommand(&command, format, ap);
-  if (len < 0) {
-    // TODO: handle formatting errors?
-    vinfolog("Redis command formatting error");
-    return nullptr;
-  }
-
-  return pipelineCommand(command, len);
-}
-
-redisReply* RedisClient::PipelineExecutor::pipelineCommand(const char* command, size_t len) const
-{
-  std::mutex mtx;
-  std::condition_variable cv;
-  std::unique_lock<std::mutex> lock(mtx);
-  std::shared_ptr<redisReply*> result = std::make_shared<redisReply*>(nullptr);
-  PipelineCommand::callback_t callback = [result, &cv](redisReply* reply) mutable {
-    *result = reply;
-    cv.notify_one();
-  };
-  d_pipelineSender.send(std::make_unique<PipelineCommand>(PipelineCommand{
-    command,
-    len,
-    callback}));
-  cv.wait(lock);
-  return *result;
-}
-
-void RedisClient::PipelineExecutor::maintenanceThread()
-{
-  setThreadName("dnsdist/redis");
-
-  for (;;) {
-    if (d_exiting) {
-      break;
-    }
-
-    bool connected = true;
-    if (d_connection.needsReconnect()) {
-      connected = d_connection.reconnect();
-    }
-
-    if (connected) {
-      auto connection = d_connection.getConnection();
-      std::list<PipelineCommand::callback_t> callbacks;
-      while (auto command = d_pipelineReceiver.receive()) {
-        if (redisAppendFormattedCommand(connection->get(), command->get()->command, command->get()->length) == REDIS_OK) {
-          callbacks.push_back(command->get()->callback);
-        }
-        else {
-          if (connection->get()->err != 0) {
-            vinfolog("Redis connection error %s", connection->get()->errstr);
-          }
-          else {
-            vinfolog("Unknown redis connection error");
-          }
-          command->get()->callback(nullptr);
-        }
-      }
-
-      for (auto callback : callbacks) {
-        void* reply;
-        if (redisGetReply(connection->get(), &reply) == REDIS_OK) {
-          callback(static_cast<redisReply*>(reply));
-        }
-        else {
-          if (connection->get()->err != 0) {
-            vinfolog("Redis connection error %s", connection->get()->errstr);
-          }
-          else {
-            vinfolog("Unknown redis connection error");
-          }
-          callback(nullptr);
-        }
-      }
-    }
-
-    std::this_thread::sleep_for(std::chrono::milliseconds(d_interval));
-  }
 };
 
-RedisClient::PipelineExecutor::~PipelineExecutor()
-{
-  d_exiting = true;
-
-  d_thread.join();
-}
-
 bool RedisKVClient::getValue(const std::string& key, std::string& value)
 {
   auto reply = d_lookupAction->getValue(*d_client, key);

pdns/dnsdistdist/redis.hh

@@ -316,22 +316,15 @@ private:
 class RedisClient
 {
 public:
-  RedisClient(const std::string& url, bool enablePipeline = true, uint32_t pipelineInterval = 10)
-  {
-    if (enablePipeline) {
-      d_executor = std::make_unique<PipelineExecutor>(url, pipelineInterval);
-    }
-    else {
-      d_executor = std::make_unique<DirectExecutor>(url);
-    }
-  }
+  RedisClient(const std::string& url) :
+    d_connection(url) {}
 
   redisReply* executeCommand(const char* format, ...) const;
   redisReply* executeCommandArgv(std::vector<std::string> args) const;
 
   const YaHTTP::URL& getUrl() const
   {
-    return d_executor->getUrl();
+    return d_connection.getUrl();
   }
 
 private:
@@ -361,68 +354,7 @@ private:
     YaHTTP::URL d_url;
   };
 
-  class Executor
-  {
-  public:
-    virtual ~Executor() = default;
-    virtual redisReply* executeCommand(const char* format, va_list ap) const = 0;
-    virtual redisReply* executeCommandArgv(int argc, const char** argv, const size_t* argvlen) const = 0;
-    virtual const YaHTTP::URL& getUrl() const = 0;
-  };
-
-  class DirectExecutor : public Executor
-  {
-  public:
-    DirectExecutor(const std::string& url) :
-      d_connection(url)
-    {
-    }
-    redisReply* executeCommand(const char* format, va_list ap) const override;
-    redisReply* executeCommandArgv(int argc, const char** argv, const size_t* argvlen) const override;
-
-    const YaHTTP::URL& getUrl() const override
-    {
-      return d_connection.getUrl();
-    }
-
-  private:
-    RedisConnection d_connection;
-  };
-
-  class PipelineExecutor : public Executor
-  {
-  public:
-    PipelineExecutor(const std::string& url, uint32_t pipelineInterval);
-    ~PipelineExecutor();
-    redisReply* executeCommand(const char* format, va_list ap) const override;
-    redisReply* executeCommandArgv(int argc, const char** argv, const size_t* argvlen) const override;
-
-    const YaHTTP::URL& getUrl() const override
-    {
-      return d_connection.getUrl();
-    }
-
-  private:
-    void maintenanceThread();
-    redisReply* pipelineCommand(const char* command, size_t len) const;
-
-    struct PipelineCommand
-    {
-      typedef std::function<void(redisReply*)> callback_t;
-      const char* command;
-      size_t length;
-      callback_t callback;
-    };
-    RedisConnection d_connection;
-    uint32_t d_interval;
-    pdns::channel::Sender<PipelineCommand> d_pipelineSender;
-    pdns::channel::Receiver<PipelineCommand> d_pipelineReceiver;
-
-    std::atomic<bool> d_exiting{false};
-    std::thread d_thread;
-  };
-
-  std::unique_ptr<Executor> d_executor;
+  RedisConnection d_connection;
 };
 
 class RedisKVClientInterface