多次并行调用函数

Question

我想多次调用具有不同 arg 值的函数。为了加快执行速度，我想在不考虑其实现的情况下调用该函数。

在循环中，我将 arg 传递给函数并现在调用它，我的程序不应等待函数执行并再次循环，而是不应认为函数尚未完成第一个循环，它应该继续并一次又一次地调用该函数，直到循环结束。

如何使用 fork 和线程来做到这一点，哪一个会更快并且代码框架会有所帮助？ 假设我的函数是 void foo(arg1,arg2)

Answer 1

如果它是一种计算密集型方法，那么为每个调用生成一个线程/进程不是一个好主意（事实上，这绝不是一个好主意！）。

您可以使用 openmp 有效地并行化 for 循环：

#pragma omp parallel for
for(int i = 0; i < N; i++)
{
    int arg1 = ...
    int arg2 = ...
    foo(arg1, arg2);
}

这将根据可用 CPU 内核的数量创建多个线程，然后在它们之间拆分迭代。您可以使用schedule 子句进一步调整调度。

Answer 2

您想要 proactor pattern. 的异步实现

这是一个示例，使用 boost::asio 和 boost::thread：

#include <iostream>
using namespace std;

#include <boost/asio.hpp>
#include <boost/thread.hpp>

// your function, with its arguments
void foo(int x, int y)
{
   cout << x << " * " << y << " = " << x * y << "\n";
}

int main()
{
   boost::asio::io_service svc;
   boost::asio::io_service::work w(svc);

   // create a thread to run the io_service proactor
   boost::thread thread(boost::bind(&boost::asio::io_service::run, &svc));

   for(int z = 0; z < 32; ++z)
   {
      // bind the function to its arguments to 
      // be called in the context of the running
      // thread
      svc.post(boost::bind(&f, 2, z));
   }

   boost::this_thread::sleep(boost::posix_time::milliseconds(1000));

   svc.stop();
   thread.join();
}

这里的好处是，如果需要，您可以通过从线程池中调用 boost::asio::io_service::run 轻松扩展。

Answer 3

就像 Tudor 推荐的那样，我也会使用来自 Intel TBB 或 Microsoft PPL 的 parallel_for 模式算法。

如果你真的想为每个函数生成一个需求线程，你可以这样做：

#include <thread>
#include <vector>

void foo(arg1, arg2) {
  //do stuff
}

int main() {

  std::vector<std::thread> threads;

  for(auto i = x; i != y; ++i)
    threads.emplace_back(foo, arg1, arg2);

  for(auto& i: threads)
    i.join();

}

Answer 4

在github 中有一个仅标头库，可用于为向量的每个元素异步调用一次函数。如果函数返回一个值，它会将结果返回到另一个向量中。它避免使用已被批评的 std::future，例如在this CppCon 演示中因为速度慢。也可以正常捕获异常。

这是一个例子：

    #include <iostream>
    #include "Lazy.h"

    template <class T1, class T2>
    double foo(T1 t1, T2 t2) {
        std::this_thread::sleep_for(std::chrono::milliseconds(1000));
        return double(t1) + double(t2);
    }


    int main() {
        constexpr int maxThreads = 16; // 0 means as many threads as there are cores

        // Make an input vector
        std::vector<std::pair<int, float>> vecIn;
        for(int i=0; i < 5; ++i)
            vecIn.push_back({10 * i, 1.1 * (i+1)});

        try {
            auto vecOut = Lazy::runForAll<maxThreads>(vecIn,
                            [](auto in){ return foo(in.first, in.second); });

            for (auto i = 0; i < vecOut.size(); ++i)
                std::cout << "foo(" << vecIn[i].first
                          << ", " << vecIn[i].second
                          << ") = " << vecOut[i] << '\n';
        } catch (...)
        { // Deal with the exception here
        }
    }

    /* Output:
    foo(0, 1.1) = 1.1
    foo(10, 2.2) = 12.2
    foo(20, 3.3) = 23.3
    foo(30, 4.4) = 34.4
    foo(40, 5.5) = 45.5
    */