c++ - 运行固定数量的线程

Question

随着c++17的新标准，我想知道是否有一个好的方法来启动一个具有固定数量线程的进程，直到一批作业完成。

你能告诉我如何实现这段代码的预期功能吗:

std::vector<std::future<std::string>> futureStore;
const int batchSize             = 1000;
const int maxNumParallelThreads = 10;
int threadsTerminated           = 0;

while(threadsTerminated < batchSize)
{
    const int& threadsRunning = futureStore.size();
    while(threadsRunning < maxNumParallelThreads)
    {
        futureStore.emplace_back(std::async(someFunction));
    }
    for(std::future<std::string>& readyFuture: std::when_any(futureStore.begin(), futureStore.end()))
    {
        auto retVal = readyFuture.get(); 
        // (possibly do something with the ret val)
        threadsTerminated++;
    }
} 

我读到，曾经有一个 std::when_any 函数，但它是一个使它进入 std 特性的特性。

当前的标准库是否支持此功能(不一定是 std::future-s)？有没有一种方法可以轻松实现它，还是我必须解决类似 this 的问题？ ？

Answer 1

在我看来这不是理想的方法:

1. 您的主线程所做的就是等待您的其他线程完成，轮询您 future 的结果。几乎以某种方式浪费了这个线程......

2. 我不知道 std::async 在多大程度上以任何合适的方式重用了线程的基础设施，因此您每次都冒着创建全新线程的风险...(除此之外您可能不会创建如果您未明确指定 std::launch::async，请参阅 here。

我个人更喜欢另一种方法:

1. 一次创建您要使用的所有线程。
2. 让每个线程运行一个循环，重复调用 someFunction()，直到达到所需任务的数量。

实现可能类似于此示例:

const int BatchSize = 20;
int tasksStarted = 0;
std::mutex mutex;
std::vector<std::string> results;

std::string someFunction()
{
    puts("worker started"); fflush(stdout);
    sleep(2);
    puts("worker done"); fflush(stdout);
    return "";
}

void runner()
{
    {
        std::lock_guard<std::mutex> lk(mutex);
        if(tasksStarted >= BatchSize)
            return;
        ++tasksStarted;
    }
    for(;;)
    {
        std::string s = someFunction();
        {
            std::lock_guard<std::mutex> lk(mutex);
            results.push_back(s);
            if(tasksStarted >= BatchSize)
                break;
            ++tasksStarted;
        }
    }
}

int main(int argc, char* argv[])
{
    const int MaxNumParallelThreads = 4;

    std::thread threads[MaxNumParallelThreads - 1]; // main thread is one, too!
    for(int i = 0; i < MaxNumParallelThreads - 1; ++i)
    {
        threads[i] = std::thread(&runner);
    }
    runner();

    for(int i = 0; i < MaxNumParallelThreads - 1; ++i)
    {
        threads[i].join();
    }

    // use results...

    return 0;
}

这样，您就不会重新创建每个线程，而是继续执行直到完成所有任务。

如果这些任务不像上面的例子那样完全相同，您可以创建一个带有纯虚函数(例如“execute”或“operator ()”)的基类 Task 并创建子类需要实现(并持有任何必要的数据)。

然后您可以将实例作为指针放入 std::vector 或 std::list(好吧，我们不会迭代，list 可能适合这里...)(否则，您将进行类型删除!)和让每个线程在完成前一个任务后删除其中一个任务(不要忘记防止竞争条件!)并执行它。一旦没有剩余任务，返回...