feat: support blpop/brpop

.github/workflows/pika.yml

@@ -20,17 +20,17 @@ jobs:
     steps:
       - uses: actions/checkout@v3
 
-      - name: cache dependencies
-        uses: actions/cache@v2
-        id: cache
-        with:
-          path: |
-            ${{ github.workspace }}/${{ env.INSTALL_LOCATION }}
-            ~/.cache/pip
-          key: ${{ runner.os }}-dependencies
+#      - name: cache dependencies
+#        uses: actions/cache@v2
+#        id: cache
+#        with:
+#          path: |
+#            ${{ github.workspace }}/${{ env.INSTALL_LOCATION }}
+#            ~/.cache/pip
+#          key: ${{ runner.os }}-dependencies
 
       - name: install Deps
-        if: ${{ steps.cache.output.cache-hit != 'true' }}
+#        if: ${{ steps.cache.output.cache-hit != 'true' }}
         run: |
           sudo apt install autoconf libprotobuf-dev protobuf-compiler -y
           sudo apt-get install -y clang-tidy-12
@@ -69,7 +69,7 @@ jobs:
         working-directory: ${{github.workspace}}/build
         run: |
           python3 ../tests/integration/pika_replication_test.py
-
+          python3 ../tests/unit/Blpop_Brpop_test.py
 
   build_on_centos:
     # The CMake configure and build commands are platform agnostic and should work equally well on Windows or Mac.

CMakeLists.txt

@@ -22,13 +22,17 @@ elseif (CMAKE_CXX_COMPILER_ID STREQUAL "GNU")
   endif()
 endif()
 
-#Sanitizer for testing stage
-#CMAKE_BUILD_TYPE must be "Debug" if you wanna use sanitizer.
+
+############# You should enable sanitizer if you are developing pika #############
+# Uncomment the following two lines to enable AddressSanitizer to detect memory leaks and other memory-related bugs.
 #set(CMAKE_BUILD_TYPE "Debug")
+#set(CMAKE_CXX_FLAGS_DEBUG "${CMAKE_CXX_FLAGS_DEBUG} -fsanitize=address  -O0 -fno-omit-frame-pointer -fno-optimize-sibling-calls")
+
+# [Notice] AddressSanitizer and ThreadSanitizer can not be enabled at the same time.
 
-#set(CMAKE_CXX_FLAGS_DEBUG "${CMAKE_CXX_FLAGS_DEBUG} -fsanitize=address")
-# [Notice] address sanitizer and thread sanitizer can not be enabled at the same time.
-#set(CMAKE_CXX_FLAGS_DEBUG "${CMAKE_CXX_FLAGS_DEBUG} -fsanitize=thread")
+# Uncomment the following two lines to enable ThreadSanitizer to detect data race and other thread-related issue.
+#set(CMAKE_BUILD_TYPE "Debug")
+#set(CMAKE_CXX_FLAGS_DEBUG "${CMAKE_CXX_FLAGS_DEBUG} -fsanitize=thread -O0 -fno-omit-frame-pointer -fno-optimize-sibling-calls")
 
 execute_process(COMMAND uname -p OUTPUT_VARIABLE HOST_ARCH)
 string(TOLOWER ${HOST_ARCH} HOST_ARCH)

include/pika_client_conn.h

@@ -55,6 +55,7 @@ class PikaClientConn : public net::RedisConn {
   bool IsPubSub() { return is_pubsub_; }
   void SetIsPubSub(bool is_pubsub) { is_pubsub_ = is_pubsub; }
   void SetCurrentTable(const std::string& db_name) { current_db_ = db_name; }
+  const std::string& GetCurrentTable() override{ return current_db_; }
   void SetWriteCompleteCallback(WriteCompleteCallback cb) { write_completed_cb_ = std::move(cb); }
 
   net::ServerThread* server_thread() { return server_thread_; }

include/pika_command.h

@@ -16,6 +16,7 @@
 #include "pstd/include/pstd_string.h"
 
 #include "include/pika_slot.h"
+#include "net/src/dispatch_thread.h"
 
 class SyncMasterSlot;
 class SyncSlaveSlot;
@@ -131,13 +132,15 @@ const std::string kCmdNamePKHRScanRange = "pkhrscanrange";
 const std::string kCmdNameLIndex = "lindex";
 const std::string kCmdNameLInsert = "linsert";
 const std::string kCmdNameLLen = "llen";
+const std::string kCmdNameBLPop = "blpop";
 const std::string kCmdNameLPop = "lpop";
 const std::string kCmdNameLPush = "lpush";
 const std::string kCmdNameLPushx = "lpushx";
 const std::string kCmdNameLRange = "lrange";
 const std::string kCmdNameLRem = "lrem";
 const std::string kCmdNameLSet = "lset";
 const std::string kCmdNameLTrim = "ltrim";
+const std::string kCmdNameBRpop = "brpop";
 const std::string kCmdNameRPop = "rpop";
 const std::string kCmdNameRPopLPush = "rpoplpush";
 const std::string kCmdNameRPush = "rpush";
@@ -398,6 +401,18 @@ class CmdRes {
   CmdRet ret_ = kNone;
 };
 
+/**
+ * Current used by:
+ * blpop,brpop
+ */
+struct UnblockTaskArgs {
+  std::string key;
+  std::shared_ptr<Slot> slot;
+  net::DispatchThread* dispatchThread;
+  UnblockTaskArgs(std::string key_, std::shared_ptr<Slot> slot_, net::DispatchThread* dispatchThread_)
+      : key(std::move(key_)), slot(slot_), dispatchThread(dispatchThread_) {}
+};
+
 class Cmd : public std::enable_shared_from_this<Cmd> {
  public:
   enum CmdStage { kNone, kBinlogStage, kExecuteStage };
@@ -449,6 +464,7 @@ class Cmd : public std::enable_shared_from_this<Cmd> {
 
   bool is_read() const;
   bool is_write() const;
+
   bool is_local() const;
   bool is_suspend() const;
   bool is_admin_require() const;

include/pika_list.h

@@ -72,6 +72,50 @@ class LLenCmd : public Cmd {
   void DoInitial() override;
 };
 
+class BlockingBaseCmd : public Cmd {
+ public:
+  BlockingBaseCmd(const std::string& name, int arity, uint16_t flag) : Cmd(name, arity, flag) {}
+
+  //blpop/brpop used start
+  struct WriteBinlogOfPopArgs{
+    BlockKeyType block_type;
+    std::string key;
+    std::shared_ptr<Slot> slot;
+    std::shared_ptr<net::NetConn> conn;
+    WriteBinlogOfPopArgs() = default;
+    WriteBinlogOfPopArgs(BlockKeyType block_type_, const std::string& key_,
+                         std::shared_ptr<Slot> slot_, std::shared_ptr<net::NetConn> conn_)
+        : block_type(block_type_), key(key_), slot(slot_), conn(conn_){}
+  };
+  void BlockThisClientToWaitLRPush(BlockKeyType block_pop_type, std::vector<std::string>& keys, int64_t expire_time);
+  void TryToServeBLrPopWithThisKey(const std::string& key, std::shared_ptr<Slot> slot);
+  static void ServeAndUnblockConns(void* args);
+  static void WriteBinlogOfPop(std::vector<WriteBinlogOfPopArgs>& pop_args);
+  void removeDuplicates(std::vector<std::string> & keys_);
+  //blpop/brpop used functions end
+};
+
+class BLPopCmd final : public BlockingBaseCmd {
+ public:
+  BLPopCmd(const std::string& name, int arity, uint16_t flag) : BlockingBaseCmd(name, arity, flag){};
+  virtual std::vector<std::string> current_key() const {
+    std::vector<std::string> res = keys_;
+    return res;
+  }
+  virtual void Do(std::shared_ptr<Slot> slot = nullptr);
+  virtual void Split(std::shared_ptr<Slot> slot, const HintKeys& hint_keys){};
+  virtual void Merge(){};
+  virtual Cmd* Clone() override { return new BLPopCmd(*this); }
+  void DoInitial() override;
+  void DoBinlog(const std::shared_ptr<SyncMasterSlot>& slot) override;
+
+ private:
+  std::vector<std::string> keys_;
+  int64_t expire_time_{0};
+  WriteBinlogOfPopArgs binlog_args_;
+  bool is_binlog_deferred_{false};
+};
+
 class LPopCmd : public Cmd {
  public:
   LPopCmd(const std::string& name, int arity, uint16_t flag) : Cmd(name, arity, flag){};
@@ -90,9 +134,10 @@ class LPopCmd : public Cmd {
   void DoInitial() override;
 };
 
-class LPushCmd : public Cmd {
+
+class LPushCmd : public BlockingBaseCmd {
  public:
-  LPushCmd(const std::string& name, int arity, uint16_t flag) : Cmd(name, arity, flag){};
+  LPushCmd(const std::string& name, int arity, uint16_t flag) : BlockingBaseCmd(name, arity, flag){};
   std::vector<std::string> current_key() const override {
     std::vector<std::string> res;
     res.push_back(key_);
@@ -209,6 +254,26 @@ class LTrimCmd : public Cmd {
   void DoInitial() override;
 };
 
+class BRPopCmd final : public BlockingBaseCmd {
+ public:
+  BRPopCmd(const std::string& name, int arity, uint16_t flag) : BlockingBaseCmd(name, arity, flag){};
+  virtual std::vector<std::string> current_key() const {
+    std::vector<std::string> res = keys_;
+    return res;
+  }
+  virtual void Do(std::shared_ptr<Slot> slot = nullptr);
+  virtual void Split(std::shared_ptr<Slot> slot, const HintKeys& hint_keys){};
+  virtual void Merge(){};
+  virtual Cmd* Clone() override { return new BRPopCmd(*this); }
+  void DoInitial() override;
+  void DoBinlog(const std::shared_ptr<SyncMasterSlot>& slot) override;
+ private:
+  std::vector<std::string> keys_;
+  int64_t expire_time_{0};
+  WriteBinlogOfPopArgs binlog_args_;
+  bool is_binlog_deferred_{false};
+};
+
 class RPopCmd : public Cmd {
  public:
   RPopCmd(const std::string& name, int arity, uint16_t flag) : Cmd(name, arity, flag){};
@@ -227,14 +292,14 @@ class RPopCmd : public Cmd {
   void DoInitial() override;
 };
 
-class RPopLPushCmd : public Cmd {
+class RPopLPushCmd : public BlockingBaseCmd {
  public:
-  RPopLPushCmd(const std::string& name, int arity, uint16_t flag) : Cmd(name, arity, flag) {
+  RPopLPushCmd(const std::string& name, int arity, uint16_t flag) : BlockingBaseCmd(name, arity, flag) {
     rpop_cmd_ = std::make_shared<RPopCmd>(kCmdNameRPop, 2, kCmdFlagsWrite | kCmdFlagsSingleSlot | kCmdFlagsList);
     lpush_cmd_ = std::make_shared<LPushCmd>(kCmdNameLPush, -3, kCmdFlagsWrite | kCmdFlagsSingleSlot | kCmdFlagsList);
   };
   RPopLPushCmd(const RPopLPushCmd& other)
-      : Cmd(other),
+      : BlockingBaseCmd(other),
         source_(other.source_),
         receiver_(other.receiver_),
         value_poped_from_source_(other.value_poped_from_source_),
@@ -265,9 +330,9 @@ class RPopLPushCmd : public Cmd {
   void DoInitial() override;
 };
 
-class RPushCmd : public Cmd {
+class RPushCmd : public BlockingBaseCmd {
  public:
-  RPushCmd(const std::string& name, int arity, uint16_t flag) : Cmd(name, arity, flag){};
+  RPushCmd(const std::string& name, int arity, uint16_t flag) : BlockingBaseCmd(name, arity, flag){};
   std::vector<std::string> current_key() const override {
     std::vector<std::string> res;
     res.push_back(key_);

src/net/include/redis_conn.h

@@ -39,6 +39,7 @@ class RedisConn : public NetConn {
   void NotifyEpoll(bool success);
 
   virtual int DealMessage(const RedisCmdArgsType& argv, std::string* response) = 0;
+  virtual const std::string& GetCurrentTable() = 0;
 
  private:
   static int ParserDealMessageCb(RedisParser* parser, const RedisCmdArgsType& argv);

src/net/include/server_thread.h

@@ -20,6 +20,7 @@
 #include "net/include/net_define.h"
 #include "net/include/net_thread.h"
 #include "net/src/net_multiplexer.h"
+#include "pstd/include/env.h"
 #include "pstd/include/pstd_mutex.h"
 #include "pstd/include/pstd_status.h"
 
@@ -111,9 +112,11 @@ const int kDefaultKeepAliveTime = 60;  // (s)
 
 class ServerThread : public Thread {
  public:
-  ServerThread(int port, int cron_interval, const ServerHandle* handle);
-  ServerThread(const std::string& bind_ip, int port, int cron_interval, const ServerHandle* handle);
-  ServerThread(const std::set<std::string>& bind_ips, int port, int cron_interval, const ServerHandle* handle);
+  ServerThread(int port, int cron_interval, const ServerHandle* handle, ServerThread* dispatcher = nullptr);
+  ServerThread(const std::string& bind_ip, int port, int cron_interval, const ServerHandle* handle,
+               ServerThread* dispatcher = nullptr);
+  ServerThread(const std::set<std::string>& bind_ips, int port, int cron_interval, const ServerHandle* handle,
+               ServerThread* dispatcher = nullptr);
 
 #ifdef __ENABLE_SSL
   /*
@@ -175,6 +178,9 @@ class ServerThread : public Thread {
   // process events in notify_queue
   virtual void ProcessNotifyEvents(const NetFiredEvent* pfe);
 
+  //dispatcher's life time should not be managed by this serverThread,
+  //so, use raw pointer instead of smart pointer
+  ServerThread* dispatcher_;
   const ServerHandle* handle_;
   bool own_handle_ = false;