• 手写能任务窃取的线程池

    目录

    function_wrapper.hpp:

    stealing_queue.hpp

    thread_pool_steal_hpp

    参考：《C++并发编程实战》

    对于thread_pool_steal.hpp的代码有改动，不然运行不了

    function_wrapper.hpp:


    //包装可调用对象，对外消除对象型别，还需要有一个函数调用符
    //
    //
    //私有成员：
    //      1.一个虚基类（没有的话在3很难定义一个指向实例化类的指针）
    //      struct impl_base{
    //              1.一个纯虚函数用来让派生类继承3执行函数
    //              2.虚析构函数
    //      }
    //      2.因为需要接受任意类型的可调用对象，所以内部需要封装一个模板类（派生自1>）
    //      template
    //      struct impl_type{
    //              1.成员变量：传入的函数
    //              2.构造函数：移动构造成员变量
    //              3.成员函数：执行函数
    //      }
    //      3.一个指向实例化后的指针（在这里看出确实需要一个基类，可以通过基类指针构
    造派生类指针）
    //公有成员：
    //      1.构造函数，因为接收的对象是任意类型，所以是模板构造函数，并且使用万能引
    用和完美转发实例化私有成员的3.
    //      2.()运算符的重载通过调用私有成员2.的3.实现
    //      3.默认构造函数 = default
    //      4.移动拷贝构造函数
    //      5.移动复制构造函数
    //   
     

    1. #ifndef _FUNCTION_WRAPPER_HPP_
    2. #define _FUNCTION_WRAPPER_HPP_
    3.  
    4. #include
    5. class function_wrapper
    6. {
    7. 	struct impl_base{
    8. 		virtual void call() = 0;
    9. 		virtual ~impl_base(){};
    10. 	};	
    11. 	template<typename Function>
    12. 	struct impl_type:impl_base{
    13. 		Function f;
    14. 		impl_type(Function&& f_):f(std::move(f_)){}
    15. 		void call(){ f(); }
    16. 	};
    17. 	std::unique_ptr impl;
    18. public:
    19. 	function_wrapper() = default;
    20. 	template<typename Function>
    21. 	function_wrapper(Function&& f):impl(new impl_type(std::move(f)))	{}
    22. 	void operator()(){
    23. 		impl->call();
    24. 	}
    25. 	function_wrapper(function_wrapper&& other):impl(std::move(other.impl)) {}
    26. 	function_wrapper& operator=(function_wrapper&& other){
    27. 		impl = std::move(other.impl);
    28. 		return *this;
    29. 	}
    30.  
    31. 	function_wrapper(function_wrapper& other) = delete;
    32. 	function_wrapper(const function_wrapper& other) = delete;
    33. 	function_wrapper& operator=(const function_wrapper& other)=delete;
    34. 	
    35. };
    36. #endif

    stealing_queue.hpp


    //可以进行任务窃取的队列
    //
    //私有成员：
    //      1.一个双端队列，pop操作在队头进行，steal操作在队尾进行，存储的内容为function_wrapper类
    //      2.互斥 保证安全，因为不存储线程，所以不需要条件变量传递线程运行信息
    //公有成员：
    //      1.默认构造函数
    //      2.push
    //      3.try_pop
    //      4.try_steal基本和pop都一样，就是弹出队尾元素//在线程池中定义一个存储窃取
    队列的vector,vector的索引代表每个线程的标识，这样每个线程就可以通过这个vector访>问窃取队列。
    //      5.empty
     

    1. #ifndef _STEALING_QUEUE_HPP
    2. #define _STEALING_QUEUE_HPP
    3.  
    4.  
    5. #include"function_wrapper.hpp"
    6. #include
    7. #include
    8. class work_stealing_queue
    9. {
    10. 	typedef function_wrapper data_type;
    11. 	std::deque the_queue;
    12. 	mutable std::mutex mut;
    13. public:
    14. 	work_stealing_queue(){}
    15. 	work_stealing_queue(const work_stealing_queue& othre) = delete;
    16. 	work_stealing_queue& operator=(const work_stealing_queue& othre) = delete;
    17. 	
    18. 	void push(data_type data){
    19. 		std::lock_guard lk(mut);
    20. 		the_queue.push_front(std::move(data));
    21. 	}
    22. 	bool try_pop(data_type& data){
    23. 		std::lock_guard lk(mut);
    24. 		if(the_queue.empty())
    25. 			return false;
    26. 		else{
    27. 			data = std::move(the_queue.front());
    28. 			the_queue.pop_front();
    29. 			return true;
    30. 		}
    31. 	}
    32. 	bool try_steal(data_type& data){
    33. 		std::lock_guard lk(mut);
    34. 		if(the_queue.empty())
    35.                         return false;
    36.                 else{
    37.                         data = std::move(the_queue.front());
    38.                         the_queue.pop_back();
    39.                         return true;
    40.                 }
    41.  
    42. 	}
    43. 	bool empty() const{
    44. 		std::lock_guard lk(mut);
    45. 		return the_queue.empty();
    46. 	}
    47. };
    48.  
    49.  
    50. #endif

    thread_pool_steal_hpp：

    //私有成员：
    //      1.控制线程正常运行的原子变量，抛出异常设置为true
    //      2.线程池的池队列，基于普通的线程安全队列
    //      3.提供索引的队列存储指向窃取队列的指针
    //      4.存放线程的队列
    //      5.封装可联结线程
    //      6.静态本地线程变量 本地线程队列
    //      7.静态本地线程变量 索引
    //
    //      8.work_thread工作函数（任务函数）
    //      9.判断能否从 本地线程队列获取任务
    //      10.判断能否从 线程池队列获取任务
    //      11.判断能否从 其他线程窃取任务
    //公有成员：
    //      1.构造函数：初始化原子变量，可联结线程类。
    //      {
    //              try
    //              {       
    //                      for()
    //                      {初始化提供索引的队列   }
    //                      for()
    //                      {初始化线程}
    //              }
    //              catch
    //      }
    //      2.提交任务函数//为了获取返回值应该返回future，传入各种函数，所以应该是模
    板函数{
    //              1.传入函数打包给pakage_task
    //              2.获取future
    //              3.判断是传入本地线程队列还是线程池队列
    //              4.return future；
    //      }
    //      3.run_package_task//work_thread的主要部分
    //      {
    //              if(判断从哪个队列获得任务)
    //              {
    //                      task()
    //              }
    //              else 交出cpu时间。
    //        }

    //        4.析构函数
     

    1. #ifndef _THREAD_POOL_STEAL_HPP_
    2. #define _THREAD_POOL_STEAL_HPP_
    3.  
    4. #include "threadsafe_queue.hpp"
    5. #include "ThreadRAII.h"
    6. #include "stealing_queue.hpp"
    7. #include "function_wrapper.hpp"
    8. #include "stealing_queue.hpp"
    9.  
    10. #include
    11. #include
    12. #include
    13. #include
    14. #include
    15. class thread_pool_steal
    16. {
    17. 	typedef function_wrapper task_type;
    18. 	std::atomic_bool done;
    19. 	threadsafe_queue pool_work_queue;
    20. 	std::vector> queues;
    21. 	std::vector threads;
    22.        	join_threads joiner;
    23. 	static thread_local work_stealing_queue* local_work_queue;
    24. 	static thread_local unsigned my_index ;
    25. 	void work_thread(unsigned index){
    26. 		my_index = index;
    27. 		local_work_queue = queues[my_index].get();
    28. 		while(!done){
    29. 			run_pending_task();
    30. 		}	
    31. 	}
    32. 	bool pop_from_local(task_type& task){
    33. 		return local_work_queue && local_work_queue->try_pop(task);
    34. 	}
    35. 	bool pop_from_pool(task_type& task){
    36. 		return pool_work_queue.try_pop(task);
    37. 	}
    38. 	bool pop_from_steal(task_type& task){
    39. 		work_stealing_queue steal_queue;
    40. 		for(unsigned i = 0; isize(); i++){
    41. 			unsigned index =(my_index + i+1)%queues.size();
    42. 			if(queues[index]->try_steal(task))
    43. 			{
    44. 				return true;
    45. 			}
    46. 		}
    47. 		return false;
    48. 	}
    49. public:
    50. 	thread_pool_steal():done(false),joiner(threads)
    51. 	{
    52. 		unsigned number = std::thread::hardware_concurrency();
    53. 		try{
    54. 			for(unsigned i = 0; i
    55. 				queues.push_back(std::unique_ptr(new work_stealing_queue));
    56. 			}
    57. 			for(unsigned i = 0; i
    58. 				threads.push_back(std::thread(&thread_pool_steal::work_thread,this,i));
    59. 			}
    60. 		}
    61. 		catch(...){
    62. 			done = true;
    63. 			throw;
    64. 		}
    65. 	}
    66. 	~thread_pool_steal(){
    67. 		done = true;
    68. 	}
    69. 	template <typename Function>
    70. 	std::future<typename std::result_of<Function()>::type> submit(Function f)
    71. 	{
    72. 		typedef typename std::result_of<Function()>::type result_type;
    73. 		std::packaged_task<result_type()> task(f);
    74. 		std::future res(task.get_future());
    75. 		int index = 0;
    76. 		for(auto& ptr:queues){
    77. 			if(ptr->empty()){
    78. 				ptr->push(std::move(task));
    79. 				index = -1;
    80. 				break;
    81. 			}
    82. 			index++;
    83. 		}
    84. 		if(index>=0){
    85. 			pool_work_queue.push(std::move(task));
    86. 		}
    87. 		return res;
    88. 	}
    89. 	void run_pending_task(){
    90. 		task_type task;
    91. 		if(pop_from_local(task) || pop_from_pool(task) || pop_from_steal(task))
    92. 		{
    93. 			task();
    94. 		}
    95. 		else{
    96. 			std::this_thread::yield();
    97. 		}
    98. 	}
    99.  
    100. };
    101. thread_local work_stealing_queue* thread_pool_steal::local_work_queue = nullptr;
    102. thread_local unsigned thread_pool_steal::my_index = 0;
    103. #endif

  • 相关阅读:
    哈希表题目：键盘行
    陈年雷司令葡萄酒中的石油笔记
    django的update和create高级操作
    Spring Cloud Alibaba 容器化部署最佳实践 | 本地部署版本详解
    联通边缘AI:打造“职业技能”，助力行业高质量发展
    按钮测试： 循环遍历 单据行明细 &执行SQL
    力扣LeatCode算法题第三题-无重复字符的最长子串
    Latex参考文献中大写字母编译后自动变成了小写，如何保持原字母大写形式
    详解--Hash（中文也称散列、哈希）
    动态规划合集
  • 原文地址：https://blog.csdn.net/qq_52758467/article/details/133685416