c++ - 并行调用 std::vector 中的函数

Question

我有一个 std::vector的 std::function<void()>像这样:

std::map<Event, std::vector<std::function<void()>>> observers_;

像这样调用每个函数:

for (const auto& obs : observers_.at(event)) obs();

我想把它变成一个并行的 for 循环。因为我正在使用 C++14 ，并且无权访问 std::execution::parallel的 C++17 ，我找到了一个小库，可以让我创建一个线程池。

如何转for (const auto& obs : observers_.at(event)) obs();转换为调用 observers_ 中每个函数的版本在平行下？我似乎无法获得正确的语法。我试过了，但这不起作用。

std::vector<std::function<void()>> vec = observers_.at(event);
ThreadPool::ParallelFor(0, vec.size(), [&](int i)
{
    vec.at(i);
});

使用以下库的示例程序:

#include <iostream>
#include <mutex>

#include "ThreadPool.hpp"
////////////////////////////////////////////////////////////////////////////////

int main()
{
    std::mutex critical;
    ThreadPool::ParallelFor(0, 16, [&] (int i)
    {
        std::lock_guard<std::mutex> lock(critical);
        std::cout << i << std::endl;
    });
    return 0;
}

线程池库。

#ifndef THREADPOOL_HPP_INCLUDED
#define THREADPOOL_HPP_INCLUDED

////////////////////////////////////////////////////////////////////////////////
#include <thread>
#include <vector>
#include <cmath>
////////////////////////////////////////////////////////////////////////////////

class ThreadPool {

public:

    template<typename Index, typename Callable>
    static void ParallelFor(Index start, Index end, Callable func) {
        // Estimate number of threads in the pool
        const static unsigned nb_threads_hint = std::thread::hardware_concurrency();
        const static unsigned nb_threads = (nb_threads_hint == 0u ? 8u : nb_threads_hint);

        // Size of a slice for the range functions
        Index n = end - start + 1;
        Index slice = (Index) std::round(n / static_cast<double> (nb_threads));
        slice = std::max(slice, Index(1));

        // [Helper] Inner loop
        auto launchRange = [&func] (int k1, int k2) {
            for (Index k = k1; k < k2; k++) {
                func(k);
            }
        };

        // Create pool and launch jobs
        std::vector<std::thread> pool;
        pool.reserve(nb_threads);
        Index i1 = start;
        Index i2 = std::min(start + slice, end);
        for (unsigned i = 0; i + 1 < nb_threads && i1 < end; ++i) {
            pool.emplace_back(launchRange, i1, i2);
            i1 = i2;
            i2 = std::min(i2 + slice, end);
        }
        if (i1 < end) {
            pool.emplace_back(launchRange, i1, end);
        }

        // Wait for jobs to finish
        for (std::thread &t : pool) {
            if (t.joinable()) {
                t.join();
            }
        }
    }

    // Serial version for easy comparison
    template<typename Index, typename Callable>
    static void SequentialFor(Index start, Index end, Callable func) {
        for (Index i = start; i < end; i++) {
            func(i);
        }
    }

};

#endif // THREADPOOL_HPP_INCLUDED

Answer 1

看来你应该简单地改变:

vec.at(i); // Only returns a reference to the element at index i

进入:

vec.at(i)(); // The second () calls the function
--- OR ---
vec[i](); // Same