Merge branch 'unstable' into fix-hash-rename

Author: dobet

.github/workflows/kiwidb.yml

@@ -71,7 +71,7 @@ jobs:
         working-directory: ${{ github.workspace }}/${{ env.BUILD_DIR }}
         # Execute tests defined by the CMake configuration.
         # See https://cmake.org/cmake/help/latest/manual/ctest.1.html for more detail
-        run: ctest
+        run: ctest --output-on-failure --timeout 600
 
       - name: Run TCL E2e Tests
         working-directory: ${{ github.workspace }}
@@ -113,7 +113,7 @@ jobs:
         working-directory: ${{ github.workspace }}/${{ env.BUILD_DIR }}
         # Execute tests defined by the CMake configuration.
         # See https://cmake.org/cmake/help/latest/manual/ctest.1.html for more detail
-        run: ctest
+        run: ctest --output-on-failure --timeout 600
 
       - name: Run TCL E2e Tests
         working-directory: ${{ github.workspace }}

.gitmodules

@@ -1,4 +1,4 @@
-# Copyright (c) 2023-present, Arana/Kiwi Community.  All rights reserved.
+# Copyright (c) 2023-present, arana-db Community.  All rights reserved.
 # This source code is licensed under the BSD-style license found in the
 # LICENSE file in the root directory of this source tree. An additional grant
 # of patent rights can be found in the PATENTS file in the same directory.

CMakeLists.txt

@@ -1,4 +1,4 @@
-Copyright (c) 2015-2023, OpenAtom Foundation
+Copyright (c) 2015-2023, arana-db Foundation
 All rights reserved.
 
 Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are met:

LICENSE.md

@@ -12,7 +12,7 @@ ExternalProject_Add(
         ${EXTERNAL_PROJECT_LOG_ARGS}
         DEPENDS gflags snappy zlib lz4 zstd
         GIT_REPOSITORY https://github.com/facebook/rocksdb.git
-        GIT_TAG v9.4.0
+        GIT_TAG v9.11.1
         GIT_SHALLOW true
         SOURCE_DIR ${ROCKSDB_SOURCES_DIR}
         CMAKE_ARGS

cmake/rocksdb.cmake

@@ -43,9 +43,9 @@ main() {
     echo "Starting code format check..."
 
     # Find all C/C++ source files
-    while IFS= read -r -d '' file; do
+    find ./src -type f \( -name "*.cpp" -o -name "*.hpp" -o -name "*.c" -o -name "*.h" -o -name "*.cc" \) -print0 | while IFS= read -r -d '' file; do
         check_file "$file"
-    done < <(find ./src -type f \( -name "*.cpp" -o -name "*.hpp" -o -name "*.c" -o -name "*.h" -o -name "*.cc" \) -print0)
+    done
 
     # Output summary
     echo "----------------------------------------"

etc/script/check_format.sh

@@ -1,4 +1,4 @@
-# Copyright (c) 2023-present, Arana/Kiwi Community.  All rights reserved.
+# Copyright (c) 2023-present, arana-db Community.  All rights reserved.
 # This source code is licensed under the BSD-style license found in the
 # LICENSE file in the root directory of this source tree. An additional grant
 # of patent rights can be found in the PATENTS file in the same directory.

src/CMakeLists.txt

@@ -1,4 +1,4 @@
-// Copyright (c) 2023-present, Arana/Kiwi Community.  All rights reserved.
+// Copyright (c) 2023-present, arana-db Community.  All rights reserved.
 // This source code is licensed under the BSD-style license found in the
 // LICENSE file in the root directory of this source tree. An additional grant
 // of patent rights can be found in the PATENTS file in the same directory
@@ -9,15 +9,12 @@
 
 #include "base_cmd.h"
 
-#include "fmt/core.h"
-
 #include "raft/raft.h"
 
 #include "common.h"
 #include "config.h"
 #include "kiwi.h"
 #include "log.h"
-#include "raft/raft.h"
 
 namespace kiwi {
 
@@ -44,16 +41,19 @@ void BaseCmd::Execute(PClient* client) {
   // read consistency (lease read) / write redirection
   if (g_config.use_raft && (HasFlag(kCmdFlagsReadonly) || HasFlag(kCmdFlagsWrite))) {
     if (!RAFT_INST.IsInitialized()) {
-      return client->SetRes(CmdRes::kErrOther, "RAFT_INST is not initialized");
+      client->SetRes(CmdRes::kErrOther, "RAFT_INST is not initialized");
+      return;
     }
 
     if (!RAFT_INST.IsLeader()) {
       auto leader_addr = RAFT_INST.GetLeaderAddress();
       if (leader_addr.empty()) {
-        return client->SetRes(CmdRes::kErrOther, std::string("-CLUSTERDOWN No Raft leader"));
+        client->SetRes(CmdRes::kErrClusterDown, "No raft leader");
+        return;
       }
 
-      return client->SetRes(CmdRes::kErrOther, fmt::format("-MOVED {}", leader_addr));
+      client->SetRes(CmdRes::kErrMoved, leader_addr);
+      return;
     }
   }
 
@@ -106,6 +106,84 @@ BaseCmd* BaseCmdGroup::GetSubCmd(const std::string& cmdName) {
   return subCmd->second.get();
 }
 
+void BaseCmd::BlockThisClientToWaitLRPush(std::vector<std::string>& keys, int64_t expire_time,
+                                          std::shared_ptr<PClient> client, BlockedConnNode::Type type) {
+  std::lock_guard<std::shared_mutex> map_lock(g_kiwi->GetBlockMtx());
+  auto& key_to_conns = g_kiwi->GetMapFromKeyToConns();
+  for (const auto& key : keys) {
+    kiwi::BlockKey blpop_key{client->GetCurrentDB(), key};
+
+    auto it = key_to_conns.find(blpop_key);
+    if (it == key_to_conns.end()) {
+      key_to_conns.emplace(blpop_key, std::make_unique<std::list<BlockedConnNode>>());
+      it = key_to_conns.find(blpop_key);
+    }
+    it->second->emplace_back(expire_time, client, type);
+  }
+}
+
+void BaseCmd::ServeAndUnblockConns(PClient* client) {
+  kiwi::BlockKey key{client->GetCurrentDB(), client->Key()};
+
+  std::lock_guard<std::shared_mutex> map_lock(g_kiwi->GetBlockMtx());
+  auto& key_to_conns = g_kiwi->GetMapFromKeyToConns();
+  auto it = key_to_conns.find(key);
+  if (it == key_to_conns.end()) {
+    // no client is waitting for this key
+    return;
+  }
+
+  auto& waitting_list = it->second;
+  std::vector<std::string> elements;
+  storage::Status s;
+
+  // traverse this list from head to tail(in the order of adding sequence) ,means "first blocked, first get served“
+  for (auto conn_blocked = waitting_list->begin(); conn_blocked != waitting_list->end();) {
+    auto BlockedClient = conn_blocked->GetBlockedClient();
+
+    if (BlockedClient->State() == ClientState::kClosed) {
+      conn_blocked = waitting_list->erase(conn_blocked);
+      g_kiwi->CleanBlockedNodes(BlockedClient);
+      continue;
+    }
+
+    switch (conn_blocked->GetCmdType()) {
+      case BlockedConnNode::Type::BLPop:
+        s = STORE_INST.GetBackend(client->GetCurrentDB())->GetStorage()->LPop(client->Key(), 1, &elements);
+        break;
+      case BlockedConnNode::Type::BRPop:
+        s = STORE_INST.GetBackend(client->GetCurrentDB())->GetStorage()->RPop(client->Key(), 1, &elements);
+        break;
+      case BlockedConnNode::Type::NotAny:
+        //! DOING NOTHING?
+        break;
+    }
+
+    if (s.ok()) {
+      BlockedClient->AppendArrayLen(2);
+      BlockedClient->AppendString(client->Key());
+      BlockedClient->AppendString(elements[0]);
+    } else if (s.IsNotFound()) {
+      // this key has no more elements to serve more blocked conn.
+      break;
+    } else {
+      BlockedClient->SetRes(CmdRes::kErrOther, s.ToString());
+    }
+    BlockedClient->SendPacket();
+    // remove this conn from current waiting list
+    conn_blocked = waitting_list->erase(conn_blocked);
+    g_kiwi->CleanBlockedNodes(BlockedClient);
+  }
+}
+
+bool BlockedConnNode::IsExpired(std::chrono::system_clock::time_point now) {
+  if (expire_time_ == 0) {
+    return false;
+  }
+  int64_t now_in_ms = std::chrono::time_point_cast<std::chrono::milliseconds>(now).time_since_epoch().count();
+  return expire_time_ <= now_in_ms;
+}
+
 bool BaseCmdGroup::DoInitial(PClient* client) {
   client->SetSubCmdName(client->argv_[1]);
   if (!subCmds_.contains(client->SubCmdName())) {

src/base_cmd.cc

@@ -1,4 +1,4 @@
-// Copyright (c) 2023-present, Arana/Kiwi Community.  All rights reserved.
+// Copyright (c) 2023-present, arana-db Community.  All rights reserved.
 // This source code is licensed under the BSD-style license found in the
 // LICENSE file in the root directory of this source tree. An additional grant
 // of patent rights can be found in the PATENTS file in the same directory
@@ -144,6 +144,8 @@ const std::string kCmdNameRPush = "rpush";
 const std::string kCmdNameRPushx = "rpushx";
 const std::string kCmdNameLPop = "lpop";
 const std::string kCmdNameRPop = "rpop";
+const std::string kCmdNameBLPop = "blpop";
+const std::string kCmdNameBRPop = "brpop";
 const std::string kCmdNameLRem = "lrem";
 const std::string kCmdNameLRange = "lrange";
 const std::string kCmdNameLTrim = "ltrim";
@@ -190,7 +192,7 @@ enum CmdFlags {
   kCmdFlagsProtected = (1 << 12),        // Don't accept in scripts
   kCmdFlagsModuleNoCluster = (1 << 13),  // No cluster mode support
   kCmdFlagsNoMulti = (1 << 14),          // Cannot be pipelined
-  kCmdFlagsExclusive = (1 << 15),        // May change Storage pointer, like Arana/Kiwi's kCmdFlagsSuspend
+  kCmdFlagsExclusive = (1 << 15),        // May change Storage pointer, like pika's kCmdFlagsSuspend
   kCmdFlagsRaft = (1 << 16),             // raft
 };
 
@@ -219,6 +221,22 @@ enum AclCategory {
   kAclCategoryRaft = (1 << 21),
 };
 
+class BlockedConnNode {
+ public:
+  enum Type { NotAny = 0, BLPop, BRPop };
+  virtual ~BlockedConnNode() = default;
+  BlockedConnNode(int64_t expire_time, std::shared_ptr<PClient> client, Type type)
+      : expire_time_(expire_time), client_(client), type_(type) {}
+  bool IsExpired(std::chrono::system_clock::time_point now = std::chrono::system_clock::now());
+  std::shared_ptr<PClient> GetBlockedClient() { return client_; }
+  Type GetCmdType() { return type_; }
+
+ private:
+  Type type_ = NotAny;
+  int64_t expire_time_ = 0;
+  std::shared_ptr<PClient> client_;
+};
+
 /**
  * @brief Base class for all commands
  * BaseCmd, as the base class for all commands, mainly implements some common functions
@@ -282,6 +300,11 @@ class BaseCmd : public std::enable_shared_from_this<BaseCmd> {
 
   uint32_t GetCmdID() const;
 
+  void ServeAndUnblockConns(PClient* client);
+
+  void BlockThisClientToWaitLRPush(std::vector<std::string>& keys, int64_t expire_time, std::shared_ptr<PClient> client,
+                                   BlockedConnNode::Type type);
+
  protected:
   // Execute a specific command
   virtual void DoCmd(PClient* client) = 0;
@@ -321,4 +344,18 @@ class BaseCmdGroup : public BaseCmd {
  private:
   std::map<std::string, std::unique_ptr<BaseCmd>> subCmds_;
 };
+
+struct BlockKey {  // this data struct is made for the scenario of multi dbs in kiwi.
+  BlockKey(int db_id, std::string key) : db_id_(db_id), key_(std::move(key)) {}
+
+  int db_id_ = -1;
+  std::string key_;
+  bool operator==(const BlockKey& p) const { return p.db_id_ == db_id_ && p.key_ == key_; }
+};
+struct BlockKeyHash {
+  std::size_t operator()(const BlockKey& k) const {
+    return std::hash<int>{}(k.db_id_) ^ std::hash<std::string>{}(k.key_);
+  }
+};
+
 }  // namespace kiwi

src/base_cmd.h

@@ -1,4 +1,4 @@
-// Copyright (c) 2023-present, Arana/Kiwi Community.  All rights reserved.
+// Copyright (c) 2023-present, arana-db Community.  All rights reserved.
 // This source code is licensed under the BSD-style license found in the
 // LICENSE file in the root directory of this source tree. An additional grant
 // of patent rights can be found in the PATENTS file in the same directory
@@ -14,12 +14,9 @@
 #include "base_cmd.h"
 #include "client.h"
 #include "config.h"
-#include "env.h"
 #include "kiwi.h"
 #include "raft/raft.h"
-#include "slow_log.h"
 #include "std/log.h"
-#include "std/std_string.h"
 
 namespace kiwi {
 
@@ -303,6 +300,7 @@ bool PClient::isPeerMaster() const {
   return repl_addr.GetIP() == PeerIP() && repl_addr.GetPort() == PeerPort();
 }
 
+// check if the client is the target of the cluster command
 bool PClient::isClusterCmdTarget() const {
   return RAFT_INST.GetClusterCmdCtx().GetPeerIp() == PeerIP() && RAFT_INST.GetClusterCmdCtx().GetPort() == PeerPort();
 }