feat: support blpop/brpop

include/pika_client_conn.h

@@ -53,6 +53,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& table_name) { current_table_ = table_name; }
+  const std::string& GetCurrentTable() override{ return current_table_; }
   void SetWriteCompleteCallback(WriteCompleteCallback cb) { write_completed_cb_ = cb; }
 
   net::ServerThread* server_thread() { return server_thread_; }

include/pika_command.h

@@ -108,13 +108,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";

include/pika_list.h

@@ -10,6 +10,7 @@
 
 #include "include/pika_command.h"
 #include "include/pika_partition.h"
+#include "net/src/dispatch_thread.h"
 
 /*
  * list
@@ -73,6 +74,29 @@ class LLenCmd : public Cmd {
   virtual void DoInitial() override;
 };
 
+class BLRPopBaseCmd : public Cmd {
+ public:
+  BLRPopBaseCmd(const std::string& name, int arity, uint16_t flag) : Cmd(name, arity, flag) {}
+  void BlockThisClientToWaitLRPush(net::BlockPopType block_pop_type);
+ protected:
+  void DoInitial() override;
+  std::vector<std::string> keys_;
+  int64_t expire_time_{0};
+};
+
+class BLPopCmd : public BLRPopBaseCmd {
+ public:
+  BLPopCmd(const std::string& name, int arity, uint16_t flag) : BLRPopBaseCmd(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<Partition> partition = nullptr);
+  virtual void Split(std::shared_ptr<Partition> partition, const HintKeys& hint_keys){};
+  virtual void Merge(){};
+  virtual Cmd* Clone() override { return new BLPopCmd(*this); }
+};
+
 class LPopCmd : public Cmd {
  public:
   LPopCmd(const std::string& name, int arity, uint16_t flag) : Cmd(name, arity, flag){};
@@ -91,9 +115,16 @@ class LPopCmd : public Cmd {
   virtual void DoInitial() override;
 };
 
-class LPushCmd : public Cmd {
+class BPopServeCmd : public Cmd {
  public:
-  LPushCmd(const std::string& name, int arity, uint16_t flag) : Cmd(name, arity, flag){};
+  BPopServeCmd(const std::string& name, int arity, uint16_t flag) : Cmd(name, arity, flag) {}
+  void TryToServeBLrPopWithThisKey(const std::string& key, std::shared_ptr<Partition> partition);
+  static void ServeAndUnblockConns(void* args);
+};
+
+class LPushCmd : public BPopServeCmd {
+ public:
+  LPushCmd(const std::string& name, int arity, uint16_t flag) : BPopServeCmd(name, arity, flag){};
   virtual std::vector<std::string> current_key() const {
     std::vector<std::string> res;
     res.push_back(key_);
@@ -210,6 +241,20 @@ class LTrimCmd : public Cmd {
   virtual void DoInitial() override;
 };
 
+class BRPopCmd : public BLRPopBaseCmd {
+ public:
+  BRPopCmd(const std::string& name, int arity, uint16_t flag) : BLRPopBaseCmd(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<Partition> partition = nullptr);
+  virtual void Split(std::shared_ptr<Partition> partition, const HintKeys& hint_keys){};
+  virtual void Merge(){};
+  virtual Cmd* Clone() override { return new BRPopCmd(*this); }
+
+};
+
 class RPopCmd : public Cmd {
  public:
   RPopCmd(const std::string& name, int arity, uint16_t flag) : Cmd(name, arity, flag){};
@@ -228,9 +273,9 @@ class RPopCmd : public Cmd {
   virtual void DoInitial() override;
 };
 
-class RPopLPushCmd : public Cmd {
+class RPopLPushCmd : public BPopServeCmd {
  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) : BPopServeCmd(name, arity, flag){};
   std::vector<std::string> current_key() const override {
     std::vector<std::string> res;
     res.push_back(source_);
@@ -247,9 +292,9 @@ class RPopLPushCmd : public Cmd {
   virtual void DoInitial() override;
 };
 
-class RPushCmd : public Cmd {
+class RPushCmd : public BPopServeCmd {
  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) : BPopServeCmd(name, arity, flag){};
   virtual std::vector<std::string> current_key() const {
     std::vector<std::string> res;
     res.push_back(key_);

include/pika_server.h

@@ -127,6 +127,7 @@ enum TaskType {
   kBgSave,
 };
 
+
 class PikaServer {
  public:
   PikaServer();
@@ -324,6 +325,8 @@ class PikaServer {
   storage::Status RewriteStorageOptions(const storage::OptionType& option_type,
                                         const std::unordered_map<std::string, std::string>& options);
 
+
+
   friend class Cmd;
   friend class InfoCmd;
   friend class PkClusterAddSlotsCmd;
@@ -360,6 +363,7 @@ class PikaServer {
   std::shared_mutex tables_rw_;
   std::map<std::string, std::shared_ptr<Table>> tables_;
 
+
   /*
    * CronTask used
    */
@@ -377,15 +381,15 @@ class PikaServer {
    * Slave used
    */
   std::string master_ip_;
-  int master_port_  = 0;
-  int repl_state_  = PIKA_REPL_NO_CONNECT;
+  int master_port_ = 0;
+  int repl_state_ = PIKA_REPL_NO_CONNECT;
   int role_ = PIKA_ROLE_SINGLE;
   int last_meta_sync_timestamp_ = 0;
   bool first_meta_sync_ = false;
   bool loop_partition_state_machine_ = false;
   bool force_full_sync_ = false;
-  bool leader_protected_mode_ = false;        // reject request after master slave sync done
-  std::shared_mutex state_protector_;         // protect below, use for master-slave mode
+  bool leader_protected_mode_ = false;  // reject request after master slave sync done
+  std::shared_mutex state_protector_;   // protect below, use for master-slave mode
 
   /*
    * Bgsave used

src/net/include/redis_conn.h

@@ -39,7 +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);
   static int ParserCompleteCb(RedisParser* parser, const std::vector<RedisCmdArgsType>& argvs);

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,9 @@ 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 +176,7 @@ class ServerThread : public Thread {
   // process events in notify_queue
   virtual void ProcessNotifyEvents(const NetFiredEvent* pfe);
 
+  ServerThread* dispatcher_;
   const ServerHandle* handle_;
   bool own_handle_ = false;
 
@@ -197,6 +199,7 @@ class ServerThread : public Thread {
    * The server event handle
    */
   virtual void HandleConnEvent(NetFiredEvent* pfe) = 0;
+
 };
 
 // !!!Attention: If u use this constructor, the keepalive_timeout_ will

src/net/src/dispatch_thread.cc

@@ -16,37 +16,37 @@ namespace net {
 
 DispatchThread::DispatchThread(int port, int work_num, ConnFactory* conn_factory, int cron_interval, int queue_limit,
                                const ServerHandle* handle)
-    : ServerThread::ServerThread(port, cron_interval, handle),
+    : ServerThread::ServerThread(port, cron_interval, handle, this),
       last_thread_(0),
       work_num_(work_num),
-      queue_limit_(queue_limit) {
+      queue_limit_(queue_limit),
+      timedTaskManager(net_multiplexer_->GetMultiplexer()) {
   for (int i = 0; i < work_num_; i++) {
     worker_thread_.emplace_back(std::make_unique<WorkerThread>(conn_factory, this, queue_limit, cron_interval));
   }
 }
 
 DispatchThread::DispatchThread(const std::string& ip, int port, int work_num, ConnFactory* conn_factory,
                                int cron_interval, int queue_limit, const ServerHandle* handle)
-    : ServerThread::ServerThread(ip, port, cron_interval, handle),
+    : ServerThread::ServerThread(ip, port, cron_interval, handle, this),
       last_thread_(0),
       work_num_(work_num),
-      queue_limit_(queue_limit) {
-
+      queue_limit_(queue_limit),
+      timedTaskManager(net_multiplexer_->GetMultiplexer()){
   for (int i = 0; i < work_num_; i++) {
     worker_thread_.emplace_back(std::make_unique<WorkerThread>(conn_factory, this, queue_limit, cron_interval));
   }
 }
 
 DispatchThread::DispatchThread(const std::set<std::string>& ips, int port, int work_num, ConnFactory* conn_factory,
                                int cron_interval, int queue_limit, const ServerHandle* handle)
-    : ServerThread::ServerThread(ips, port, cron_interval, handle),
+    : ServerThread::ServerThread(ips, port, cron_interval, handle, this),
       last_thread_(0),
       work_num_(work_num),
-      queue_limit_(queue_limit) {
-
+      queue_limit_(queue_limit),
+      timedTaskManager(net_multiplexer_->GetMultiplexer()){
   for (int i = 0; i < work_num_; i++) {
     worker_thread_.emplace_back(std::make_unique<WorkerThread>(conn_factory, this, queue_limit, cron_interval));
-
   }
 }