使用boost将单线程转为多线程

Question

我正在尝试将代码从单线程转换为多线程（例如，创建 6 个线程而不是 1 个），同时确保它们都开始和结束而不会相互干扰。有什么方法可以做到这一点？我可以只做一个创建线程直到 i

#include <iostream>
#include <boost/thread.hpp>
#include <boost/date_time.hpp>

void workerFunc()
{
    boost::posix_time::seconds workTime(3);

    std::cout << "Worker: running" << std::endl;

    // Pretend to do something useful...
    boost::this_thread::sleep(workTime);

    std::cout << "Worker: finished" << std::endl;
}

int main(int argc, char* argv[])
{
    std::cout << "main: startup" << std::endl;

    boost::thread workerThread(workerFunc);

    std::cout << "main: waiting for thread" << std::endl;

    workerThread.join();

    std::cout << "main: done" << std::endl;

    system("pause");
    return 0;
}

Answer 1

是的，这当然是可能的。由于您不希望它们之间有任何干扰，因此请为它们提供唯一的数据以使用，这样您就不需要将对该数据的访问与std::mutex 同步或使其成为std::atomic。为了进一步减少线程间的干扰，按照std::hardware_destructive_interference_size对齐数据。

您可以使用boost::thread::hardware_concurrency() 获取当前系统上可用的硬件线程数，这样您就不必硬编码要运行的线程数。

可以使用std::ref 传递对线程的引用（否则线程将获得数据副本的引用）。

在这里，我创建了一个 std::list 的线程和一个 std::vector 的数据来处理。

#include <cstdint> // std::int64_t
#include <iostream>
#include <list>
#include <new> // std::hardware_destructive_interference_size
#include <vector>
#include <boost/thread.hpp>

unsigned hardware_concurrency() {
    unsigned rv = boost::thread::hardware_concurrency();
    if(rv == 0) rv = 1; // fallback if hardware_concurrency returned 0
    return rv;
}

// if you don't have hardware_destructive_interference_size, use something like this
// instead:
//struct alignas(64) data {
struct alignas(std::hardware_destructive_interference_size) data {
    std::int64_t x;
};

void workerFunc(data& d) {
    // work on the supplied data
    for(int i = 0; i < 1024*1024-1; ++i) d.x -= i;
    for(int i = 0; i < 1024*1024*1024-1; ++i) d.x += i;
}

int main() {
    std::cout << "main: startup" << std::endl;

    size_t number_of_threads = hardware_concurrency();
    std::list<boost::thread> threads;
    std::vector<data> dataset(number_of_threads);

    // create the threads 
    for(size_t idx = 0; idx < number_of_threads; ++idx)
        threads.emplace_back(workerFunc, std::ref(dataset[idx]));

    std::cout << "main: waiting for threads" << std::endl;

    // join all threads
    for(auto& th : threads) th.join();
    // display results
    for(const data& d : dataset) std::cout << d.x << "\n";

    std::cout << "main: done" << std::endl;
}

如果您使用的是 C++11（或更高版本），我建议您改用 std::thread。

Answer 2

启动和停止一堆 Boost 线程

std::vector<boost::thread> threads;
for (int i = 0; i < numberOfThreads; ++i) {
  boost::thread t(workerFunc);
  threads.push_back(std::move(t));
}

for (auto& t : threads) {
  t.join();
}

请记住，join() 不会终止线程，它只会等待它们完成。

同步

如果多个线程访问相同的数据并且其中至少有一个正在写入数据，则需要互斥锁。您可以使用互斥锁来确保多个线程进入代码的关键部分。示例：

std::queue<int> q;
std::mutex q_mu;

void workerFunc1() {
  // ...
  {
    std::lock_guard<std::mutex> guard(q_mu);
    q.push(foo);
  } // lock guard goes out of scope and automatically unlocks q_mu
  // ...
}

void workerFunc2() {
  // ...
  {
    std::lock_guard<std::mutex> guard(q_mu);
    foo = q.pop();
  } // lock guard goes out of scope and automatically unlocks q_mu
  // ...
}

这可以防止未定义的行为，例如从尚未完全写入的队列中读取项目。小心 - 数据竞争可能会使您的程序崩溃或损坏您的数据。我经常使用Thread Sanitizer 或Helgrind 之类的工具来确保我不会错过任何东西。如果您只想将结果传回主程序，但不需要在线程之间共享数据，您可能需要考虑使用std::promise 和std::future。

Answer 3

是的，产生新线程可以通过一个简单的循环来完成。不过，您必须牢记以下几点：

1. 如果线程将对共享数据进行操作，则需要使用互斥锁、原子或其他方式对其进行保护，以避免数据竞争和未定义行为（请记住，即使像 int 这样的原始类型也必须被包装根据标准使用原子或互斥体）。
2. 您必须确保最终在其对象超出范围之前在每个衍生线程上调用 join() 或 detach()，以防止其突然终止。
3. 最好在主线程上进行一些计算，同时等待工作线程有效地利用这段时间，而不是浪费它。
4. 您通常希望生成的线程数少于您希望的总线程数，因为程序开始运行时默认使用一个线程（主线程）。