Intel TBBの並列コンテナによる安全でスケーラブルな並列処理

はじめに

　マルチスレッドアプリケーションは、コーディングもテストもデバッグも難しいことで知られています。しかし、マルチコアのデスクトップシステムとラップトップシステムに秘められた高いパフォーマンスを最大限に利用するために、開発者たちはアプリケーションをスレッド化するという難題に取り組んでいます。マルチスレッドアプリケーション開発の困難な問題に効く万能薬はありませんが、既存のライブラリとツールを活用すれば、この過渡期の負担を大幅に軽減することができます。

　本稿では、C++アプリケーションをスレッド化するときに遭遇する正確さとパフォーマンスの問題の主な原因の1つ、すなわちスレッドセーフでないコンテナクラスの使用について考察することにします。まず、この問題がなぜ起きるのかを例で示し、それからIntel Threading Building Blocks（Intel TBB）ライブラリの並列コンテナクラスについて説明します。このライブラリはマルチスレッドアプリケーションの開発を支援すべく設計されたC++テンプレートライブラリです。TBBの並列コンテナクラスを利用すると、アプリケーションにスケーラブルな並列処理を安全に追加することができます。

あなたのコンテナはスレッドセーフか？

　多くの開発者は、C++の標準テンプレートライブラリ（STL）の実装に含まれているコンテナクラスか、自作のコンテナクラスに頼っています。しかし困ったことに、こうしたライブラリはスレッドセーフでないことが多いのです。STLの仕様では、スレッドやマルチスレッドコードで使用する際のコンテナクラスの必須動作やスレッドについて何も言及していません。そのため、これらのSTLコンテナクラスの実装がスレッドセーフでないという事態が一般化しているのです。

　例えば、STLのmap<string, MyClass>を使用する場合を考えてみましょう。

　たとえ異なる2つのキーに関連付けられた異なる2つの値を上記のコードで修正するとしても、大部分のSTL実装は正しい動作を保証しません。これらの操作を同期なしに並列的に実行すると、マップが破損する可能性があります。スレッドセーフに対する要件の指定がないと、異なる2つのマップにアクセスすることでデータ破損を招く可能性さえあります。

　もちろん、上記のコードをスレッドセーフにするようなやり方で、STLテンプレートクラスのmapを実装することは可能です。残念ながら、よく使われるmap操作シーケンスの中には、スレッドセーフなやり方で実装できないものがあります。それぞれの操作を単独で実行すればスレッドセーフになるかもしれませんが、シリアルコードの中でよく使われるシーケンスは予期せぬ結果を引き起こすことがあります。例えば、2つのスレッドが次のコードでマップ内の同じ要素を操作したらどうなるでしょう。

　Thread 0によって実行されるコードは2つの操作を実行します。まずoperator []を呼び出して、"Key1"に関連付けられたオブジェクトへの参照を取得します。このキーがマップ内になければ、operator []はMyClass型のオブジェクトを格納するためのスペースを割り当て、このキーと関連付けます。次にoperator =が呼び出され、取得した参照が指し示すオブジェクトにMyClassの一時的なインスタンスがコピーされます。

　望ましい結果は、"Key1"がマップ内に現れないか、MyClass()のインスタンスと対になることです。しかし、ユーザーによって挿入される同期がないと、たとえ各操作が単独ではスレッドセーフであっても、これ以外の結果も起こり得ます。Thread 1によって呼び出されるメソッドeraseが、Thread 0によるoperator []呼び出しとoperator =呼び出しの間に生じる能性があります。その場合、Thread 0は削除されたオブジェクトに対してoperator =を呼び出そうとし、これは誤った動作になります。このようなマルチスレッド化のバグは「競合状態」として知られています。この動作はどちらのスレッドが先に操作を実行するかにかかっています。

　この例のような競合の難しい点の1つは、動作が予測不能（非決定論的）だということです。このコードをテストで実行しても、Thread 1からのerase呼び出しが、いつもThread 0のフェッチと更新の間に入ることもあります。そのため、このようなバグはテストをすり抜け、検証済みのリリースコードの中に潜伏し、ユーザーのシステムでいつでも表に現れる可能性があります。

　スレッドフレンドリでないコンテナクラスを使用する際に、こうしたバグを避け、正確さを確保するため、開発者たちは各コンテナの使用時に必ずロックをかけ、一度に1つのスレッドしかアクセスできないようにしています。このような粒度の粗い同期アプローチでは、アプリケーションで使用できる並列処理が制限され、またアクセスポイントごとにコードを追加することになるため、複雑さが増します。しかし、こうした既存のライブラリを利用する以上、これは支払わなければならない代価です。

代替策：Intel TBBのコンテナクラス

　スレッドセーフでないコンテナを粒度の粗い同期でラップする以外にも、別の方法があります。Intel TBBライブラリはスレッドを使用するC++用のランタイムベース並列プログラミングモデルです。ライブラリが提供するスケーラブルな並列アルゴリズムに加え、Intel TBBはマップ、キュー、ベクタのコンテナクラスの安全でスケーラブルな実装を提供します。これらのテンプレートは、ユーザースレッド化コードで直接使用することも、ライブラリに含まれている並列アルゴリズムと共に使用することもできます。

　上述したように、一部のコンテナクラスの標準インターフェイスは本質的にスレッドフレンドリではありません。そのため、Intel TBBの並列コンテナは対応するSTLコンテナの単純な代替品にはなりません。そこで、Intel TBBはSTLの精神に従いながらも、スレッドセーフを保証する必要があるところでは修正されたインターフェイスを提供するのです。

　Intel TBBライブラリのすべての並列コンテナは「粒度の細かいロック」を使って実装されています。コンテナに対してメソッドが呼び出されたときは、データ構造の中で、そのメソッドが扱う部分だけがロックされるので、他の部分には複数のスレッドが同時にアクセスできます。

　以下では、concurrent_hash_map、concurrent_queue、concurrent_vectorの各テンプレートクラスについて説明し、安全でないコンテナの使用を検知して置き換える方法を示します。

クラス：concurrent_hash_map< Key, T, HashCompare >

　Intel TBBのテンプレートクラスconcurrent_hash_mapは、対応するSTLのコンテナに似ていますが、要素への同時アクセスを許します。これは、Key型のキーをT型の値にマップするハッシュテーブルです。特性型のHashCompareは、マッピングで使用するハッシュ関数と、2つのキーが等しいかどうか評価する関数を定義します。これらの関数は、等しい2つのキーから同じハッシュコードが生成されることに同意していなければならないので、一緒に使用されます。

　文字列型のキーに対する特性クラスの例を以下に示します。

struct my_hash_compare {
    static size_t hash( const string& x ) {
        size_t h = 0;
        for( const char* s = x.c_str(); *s; ++s )
            h = (h*17)^*s;
        return h;
    }
    //! True if strings are equal
    static bool equal( const string& x, const string& y ) {
        return x==y;

};

　concurrent_hash_mapは、std::pair<const Key,T>型の要素のコンテナとして機能します。一般にコンテナ要素にアクセスするのは、それを更新または読み取ろうとするときです。テンプレートクラスconcurrent_hash_mapは、この2つの目的を、それぞれaccessorクラスとconst_accessorクラスによってサポートします。これらはスマートポインタとして機能し、先ほどのSTLのmapの例にはなかった、要素へのアトミックなアクセスを可能にします。

　accessorは更新（書き込み）アクセスを表します。これが要素を指している限り、テーブル内のそのキーを参照する他のすべての試みはaccessorが済むまでブロックします。const_accessorもよく似ていますが、こちらは「読み取り専用」アクセスを表しています。同時に複数のconst_accessorsが同じ要素を指すことが許されるので、要素が頻繁に読み取られ、あまり頻繁に更新されない状況では同時並列性が大幅に向上します。

　concurrent_hash_map要素へのアクセスは、主としてinsert、find、eraseの各メソッドによって行います。findメソッドとinsertメソッドは、引数としてaccessorまたはconst_accessorを取ります。選択した引数によって、concurrent_hash_mapは「更新」と「読み取り専用」のどちらのアクセスが要求されているのかを判断します。メソッドが復帰すると、accessorまたはconst_acessorが破棄されるまで、そのアクセスが持続します。removeメソッドは書き込みアクセスを暗黙に要求します。従って、残存している他のアクセスが終了するのを待ち、それからキーを削除します。

　次のコードは、concurrent_hash_mapに新しい要素を挿入する例です。

concurrent_hash_map<string, MyClass, my_hash_compare> string_table;

void insert_into_string_table ( string &key, MyClass &m ) {

    // create an accessor that will act as a smart pointer
    // for write access
    concurrent_hash_map<string, MyClass, my_hash_compare>::accessor a;

    // call insert to create a new element, or return an existing
    // element if one exists.
    // accessor a locks this element for exclusive use by this thread
    string_table.insert( a, key );

    // modify the value held by the pair
    a->second = m;

    // the accessor "a" releases the lock on the element when it is
    // destroyed at the end of the scope
}

concurrent_queue< Key, T, HashCompare >

　Intel TBBのテンプレートクラスconcurrent_queue<T>は、T型の値を持つ並列キューを実装しています。このキューに対しては、同時に複数のスレッドがプッシュ/ポップを行えます。このキューはデフォルトでは無制限ですが、最大容量を設定することで制限することができます。

　一般にキューは先入れ先出し方式のデータ構造であり、シングルスレッドプログラムでは、この厳格な順序付けをサポートするようにキューを設計できます。しかし、同時に複数のスレッドがプッシュ/ポップを行う場合、何をもって「先」とするかはあまり明確でなくなります。Intel TBBのテンプレートクラスconcurrent_queueは、あるスレッドが2つの値をプッシュし、別のスレッドがこの2つの値をポップする場合に、これらの値がプッシュされたときと同じ順序でポップされることを保証します。異なるスレッドによってプッシュされる値のインタリーブは制限されません。

　並列キューはプロデューサ-コンシューマアプリケーションでよく使われます。このようなアプリケーションでは、あるスレッドで生成されるデータが別のスレッドで消費されます。この種のアプリケーションを柔軟にサポートするため、Intel TBBはポップのブロッキングバージョンと非ブロッキングバージョンを提供しています。pop_if_presentメソッドは非ブロッキングです。このメソッドは値のポップを試みますが、キューが空なら、すぐに復帰します。それに対し、popメソッドはアイテムが利用可能になるまでブロックし、利用可能になったら、アイテムをキューからポップします。

　次の例では、concurrent_queue<int>による2つのスレッド間の通信にブロッキングメソッドのpopを使用しています。Thread 1は、Thread 0によってキューにプッシュされた各値を、それらが利用可能になったときに出力します。

　STLの大部分のコンテナと違って、concurrent_queue::size_typeは「符号付き」整数型であり、符号なしではありません。その理由は、concurrent_queue::size()が、開始されたプッシュ操作の数から開始されたポップ操作の数を差し引いたものとして定義されているからです。ポップの数がプッシュの数を上回ると、size()が負になります。例えば、concurrent_queueが空で、未決着のポップ操作がn個あるとすれば、size()は~nを返します。この方法によって、プロデューサはキューで待機しているコンシューマの数を簡単に知ることができます。

concurrent_vector< Key, T, HashCompare >

　TBBで定義されている最後の並列コンテナはテンプレートクラスconcurrent_vectorです。concurrent_vectorは動的に拡大できる配列であり、拡大しながら同時にベクタ内の要素へのアクセスを可能にします。

　concurrent_vectorクラスは安全な同時拡大を可能にするために、grow_byとgrow_to_at_leastという2つのメソッドを定義しています。grow_by(n)では、ベクタに連続するn個の要素を安全に追加することができ、追加した最初の要素のインデックスが返されます。各要素はT()で初期化されます。grow_to_at_least(n)は、ベクタの現在のサイズ（要素数）がn未満であれば、ベクタのサイズをnに拡大します。

　次のルーチンは、共有ベクタにC文字列を安全に追加します。

void Append ( concurrent_vector<char>& vector, const char* string) {
    size_t n = strlen(string) + 1;
    memcpy( &vector[vector_grow_by(n)], string, n+1 );
}

　size()メソッドはベクタ内の要素数を返します。これにはgrow_byメソッドとgrow_to_at_leastメソッドによって同時並列的に構築中の要素も含まれている可能性があります。従って、反復子がend()の現在値を超えない限り、concurrent_vectorの拡大中に反復子を使用しても大丈夫です。しかし、反復子が同時並列的に構築中の要素を参照している可能性があります。Intel TBBのconcurrent_vectorを使用するときに構築とアクセスを同期化するのは開発者の責任です。

安全でないコンテナの使用を検知して置き換える方法

　既に述べたように、TBBのコンテナを使用すると、STLなどの他のライブラリを使ったのでは安全に実現できない（または粒度の粗い同期でラップしなければ安全が保証されない）ような並列処理が可能になります。とはいえ、これらの並列コンテナには、スレッドセーフでないコンテナよりもオーバーヘッドが大きいという難点があります。そのため、これらのスレッドセーフなコンテナを使用するのは、スレッドセーフを保証する必要がある場合か、同時並列性を高める必要がある場合に限るべきです。

　従って、「スレッドセーフでないか同時並列性の低いコンテナクラスの使用をどうやって検知するか」というのが最初に検討すべき課題となります。以降では、文字列カウントの例を使って、Intel TBBと共にIntel Threading Tools、Intel Thread Checker、Intel Thread Profiler（http://www.intel.com/software/products/threadingを参照）をどのように使用すればこうした厄介なコンテナクラスを検知して置き換えることができるかを示します。

文字列カウントの例

　次のコードでは、STLのmap<string, int> tableを使って、Data配列内で異なる文字列の出現回数を数えています。このコードは、Win32スレッド（NThread）を作成し、各スレッドにData配列のチャンクを処理させることで並列化されています。

　CountOccurrencesメソッドはスレッドを作成し、各スレッドに処理すべき範囲の配列要素を渡します。tallyメソッドは実際の計算を行います。ここで各スレッドがDataの担当部分を反復処理し、STLのmapテーブル内の対応する要素をインクリメントします。この例の30行目で、出現回数を数えるのに要した合計時間ならびに調べた文字列の合計数（これは常にデータセットサイズNに等しくならなければならない）を出力します。リスト1に、この例のすべてのバージョンの完全なコードを示します。

00: const size_t N = 1000000;
01: static string Data[N];
02: typedef map<string,int> StringTable;
03: StringTable table;

04: DWORD WINAPI tally ( LPVOID arg ) {
05:    pair<int, int> *range =
06:      reinterpret_cast< pair<int,int> * >(arg);
07:    for( int i=range->first; i < range->second; ++i )
08:        table[Data[i]] += 1;
09:    return 0;
10: }

11: static void CountOccurrences() {
12:  HANDLE worker[8];
13:  pair<int,int> range[8];

14:  int g = static_cast<int>(ceil(static_cast<double>(N) / NThread));

15:  tick_count t0 = tick_count::now();
16:  for (int i = 0; i < NThread; i++) {
17:    int start = i * g;
18:    range[i].first = start;
29:    range[i].second = (start + g) < N ? start + g : N;
20:    worker[i] =
21:      CreateThread( NULL, 0, tally, &range[i], 0, NULL );
22:  }
23:  WaitForMultipleObjects ( NThread, worker, TRUE, INFINITE );
24:  tick_count t1 = tick_count::now();

25:  int n = 0;
26:  for( StringTable::iterator i=table.begin();
27:          i!=table.end(); ++i )
28:    n+=i->second;
39:
30:  printf("total=%d time = %g
",n,(t1-t0).seconds());
31: }

/*================================================
Copyright 2005-2006 Intel Corporation.  All Rights Reserved.

The source code contained or described herein and all documents related
to the source code ("Material") are owned by Intel Corporation or its
suppliers or licensors.  Title to the Material remains with Intel
Corporation or its suppliers and licensors.  The Material is protected
by worldwide copyright laws and treaty provisions.  No part of the
Material may be used, copied, reproduced, modified, published, uploaded,
posted, transmitted, distributed, or disclosed in any way without
Intel’s prior express written permission.

No license under any patent, copyright, trade secret or other
intellectual property right is granted to or conferred upon you by
disclosure or delivery of the Materials, either expressly, by
implication, inducement, estoppel or otherwise.  Any license under such
intellectual property rights must be express and approved by Intel in
writing.
=================================================*/

#include "tbb/concurrent_hash_map.h"
#include "tbb/scalable_allocator.h"
#include "tbb/tick_count.h"

#include <windows.h>
#include <cmath.h>

#include <string>
#include <utility>
#include <cstdio>

#if !defined(USE_TBB)
#include <map>
#endif

using namespace tbb;
using namespace std;

//! Set to true to counts.
static bool Verbose = false;
static int NThread = 1;

#if defined(USE_FAST_STRINGS)
typedef basic_string<char, char_traits<char>,
    scalable_allocator<char> > FastString;
#else
typedef string FastString;
#endif

const size_t N = 1000000;
static FastString Data[N];

#if defined(USE_TBB)

//! Structure that defines hashing and
//  comparison operations for user’s type.
struct MyHashCompare {
    static size_t hash( const FastString& x ) {
        size_t h = 0;
        for( const char* s = x.c_str(); *s; s++ )
            h = (h*17)^*s;
        return h;
    }
    //! True if strings are equal
    static bool equal( const FastString& x, const FastString& y ) {
        return x==y;
    }
};

typedef concurrent_hash_map<FastString,int,MyHashCompare> StringTable;
#else
typedef map<FastString,int> StringTable;
#endif

StringTable table;

#if defined(USE_MUTEX)
HANDLE table_mutex;
#endif

#if defined(USE_CS)
CRITICAL_SECTION table_cs;
#endif

DWORD WINAPI tally ( LPVOID arg ) {
    pair<int, int> *range = reinterpret_cast< pair<int,int> * >(arg);

    for( int i=range->first; i < range->second; ++i ) {

#if defined(USE_TBB)
        StringTable::accessor a;
        table.insert( a, Data[i] );
        a->second += 1;
#else

#if defined(USE_MUTEX)
        WaitForSingleObject ( table_mutex, INFINITE );
#endif

#if defined(USE_CS)
        EnterCriticalSection(&table_cs);
#endif

        table[Data[i]] += 1;

#if defined(USE_MUTEX)
        ReleaseMutex ( table_mutex );
#endif

#if defined(USE_CS)
        LeaveCriticalSection(&table_cs);
#endif

#endif
    }

    return 0;
}

static void CountOccurrences() {
    HANDLE worker[8];
    pair<int,int> range[8];

    int g = static_cast<int>(ceil(static_cast<double>(N) / NThread));
    int r = N%NThread;

#if defined(USE_MUTEX)
    table_mutex = CreateMutex( NULL, false, NULL );
#endif

#if defined(USE_CS)
    InitializeCriticalSection(&table_cs);
#endif

    tick_count t0 = tick_count::now();

    for (int i = 0; i < NThread; i++) {
      int start = i * g;
      range[i].first = start;
      range[i].second = (start + g) < N ? start + g : N;
      worker[i] = CreateThread( NULL, 0, tally, &range[i], 0, NULL );
    }
    WaitForMultipleObjects ( NThread, worker, TRUE, INFINITE );

    tick_count t1 = tick_count::now();

    int n = 0;
    for( StringTable::iterator i=table.begin(); i!=table.end(); ++i ) {
        if( Verbose )
            printf("%s %d
",i->first.c_str(),i->second);
        n+=i->second;
    }
    printf("total=%d time = %g
",n,(t1-t0).seconds());

#if defined(USE_CS)
    DeleteCriticalSection(&table_cs);
#endif
}

static const FastString Adjective[] = {
    "sour",
    "sweet",
    "bitter",
    "salty",
    "big",
    "small"
};

static const FastString Noun[] = {
    "apple",
    "banana",
    "cherry",
    "date",
    "eggplant",
    "fig",
    "grape",
    "honeydew",
    "icao",
    "jujube"
};

static void CreateData() {
    size_t n_adjective = sizeof(Adjective)/sizeof(Adjective[0]);
    size_t n_noun = sizeof(Noun)/sizeof(Noun[0]);
    for( int i=0; i<N; ++i ) {
        Data[i] = Adjective[rand()%n_adjective];
        Data[i] += " ";
        Data[i] += Noun[rand()%n_noun];
    }
}

static void ParseCommandLine( int argc, char* argv[] ) {
    int i = 1;
    if( i<argc && strcmp( argv[i], "verbose" )==0 ) {
        Verbose = true;
        ++i;
    }
    if( i<argc && !isdigit(argv[i][0]) ) {
        fprintf(stderr,"Usage: %s [verbose] number-of-threads
",
                argv[0]);
        exit(1);
    }
    if( i<argc ) NThread = strtol(argv[i++],0,0);
}

int main( int argc, char* argv[] ) {
    srand(2);
    ParseCommandLine( argc, argv );
    CreateData();
    CountOccurrences();
}

安全でないコンテナの使用を検知する

　この例をMicrosoft Visual Studio .NET 2005でコンパイルし、4プロセッサのIntel Xeonサーバー上で4つのスレッドによって実行してください。そうすれば、何か問題があることがすぐに分かるでしょう。31行目によって報告される合計値が実行のたびに変わり、決してN（1000000）と等しくなりません。競合状態やその他のスレッド化エラーがあるのかもしれません。

　この点を確認するため、潜在的なデータ競合やデッドロックやその他のスレッド化エラーを検出するツール（Intel Thread Checker）でこの例を実行してみました。Intel Thread Checkerにより、スレッド間の潜在的な依存関係を検知し、その依存関係をエラーの元になっているソース行にマップすることができました。これらの潜在的な依存関係は、すべて同じ箇所（8行目）を指していました。これは、スレッドがSTLのmapにアクセスするところです（図1を参照）。

図1　Intel Thread Checkerツールは依存関係を検知して、潜在的な競合を容易に取り除けるようにする

スケーラブルなソリューションを見つける

　図1を見ると、mapへのアクセスを同期化する必要があることが明らかになります。まず、単一のWin32 MutexまたはCRITICAL_SECTIONオブジェクトを使ってアクセスを監視する処理のパフォーマンスを調べます。既に述べたように、粒度の粗い同期は、スレッドセーフでないライブラリを使ってスレッドセーフを保証する唯一の方法です。

　Win32 Mutexは全プロセスから見えるカーネルオブジェクトです。単一のMutexオブジェクトでSTLのmapへの各アクセスを監視すれば、スレッドセーフが保証されますが、そうするとオーバーヘッドも非常に大きくなります。それに対し、Win32 CRITICAL_SECTIONは軽量で、ユーザー空間のプロセス内ミューテックスオブジェクトなので、こちらの方が好ましい選択肢です。

　図2に、STLのmapを使って同期化を行うバージョンのパフォーマンスを、同じコードをシーケンシャルに実装した場合を1として示します。

図2　3つのオプションのいずれも許容できるパフォーマンスをもたらさない

　同期がなければ、STLのmapは良好なパフォーマンスをもたらしますが、同時アクセスが行われると問題が起き、誤った結果を引き起こします。Win32 Mutexはコストが高いため、予想どおり劣悪なパフォーマンスをもたらします。単一のCRITICAL_SECTIONオブジェクトと粒度の粗い同期を用いた場合も、パフォーマンスは低下します。スレッドセーフなSTL mapベースの実装は、どちらも元のシーケンシャルな実装よりも実行時間が長くなります。粒度の粗い同期は正確さを復活させますが、同時並行性がないためスケーラビリティが低下します。

　図3は、STLのmapをTBBのconcurrent_hash_mapに置き換えたときのパフォーマンスを示しています。

図3　Intel TBBのconcurrent_hash_mapはSTLのmapのパフォーマンスを向上させる

　concurrent_hash_mapを使用する場合のパフォーマンスと、STLのmapで同期を行わない場合のパフォーマンスとを比較すると、Intel TBBによって実現されるスレッドセーフにはそれなりに代償が伴うことが明らかになります。しかしこの実装は、他のスレッドセーフな実装とは異なり、1よりも大きなスピードアップをもたらすだけの同時並行性を実現します。

　つまり、TBBの並列コンテナの同時並行性を利用すると、安全でないコンテナに粒度の粗いロックを適用する方法ではとても実現できない高いパフォーマンスが可能になります。

　これからはますます多くの開発者が、マルチコアシステムへの移行という避けがたい事態に直面することになるでしょう。こうした開発者にとっては、マルチスレッドアプリケーション開発でエラーを起きにくくするためのツールを探求することが絶対必要になります。TBBに含まれている並列コンテナクラスもそうしたものの1つであり、マルチコアへの移行を容易にしてくれます。これらの並列コンテナクラスが提供する安全でスケーラブルな実装のおかげで、C++ STLで定義されているようなスレッドフレンドリでないコンテナを使わずに済み、このようなコンテナに伴う諸々の問題を回避できるからです。