在 C++ 中使用多线程进行重复模拟答案

【问题标题】：Repetitive simulations with multithreading in C++在 C++ 中使用多线程进行重复模拟
【发布时间】：2020-08-03 20:24:30
【问题描述】：

我想多次模拟一个场景，即 repCnt = 100。为了加快进程，我想使用多个线程，它们并行工作并在必须将结果记录到文件中时使用互斥锁。

他们需要通过以 NUM_THREADS = 4 为一组工作来从重复总数中扣除。使用互斥锁的日志记录部分很容易，但我不知道主循环应该是什么样子。

这是一个开始：

NUM_THREADS = 4

void simulate(struct argType arg) {

}


int main() {
    // ... Some code here ... 

    vector<thread> vecOfThreads;

    for (rep = 0; rep < repCnt; rep++) {

        // Here they should work in groups of 4, i.e., rep = 1, 2, 3, 4

        // They need to call the simulate(struct argType) function while they are working

        // Once a thread is done, it should get the next task from the loop, i.e., rep = 5

    } 

    return 0;
}

当我在 C++ 中搜索线程池时，我发现所有的类和方法太多。应该有一种更快的方法来完成我想要实现的目标。有人可以帮我写出最简单最短的 C++ 代码吗？

【问题讨论】：

arg 中有什么内容？每次通话都一样吗？
所有快速简单的解决方案都将涉及在每次循环迭代中创建和销毁线程。该标准不能很好地支持线程池，并且在 Networking TS 或类似的东西被合并之前不会支持。你可以使用像 Boost.Asio 这样的库吗？还是平台特定的线程池？
您的编译器是否支持 OpenMP？在那里，您可以轻松地创建一个并行区域并在循环中定义 tasks。不幸的是，C++ 线程并不支持这么简单的解决方案。
这能回答你的问题吗？ Thread pooling in C++11
这里有一些非常简单的线程队列代码here。您可以根据需要对其进行调整。

标签： c++ multithreading threadpool

【解决方案1】：

仅使用标题cxxpool:

#include "cxxpool.h"
#include <iostream>
#include <mutex>
#include <thread>

std::mutex cout_mutex;

struct some_arg {
    int i;
};

void simulate(const some_arg arg) {
    std::lock_guard<std::mutex> lock(cout_mutex);
    std::cout << "Hello from task #" << arg.i << " and thread " << std::this_thread::get_id() << std::endl;
}

int main() {
    cxxpool::thread_pool pool{4};
    for (int i = 0; i < 100; i++) {
        pool.push(simulate, some_arg{i});
    }
}

【讨论】：

即使我不想使用额外的标题、类等，我也试过了。但它在运行时崩溃。我怀疑这是赛车状况，但事实并非如此。我在代码的关键部分之前返回，但它仍然崩溃。
这个确切的代码？它可能在哪里产生竞争条件？
不，我将它改编为我的代码。需要互斥锁的区域被注释掉。所以现在它是一个代码，只模拟但不记录任何东西。不过，它对我来说还是崩溃了
我只能重复一遍：请附加一个调试器并找出它崩溃的地方，而不是将您的程序视为黑匣子。
我重新配置了调试环境，使其正常工作。我能够检测到它崩溃的原因。谢谢。

【解决方案2】：

一种简单的方法是在主线程中使用三个 for 循环，在工作线程中使用一个 while 循环：

在主线程中：

1) 用您的工作单元排满队列

2) 创建工作线程并将它们存储在向量中

3) 等待线程完成

在工作线程中：

在一个循环中，尝试使用互斥锁一次将一个工作单元从工作队列中取出。如果队列为空，则所有处理都完成。

#include <iostream>
#include <string>
#include <vector>
#include <queue>
#include <thread>
#include <mutex>

struct Args {
    int num;

    Args(int num) : num(num) {}
};

void worker(std::queue<Args>& work, std::mutex& workMutex, std::vector<Args>& results, std::mutex& resultMutex) {
    while(true) {
           Args arg(-1);

           // dequeue work unit
           {
               std::lock_guard<std::mutex> lock(workMutex);
               if (work.empty()) break;
               arg = work.front();
               work.pop();
           }

           // process arg
           Args result(arg.num + 1);

           // save your results
           {
                std::lock_guard<std::mutex> lock(resultMutex);
                results.push_back(result);
           }
    }
}

int main() {
    std::vector<std::thread> threads;
    std::queue<Args> work;
    std::mutex workMutex;
    std::vector<Args> results;
    std::mutex resultMutex;

    const int NUMBER_OF_THREADS = 4;
    const int NUMBER_OF_WORK_UNITS = 100;

    // enqueue work units
    for (int i = 0; i < NUMBER_OF_WORK_UNITS; ++i) {
        work.push(Args(i));
    }

    // create worker threads
    for (int i = 0; i < NUMBER_OF_THREADS; ++i) {
        threads.emplace_back(worker, std::ref(work), std::ref(workMutex), std::ref(results), std::ref(resultMutex));
    }

    // wait for threads completing all work units
    for (std::thread& t : threads) {
        t.join();
    }

    // process results
    for (const Args& result : results) {
        std::cout << result.num << "\n";
    }
}

【讨论】：

【解决方案3】：

恕我直言，最简单的解决方案是使用 OpenMP - 不需要外部库，几乎所有编译器都支持它。

这个页面很好地介绍了 OpenMP：http://jakascorner.com/blog/2016/04/omp-introduction.html 此处提供了有关循环并行性的更多详细信息：https://pages.tacc.utexas.edu/~eijkhout/pcse/html/omp-loop.html

【讨论】：