PikiwiDB 架构入门

[luky116]

PikiwiDB 核心组件

1. PikiwiDB：主对象，pikiwidb 整个流程的入口对象。他的核心成员有如下：
   1. worker_threads_：负责接收用户的请求，将 redis 命令处理结果推送给 client。每个 client 进来，会记录当前总的连接，并和 workerloops 的大小取模选择其中一个线程使用，即，worker_threads_是被复用的。（备注：worker_threads_的回调函数，使用的是PikiwiDB::OnNewConnection 方法。）
   2. slave_threads_：负责处理 slave 的请求，目前为使用Link
   3. cmd_threads_：用来处理命令（下面解释）

 private:
  pikiwidb::WorkIOThreadPool worker_threads_;
  pikiwidb::IOThreadPool slave_threads_;
  pikiwidb::CmdThreadPool cmd_threads_;

2. WorkIOThreadPool：网络相关的线程，负责接收网络请求，以及将处理结果返回给client
   1. writeThreads_：实处理请求的线程数量
   2. writeQueue_：待返回的 client 对象
   3. writeCond_：阻塞直到有待处理的请求，才会被唤醒

 private:
  std::vector<std::thread> writeThreads_;
  std::vector<std::unique_ptr<std::mutex>> writeMutex_;
  std::vector<std::unique_ptr<std::condition_variable>> writeCond_;
  std::vector<std::deque<std::shared_ptr<PClient>>> writeQueue_;
  std::atomic<uint64_t> counter_ = 0;
  bool writeRunning_ = true;

3. CmdThreadPool：里面存放了所有的待处理的 redis 命令，核心成员如下
   1. fast_tasks_：存放要处理的快 redis 命令，目前所有命令都是快命令；
   2. slow_tasks_：存放要处理的慢 redis 命令，暂未开启使用；
   3. threads_：等于 fast_tasks_+slow_tasks_，用来用来做异步并发处理 fasttasks和 slow_tasks_中的命令，线程中实际跑的是 CmdWorkThreadPoolWorker 类型；
   4. fast_condition_/slow_condition_：CmdThreadPoolTask 用来阻塞和唤醒进行命令处理工作；

private:
    std::deque<std::shared_ptr<CmdThreadPoolTask>> fast_tasks_;  // fast task queue
    std::deque<std::shared_ptr<CmdThreadPoolTask>> slow_tasks_;  // slow task queue
    
    std::vector<std::thread> threads_;
    std::vector<std::shared_ptr<CmdWorkThreadPoolWorker>> workers_;
    std::string name_;  // thread pool name
    int fast_thread_num_ = 0;
    int slow_thread_num_ = 0;
    std::mutex fast_mutex_;
    std::condition_variable fast_condition_;
    std::mutex slow_mutex_;
    std::condition_variable slow_condition_;
    std::atomic_bool stopped_ = false;

3. CmdWorkThreadPoolWorker：负责实际处理 redis 命令。CmdWorkThreadPoolWorker 的核心成员如下：
   1. self_task_：存放的是待本线程要处理的 redis 命令，它是以 CmdThreadPoolTask 的方式存放的
   2. pool_：即上面介绍的 CmdThreadPool，selftask中的数据来自于 pool_；
   3. cmd_table_manager_：存放 redis 命令对象和名字的映射关系。

protected:
  std::vector<std::shared_ptr<CmdThreadPoolTask>> self_task_;  // the task that the worker get from the thread pool
  CmdThreadPool *pool_ = nullptr;
  const int once_task_ = 0;  // the max task num that the worker can get from the thread pool
  const std::string name_;
  bool running_ = true;

  pikiwidb::CmdTableManager cmd_table_manager_;

4. CmdThreadPoolTask：对 redis 命令的封装和用户 client 的封装，Run 方法用来处理命令，client_ 用来将梳理结果返回给用户。

class CmdThreadPoolTask {
 public:
  CmdThreadPoolTask(std::shared_ptr<PClient> client) : client_(std::move(client)) {}
  void Run(BaseCmd *cmd);
  const std::string &CmdName();
  std::shared_ptr<PClient> Client();

 private:
  std::shared_ptr<PClient> client_;
};

核心流程

命令处理

命令的实际处理流程，由多个 CmdWorkThreadPoolWorker 来完成。PikiwiDB 在启动的时候，会根据配置的 fast_thread_num_和slow_thread_num_参数，启动多个 CmdWorkThreadPoolWorker 线程来监听 CmdThreadPool中待处理的 redis 命令：

void CmdThreadPool::Start() {
  for (int i = 0; i < fast_thread_num_; ++i) {
    auto fastWorker = std::make_shared<CmdFastWorker>(this, 2, "fast worker" + std::to_string(i));
    std::thread thread(&CmdWorkThreadPoolWorker::Work, fastWorker);
    threads_.emplace_back(std::move(thread));
    workers_.emplace_back(fastWorker);
    INFO("fast worker [{}] starting ...", i);
  }
  for (int i = 0; i < slow_thread_num_; ++i) {
    auto slowWorker = std::make_shared<CmdSlowWorker>(this, 2, "slow worker" + std::to_string(i));
    std::thread thread(&CmdWorkThreadPoolWorker::Work, slowWorker);
    threads_.emplace_back(std::move(thread));
    workers_.emplace_back(slowWorker);
    INFO("slow worker [{}] starting ...", i);
  }
}

CmdWorkThreadPoolWorker 的 Work 方法会被 thread 异步启动，核心流程如下：

1. 调用 LoadWork 方法进行阻塞，直到 pool 中有请求进来；
2. 依次从 self_task_队列中获取待处理的 redis 命令
3. 根据 cmd_name 获取 cmd 对象，然后处理命令；如果命令不存在直接返回错误
4. 处理完成后，调用 PushWriteTask 将请求写入到返回队列中，待异步触发返回

void CmdWorkThreadPoolWorker::Work() {
  while (running_) {
    LoadWork();
    for (const auto &task : self_task_) { // 处理 self_task_ 中的数据
      if (task->Client()->State() != ClientState::kOK) {  // the client is closed
        continue;
      }
      auto [cmdPtr, ret] = cmd_table_manager_.GetCommand(task->CmdName(), task->Client().get());

      if (!cmdPtr) {
        if(ret == CmdRes::kUnknownCmd) {
          task->Client()->SetRes(CmdRes::kErrOther, "unknown command '" + task->CmdName() + "'");
        } else if(ret == CmdRes::kUnknownSubCmd) {
          task->Client()->SetRes(CmdRes::kErrOther, "unknown sub command '" + task->Client().get()->argv_[1] + "'");
        } else {
          task->Client()->SetRes(CmdRes::kInvalidParameter);
        }
        g_pikiwidb->PushWriteTask(task->Client()); // 将要返回的 client 放到队列中
        continue;
      }

      if (!cmdPtr->CheckArg(task->Client()->ParamsSize())) {
        task->Client()->SetRes(CmdRes::kWrongNum, task->CmdName());
        g_pikiwidb->PushWriteTask(task->Client());
        continue;
      }
      task->Run(cmdPtr);
      g_pikiwidb->PushWriteTask(task->Client());
    }
    self_task_.clear();
  }
  INFO("worker [{}] goodbye...", name_);
}

LoadWork 的核心逻辑如下：

1. 循环遍历 pool_->fast_tasks 中是否存在待处理的 redis 命令，没有则睡眠；
2. Pclient 中如果收到待处理的 redis 命令，则会唤醒此流程；
3. 将 pool 中获取指定数量的 redis 命令，放到 work 的 self_task_中，并返回；
4. 接下来，就到了 Work 中的流程，执行 redis 命令。

[AlexStocks]

@lqxhub 代码解析

日期: 2024-12-28 21:37:38
录制文件：https://meeting.tencent.com/crm/KeYVZDYW15
密码：YDEN