Merge branch 'unstable' into unstable

Author: LindaSummer

kvrocks.conf

@@ -193,6 +193,18 @@ replication-connect-timeout-ms 3100
 # future 'clusterx setnodes' commands because the replication thread is blocked on recv.
 replication-recv-timeout-ms 3200
 
+# Maximum bytes to buffer before sending replication data to replicas.
+# The master will pack multiple write batches into one bulk to reduce network overhead,
+# but will send immediately if the bulk size exceeds this limit.
+# Default: 16KB (16384 bytes)
+replication-delay-bytes 16384
+
+# Maximum number of updates to buffer before sending replication data to replicas.
+# The master will pack multiple write batches into one bulk to reduce network overhead,
+# but will send immediately if the number of updates exceeds this limit.
+# Default: 16 updates
+replication-delay-updates 16
+
 # TCP listen() backlog.
 #
 # In high requests-per-second environments you need an high backlog in order

src/cluster/replication.cc

@@ -56,6 +56,14 @@
 #include <openssl/ssl.h>
 #endif
 
+FeedSlaveThread::FeedSlaveThread(Server *srv, redis::Connection *conn, rocksdb::SequenceNumber next_repl_seq)
+    : srv_(srv),
+      conn_(conn),
+      next_repl_seq_(next_repl_seq),
+      req_(srv),
+      max_delay_bytes_(srv->GetConfig()->max_replication_delay_bytes),
+      max_delay_updates_(srv->GetConfig()->max_replication_delay_updates) {}
+
 Status FeedSlaveThread::Start() {
   auto s = util::CreateThread("feed-replica", [this] {
     sigset_t mask, omask;
@@ -154,6 +162,16 @@ void FeedSlaveThread::readCallback(bufferevent *bev, [[maybe_unused]] void *ctx)
   }
 }
 
+bool FeedSlaveThread::shouldSendGetAck(rocksdb::SequenceNumber seq) {
+  rocksdb::SequenceNumber largest_unblockable_seq = srv_->LargestTargetSeqToWakeup(seq);
+  if (largest_unblockable_seq > last_getack_seq_) {
+    last_getack_seq_ = largest_unblockable_seq;
+    return true;
+  }
+
+  return false;
+}
+
 void FeedSlaveThread::loop() {
   // is_first_repl_batch was used to fix that replication may be stuck in a dead loop
   // when some seqs might be lost in the middle of the WAL log, so forced to replicate
@@ -194,18 +212,23 @@ void FeedSlaveThread::loop() {
     // 3. To avoid master don't send replication stream to slave since of packing
     //    batches strategy, we still send batches if current batch sequence is less
     //    kMaxDelayUpdates than latest sequence.
-    if (is_first_repl_batch || batches_bulk.size() >= kMaxDelayBytes || updates_in_batches >= kMaxDelayUpdates ||
-        srv_->storage->LatestSeqNumber() - batch.sequence <= kMaxDelayUpdates) {
+    if (is_first_repl_batch || batches_bulk.size() >= max_delay_bytes_ || updates_in_batches >= max_delay_updates_ ||
+        srv_->storage->LatestSeqNumber() - batch.sequence <= max_delay_updates_) {
+      if (shouldSendGetAck(batch.sequence)) {
+        batches_bulk += redis::BulkString("_getack");
+      }
+
       // Send entire bulk which contain multiple batches
       auto s = util::SockSend(conn_->GetFD(), batches_bulk, conn_->GetBufferEvent());
       if (!s.IsOK()) {
         error("Write error while sending batch to slave: {}. batches: 0x{}", s.Msg(), util::StringToHex(batches_bulk));
         Stop();
         return;
       }
+
       is_first_repl_batch = false;
       batches_bulk.clear();
-      if (batches_bulk.capacity() > kMaxDelayBytes * 2) batches_bulk.shrink_to_fit();
+      if (batches_bulk.capacity() > max_delay_bytes_ * 2) batches_bulk.shrink_to_fit();
       updates_in_batches = 0;
     }
     curr_seq = batch.sequence + batch.writeBatchPtr->Count();
@@ -608,22 +631,29 @@ ReplicationThread::CBState ReplicationThread::tryPSyncReadCB(bufferevent *bev) {
   }
 }
 
-void ReplicationThread::sendReplConfAck(bufferevent *bev) {
-  SendString(bev, redis::ArrayOfBulkStrings({"replconf", "ack", std::to_string(storage_->LatestSeqNumber())}));
+void ReplicationThread::sendReplConfAck(bufferevent *bev, bool force) {
+  int64_t now = util::GetTimeStamp();
+
+  // If force is true, always send ack. Otherwise, check if it has been 1s from last ack
+  if (force || (now - last_ack_time_secs_) >= 1) {
+    SendString(bev, redis::ArrayOfBulkStrings({"replconf", "ack", std::to_string(storage_->LatestSeqNumber())}));
+    last_ack_time_secs_ = now;
+  }
 }
 
 ReplicationThread::CBState ReplicationThread::incrementBatchLoopCB(bufferevent *bev) {
   repl_state_.store(kReplConnected, std::memory_order_relaxed);
   auto input = bufferevent_get_input(bev);
   bool data_written = false;
+  bool force_ack = false;
   while (true) {
     switch (incr_state_) {
       case Incr_batch_size: {
         // Read bulk length
         UniqueEvbufReadln line(input, EVBUFFER_EOL_CRLF_STRICT);
         if (!line) {
           if (data_written) {
-            sendReplConfAck(bev);
+            sendReplConfAck(bev, force_ack);
           }
           return CBState::AGAIN;
         }
@@ -639,7 +669,7 @@ ReplicationThread::CBState ReplicationThread::incrementBatchLoopCB(bufferevent *
         // Read bulk data (batch data)
         if (incr_bulk_len_ + 2 > evbuffer_get_length(input)) {  // If data not enough
           if (data_written) {
-            sendReplConfAck(bev);
+            sendReplConfAck(bev, force_ack);
           }
           return CBState::AGAIN;
         }
@@ -654,11 +684,18 @@ ReplicationThread::CBState ReplicationThread::incrementBatchLoopCB(bufferevent *
           // master would send the ping heartbeat packet to check whether the slave was alive or not,
           // don't write ping to db here.
           if (data_written) {
-            sendReplConfAck(bev);
+            sendReplConfAck(bev, force_ack);
           }
           return CBState::AGAIN;
         }
 
+        if (bulk_string == "_getack") {
+          // master would send the _getack command to the master to get acknowledgment
+          // don't write _getack to db here.
+          force_ack = true;
+          continue;
+        }
+
         rocksdb::WriteBatch batch(std::move(bulk_string));
 
         auto s = storage_->ReplicaApplyWriteBatch(&batch);

src/cluster/replication.h

@@ -64,8 +64,7 @@ using FetchFileCallback = std::function<void(const std::string &, uint32_t)>;
 
 class FeedSlaveThread {
  public:
-  explicit FeedSlaveThread(Server *srv, redis::Connection *conn, rocksdb::SequenceNumber next_repl_seq)
-      : srv_(srv), conn_(conn), next_repl_seq_(next_repl_seq), req_(srv) {}
+  explicit FeedSlaveThread(Server *srv, redis::Connection *conn, rocksdb::SequenceNumber next_repl_seq);
   ~FeedSlaveThread() = default;
 
   Status Start();
@@ -86,14 +85,18 @@ class FeedSlaveThread {
   // used to parse the ack response from the slave
   redis::Request req_;
   std::atomic<rocksdb::SequenceNumber> ack_seq_ = 0;
+  rocksdb::SequenceNumber last_getack_seq_ = 0;
+  int64_t last_ack_time_secs_ = 0;
 
-  static const size_t kMaxDelayUpdates = 16;
-  static const size_t kMaxDelayBytes = 16 * 1024;
+  // Configurable delay limits
+  size_t max_delay_bytes_;
+  size_t max_delay_updates_;
 
   void loop();
   void checkLivenessIfNeed();
   void readCallback(bufferevent *bev, void *ctx);
   static void staticReadCallback(bufferevent *bev, void *ctx);
+  bool shouldSendGetAck(rocksdb::SequenceNumber seq);
 };
 
 class ReplicationThread : private EventCallbackBase<ReplicationThread> {
@@ -161,6 +164,7 @@ class ReplicationThread : private EventCallbackBase<ReplicationThread> {
   engine::Storage *storage_ = nullptr;
   std::atomic<ReplState> repl_state_;
   std::atomic<int64_t> last_io_time_secs_ = 0;
+  int64_t last_ack_time_secs_ = 0;
   bool next_try_old_psync_ = false;
   bool next_try_without_announce_ip_address_ = false;
 
@@ -202,8 +206,6 @@ class ReplicationThread : private EventCallbackBase<ReplicationThread> {
   CBState fullSyncWriteCB(bufferevent *bev);
   CBState fullSyncReadCB(bufferevent *bev);
 
-  void sendReplConfAck(bufferevent *bev);
-
   Status sendAuth(int sock_fd, ssl_st *ssl);
   Status fetchFile(int sock_fd, evbuffer *evbuf, const std::string &dir, const std::string &file, uint32_t crc,
                    const FetchFileCallback &fn, ssl_st *ssl);
@@ -214,6 +216,8 @@ class ReplicationThread : private EventCallbackBase<ReplicationThread> {
   static bool isWrongPsyncNum(std::string_view err);
   static bool isUnknownOption(std::string_view err);
 
+  void sendReplConfAck(bufferevent *bev, bool force = false);
+
   Status parseWriteBatch(const rocksdb::WriteBatch &write_batch);
 };
 

src/commands/cmd_server.cc

@@ -102,13 +102,23 @@ class CommandNamespace : public Commander {
       Status s = srv->GetNamespace()->Del(args_[2]);
       *output = s.IsOK() ? redis::RESP_OK : redis::Error(s);
       warn("Deleted namespace: {}, addr: {}, result: {}", args_[2], conn->GetAddr(), s.Msg());
+    } else if (args_.size() == 2 && sub_command == "current") {
+      *output = redis::BulkString(conn->GetNamespace());
     } else {
-      return {Status::RedisExecErr, "NAMESPACE subcommand must be one of GET, SET, DEL, ADD"};
+      return {Status::RedisExecErr, "NAMESPACE subcommand must be one of GET, SET, DEL, ADD and CURRENT"};
     }
     return Status::OK();
   }
 };
 
+static uint64_t GenerateNamespaceFlag(uint64_t flags, const std::vector<std::string> &args) {
+  if (args.size() >= 2 && util::EqualICase(args[1], "current")) {
+    return flags & ~kCmdAdmin;
+  }
+
+  return flags;
+}
+
 class CommandKeys : public Commander {
  public:
   Status Execute(engine::Context &ctx, Server *srv, Connection *conn, std::string *output) override {
@@ -1539,49 +1549,48 @@ class CommandFlushBlockCache : public Commander {
   }
 };
 
-REDIS_REGISTER_COMMANDS(Server, MakeCmdAttr<CommandAuth>("auth", 2, "read-only ok-loading auth", NO_KEY),
-                        MakeCmdAttr<CommandPing>("ping", -1, "read-only", NO_KEY),
-                        MakeCmdAttr<CommandSelect>("select", 2, "read-only", NO_KEY),
-                        MakeCmdAttr<CommandInfo>("info", -1, "read-only ok-loading", NO_KEY),
-                        MakeCmdAttr<CommandRole>("role", 1, "read-only ok-loading", NO_KEY),
-                        MakeCmdAttr<CommandConfig>("config", -2, "read-only admin", NO_KEY, GenerateConfigFlag),
-                        MakeCmdAttr<CommandNamespace>("namespace", -3, "read-only admin", NO_KEY),
-                        MakeCmdAttr<CommandKeys>("keys", 2, "read-only slow", NO_KEY),
-                        MakeCmdAttr<CommandFlushDB>("flushdb", 1, "write no-dbsize-check exclusive", NO_KEY),
-                        MakeCmdAttr<CommandFlushAll>("flushall", 1, "write no-dbsize-check exclusive admin", NO_KEY),
-                        MakeCmdAttr<CommandDBSize>("dbsize", -1, "read-only", NO_KEY),
-                        MakeCmdAttr<CommandSlowlog>("slowlog", -2, "read-only", NO_KEY),
-                        MakeCmdAttr<CommandKProfile>("kprofile", -3, "read-only admin", NO_KEY),
-                        MakeCmdAttr<CommandPerfLog>("perflog", -2, "read-only", NO_KEY),
-                        MakeCmdAttr<CommandClient>("client", -2, "read-only", NO_KEY),
-                        MakeCmdAttr<CommandMonitor>("monitor", 1, "read-only no-multi no-script", NO_KEY),
-                        MakeCmdAttr<CommandShutdown>("shutdown", 1, "read-only exclusive no-multi no-script admin",
-                                                     NO_KEY),
-                        MakeCmdAttr<CommandQuit>("quit", 1, "read-only", NO_KEY),
-                        MakeCmdAttr<CommandScan>("scan", -2, "read-only", NO_KEY),
-                        MakeCmdAttr<CommandRandomKey>("randomkey", 1, "read-only", NO_KEY),
-                        MakeCmdAttr<CommandDebug>("debug", -2, "read-only exclusive", NO_KEY, CommandDebug::FlagGen),
-                        MakeCmdAttr<CommandCommand>("command", -1, "read-only", NO_KEY),
-                        MakeCmdAttr<CommandEcho>("echo", 2, "read-only", NO_KEY),
-                        MakeCmdAttr<CommandTime>("time", 1, "read-only ok-loading", NO_KEY),
-                        MakeCmdAttr<CommandDisk>("disk", 3, "read-only", 2, 2, 1),
-                        MakeCmdAttr<CommandMemory>("memory", 3, "read-only", 2, 2, 1),
-                        MakeCmdAttr<CommandHello>("hello", -1, "read-only ok-loading auth", NO_KEY),
-                        MakeCmdAttr<CommandRestore>("restore", -4, "write", 1, 1, 1),
-
-                        MakeCmdAttr<CommandCompact>("compact", 1, "read-only no-script", NO_KEY),
-                        MakeCmdAttr<CommandBGSave>("bgsave", 1, "read-only no-script admin", NO_KEY),
-                        MakeCmdAttr<CommandLastSave>("lastsave", -1, "read-only admin", NO_KEY),
-                        MakeCmdAttr<CommandFlushBackup>("flushbackup", 1, "read-only no-script admin", NO_KEY),
-                        MakeCmdAttr<CommandSlaveOf>("slaveof", 3, "read-only exclusive no-script admin", NO_KEY),
-                        MakeCmdAttr<CommandSlaveOf>("replicaof", 3, "read-only exclusive no-script admin", NO_KEY),
-                        MakeCmdAttr<CommandStats>("stats", 1, "read-only", NO_KEY),
-                        MakeCmdAttr<CommandRdb>("rdb", -3, "write exclusive admin", NO_KEY),
-                        MakeCmdAttr<CommandReset>("reset", 1, "ok-loading bypass-multi no-script", NO_KEY),
-                        MakeCmdAttr<CommandApplyBatch>("applybatch", -2, "write no-multi", NO_KEY),
-                        MakeCmdAttr<CommandDump>("dump", 2, "read-only", 1, 1, 1),
-                        MakeCmdAttr<CommandPollUpdates>("pollupdates", -2, "read-only admin", NO_KEY),
-                        MakeCmdAttr<CommandSST>("sst", -3, "write exclusive admin", 1, 1, 1),
-                        MakeCmdAttr<CommandFlushMemTable>("flushmemtable", -1, "exclusive write", NO_KEY),
-                        MakeCmdAttr<CommandFlushBlockCache>("flushblockcache", 1, "exclusive write", NO_KEY), )
+REDIS_REGISTER_COMMANDS(
+    Server, MakeCmdAttr<CommandAuth>("auth", 2, "read-only ok-loading auth", NO_KEY),
+    MakeCmdAttr<CommandPing>("ping", -1, "read-only", NO_KEY),
+    MakeCmdAttr<CommandSelect>("select", 2, "read-only", NO_KEY),
+    MakeCmdAttr<CommandInfo>("info", -1, "read-only ok-loading", NO_KEY),
+    MakeCmdAttr<CommandRole>("role", 1, "read-only ok-loading", NO_KEY),
+    MakeCmdAttr<CommandConfig>("config", -2, "read-only admin", NO_KEY, GenerateConfigFlag),
+    MakeCmdAttr<CommandNamespace>("namespace", -2, "read-only admin", NO_KEY, GenerateNamespaceFlag),
+    MakeCmdAttr<CommandKeys>("keys", 2, "read-only slow", NO_KEY),
+    MakeCmdAttr<CommandFlushDB>("flushdb", 1, "write no-dbsize-check exclusive", NO_KEY),
+    MakeCmdAttr<CommandFlushAll>("flushall", 1, "write no-dbsize-check exclusive admin", NO_KEY),
+    MakeCmdAttr<CommandDBSize>("dbsize", -1, "read-only", NO_KEY),
+    MakeCmdAttr<CommandSlowlog>("slowlog", -2, "read-only", NO_KEY),
+    MakeCmdAttr<CommandKProfile>("kprofile", -3, "read-only admin", NO_KEY),
+    MakeCmdAttr<CommandPerfLog>("perflog", -2, "read-only", NO_KEY),
+    MakeCmdAttr<CommandClient>("client", -2, "read-only", NO_KEY),
+    MakeCmdAttr<CommandMonitor>("monitor", 1, "read-only no-multi no-script", NO_KEY),
+    MakeCmdAttr<CommandShutdown>("shutdown", 1, "read-only exclusive no-multi no-script admin", NO_KEY),
+    MakeCmdAttr<CommandQuit>("quit", 1, "read-only", NO_KEY), MakeCmdAttr<CommandScan>("scan", -2, "read-only", NO_KEY),
+    MakeCmdAttr<CommandRandomKey>("randomkey", 1, "read-only", NO_KEY),
+    MakeCmdAttr<CommandDebug>("debug", -2, "read-only exclusive", NO_KEY, CommandDebug::FlagGen),
+    MakeCmdAttr<CommandCommand>("command", -1, "read-only", NO_KEY),
+    MakeCmdAttr<CommandEcho>("echo", 2, "read-only", NO_KEY),
+    MakeCmdAttr<CommandTime>("time", 1, "read-only ok-loading", NO_KEY),
+    MakeCmdAttr<CommandDisk>("disk", 3, "read-only", 2, 2, 1),
+    MakeCmdAttr<CommandMemory>("memory", 3, "read-only", 2, 2, 1),
+    MakeCmdAttr<CommandHello>("hello", -1, "read-only ok-loading auth", NO_KEY),
+    MakeCmdAttr<CommandRestore>("restore", -4, "write", 1, 1, 1),
+
+    MakeCmdAttr<CommandCompact>("compact", 1, "read-only no-script", NO_KEY),
+    MakeCmdAttr<CommandBGSave>("bgsave", 1, "read-only no-script admin", NO_KEY),
+    MakeCmdAttr<CommandLastSave>("lastsave", -1, "read-only admin", NO_KEY),
+    MakeCmdAttr<CommandFlushBackup>("flushbackup", 1, "read-only no-script admin", NO_KEY),
+    MakeCmdAttr<CommandSlaveOf>("slaveof", 3, "read-only exclusive no-script admin", NO_KEY),
+    MakeCmdAttr<CommandSlaveOf>("replicaof", 3, "read-only exclusive no-script admin", NO_KEY),
+    MakeCmdAttr<CommandStats>("stats", 1, "read-only", NO_KEY),
+    MakeCmdAttr<CommandRdb>("rdb", -3, "write exclusive admin", NO_KEY),
+    MakeCmdAttr<CommandReset>("reset", 1, "ok-loading bypass-multi no-script", NO_KEY),
+    MakeCmdAttr<CommandApplyBatch>("applybatch", -2, "write no-multi", NO_KEY),
+    MakeCmdAttr<CommandDump>("dump", 2, "read-only", 1, 1, 1),
+    MakeCmdAttr<CommandPollUpdates>("pollupdates", -2, "read-only admin", NO_KEY),
+    MakeCmdAttr<CommandSST>("sst", -3, "write exclusive admin", 1, 1, 1),
+    MakeCmdAttr<CommandFlushMemTable>("flushmemtable", -1, "exclusive write", NO_KEY),
+    MakeCmdAttr<CommandFlushBlockCache>("flushblockcache", 1, "exclusive write", NO_KEY), )
 }  // namespace redis

src/config/config.cc

@@ -203,6 +203,8 @@ Config::Config() {
       {"slave-read-only", false, new YesNoField(&slave_readonly, true)},
       {"replication-connect-timeout-ms", false, new IntField(&replication_connect_timeout_ms, 3100, 0, INT_MAX)},
       {"replication-recv-timeout-ms", false, new IntField(&replication_recv_timeout_ms, 3200, 0, INT_MAX)},
+      {"replication-delay-bytes", false, new IntField(&max_replication_delay_bytes, 16 * 1024, 1, INT_MAX)},
+      {"replication-delay-updates", false, new IntField(&max_replication_delay_updates, 16, 1, INT_MAX)},
       {"use-rsid-psync", true, new YesNoField(&use_rsid_psync, false)},
       {"profiling-sample-ratio", false, new IntField(&profiling_sample_ratio, 0, 0, 100)},
       {"profiling-sample-record-max-len", false, new IntField(&profiling_sample_record_max_len, 256, 0, INT_MAX)},

src/config/config.h

@@ -120,6 +120,8 @@ struct Config {
   int slave_priority = 100;
   int replication_connect_timeout_ms = 3100;
   int replication_recv_timeout_ms = 3200;
+  int max_replication_delay_bytes = 16 * 1024;  // 16KB default
+  int max_replication_delay_updates = 16;       // 16 updates default
   int max_db_size = 0;
   int max_replication_mb = 0;
   int max_io_mb = 0;

src/search/executors/filter_executor.h

@@ -78,6 +78,7 @@ struct QueryExprEvaluator {
 
   StatusOr<bool> Visit(TagContainExpr *v) const {
     auto val = GET_OR_RET(ctx->Retrieve(ctx->db_ctx, row, v->field->info));
+    if (val.IsNull()) return false;
 
     CHECK(val.Is<kqir::StringArray>());
     auto tags = val.Get<kqir::StringArray>();
@@ -93,6 +94,7 @@ struct QueryExprEvaluator {
 
   StatusOr<bool> Visit(NumericCompareExpr *v) const {
     auto l_val = GET_OR_RET(ctx->Retrieve(ctx->db_ctx, row, v->field->info));
+    if (l_val.IsNull()) return false;
 
     CHECK(l_val.Is<kqir::Numeric>());
     auto l = l_val.Get<kqir::Numeric>();
@@ -118,6 +120,7 @@ struct QueryExprEvaluator {
 
   StatusOr<bool> Visit(VectorRangeExpr *v) const {
     auto val = GET_OR_RET(ctx->Retrieve(ctx->db_ctx, row, v->field->info));
+    if (val.IsNull()) return false;
 
     CHECK(val.Is<kqir::NumericArray>());
     auto l_values = val.Get<kqir::NumericArray>();

src/search/executors/topn_executor.h

@@ -58,6 +58,9 @@ struct TopNExecutor : ExecutorNode {
 
         auto get_order = [this](RowType &row) -> StatusOr<double> {
           auto order_val = GET_OR_RET(ctx->Retrieve(ctx->db_ctx, row, topn->order->field->info));
+          // TODO(twice): here we return NaN if this field is not found,
+          // but we should consider to just skip this row instead.
+          if (order_val.IsNull()) return std::nan("");
           CHECK(order_val.Is<kqir::Numeric>());
           return order_val.Get<kqir::Numeric>();
         };