PPLの並列コンテナ
前置きが長くなりました。複数のスレッド間でのデータのやり取りに使える並列コンテナを提供してくれるライブラリの一つが Microsoft PPL（Parallel Patterns Library）で、Visual Studioに同梱されています。PPLは以下の並列コンテナをサポートしています：
- concurrent_vector
- concurrent_queue
- concurrent_unordered_map
- concurrent_unordered_multimap
- concurrent_unordered_set
- concurrent_unordered_multiset
補足
Linux（とWindows、OS X、Android）での並列コンテナならIntel TBB:Threading Building Blocksがオススメ。TBBはPPLが提供する上記に加え、concurrent_bounded_queue／concurrent_priority_queueが使えます。これら並列コンテナの置かれた名前空間は、PPLはConcurrency（またはconcurrency）、TBBではtbbです。
std::vectorでは勘定の合わなくなった素数を求めるコードのstd::vectorをconcurrency::concurrent_vectorに差し替えて試してみます。
ご覧のとおり、mutexで排他しなくても正しく動いてくれてますね。さらに、要素を1つずつpush_back()したときの各要素のアドレスは：
いったんコンテナに格納された要素のアドレスはstd::vectorとは異なり要素数が増えても変化しません。つまりイテレータが無効にならないってことですから、他のスレッドがイテレータを介して要素をアクセスしても安全なんです。ただしその代償として、要素の並びが不連続です。なので std::vectorのように、
int* p = &c[0]; // 先頭要素を指すポインタ
for ( int i = 0; i < c.size(); ++i ) {
// *pに対してなんかする
++p; // 次の要素へ
}
のような、要素の連続性をアテにしたコードは正しく動いてくれません。
Microsoftの開発情報：MSDNにはconcurrent_vectorの特徴およびvectorとの違いについて解説されてはいるものの、機械翻訳のせいかまるで日本語になっていません。
英語原文によりますと：
- 要素の追加と参照およびイテレータによる走査は同時実行セーフ（concurrency-safe：複数のスレッドから同時に行っても安全）である。
- 要素の追加は末尾への挿入を行うpush_backのみであり、insertを持たない。
- 要素の追加：push_backにおいて、moveセマンティクスをサポートしない。
- 要素の削除：erase、pop_backを持たない。全要素を削除するclearはあるが同時実行セーフではない。
- 各要素はメモリ上で連続していない。
- 要素数を拡大するgrow_by、glow_at_leastは同時実行セーフである。そのかわりresizeを持たない。
- 要素の追加あるいは要素数の拡大時、要素の移動／再配置を行わない。
- bool型に対する特殊化（concurrent_vector<bool>）は定義されていない。
とのこと。
concurrent_queue
そんなわけで、concurrent_vectorは同時実行セーフが至上命令であるがために要素数が増える一方のコンテナです。
対して要素数が増減する並列コンテナ：concurrent_queueを紹介しておきましょう。
concurent_queueはその名のとおり、同時実行セーフなキュー（待ち行列／FirstIn-FirstOutバッファ）です。列の長さ（要素数）はメモリの許す限り無制限。サンプルを1つ用意しました。
この世にあるおよそあらゆるアプリケーションは入力／計算／出力の一連の処理を行います。入力部をproduce（生産）、計算と出力部をconsume（消費）と呼ぶことにします。produceに必要な時間は平均5ms、consumeは平均50msとすると、"produceののちconsume"をくりかえしたときの一回当たりの平均処理時間は55msですわね。コードを書いて確かめます。
#include <iostream>
#include <thread>
#include <chrono>
#include <random>
const int N = 100;
std::mt19937 gen;
std::uniform_int_distribution<> dist(1,100);
auto r = [&]() { return dist(gen); }; // 1～100の一様乱数
// 生産者
int produce() {
int val = r();
std::this_thread::sleep_for(std::chrono::milliseconds(val/10));
return val;
}
// 消費者
void consume(int) {
int val = r();
std::this_thread::sleep_for(std::chrono::milliseconds(val));
}
int main() {
auto start = std::chrono::high_resolution_clock::now();
for ( int i = 0; i < N; ++i ) {
consume(produce());
}
auto stop = std::chrono::high_resolution_clock::now();
std::cout << std::chrono::duration_cast<std::chrono::microseconds>(stop - start).count() / 1000.0f << " [ms]\n";
}
100回繰り返したんなら……まぁこんなところでしょう。
このときproduceは、そのあとに続くconsumeの遅さにひきずられてフル稼働できていません。consumeが終わるまで次のproduceが待たされています。produceとconsumeをそれぞれ別のスレッドで実行し、produceからconsumeへのデータの引き渡しにconcurrent_queueを使います。produceとconsumeの処理時間の差を待ち行列に吸収してもらおうってタクラミです。
#include <iostream>
#include <thread>
#include <future>
#include <chrono>
#include <random>
#include <concurrent_queue.h>
const int N = 100;
std::mt19937 gen;
std::uniform_int_distribution<> dist(1,100);
auto r = [&]() { return dist(gen); };
concurrency::concurrent_queue<int> pc_queue;
void produce() {
for ( int i = 0; i < N; ++i ) {
int val = r();
std::this_thread::sleep_for(std::chrono::milliseconds(val/10));
pc_queue.push(val);
}
pc_queue.push(-1);
}
int consume() {
int val;
int count = 0;
while ( true ) {
while ( !pc_queue.try_pop(val) ) {
std::this_thread::yield();
}
if ( val < 0 ) {
pc_queue.push(-1);
return count;
}
std::this_thread::sleep_for(std::chrono::milliseconds(r()));
++count;
}
}
int main() {
auto start = std::chrono::high_resolution_clock::now();
std::thread thr_produce(produce);
std::future<int> consume = std::async(consume);
thr_produce.join();
auto mid = std::chrono::high_resolution_clock::now();
if ( consume.get() != N ) {
std::cout << "oops!\n";
}
auto stop = std::chrono::high_resolution_clock::now();
std::cout << std::chrono::duration_cast<std::chrono::microseconds>(mid - start).count() / 1000.0f << " [ms] produce\n";
std::cout << std::chrono::duration_cast<std::chrono::microseconds>(stop - start).count() / 1000.0f << " [ms] consume\n";
}
うまいこと動いてくれました。produceは早々に完了してますし、consumeもproduceからのデータ到着を待つ頻度が減ったことで少しばかり速くなっています。
produceが生産したデータの順序を維持しなくても構わないなら、consumeを複数起こしてパフォーマンスを上げることも可能です。このときconcurrent_queueを複数のスレッドが読み出すことになります。
int main() {
auto start = std::chrono::high_resolution_clock::now();
std::thread thr_produce(produce);
std::future<int> consume1 = std::async(consume);
std::future<int> consume2 = std::async(consume);
thr_produce.join();
auto mid = std::chrono::high_resolution_clock::now();
if ( consume1.get() + consume2.get() != N ) {
std::cout << "oops!\n";
}
auto stop = std::chrono::high_resolution_clock::now();
std::cout << std::chrono::duration_cast<std::chrono::microseconds>(mid - start).count() / 1000.0f << " [ms] produce\n";
std::cout << std::chrono::duration_cast<std::chrono::microseconds>(stop - start).count() / 1000.0f << " [ms] consume\n";
}
consumeスレッドが2つ独立に動いたことで処理時間はほぼ半分、同時実行セーフな並列コンテナのおかげでデータの欠落もありません。
処理時間の異なるスレッド間に待ち行列を挟むことで時間差を吸収する例を紹介しました。もう一つの解法にパイプライン（pipeline）と呼ばれるからくりがあります。前段と後段でそれぞれスレッドを起こすのではなく、前段/後段の一連の処理を行うスレッドを複数起こす方法で、『インテル TBBによるパイプライン処理』で解説しています。併せてご一読を。
