From 7e302bba477109b3e401299c9d15495aed5cfdf4 Mon Sep 17 00:00:00 2001 From: jkjk Date: Sun, 31 Mar 2024 15:37:04 +0800 Subject: [PATCH 1/4] =?UTF-8?q?=E6=B7=BB=E5=8A=A0=E8=A7=A3=E8=AF=BB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- read.md | 104 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 104 insertions(+) create mode 100644 read.md diff --git a/read.md b/read.md new file mode 100644 index 0000000..0178354 --- /dev/null +++ b/read.md @@ -0,0 +1,104 @@ +## 以这个代码为例 https://github.com/progschj/ThreadPool +## 数据结构: +```cpp +private: + // need to keep track of threads so we can join them + std::vector< std::thread > workers; + // the task queue + std::queue< std::function > tasks; + + // synchronization + std::mutex queue_mutex; + std::condition_variable condition; + bool stop; +``` +可以看到一个vector存放线程，一个队列存放任务队列，一把锁控制并发，一个条件变量，一个布尔变量控制整个线程池的任务执行与否 + +## 构造函数: +```cpp +// the constructor just launches some amount of workers +inline ThreadPool::ThreadPool(size_t threads) + : stop(false) +{ + for(size_t i = 0;i task; + { + std::unique_lock lock(this->queue_mutex); + this->condition.wait(lock, + [this]{ return this->stop || !this->tasks.empty(); }); + if(this->stop && this->tasks.empty()) + return; + task = std::move(this->tasks.front()); + this->tasks.pop(); + } + task(); + } + } + ); +} +``` +1. `emplace_back` 与`push_back`不同，`emplace_back` 直接在`vector`尾部构造元素，省了一次额外复制或移动操作 [emplace_back参考](https://zh.cppreference.com/w/cpp/container/vector/emplace_back) +2. 传入 `lambda` 表达式，函数内容是 启动`threads`个线程，每个线程都循环读取任务队列的任务，直到队列任务被执行完或者 `stop` 被置为 `true` 。 +3. 通过条件变量来触发线程运行，值得一提的是为了检测虚假唤醒，加入了条件判断 如果没有停止或者还有任务，就运行线程。 +4. 通过`move`转移生命周期控制权到`task`，之后执行`task` + +## 析构函数： +```cpp +// the destructor joins all threads +inline ThreadPool::~ThreadPool() +{ + { + std::unique_lock lock(queue_mutex); + stop = true; + } + condition.notify_all(); + for(std::thread &worker: workers) + worker.join(); +} +``` + +1. 第一个作用域，首先获取锁，然后`stop`置为`true` +2. 条件变量通知所有线程 +3. 所有线程`join`。析构的时候阻塞，等待所有线程执行完。 + + +## enqueue 函数: +```cpp +// add new work item to the pool +template +auto ThreadPool::enqueue(F&& f, Args&&... args) + -> std::future::type> +{ + using return_type = typename std::result_of::type; + + auto task = std::make_shared< std::packaged_task >( + std::bind(std::forward(f), std::forward(args)...) + ); + + std::future res = task->get_future(); + { + std::unique_lock lock(queue_mutex); + + // don't allow enqueueing after stopping the pool + if(stop) + throw std::runtime_error("enqueue on stopped ThreadPool"); + + tasks.emplace([task](){ (*task)(); }); + } + condition.notify_one(); + return res; +} +``` + +1. 定义模板, 关于模板不是很懂 +2. `emplace`推入新元素到队列结尾。原位构造元素，即不进行移动或复制操作 +3. 为什么用一个`lambda`函数包一层而不是直接把 `(*task)()` 推进去? + + 1. 首先明确`task`是什么类型？是`shared_ptr`, 指向对象是函数。 + 2. 因为加入`tasks`的元素`function Date: Sun, 31 Mar 2024 15:56:29 +0800 Subject: [PATCH 2/4] =?UTF-8?q?=E6=B7=BB=E5=8A=A0=E5=AE=9E=E7=8E=B0?= =?UTF-8?q?=E6=80=9D=E8=B7=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- read.md | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/read.md b/read.md index 0178354..4214f17 100644 --- a/read.md +++ b/read.md @@ -1,4 +1,8 @@ ## 以这个代码为例 https://github.com/progschj/ThreadPool + +## 实现思路： +线程池中持有线程和任务队列，使用线程和锁的模型。初始化的时候创建一些线程，之后推入一些任务，每次推入后通过条件变量触发某个阻塞的线程执行。线程池析构的时候`Unblocks all threads currently waiting for *this.`。 线程和任务队列的关系是每个线程都会从任务队列读取没被执行的任务，然后执行。 + ## 数据结构: ```cpp private: From 845d3f95d5477573a8d42e09ff0f01246fdb6a63 Mon Sep 17 00:00:00 2001 From: jkjk Date: Sun, 31 Mar 2024 15:59:33 +0800 Subject: [PATCH 3/4] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E6=8F=8F=E8=BF=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- read.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/read.md b/read.md index 4214f17..96ea6d1 100644 --- a/read.md +++ b/read.md @@ -105,4 +105,4 @@ auto ThreadPool::enqueue(F&& f, Args&&... args) 1. 首先明确`task`是什么类型？是`shared_ptr`, 指向对象是函数。 2. 因为加入`tasks`的元素`function Date: Wed, 22 May 2024 14:51:27 +0800 Subject: [PATCH 4/4] fix typo --- read.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/read.md b/read.md index 96ea6d1..51d884c 100644 --- a/read.md +++ b/read.md @@ -104,5 +104,5 @@ auto ThreadPool::enqueue(F&& f, Args&&... args) 3. 为什么用一个`lambda`函数包一层而不是直接把 `(*task)()` 推进去? 1. 首先明确`task`是什么类型？是`shared_ptr`, 指向对象是函数。 - 2. 因为加入`tasks`的元素`function`的要求](https://zh.cppreference.com/w/cpp/utility/functional/function)） 4. 解除某个被阻塞线程的阻塞 \ No newline at end of file