C++ 并发编程实战 第九章

目录

9.1 线程池 

9.1.1 最简易可行的线程池

9.1.2 等待提交给线程池的任务完成运行

9.1.3等待其他任务完成的任务

9.1.4 避免任务队列上的争夺

9.1.5 任务窃取

9.2 中断线程

9.2.1 发起一个线程，以及把他中断

9.2.2 检测线程是否被中断

9.2.3 中断条件变量上的等待

9.2.4 中断条件变量std::condition_variable_any上的等待

9.2.5 中断其他阻塞型等待

9.2.6 处理中断

9.2.7 在应用程序推出时中断后台任务

9.3 小结

---

参考：https://github.com/xiaoweiChen/CPP-Concurrency-In-Action-2ed-2019/blob/master/content/chapter9/9.1-chinese.md

9.1 线程池 

大多数系统中，将每个任务指定给某个线程是不切实际的，不过可以利用并发性，进行并发执行。线程池提供了这样的功能，将提交到线程池中的任务并发执行，提交的任务将会挂在任务队列上。工作线程会从队列中的获取任务，当任务执行完成后，再从任务队列中获取下一个任务。

创建一个线程池时，会遇到几个关键性的设计问题，比如：可使用的线程数量，高效的任务分配方式，以及是否需要等待一个任务完成。

9.1.1 最简易可行的线程池

代码9.1 简单的线程池

class thread_pool
{std::atomic_bool done;thread_safe_queue<std::function<void()> > work_queue;  // 1std::vector<std::thread> threads;  // 2join_threads joiner;  // 3void worker_thread(){while(!done)  // 4{std::function<void()> task;if(work_queue.try_pop(task))  // 5{task();  // 6}else{std::this_thread::yield();  // 7}}}public:thread_pool():done(false),joiner(threads){unsigned const thread_count=std::thread::hardware_concurrency();  // 8try{for(unsigned i=0;i<thread_count;++i){threads.push_back( std::thread(&thread_pool::worker_thread,this));  // 9}}catch(...){done=true;  // 10throw;}}~thread_pool(){done=true;  // 11}template<typename FunctionType>void submit(FunctionType f){work_queue.push(std::function<void()>(f));  // 12}
};

这样简单的线程池就完成了，特别是任务没有返回值，或需要执行阻塞操作的任务。很多情况下，这样的线程池是不够用的，其他情况使用这样简单的线程池可能会出现问题，比如：死锁。同样，在简单例子中使用std::async能提供更好的功能。

9.1.2 等待提交给线程池的任务完成运行

第8章中的例子中，线程间的任务划分完成后，代码会显式生成新线程，主线程通常是等待新线程在返回调用之后结束，确保所有任务都完成。使用线程池就需要等待任务提交到线程池中，而非直接提交给单个线程。与基于std::async的方法类似，使用代码9.1中的简单线程池，使用第4章中提到的工具：条件变量和future。虽然会增加代码的复杂度，不过要比直接对任务进行等待好很多。

通过增加线程池的复杂度，可以直接等待任务完成。使用submit()函数返回对任务描述的句柄，可用来等待任务的完成。任务句柄会用条件变量或future进行包装，从而简化线程池的实现。

一种特殊的情况是，执行任务的线程需要返回结果到主线程上进行处理。本这种情况下，需要用future对最终的结果进行转移。代码9.2展示了对简单线程池的修改，通过修改就能等待任务完成，以及在工作线程完成后，返回一个结果到等待线程中去，不过std::packaged_task<>实例是不可拷贝的，仅可移动，所以不能再使用std::function<>来实现任务队列，因为std::function<>需要存储可复制构造的函数对象。包装一个自定义函数，用来处理可移动的类型，就是一个带有函数操作符的类型擦除类。只需要处理没有入参的函数和无返回的函数即可，所以这只是一个简单的虚函数调用。

代码9.2 可等待任务的线程池

class function_wrapper
{struct impl_base {virtual void call()=0;virtual ~impl_base() {}};std::unique_ptr<impl_base> impl;template<typename F>struct impl_type: impl_base{F f;impl_type(F&& f_): f(std::move(f_)) {}void call() { f(); }};
public:template<typename F>function_wrapper(F&& f):impl(new impl_type<F>(std::move(f))){}void operator()() { impl->call(); }function_wrapper() = default;function_wrapper(function_wrapper&& other):impl(std::move(other.impl)){}function_wrapper& operator=(function_wrapper&& other){impl=std::move(other.impl);return *this;}function_wrapper(const function_wrapper&)=delete;function_wrapper(function_wrapper&)=delete;function_wrapper& operator=(const function_wrapper&)=delete;
};class thread_pool
{thread_safe_queue<function_wrapper> work_queue;  // 使用function_wrapper，而非使用std::functionvoid worker_thread(){while(!done){function_wrapper task;if(work_queue.try_pop(task)){task();}else{std::this_thread::yield();}}}
public:template<typename FunctionType>std::future<typename std::result_of<FunctionType()>::type>  // 1submit(FunctionType f){typedef typename std::result_of<FunctionType()>::typeresult_type;  // 2std::packaged_task<result_type()> task(std::move(f));  // 3std::future<result_type> res(task.get_future());  // 4work_queue.push(std::move(task));  // 5return res;  // 6}// 和之前一样
};

9.1.3等待其他任务完成的任务

最简单的方法就是在thread_pool中添加一个新函数，来执行任务队列上的任务，并对线程池进行管理。高级线程池的实现可能会在等待函数中添加逻辑，或等待其他函数来处理这个任务，优先的任务会让其他的任务进行等待。下面代码中的实现，就展示了一个新run_pending_task()函数，对于快速排序的修改将会在代码9.5中展示。

代码9.4 run_pending_task()函数实现

void thread_pool::run_pending_task()
{function_wrapper task;if(work_queue.try_pop(task)){task();}else{std::this_thread::yield();}
}

下面快速排序算法的实现要比代码8.1中版本简单许多，因为所有线程管理逻辑都移到线程池中了。

代码9.5 基于线程池的快速排序实现

template<typename T>
struct sorter  // 1
{thread_pool pool;  // 2std::list<T> do_sort(std::list<T>& chunk_data){if(chunk_data.empty()){return chunk_data;}std::list<T> result;result.splice(result.begin(),chunk_data,chunk_data.begin());T const& partition_val=*result.begin();typename std::list<T>::iterator divide_point=std::partition(chunk_data.begin(),chunk_data.end(),[&](T const& val){return val<partition_val;});std::list<T> new_lower_chunk;new_lower_chunk.splice(new_lower_chunk.end(),chunk_data,chunk_data.begin(),divide_point);std::future<std::list<T> > new_lower=  // 3pool.submit(std::bind(&sorter::do_sort,this,std::move(new_lower_chunk)));std::list<T> new_higher(do_sort(chunk_data));result.splice(result.end(),new_higher);while(!new_lower.wait_for(std::chrono::seconds(0)) ==std::future_status::timeout){pool.run_pending_task();  // 4}result.splice(result.begin(),new_lower.get());return result;}
};template<typename T>
std::list<T> parallel_quick_sort(std::list<T> input)
{if(input.empty()){return input;}sorter<T> s;return s.do_sort(input);
}

9.1.4 避免任务队列上的争夺

为了避免乒乓缓存，每个线程建立独立的任务队列。这样，每个线程就会将新任务放在自己的任务队列上，并且当线程上的任务队列没有任务时，去全局的任务列表中取任务。下面列表中的实现，使用了一个thread_local变量，来保证每个线程都拥有自己的任务列表(如全局列表那样)。

代码9.6 线程池——线程具有本地任务队列        

class thread_pool
{thread_safe_queue<function_wrapper> pool_work_queue;typedef std::queue<function_wrapper> local_queue_type;  // 1static thread_local std::unique_ptr<local_queue_type>local_work_queue;  // 2void worker_thread(){local_work_queue.reset(new local_queue_type);  // 3while(!done){run_pending_task();}}public:template<typename FunctionType>std::future<typename std::result_of<FunctionType()>::type>submit(FunctionType f){typedef typename std::result_of<FunctionType()>::type result_type;std::packaged_task<result_type()> task(f);std::future<result_type> res(task.get_future());if(local_work_queue)  // 4{local_work_queue->push(std::move(task));}else{pool_work_queue.push(std::move(task));  // 5}return res;}void run_pending_task(){function_wrapper task;if(local_work_queue && !local_work_queue->empty())  // 6{task=std::move(local_work_queue->front());local_work_queue->pop();task();}else if(pool_work_queue.try_pop(task))  // 7{task();}else{std::this_thread::yield();}}
// rest as before
};

9.1.5 任务窃取

任务分配不均时，造成的结果就是：某个线程本地队列中有很多任务的同时，其他线程无所事事。例如：举一个快速排序的例子，一开始的数据块能在线程池上被处理，因为剩余部分会放在工作线程的本地队列上进行处理，这样的使用方式也违背使用线程池的初衷。

幸好这个问题有解：本地工作队列和全局工作队列上没有任务时，可从别的线程队列中窃取任务。

代码9.7 基于锁的任务窃取队列

class work_stealing_queue
{
private:typedef function_wrapper data_type;std::deque<data_type> the_queue;  // 1mutable std::mutex the_mutex;public:work_stealing_queue(){}work_stealing_queue(const work_stealing_queue& other)=delete;work_stealing_queue& operator=(const work_stealing_queue& other)=delete;void push(data_type data)  // 2{std::lock_guard<std::mutex> lock(the_mutex);the_queue.push_front(std::move(data));}bool empty() const{std::lock_guard<std::mutex> lock(the_mutex);return the_queue.empty();}bool try_pop(data_type& res)  // 3{std::lock_guard<std::mutex> lock(the_mutex);if(the_queue.empty()){return false;}res=std::move(the_queue.front());the_queue.pop_front();return true;}bool try_steal(data_type& res)  // 4{std::lock_guard<std::mutex> lock(the_mutex);if(the_queue.empty()){return false;}res=std::move(the_queue.back());the_queue.pop_back();return true;}
};

这就说明每个线程中的“队列”是一个后进先出的栈，最新推入的任务将会第一个执行。从缓存角度来看，这将对性能有所提升，因为任务相关的数据一直存于缓存中，要比提前将任务相关数据推送到栈上好。同样，这种方式很好的映射到某个算法上，例如：快速排序。之前的实现中，每次调用do_sort()都会推送一个任务到栈上，并且等待这个任务执行完毕。通过对最新推入任务的处理，就可以保证在将当前所需数据块处理完成前，其他任务是否需要这些数据块，从而可以减少活动任务的数量和栈的使用次数。try_steal()从队列末尾获取任务，为了减少与try_pop()之间的竞争。使用在第6、7章中的所讨论的技术来让try_pop()和try_steal()并发执行。

现在拥有了一个很不错的任务队列，并且支持窃取。那如何在线程池中使用这个队列呢？这里简单的展示一下。

代码9.8 使用任务窃取的线程池

class thread_pool
{typedef function_wrapper task_type;std::atomic_bool done;thread_safe_queue<task_type> pool_work_queue;std::vector<std::unique_ptr<work_stealing_queue> > queues;  // 1std::vector<std::thread> threads;join_threads joiner;static thread_local work_stealing_queue* local_work_queue;  // 2static thread_local unsigned my_index;void worker_thread(unsigned my_index_){my_index=my_index_;local_work_queue=queues[my_index].get();  // 3while(!done){run_pending_task();}}bool pop_task_from_local_queue(task_type& task){return local_work_queue && local_work_queue->try_pop(task);}bool pop_task_from_pool_queue(task_type& task){return pool_work_queue.try_pop(task);}bool pop_task_from_other_thread_queue(task_type& task)  // 4{for(unsigned i=0;i<queues.size();++i){unsigned const index=(my_index+i+1)%queues.size();  // 5if(queues[index]->try_steal(task)){return true;}}return false;}public:thread_pool():done(false),joiner(threads){unsigned const thread_count=std::thread::hardware_concurrency();try{for(unsigned i=0;i<thread_count;++i){queues.push_back(std::unique_ptr<work_stealing_queue>(  // 6new work_stealing_queue));threads.push_back(std::thread(&thread_pool::worker_thread,this,i));}}catch(...){done=true;throw;}}~thread_pool(){done=true;}template<typename FunctionType>std::future<typename std::result_of<FunctionType()>::type> submit(FunctionType f){ typedef typename std::result_of<FunctionType()>::type result_type;std::packaged_task<result_type()> task(f);std::future<result_type> res(task.get_future());if(local_work_queue){local_work_queue->push(std::move(task));}else{pool_work_queue.push(std::move(task));}return res;}void run_pending_task(){task_type task;if(pop_task_from_local_queue(task) ||  // 7pop_task_from_pool_queue(task) ||  // 8pop_task_from_other_thread_queue(task))  // 9{task();}else{std::this_thread::yield();}}
};

9.2 中断线程

9.2.1 发起一个线程，以及把他中断

代码9.9 interruptible_thread的基本实现

class interrupt_flag
{
public:void set();bool is_set() const;
};
thread_local interrupt_flag this_thread_interrupt_flag;  // 1class interruptible_thread
{std::thread internal_thread;interrupt_flag* flag;
public:template<typename FunctionType>interruptible_thread(FunctionType f){std::promise<interrupt_flag*> p;  // 2internal_thread=std::thread([f,&p]{  // 3p.set_value(&this_thread_interrupt_flag);f();  // 4});flag=p.get_future().get();  // 5}void interrupt(){if(flag){flag->set();  // 6}}
};

9.2.2 检测线程是否被中断

9.2.3 中断条件变量上的等待

代码9.11 为std::condition_variable在interruptible_wait中使用超时

class interrupt_flag
{std::atomic<bool> flag;std::condition_variable* thread_cond;std::mutex set_clear_mutex;public:interrupt_flag():thread_cond(0){}void set(){flag.store(true,std::memory_order_relaxed);std::lock_guard<std::mutex> lk(set_clear_mutex);if(thread_cond){thread_cond->notify_all();}}bool is_set() const{return flag.load(std::memory_order_relaxed);}void set_condition_variable(std::condition_variable& cv){std::lock_guard<std::mutex> lk(set_clear_mutex);thread_cond=&cv;}void clear_condition_variable(){std::lock_guard<std::mutex> lk(set_clear_mutex);thread_cond=0;}struct clear_cv_on_destruct{~clear_cv_on_destruct(){this_thread_interrupt_flag.clear_condition_variable();}};
};void interruptible_wait(std::condition_variable& cv,std::unique_lock<std::mutex>& lk)
{interruption_point();this_thread_interrupt_flag.set_condition_variable(cv);interrupt_flag::clear_cv_on_destruct guard;interruption_point();cv.wait_for(lk,std::chrono::milliseconds(1));interruption_point();
}

9.2.4 中断条件变量std::condition_variable_any上的等待

代码9.12 为std::condition_variable_any设计的interruptible_wait

class interrupt_flag
{std::atomic<bool> flag;std::condition_variable* thread_cond;std::condition_variable_any* thread_cond_any;std::mutex set_clear_mutex;public:interrupt_flag(): thread_cond(0),thread_cond_any(0){}void set(){flag.store(true,std::memory_order_relaxed);std::lock_guard<std::mutex> lk(set_clear_mutex);if(thread_cond){thread_cond->notify_all();}else if(thread_cond_any){thread_cond_any->notify_all();}}template<typename Lockable>void wait(std::condition_variable_any& cv,Lockable& lk){struct custom_lock{interrupt_flag* self;Lockable& lk;custom_lock(interrupt_flag* self_,std::condition_variable_any& cond,Lockable& lk_):self(self_),lk(lk_){self->set_clear_mutex.lock();  // 1self->thread_cond_any=&cond;  // 2}void unlock()  // 3{lk.unlock();self->set_clear_mutex.unlock();}void lock(){std::lock(self->set_clear_mutex,lk);  // 4}~custom_lock(){self->thread_cond_any=0;  // 5self->set_clear_mutex.unlock();}};custom_lock cl(this,cv,lk);interruption_point();cv.wait(cl);interruption_point();}// rest as before
};template<typename Lockable>
void interruptible_wait(std::condition_variable_any& cv,Lockable& lk)
{this_thread_interrupt_flag.wait(cv,lk);
}

9.2.5 中断其他阻塞型等待

9.2.6 处理中断

9.2.7 在应用程序推出时中断后台任务

试想在桌面上查找一个应用。这就需要与用户互动，应用的状态需要能在显示器上显示，就能看出应用有什么改变。为了避免影响GUI的响应时间，通常会将处理线程放在后台运行。后台进程需要一直执行，直到应用退出。后台线程会作为应用启动的一部分被启动，并且在应用终止的时候停止运行。通常这样的应用只有在机器关闭时才会退出，因为应用需要更新应用最新的状态，就需要全时间运行。在某些情况下，当应用关闭，需要使用有序的方式将后台线程关闭，其中一种方式就是中断。

下面代码中为一个系统实现了简单的线程管理部分。

代码9.13 后台监视文件系统

std::mutex config_mutex;
std::vector<interruptible_thread> background_threads;void background_thread(int disk_id)
{while(true){interruption_point();  // 1fs_change fsc=get_fs_changes(disk_id);  // 2if(fsc.has_changes()){update_index(fsc);  // 3}}
}void start_background_processing()
{background_threads.push_back(interruptible_thread(background_thread,disk_1));background_threads.push_back(interruptible_thread(background_thread,disk_2));
}int main()
{start_background_processing();  // 4process_gui_until_exit();  // 5std::unique_lock<std::mutex> lk(config_mutex);for(unsigned i=0;i<background_threads.size();++i){background_threads[i].interrupt();  // 6}for(unsigned i=0;i<background_threads.size();++i){background_threads[i].join(); // 7}
}

9.3 小结

本章中了解各种线程管理的高级技术：线程池和中断线程。也了解了如何使用本地任务队列，使用任务窃取的方式减小同步开销，提高线程池的吞吐量，等待子任务完成的同时执行队列中其他任务，从而来避免死锁。

还有，使用线程去中断另一个处理线程的各种方式，比如：使用特定的断点和函数执行中断，要不就是使用某种方法对阻塞等待进行中断。