c++11 async<>，可用内核数量未知

Question

我的 C++ 代码对时间序列数据 (t2 >> t1) 计算非常大的积分。积分是固定长度的，当前存储在 [m x 2] double 列数组中。第一列是时间。第 2 列是正在整合的信号。代码在四核或八核机器上运行。

对于具有 k 个内核的机器，我想:

• 分拆 k-1 个工作进程(每个剩余核心一个)以评估积分的各个部分(梯形积分)并将其结果返回到等待的主线程。
• 无需深度复制原始数组的部分即可实现上述目标。
• 实现 C++11 异步模板以实现可移植性

如何在不硬编码可用内核数量的情况下实现上述目标？

我目前使用的是 VS 2012。

清晰度更新:

例如，这是粗略的伪代码

data is [100000,2] double

result = MyIntegrator(data[1:50000,1:2]) + MyIntegrator(data[50001:100000, 1:2]); 

我需要在单独的线程中评估 MyIntegrator() 函数。主线程等待这两个结果。

Answer 1

这是对问题进行多线程集成的源代码。

#include <vector>
#include <memory>
#include <future>
#include <iterator>
#include <iostream>

struct sample {
  double duration;
  double value;
};
typedef std::pair<sample*, sample*> data_range;
sample* begin( data_range const& r ) { return r.first; }
sample* end( data_range const& r ) { return r.second; }

typedef std::unique_ptr< std::future< double > > todo_item;

double integrate( data_range r ) {
  double total = 0.;
  for( auto&& s:r ) {
    total += s.duration * s.value;
  }
  return total;
}

todo_item threaded_integration( data_range r ) {
  return todo_item( new std::future<double>( std::async( integrate, r )) );
}
double integrate_over_threads( data_range r, std::size_t threads ) {
  if (threads > std::size_t(r.second-r.first))
    threads = r.second-r.first;
  if (threads == 0)
    threads = 1;
  sample* begin = r.first;
  sample* end = r.second;

  std::vector< std::unique_ptr< std::future< double > > > todo_list;

  sample* highwater = begin;

  while (highwater != end) {
    sample* new_highwater = (end-highwater)/threads+highwater;
    --threads;
    todo_item item = threaded_integration( data_range(highwater, new_highwater) );
    todo_list.push_back( std::move(item) );
    highwater = new_highwater;
  }
  double total = 0.;
  for (auto&& item: todo_list) {
    total += item->get();
  }
  return total;
}

sample data[5] = {
  {1., 1.},
  {1., 2.},
  {1., 3.},
  {1., 4.},
  {1., 5.},
};
int main() {
  using std::begin; using std::end;
  double result = integrate_over_threads( data_range( begin(data), end(data) ), 2 );
  std::cout << result << "\n";
}

它需要一些修改才能完全按照您指定的格式读取数据。

但是你可以用 std::thread::hardware_concurrency() 作为线程数来调用它，它应该可以工作。

(特别是，为了简单起见，我有成对的 (duration, value) 而不是 (time, value)，但这只是一个次要细节)。