並列処理のためのC++ライブラリ拡張
～C++17: 並列STLの概要

「ParallelSTL」 ── 近未来の並列STL

　 マクラがえらく長くなりましたが、ここからが本題。2017年のお披露目を目論む次期標準C++規格:C++17に導入が検討されている標準ライブラリの一つに、通称"ParallelSTL"があります。並列STLの名のとおり、STLのアルゴリズム群の多くを並列処理に対応させようという意欲的なライブラリで、その草案（N3960 "Working Draft, Technical Specification for C++ Extensions for Parallelism"／PDF）はGPGPUによる並列コンピューティングの担い手:NVIDIAの手によるものです。NVIDIAのGPGPUフレームワーク:CUDAが内包する並列アルゴリズム・ライブラリ:Thrust（旧komrade）がベースになっていると聞きました。

　このParallelSTL、紙の上だけじゃなく実装されて動くモノがないかと探してみたらありました。Microsoft版ParallelSTLがCodePlexで公開されています。Microsoft版ParallelSTLはまだ開発途中でバージョン番号も付けられておらず、業務に使うわけにはいかないのシロモノですが、近未来の並列アルゴリズム・ライブラリの早期体験版として試す価値は大いにありそうです。バージョン番号が付いてないくらいですからダウンロード・パッケージも見当たりません。リポジトリからzipで固められたコード・セットを拾ってくることになります。

　zipを解いてVisual Studio 2013でソリューション:ParallelSTL.slnを開き、プロジェクト:ParallelSTLDesktopをビルドすれば、Release／DebugディレクトリにParallelSTL.dll／libが生成されます。これとincludeディレクトリ配下のヘッダ群を利用することになります。

　PPLを用いた先ほどのコードにParallelSTLのsortを追加したものを示します。ヘッダと名前空間はParallelSTLの仕様自体が実験／草案段階にあるためそれぞれ<experimental/algoritm>, std::experimental::parallelとなっていますが、C++17正式採用のあかつきには<algorithm>, atd::parallelとなるはずです。

list03　sort_ParallelSTL.cpp

#include <ppl.h>     // parallel_sort
#include <experimental/algorithm> // parallel::sort

#include <iostream>  // cout, endl
#include <chrono>    // clock, time_point, duration
#include <vector>    // vector
#include <array>     // array;
#include <numeric>   // iota

using namespace std;

// 関数fの実行時間を計測する
template<typename Function>
void measure(Function f) { /* 省略 */ }

int main() {
  const int N = 4000000;
  vector<int> data(N);
  iota(begin(data), end(data), 0);

  reverse(begin(data), end(data));
  cout << "std::sort:      ";
  measure([&]() { sort(begin(data), end(data)); });

  reverse(begin(data), end(data));
  cout << "parallel_sort:  ";
  measure([&]() { concurrency::parallel_sort(begin(data), end(data)); });

  reverse(begin(data), end(data));
  cout << "parallel::sort: ";
  measure([&]() { experimental::parallel::sort(experimental::parallel::par,
                                               begin(data), end(data)); });
}

　……よしよし、PPLのparallel_sortと同等のスピードで動いてます。

　ParallelSTLが提供する並列アルゴリズムは以下のとおり、STLアルゴリズムの多くに対応した並列版を提供してくれています。

table01

adjacent_find	all_of	any_of	copy
copy_if	copy_n	count	count_if
equal	exclusive_scan	fill	fill_n
find	find_end	find_first_of	find_if
find_if_not	for_each	for_each_n	generate
generate_n	includes	inclusive_scan	inplace_merge
is_heap	is_partitioned	is_sorted	is_sorted_until
lexicographical_compare	max_element	merge	min_element
minmax_element	mismatch	move	none_of
nth_element	partial_sort	partial_sort_copy	partition
partition_copy	reduce	remove	remove_copy
remove_copy_if	reverse	reverse_copy	rotate
rotate_copy	search	search_n	set_difference
set_intersection	set_symmetric_difference	set_union	sort
stabe_partition	stable_sort	swap_ranges	transform
uninitialized_copy	uninitialized_copy_n	uninitialized_fill	uninitialized_fill_n
unique	unique_copy

　並列アルゴリズムはどれも、第一引数に実行ポリシー（execution_policy）を与えます。実行ポリシーは次の3つ：

• parallel::seq 単一スレッドによる順次（sequential）実行
• parallel::par 複数スレッドによる並列（parallel）実行
• parallel::vec ベクトル化命令によるベクトル（vectorized）実行

　並列アルゴリズムは与えられた実行ポリシーに応じて、順次／並列／ベクトル実行による処理を行います。Microsoft版ParallelSTLは、現時点ではparallel::vecをサポートしていません。これがサポートされれば、Intel SSEやAVXなどのベクトル化命令を利用した高速化が期待できます。

　また、実行ポリシーは実行時に動的に変更することができます。例えば、データ数が1000未満のときは並列実行だと余計に遅くなってしまうことが分かっているなら：

list04

vector<int> data = ...;

parallel::execution_policy exec = parallel::par;
if ( data.size() < 1000 ) {
  exec = parallel::seq;
}

parallel::sort(exec, data.begin(), data.end();

のように、実行ポリシーの動的コンテナ:execution_policyのナカミを差し替えることで、状況に応じた実行ポリシーで処理することができます。

　ParallelSTLの並列アルゴリズムを使う際の注意点は、

• 処理の順序(前後関係)に依存してはならない
• データの競合(data race)が起こりうる

など、マルチスレッドと同じ落とし穴にハマらないこと。例えば絶対値の総和を求める以下のコード：

list05　data_race.cpp

#include <experimental/algorithm>
#include <iostream>
#include <iterator>

using namespace std;
using namespace std::experimental;

int main() {
  const size_t N = 10000;
  int data[N];
  // -1, 1, -1, 1, -1 ... で埋める
  for ( size_t i = 0; i < N; ++i ) data[i] = i % 2 ? 1 : -1;
  
  int abs_sum = 0;
  parallel::for_each(parallel::par,
    begin(data), end(data),
    [&](int item) {
      abs_sum += item > 0 ? item : -item;
    }
  );
  cout << abs_sum << endl;
}

　この結果が10000になることを期待しちゃいけません。parallel::for_eachに与えた関数オブジェクト（ここではラムダ式）は複数のスレッドで同時に動くのですから、abs_sumに足しこむ際にデータ競合を起こすでしょう。

　mutexで排他するか、

list06

mutex mtx;
int abs_sum = 0;
parallel::for_each(parallel::par,
  begin(data), end(data),
  [&](int item) {
    lock_guard<mutex> guard(mtx);
    abs_sum += item > 0 ? item : -item;
  }
);

　atomic変数を使わないとね。

list07

atomic<int> abs_sum = 0;
parallel::for_each(parallel::par,
  begin(data), end(data),
  [&](int item) {
    abs_sum += item > 0 ? item : -item;
  }
);

　それともう一つ、例外の扱いが少しばかり特殊です。さきほどのparallel::for_eachに与えた関数オブジェクトの中で例外をthrowすると、並列アルゴリズム内で作られたスレッドの中から外へ、例外がスレッドをまたいで伝播することになります。加えて複数のスレッドから同時に例外が飛んでくるかもしれません。これらに対処すべく、関数オブジェクト内でthrowされた例外はstd::exception_ptrに変換され、さらに複数の例外を内包するparallel::exception_listに追加されてthrowされます。

　従って、呼び出し元で例外を捕まえるには、例えばこんなコードになりますか。

list08

try {
  parallel::for_each(parallel::par,
    begin(data), end(data),
    [](int item) {
       // この中で foo_error/bar_error が
       // throwされる
    });
} catch ( std::bad_alloc& memory_error ) {
  ...
} catch ( parallel::exception_list& errors ) {
  for ( std::exception_ptr e : errors ) {
    try {
      std::rethrow_exception(e);
    } catch ( foo_error& ferr ) {
      // ここでfoo_errorに対処する
    } catch ( bar_error& berr ) {
      // ここでbar_errorに対処する
    }
  }
}

　……なにやら面白いことになってきました。並列アルゴリズムの草案を提出したNVIDIAがCUDAを、あるいはMicrosoftがC++ AMPを使ってGPGPUによる"超"並列アルゴリズムの実装をリリースしてくれるかもしれません。C++17が今から楽しみです。