Rocksdb 的优秀代码（三）-- 工业级 线程池实现分享

前言

Rocksdb 作为一个第三方库的形态嵌入到各个存储系统之中存储元数据，当rocksdb被使用的时候其内部会自启动一些线程，随着需要处理的用户数据越来越多，为了保证性能，rocksdb会让这一些线程也会不断增加。而在分布式存储场景，往往一个机器节点会有很多rocksdb实例（64个实例，每一个实例都会有compaction/flush线程），这个时候在Rocksdb内部使用合理的线程管理方式会节省系统CPU调度资源。

所以Rocksdb自实现的Thread Pool就是为了更好得管理Rocksdb内部线程，除了一些基本的线程调度之外，还会有可控制的线程优先级的调度，因为大多数场景Rocksdb让Flush线程的优先级高于Compaction线程，而有的场景则需要Compaction的优先级高于Flush，为了更快速的compaction清理掉旧数据。

接下来简单看一下Rocksdb 线程池的基本实现，本人已经将该线程池代码摘出来单独维护，可作为一个独立线程池去调度。

https://github.com/BaronStack/ThreadPool

线程池存在的目的 正如上面Rocksdb使用线程池的目的一样， 能够更加方便得管理我们应用中的线程，包括但不限于：线程创建，线程资源约束，线程优先级调度，线程销毁 等。

1. Rocksdb线程池概览

Rocksdb 实现的线程池支持的特性：

• 创建/销毁线程
• 动态增加、减少线程池线程数目上限（线程池数目需要设置上限，因为Compaction/Flush占用的资源也不能无限增加，需根据实际的Rocksdb 写入量来动态增加）
• 支持动态调整 线程CPU 和 I/O优先级（为了暴露足够的接口给用户，来让用户选择两个功能调度的优先顺序）

2. Rocksdb 线程池实现

2.1 基本数据结构

// 线程池核心的数据结构
struct Impl {private:bool low_io_priority_;  // I/O 优先级bool low_cpu_priority_; // CPU 优先级Env::Priority priority_; // 线程优先级Env*         env_;       // 获取当前线程池的环境变量int total_threads_limit_; // 线程池线程总数std::atomic_uint queue_len_;  // 当前线程池中执行线程的排队长度bool exit_all_threads_; // 清理线程池时会调度所有未执行的线程bool wait_for_jobs_to_complete_; // 等待所有线程池的线程执行完毕// Entry per Schedule()/Submit() callstruct BGItem {void* tag = nullptr;std::function<void()> function; // 执行函数std::function<void()> unschedFunction; // 不执行函数};using BGQueue = std::deque<BGItem>;BGQueue       queue_; // deque 保存线程池中调度的线程相关的信息：线程函数、函数参数std::mutex               mu_;std::condition_variable  bgsignal_; // 条件变量，唤醒正在睡眠的线程std::vector<port::Thread> bgthreads_; // 保存需要调度的线程
}

线程池类:

class ThreadPoolImpl : public ThreadPool {private:std::unique_ptr<Impl>   impl_;// 线程池核心数据结构
};

2.2 线程池创建

Rocksdb维护了一个Env 类，这个类再同一个进程中的多个rocksdb实例之间是能够共享的。所以Rocksdb将这个类作为线程池的入口，从而让Flush/Compaction 这样的线程调度过程中，多个db可以只使用同一个线程池。

Rocksdb实现了多个环境变量:HdfsEnv，PosixEnv等，方便Rocksdb的文件操作/线程操作 接口在不同的环境平台下进行扩展，当然如果用户变更了新的平台，只需要支持Env基类的接口，就能扩展到用户的新平台。

Env默认实例是PosixEnv，为了保证多db实例间共享同一个环境变量，PosixEnv仅维护一个单例。

// 创建Env,初始化几个类的单例
// 这里注意调用的顺序，先调用ThreadLocalPtr实例的初始化，再调用PosixEnv的
// 这样在Env析构的时候能够反方向析构，从而保证ThreadLocal的信息最后一个被清理
Env* Env::Default() {ThreadLocalPtr::InitSingletons(); // Threadlocal 实例数据，用来访问当前db实例运行的线程状态信息CompressionContextCache::InitSingleton();INIT_SYNC_POINT_SINGLETONS();static PosixEnv default_env; // 创建posix env	return &default_env;
}

紧接着通过 PosixEnv的构造函数创建线程池

// 根据Env设置的线程优先级，为每一个优先级创建一个线程池（方便优先级线程池的调度）
// 创建多个线程池: enum Priority { BOTTOM, LOW, HIGH, USER, TOTAL };
std::vector<ThreadPoolImpl> thread_pools_;PosixEnv::PosixEnv(): checkedDiskForMmap_(false),forceMmapOff_(false),page_size_(getpagesize()),thread_pools_(Priority::TOTAL),allow_non_owner_access_(true) {ThreadPoolImpl::PthreadCall("mutex_init", pthread_mutex_init(&mu_, nullptr));// 根据优先级创建线程池，默认创建四个线程池，但一般只会用到两个(LOW,HIGH)for (int pool_id = 0; pool_id < Env::Priority::TOTAL; ++pool_id) {thread_pools_[pool_id].SetThreadPriority(static_cast<Env::Priority>(pool_id));// This allows later initializing the thread-local-env of each thread.thread_pools_[pool_id].SetHostEnv(this);}thread_status_updater_ = CreateThreadStatusUpdater();
}

2.3 线程池 调度线程执行

线程池调度栈如下：从入口到具体的线程函数的执行

Env::Schedule() // Env对外接口PosixEnv::Schedule()ThreadPoolImpl::Schedule() // 线程池的调度入口ThreadPoolImpl::Impl::Submit() // 将线程函数、参数、线程回收函数封装，添加到待调度队列queue_ThreadPoolImpl::Impl::StartBGThreads() ThreadPoolImpl::Impl::BGThreadWrapper() // 更新当前执行的线程状态并启动一个调度队列中的线程ThreadPoolImpl::Impl::BGThread()// 从待调度队列queue_中调度线程func() // 执行线程函数

Env的实例调用Schedule接口，接收待调度的线程执行函数，参数，所属优先级线程池，以及线程销毁函数及其参数。

  virtual void Schedule(void (*function)(void* arg), void* arg,Priority pri = LOW, void* tag = nullptr,void (*unschedFunction)(void* arg) = nullptr) = 0;

后续会执行到ThreadPoolImpl::Impl::Submit()

void ThreadPoolImpl::Impl::Submit(std::function<void()>&& schedule,std::function<void()>&& unschedule, void* tag) {// 后续需要更新当前线程池的线程调度队列，需要保证更新过程的原子性std::lock_guard<std::mutex> lock(mu_);// 需要销毁线程池了，不接受新的线程加入if (exit_all_threads_) {return;}// 启动线程StartBGThreads();// 更新线程函数相关的信息 到线程调度队列尾部（双端队列）queue_.push_back(BGItem());// 更新auto& item = queue_.back();item.tag = tag;item.function = std::move(schedule);item.unschedFunction = std::move(unschedule);queue_len_.store(static_cast<unsigned int>(queue_.size()),std::memory_order_relaxed);// 如果正在执行的线程没有超过线程池线程数限制，则唤醒一个正在休眠的线程if (!HasExcessiveThread()) {// Wake up at least one waiting thread.bgsignal_.notify_one();} else { // 。。。这个逻辑不太懂，超过限制之后 不应该就不唤醒了吗？// Need to wake up all threads to make sure the one woken// up is not the one to terminate.WakeUpAllThreads();}
}

后续的执行就是按照以上调用栈进行的，从线程调度队列头部取线程函执行。

2.4 线程池销毁线程

线程池的销毁也就是Env变量的析构函数，db被destory或者close，则会进入该逻辑，Env的默认环境变量是PosixEnv，即Env的子类。则会先调用PosixEnv 的析构函数，其中线程池相关的清理逻辑：

整体的调用栈如下：

~PosixEnv()ThreadPoolImpl::JoinAllThreads() ThreadPoolImpl::Impl::JoinThreads()

在析构函数中调用相关的线程清理工作：

~PosixEnv() override {// 通过Posix startthread 的接口调度的线程函数并发执行完毕for (const auto tid : threads_to_join_) {pthread_join(tid, nullptr);}// 让不同优先级线程池中待执行线程执行完for (int pool_id = 0; pool_id < Env::Priority::TOTAL; ++pool_id) {thread_pools_[pool_id].JoinAllThreads();}// 放置Posix析构过程中不应该thread_status_updater_ ，防止一些子线程更新线程状态出错// Delete the thread_status_updater_ only when the current Env is not// Env::Default().  This is to avoid the free-after-use error when// Env::Default() is destructed while some other child threads are// still trying to update thread status.if (this != Env::Default()) {delete thread_status_updater_;}
}

其中JoinAllThreads函数用来唤醒所有子线程的执行，并设置标记防止接收新的线程

void ThreadPoolImpl::Impl::JoinThreads(bool wait_for_jobs_to_complete) {std::unique_lock<std::mutex> lock(mu_);assert(!exit_all_threads_);wait_for_jobs_to_complete_ = wait_for_jobs_to_complete;// 原子（加锁）方式更新如下变量，用作在submit函数中屏蔽接收新的线程exit_all_threads_ = true;// prevent threads from being recreated right after they're joined, in case// the user is concurrently submitting jobs.// 重置线程池的线程上限，防止用户并发调用submit添加待调度线程total_threads_limit_ = 0;lock.unlock();bgsignal_.notify_all(); //唤醒所有等待在bgsignal_的线程for (auto& th : bgthreads_) {// join 执行，直到执行完。th.join();}bgthreads_.clear();exit_all_threads_ = false;wait_for_jobs_to_complete_ = false;
}

2.5 线程池优先级调度

之前说过Rocksdb线程池支持 用户针对不同LOW/HIGH 线程池的I/O或者CPU的优先级设置。

比如 设置LOW线程池具有更低的I/O优先级和CPU优先级

target_->LowerThreadPoolIOPriority(Env::Priority::LOW);
target_->LowerThreadPoolCPUPriority(Env::Priority::LOW);

具体底层的设置方式是针对之前提到的线程数据结构中的两个参数Impl::low_io_priority_和Impl::low_c pu_priority_进行置位true。在ThreadPoolImpl::Impl::BGThread调度函数执行之前，会通过系统调用setpriority和syscall(SYS_ioprio_set,,,）设置当前线程的I/O和CPU优先级。

void ThreadPoolImpl::Impl::BGThread(size_t thread_id) {bool low_io_priority = false;bool low_cpu_priority = false;while (true) {// Wait until there is an item that is ready to runstd::unique_lock<std::mutex> lock(mu_);...bool decrease_io_priority = (low_io_priority != low_io_priority_);bool decrease_cpu_priority = (low_cpu_priority != low_cpu_priority_);lock.unlock();#ifdef OS_LINUX// Linux 系统支持 设置CPU优先级if (decrease_cpu_priority) {setpriority(PRIO_PROCESS,// Current thread.0,// Lowest priority possible.19);low_cpu_priority = true;}if (decrease_io_priority) {
#define IOPRIO_CLASS_SHIFT (13)
#define IOPRIO_PRIO_VALUE(class, data) (((class) << IOPRIO_CLASS_SHIFT) | data)// Put schedule into IOPRIO_CLASS_IDLE class (lowest)// These system calls only have an effect when used in conjunction// with an I/O scheduler that supports I/O priorities. As at// kernel 2.6.17 the only such scheduler is the Completely// Fair Queuing (CFQ) I/O scheduler.// To change scheduler://  echo cfq > /sys/block/<device_name>/queue/schedule// Tunables to consider://  /sys/block/<device_name>/queue/slice_idle//  /sys/block/<device_name>/queue/slice_sync// 设置I/O优先级syscall(SYS_ioprio_set, 1,  // IOPRIO_WHO_PROCESS0,                  // current threadIOPRIO_PRIO_VALUE(3, 0));low_io_priority = true;}
#else// 非Linux系统的话就不做任何处理了，仅仅保证变量被使用而已，防止编译warning (void)decrease_io_priority;  // avoid 'unused variable' error(void)decrease_cpu_priority;
#endiffunc();}
}

2.6 动态调整线程池 线程数目上限

支持动态调整线程池可调度的线程数目上限，这个能够限制线程池资源的占用，主要用作Rocksdb 中调整Flush和Compaction的各自所处的HIGH和LOW线程池中的线程数目上限。能够根据db的工作负载，动态增加或者减少线程池中可调度的线程数目。

void ThreadPoolImpl::IncBackgroundThreadsIfNeeded(int num) {impl_->SetBackgroundThreadsInternal(num, false);
}void ThreadPoolImpl::Impl::SetBackgroundThreadsInternal(int num,bool allow_reduce) {std::unique_lock<std::mutex> lock(mu_);// 如果线程池已经要被销毁了，就不用增加线程池的调度线程数目上限了if (exit_all_threads_) {lock.unlock();return;}// 增加线程数目或者减少线程数目// 唤醒休眠的线程并调度后台线程继续执行。if (num > total_threads_limit_ ||(num < total_threads_limit_ && allow_reduce)) {total_threads_limit_ = std::max(0, num);WakeUpAllThreads();StartBGThreads();}
}

3. 总结

到此整个线程池的基本实现就描述完成了，这是一个非常成熟的线程池（经历过接近十年的工业级考验，2012年facebook开始开发rocksdb），规模虽小，但五脏俱全。其能够支撑引擎级别的线程调度压力，保证引擎的核心逻辑flush和compaction的高效调度。

目前该线程池的独立实现已经放在了https://github.com/BaronStack/ThreadPool 中，拥有完备的线程池调度/销毁，优先级配置，欢迎star。