用于装配的调度系统

Question

我想用多个处理器并行化一个线性运算(将一个复杂的数学函数拟合到某个数据集)。

假设我的机器中有8个内核，我想要容纳1000个数据集。我期望的是某个系统将1000个数据集作为一个队列，并将它们发送到8个核心进行处理，因此它首先从1000个数据集中的前8个作为FIFO。每个数据集的拟合时间通常都不同，因此要拟合的8个数据集中的一些数据集可能比其他数据集花费的时间更长。我希望从系统中保存拟合数据集的结果，然后继续从大队列(1000个数据集)中为完成的每个线程获取新的数据集。这必须继续，直到处理完整个1000个数据集。然后我就可以继续我的程序了。

这样的系统叫什么？在C++上有这样的模型吗？

我使用OpenMP并行化，并使用先进的C++技术，如模板和多态性。

感谢您的努力。

Answer 1

您可以将OpenMP parallel用于动态调度任务或OpenMP任务。两者都可以用来并行化每次迭代需要不同的时间来完成的情况。动态计划用于：

#pragma omp parallel
{
   Fitter fitter;
   fitter.init();
   #pragma omp for schedule(dynamic,1)
   for (int i = 0; i < numFits; i++)
      fitter.fit(..., &results[i]);
}

schedule(dynamic,1)使每个线程一次执行一个迭代，并且线程永远不会空闲，除非没有更多的迭代需要处理。

对于任务：

#pragma omp parallel
{
   Fitter fitter;
   fitter.init();
   #pragma omp single
   for (int i = 0; i < numFits; i++)
   {
      #pragma omp task
      fitter.fit(..., &results[i]);
   }
   #pragma omp taskwait
   // ^^^ only necessary if more code before the end of the parallel region
}

这里，其中一个线程运行了一个生成1000个OpenMP任务的for循环。OMP任务保存在队列中，并由空闲线程处理。它的工作原理有点类似于动态for循环，但允许在代码构造中有更大的自由度(例如，对于任务，您可以并行化递归算法)。taskwait构造等待完成所有挂起的任务。它隐含在并行区域的末尾，因此只有在并行区域的末尾之前有更多的代码跟随时，它才是真正必要的。

在这两种情况下，对fit()的每次调用都将在不同的线程中完成。您必须确保拟合一组参数不会影响拟合其他集合，例如fit()是线程安全的方法/函数。这两种情况还要求执行fit()的时间比OpenMP构造的开销高得多。

OpenMP任务需要兼容OpenMP 3.0的编译器。这将排除所有版本的MS VC++ (即使是在VS2012中的版本)，如果您碰巧在Windows上开发的话。

如果您希望每个线程只初始化一个fitter实例，那么您应该采取一些不同的方法，例如，将fitter对象设置为全局对象并设置为threadprivate

#include <omp.h>

Fitter fitter;
#pragma omp threadprivate(fitter)

...

int main()
{
   // Disable dynamic teams
   omp_set_dynamic(0);

   // Initialise all fitters once per thread
   #pragma omp parallel
   {
      fitter.init();
   }

   ...

   #pragma omp parallel
   {
      #pragma omp for schedule(dynamic,1)
      for (int i = 0; i < numFits; i++)
         fitter.fit(..., &results[i]);
   }

   ...

   return 0;
 }

这里的fitter是Fitter类的全局实例。omp threadprivate指令指示编译器将其放入线程本地存储中，例如使其成为每个线程的全局变量。这些在不同的并行区域之间持续存在。您还可以在static局部变量上使用omp threadprivate。这些也在不同的并行区域之间存在(但仅在同一函数中)：

#include <omp.h>

int main()
{
   // Disable dynamic teams
   omp_set_dynamic(0);

   static Fitter fitter; // must be static
   #pragma omp threadprivate(fitter)

   // Initialise all fitters once per thread
   #pragma omp parallel
   {
      fitter.init();
   }

   ...

   #pragma omp parallel
   {
      #pragma omp for schedule(dynamic,1)
      for (int i = 0; i < numFits; i++)
         fitter.fit(..., &results[i]);
   }

   ...

   return 0;
 }

omp_set_dynamic(0)调用禁用动态团队，即每个并行区域将始终使用OMP_NUM_THREADS环境变量指定的任意多个线程执行。

Answer 2

您基本上需要的是一个工作者池(或线程池)，它们从队列中获取一个作业，处理它，然后继续处理另一个作业。OpenMP提供了不同的方法来处理这类任务，例如障碍(所有工人都运行到某个点，只有在满足特定要求时才继续运行)或减少，以便在工人设法计算各自的部分后将值累积到全局变量中。

您的问题非常广泛，但我可以给您的另一个提示是了解一下MapReduce范式。在这个范例中，一个函数被映射到一个数据集上，并且结果被排序到存储桶中，这些存储桶使用另一个函数(也可能是同一个函数)进行简化。在您的例子中，这意味着您的每个处理器/核心/节点将一个给定函数映射到其分配的一组数据上，并将结果存储桶发送到负责组合它的另一个节点。我猜如果你想在C++中使用MapReduce，而不是使用特定的MapReduce框架，你就必须研究MPI。当你在一个节点上运行程序时，也许你可以用OpenMP做一些类似的事情，所以在网上搜索可能会有所帮助。

TL;DR搜索工作者池(线程池)、障碍和MapReduce。