C++线程池

Posted 2021-07-17 小丑快学习

线程池

• • • 1.任务队列
    • 2.可窃取任务的队列
    • 3.线程池的实现
    • 4.线程销毁

代码来自 <<c++ Concurence in Action>>
所谓线程池，就是一个由多个线程组成的数组或者队列，在相应的队列中不断取出任务进行执行。将多个线程此存放于一个数组中可以减少不断地产生和销毁线程带来的开销。下面便是一个由c++实现的线程池。更多的细节请参阅相关书籍。

1.任务队列

任务队列就是就是一个共享的队列，其需要实现线程的安全的访问，并且不能在访问时产生死锁。线程安全的队列可以参考文章：
线程安全队列
该链接中有对线程安全队列的代码的详细注释以及完整源码。

2.可窃取任务的队列

如果所有得线程都到统一个共享队列中去加载任务进行执行，当线程很多时，就会导致访问共享队列变得缓慢，因此，为了加快线程得访问速度，为每一个线程设置一个独立得队列，该线程优先执行该队列中的任务，倘若该队列中没有任务，则取全局共享队列中得任务执行，倘若共享队列为空，则取其他线程得专属任务队列中窃取任务进行执行，这样就可以减少因为访问同一个共享变量而带来的负担。因此，可供窃取的队列应该要满足可以安全的窃取任务。代码如下：

class work_stealing_queue
{
private:
	typedef function_wrapper data_type;//一个可调用对象，这里直接考虑为一个function类型
	std::deque<data_type> the_queue; // 1 对deque进行封装以实现对安全队列的访问
	mutable 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) // 有任务返回true，并将结果存放于可调用对象res中。
	{
		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 和try_pop的不同在于这是从尾部进行任务的窃取
	{
		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;
	}
};

由以上我们知道任队列中存放的其实是可调用对象，也就是相当于一个函数指针，这个可调用的对象将会传递给一个线程进行执行，执行结果可以放于future中，最后由主线程对执行结果进行整合。所以我们需要每一个执行的结果存放于一个future中，因此，我们需要将每一个任务（可调用对象）放置于一个packaged_task，然后通过packaged_task对应的future获取结果，但是packaged_task是仅支持移动语义的，因此，我们的task也仅能是可以移动的，因此，我们需要将其封装为仅有一移动义的可调用对象。也就是队列中的function_wrapper类。

//仅支持移动赋值操作的function类
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;
};

3.线程池的实现

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; // 1 线程专属的队列，可窃取任务
		std::vector<std::thread> threads; //线程数组
	join_threads joiner;//RAII手法，保证线程能够回收,见下文
	static thread_local work_stealing_queue* local_work_queue; // 2 每个线程都有一个指针指向一个可窃取任务的队列
		static thread_local unsigned my_index;//每一个线程都有一个线程专属队列的索引
	
	//传入每个线程的函数，即每个线程应该不断的去执行任务
	void worker_thread(unsigned my_index_)
	{
		my_index = my_index_;
		local_work_queue = queues[my_index].get(); // 3 获取线专属队列
		while (!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(); // 5 为防止每次都从同一个线程中窃取任务
			if (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>(
					new work_stealing_queue));//生成线程专属队列
				threads.push_back(//生成线程
					std::thread(&thread_pool::worker_thread, this, i));
			}
		}
		catch (...)
		{
			done = true;
			throw;
		}
	}
	~thread_pool()
	{
		done = true;
	}
	//提交任务到队列，并返回相应的future，future中保存执行结果
	template<typename FunctionType>
	std::future<typename std::result_of<FunctionType()>::type>
		submit(
			FunctionType f)
	{
		typedef typename std::result_of<FunctionType()>::type result_type;//获取FunctionType()的返回值
		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) || // 7 
			pop_task_from_pool_queue(task) || // 8 
			pop_task_from_other_thread_queue(task)) // 9 
		{
			task();
		}
		else
		{
			std::this_thread::yield();
		}
	}
};
//静态变量需要在类的外部进行初始化
thread_local work_stealing_queue* thread_pool::local_work_queue;
thread_local unsigned thread_pool::my_index;

4.线程销毁

线程的销毁应该注意异常安全，也就是我们需要防止当主线程发生异常时其他线程将没法回收的情况，因此我们采用RAII的手段进行管理，

class join_threads
{
    std::vector<std::thread>& threads;
public:
    explicit join_threads(std::vector<std::thread>& threads_) :
        threads(threads_)
    {}
    ~join_threads()
    {
        for (unsigned long i = 0; i < threads.size(); ++i)
        {
            if (threads[i].joinable())
                threads[i].join();
        }
    }
};

在join_threads的析构函数中我们对线程进行回收，因此，即使打当有异常发生时，我们的join_threads对象在析构时就会将线程进行回收，因此不会带来异常安全问题。