我是 C++ 新手，并试图了解多线程。我已经涵盖了基础知识。现在想象一下这种情况：
比如说，我有 N 个任务想要尽快完成。这很容易，只需启动 N 个线程并向后倾斜。但我不确定这是否适用于 N=200 或更多。
所以我想说：我有 N 个任务，我想启动有限数量的 M 个工作线程。 一旦之前的线程之一完成，我如何安排将任务发布到新线程？
或者所有这些都由操作系统或运行时处理，我完全不用担心，即使 N 变得真的很大？
【问题讨论】：
标签： c++ multithreading
- 我有 N 个任务，我想启动有限数量的 M 个工作线程。
- 如何安排一次任务向新线程发出一次 之前的一个主题已经结束？
M，同时考虑系统中可用的线程数 (hardware_concurrency)。counting_semaphore 确保在没有可用线程池插槽时不会启动任务。N 任务，获取线程池槽，运行任务，并释放线程池槽。请注意，由于任务是异步启动的，您将能够让M 任务并行运行。#include <future> // async
#include <iostream> // cout
#include <semaphore> // counting_semaphore
#include <vector>
static const size_t THREAD_POOL_SIZE_DEFAULT{ std::thread::hardware_concurrency() };
static const size_t THREAD_POOL_SIZE_MAX{ std::thread::hardware_concurrency() * 2 };
static const size_t NUM_TASKS_DEFAULT{ 20 };
template <typename F>
void run_tasks(
F&& f,
size_t thread_pool_size = THREAD_POOL_SIZE_DEFAULT,
size_t num_tasks = NUM_TASKS_DEFAULT)
{
thread_pool_size = std::min(thread_pool_size, THREAD_POOL_SIZE_MAX);
std::counting_semaphore task_slots(thread_pool_size);
auto futures{ std::vector<std::future<void>>(num_tasks) };
auto task_results{ std::vector<int>(num_tasks) };
// We can run thread_pool_size tasks in parallel
// If all task slots are busy, we have to wait for a task to finish
for (size_t i{ 0 }; i < num_tasks; ++i)
{
// Wait for a task slot to be free
task_slots.acquire();
futures[i] = std::async(
std::launch::async,
[i, &f, &task_result = task_results[i], &task_slots]() {
// Execute task
task_result = std::forward<F>(f)(i);
// Release the task slot
task_slots.release();
}
);
}
// Wait for all the tasks to finish
for (auto& future : futures) { future.get(); };
for (auto& result: task_results) { std::cout << result << " "; }
}
int main()
{
run_tasks([](int i) { return i * i; }, 4, 20);
}
【讨论】：
这是我对线程池的看法（尚未广泛调试）。在 main 中，它使用硬件允许的最大线程数启动一个线程池（Ted Lyngmo 指的是）
因为这个线程池还允许调用者取回异步启动调用的结果，所以涉及到很多事情
#include <cassert>
#include <condition_variable>
#include <exception>
#include <iostream>
#include <mutex>
#include <future>
#include <thread>
#include <vector>
#include <queue>
//=====================================================================================================================================
namespace details
{
// task_itf is something the threadpool can call to start a scheduled function call
// independent of argument and/or return value types
class task_itf
{
public:
virtual void execute() = 0;
};
//-------------------------------------------------------------------------------------------------------------------------------------
// A task is a container for a function call + arguments a future.
// but is already specialized for the return value type of the function call
// which the future also needs
//
template<typename retval_t>
class task final :
public task_itf
{
public:
template<typename lambda_t>
explicit task(lambda_t&& lambda) :
m_task(lambda)
{
}
std::future<retval_t> get_future()
{
return m_task.get_future();
}
std::shared_future<retval_t> get_shared_future()
{
return std::shared_future<retval_t>(m_task.get_future());
}
virtual void execute() override
{
m_task();
}
private:
std::packaged_task<retval_t()> m_task;
};
class stop_exception :
public std::exception
{
};
}
//-------------------------------------------------------------------------------------------------------------------------------------
// actual thread_pool class
class thread_pool_t
{
public:
// construct a thread_pool with specified number of threads.
explicit thread_pool_t(const std::size_t size) :
m_stop{ false }
{
std::condition_variable signal_started;
std::atomic<std::size_t> number_of_threads_started{ 0u };
for (std::size_t n = 0; n < size; ++n)
{
// move the thread into the vector, no need to copy
m_threads.push_back(std::move(std::thread([&]()
{
{
number_of_threads_started++;
signal_started.notify_all();
}
thread_loop();
})));
}
// wait for all threads to have started.
std::mutex mtx;
std::unique_lock<std::mutex> lock{ mtx };
signal_started.wait(lock, [&] { return number_of_threads_started == size; });
}
// destructor signals all threads to stop as soon as they are done.
// then waits for them to stop.
~thread_pool_t()
{
{
std::unique_lock<std::mutex> lock(m_queue_mutex);
m_stop = true;
}
m_wakeup.notify_all();
for (auto& thread : m_threads)
{
thread.join();
}
}
// pass a function asynchronously to the threadpool
// this function returns a future so the calling thread
// my synchronize with a result if it so wishes.
template<typename lambda_t>
auto async(lambda_t&& lambda)
{
using retval_t = decltype(lambda());
auto task = std::make_shared<details::task<retval_t>>(lambda);
queue_task(task);
return task->get_shared_future();
}
// let the threadpool run the function but wait for
// the threadpool thread to finish
template<typename lambda_t>
auto sync(lambda_t&& lambda)
{
auto ft = async(lambda);
return ft.get();
}
void synchronize()
{
sync([] {});
}
private:
void queue_task(const std::shared_ptr<details::task_itf>& task_ptr)
{
{
std::unique_lock<std::mutex> lock(m_queue_mutex);
m_queue.push(task_ptr);
}
// signal only one thread, first waiting thread to wakeup will run the next task.
m_wakeup.notify_one();
}
std::shared_ptr<details::task_itf> get_next_task()
{
static auto pred = [this] { return (m_stop || (m_queue.size() > 0)); };
std::unique_lock<std::mutex> lock(m_queue_mutex);
while (!pred())
{
m_wakeup.wait(lock, pred);
}
if (m_stop)
{
// use exception to break out of the mainloop
throw details::stop_exception();
}
auto task = m_queue.front();
m_queue.pop();
return task;
}
void thread_loop()
{
try
{
while (auto task = get_next_task())
{
task->execute();
}
}
catch (const details::stop_exception&)
{
}
}
std::vector<std::thread> m_threads;
std::mutex m_queue_mutex;
std::queue<std::shared_ptr<details::task_itf>> m_queue;
std::condition_variable m_wakeup;
bool m_stop;
};
//-----------------------------------------------------------------------------
int main()
{
thread_pool_t thread_pool{ std::thread::hardware_concurrency() };
for (int i = 0; i < 200; i++)
{
// just schedule asynchronous calls, returned futures are not used in this example
thread_pool.async([i]
{
std::cout << i << " ";
});
}
// this threadpool will not by default wait until all work is finished
// but stops processing when destructed.
// a call to synchronize will block until all work is done that is queued up till this moment.
thread_pool.synchronize();
std::cout << "\nDone...\n";
return 0;
}
【讨论】：
std::thread::hardware_concurrancy 可能有助于确定您想要多少线程。如果它返回除0 之外的任何内容，则它是可以同时运行的并发线程数。它通常是 CPU 内核的数量乘以每个内核可以运行的超线程的数量。 12 核和 2 HT:s/core 为 24。超过这个数字可能只会减慢一切。
您可以创建一个备用线程池来处理您的命令，因为创建线程有点昂贵。如果您有 1000000 个任务要处理，那么您希望 24 个线程（在此示例中）始终处于运行状态。
尽管这是一个非常常见的场景，并且自 C++17 以来，许多标准算法（如 std::for_each）都添加了一个附加功能，以使它们根据执行策略执行。如果您希望它并行执行，它将使用内置线程池（很可能）来完成任务。
例子：
#include <algorithm>
#include <execution>
#include <vector>
struct Task {
some_type data_to_work_on;
some_type result;
};
int main() {
std::vector<Task> tasks;
std::for_each(std::execution::par, tasks.begin(), tasks.end(), [](Task& t) {
// work on task `t` here
});
// all tasks done, check the result in each.
}
【讨论】：
不，您不想创建 200 个线程。虽然它可能工作得很好，但创建线程涉及大量的处理开销。相反，您需要一个“任务队列”系统，其中一个工作线程池（通常大小等于 CPU 内核的数量）从需要完成的事情的共享队列中提取。英特尔 TBB 包含一个常用的任务队列实现，但也有其他实现。
【讨论】：