組み立てラインでの作業: パイプライン

パイプライン は、従来の生産組み立てラインを模倣した一般的な並列パターンです。データは一連のパイプライン・フィルターを流れ、各フィルターではいくつかの方法でデータが処理されます。データの入力ストリームが与えられると、これらのフィルターのいくつかは並列に操作できるものがあります。例えば、ビデオ処理では、フレーム上のいくつかの操作はほかのフレームに依存しないため、複数のフレーム上で同時に操作できます。一方、前のフレームを最初に処理しなければならない操作もあります。

インテル® スレッディング・ビルディング・ブロック (インテル® TBB) の pipeline クラスと filter クラスは、パイプライン・パターンを実装します。単純なテキスト処理サンプルで、pipeline と filter を使用して並列処理を行う方法を説明します。サンプルは、テキストファイルを読み取り、テキストの 10 進数を 2 乗して、修正したテキストを新しいファイルに書き込みます。次にパイプラインの図を示します。

ファイル I/O はシーケンシャルであると仮定します。2 乗フィルターは並列に実行できます。つまり、n チャンクを非常に速くシリアルに読み取ることができれば、それらが適切な順で出力ファイルに書き込まれている限り、n チャンクをそれぞれ並列に処理できます。I/O はシーケンシャルであるため、入出力のフォーマットは中間のフィルターに移動して、並列にできます。

並列スケジュールのオーバーヘッドの影響を考慮して、フィルターは複数のテキストを処理します。各入力チャンクは約 4,000 文字です。各チャンクは、TextSlice クラスのインスタンスで表されています。

// テキストのスライスを保持
/** C++ 宣言はメモリーのはるかに大きなオブジェクトのヘッダーのみを表すため、
   複数のメソッドをここで使用してインスタンスを割り当て/解放する必要がある */
class TextSlice {
    // シーケンスの最後の 1 つ前の文字へのポインター
    char* logical_end;
    // シーケンスの最後の 1 つ前の利用可能なバイトへのポインター
    char* physical_end;
public:
    // max_size までの文字を保持できる TextSlice オブジェクトを割り当てる
    static TextSlice* allocate( size_t max_size ) {
        // null ターミネーター文字のためにスペースを +1 残す
        TextSlice* t = (TextSlice*)tbb::tbb_allocator<char>().allocate( sizeof(TextSlice)+max_size+1 );
        t->logical_end = t->begin();
        t->physical_end = t->begin()+max_size;
        return t;
    }
    // この TextSlice オブジェクトを解放する
    void free() {
        tbb::tbb_allocator<char>().deallocate((char*)this, sizeof(TextSlice)+(physical_end-begin())+1);
    }
    // シーケンスの最初へのポインター
    char* begin() {return (char*)(this+1);}
    // シーケンスの最後の 1 つ前の文字へのポインター
    char* end() {return logical_end;}
    // シーケンスの長さ
    size_t size() const {return logical_end-(char*)(this+1);}
    // シーケンスに追加できる最大文字数
    size_t avail() const {return physical_end-logical_end;}
    // このシーケンスにシーケンス [first,last) を追加
    void append( char* first, char* last ) {
        memcpy( logical_end, first, last-first );
        logical_end += last-first;
    }
    // end() を指定された値に設定
    void set_end( char* p ) {logical_end=p;}
};

次のコードは、パイプラインを構築して実行するトップレベルのコードです。TextSlice をコピーするオーバーヘッドを回避するため、TextSlice オブジェクトはポインターを使用してフィルター間で渡されます。

void RunPipeline( int ntoken, FILE* input_file, FILE* output_file ) {
    tbb::parallel_pipeline(
        ntoken,
        tbb::make_filter<void,TextSlice*>(
            tbb::filter::serial_in_order, MyInputFunc(input_file) )
    &
        tbb::make_filter<TextSlice*,TextSlice*>(
            tbb::filter::parallel, MyTransformFunc() )
    &
        tbb::make_filter<TextSlice*,void>(
            tbb::filter::serial_in_order, MyOutputFunc(output_file) ) );
} 

parallel_pipeline メソッドの ntoken 引数は、並列処理のレベルを制御します。概念的には、トークンはパイプラインを流れます。シリアル・インオーダー・フィルターでは、各トークンはシリアルで順番に処理されます。並列フィルターでは、複数のトークンを並列に処理することができます。トークンの数に制限がない場合、出力フィルターの処理が追いつかないため、トークンを取得する中間フィルターで問題が発生することがあります。通常、この状況では、中間フィルターで不適切にリソースが消費されます。parallel_pipeline メソッドの引数は、処理できるトークンの最大数を指定します。いったんこの制限に達すると、別のトークンが出力フィルターで破棄されるまで、パイプラインは入力フィルターで新しいトークンを作成しません。

第 2 引数は、フィルターのシーケンスを指定します。各フィルターは、make_filter<inputType,outputType>(mode,functor) 関数で構築されます。

• inputType は、フィルターの入力値の型を指定します。入力フィルターの型は void です。

• outputType は、フィルターの出力値の型を指定します。出力フィルターの型は void です。

• mode は、フィルターがアイテムを処理する方法 (並列、シリアル・インオーダー、アウトオブオーダー) を指定します。

• functor は、入力値から出力値を生成する方法を指定します。

フィルターは operator& で連結されます。2 つのフィルターを連結する場合、最初のフィルターの outputType と 2 番目のフィルターの inputType が一致していなければなりません。

フィルターは事前に構築して連結できます。前のサンプルの等価バージョンを次に示します。

void RunPipeline( int ntoken, FILE* input_file, FILE* output_file ) {
    tbb::filter_t<void,TextSlice*> f1( tbb::filter::serial_in_order, 
                                       MyInputFunc(input_file) );
    tbb::filter_t<TextSlice*,TextSlice*> f2(tbb::filter::parallel, 
                                            MyTransformFunc() );
    tbb::filter_t<TextSlice*,void> f3(tbb::filter::serial_in_order, 
                                      MyOutputFunc(output_file) );
    tbb::filter_t<void,void> f = f1 & f2 & f3;
    tbb::parallel_pipeline(ntoken,f);
}

フィルターはシーケンシャル・ファイルからチャンクを読み、出力フィルターはチャンクを同じ順序で書かなければならないため、このサンプルの入力フィルターは serial_in_order でなければなりません。serial_in_order フィルターはすべて、アイテムを同じ順序で処理します。そのため、アイテムが MyOutputFunc に MyInputFunc と異なる順序で到達した場合、パイプラインはその前のアイテムが処理されるまで、MyOutputFunc::operator() の呼び出しを自動的に遅延します。順序を保存しない、別の種類のシリアルフィルター serial_out_of_order もあります。

中間フィルターは純粋にローカルデータを処理するため、この構文の任意の数の呼び出しは同時に実行できます。したがって、このフィルターは並列フィルターとして指定されます。

では、各フィルターの構文について詳細に説明します。出力構文は最も単純です。この構文で行う処理は、TextSlice をファイルに書くことと TextSlice を解放することだけです。

// TextSlice をファイルに書き込む構文
class MyOutputFunc {
    FILE* my_output_file;
public:
    MyOutputFunc( FILE* output_file );
    void operator()( TextSlice* item ) const;
};
 
MyOutputFunc::MyOutputFunc( FILE* output_file ) :
    my_output_file(output_file)
{
}
 
void MyOutputFunc::operator()( TextSlice* out ) const {
    size_t n = fwrite( out->begin(), 1, out->size(), my_output_file );
    if( n!=out->size() ) {
        fprintf(stderr,"Can't write into file '%s'\n", OutputFileName);
        exit(1);
    }
    out->free();
} 

operator() メソッドは TextSlice を処理します。out 引数は、処理される TextSlice を指します。この構文はパイプラインの最後のフィルターに使用されるため、void を返します。

中間フィルターの構文は似ていますが、少し複雑です。この構文は、生成する TextSlice のポインターを返します。

// 10 進数を 2 乗に変更する構文
class MyTransformFunc {
public:
    TextSlice* operator()( TextSlice* input ) const;
};
 
TextSlice* MyTransformFunc::operator()( TextSlice* input ) const {
    // 数が入力の最後の場合でも strtol が正しく動作するように null ターミネーターを追加
    *input->end() = '\0';
    char* p = input->begin();
    TextSlice* out = TextSlice::allocate( 2*MAX_CHAR_PER_INPUT_SLICE );
    char* q = out->begin();
    for(;;) {
        while( p<input->end() && !isdigit(*p) )
            *q++ = *p++;
        if( p==input->end() )
            break;
        long x = strtol( p, &p, 10 );
        // 注: ここでは入力文字列長の 2 倍の長さが用意されていて、
        // 負でない整数 n の 2 乗の桁は n の 2 倍以上にならないため、
        // オーバーフローをチェックする必要はない
        long y = x*x;
        sprintf(q,"%ld",y);
        q = strchr(q,0);
    }
    out->set_end(q);
    input->free();
    return out;
} 

数が境界を超えていないことを保証しなければならないため、入力構文は最も複雑です。数が境界を越えて次のスライスにも含まれる場合、不完全な数を次のスライスにコピーします。また、入力の最後にいつ到達するかを示す必要もあります。このために、stop() メソッドを型 flow_control の特別な引数で呼び出します。この慣用句は、パイプラインの最初のフィルターに使用する構文に必要です。次のコードで太字で示されている部分です。

TextSlice* next_slice = NULL;

class MyInputFunc {
public:
    MyInputFunc( FILE* input_file_ );
    MyInputFunc( const MyInputFunc& f ) : input_file(f.input_file) { }
    ~MyInputFunc();
    TextSlice* operator()( tbb::flow_control& fc ) const;
private:
    FILE* input_file;
};
 
MyInputFunc::MyInputFunc( FILE* input_file_ ) :
    input_file(input_file_) { }
 
MyInputFunc::~MyInputFunc() {
}
 
TextSlice* MyInputFunc::operator()( tbb::flow_control& fc ) const {
    // 次のスライスで利用可能なスペースに文字を読む
    if( !next_slice )
        next_slice = TextSlice::allocate( MAX_CHAR_PER_INPUT_SLICE );
    size_t m = next_slice->avail();
    size_t n = fread( next_slice->end(), 1, m, input_file );
    if( !n && next_slice->size()==0 ) {
        // これ以上処理する文字はない
        fc.stop();
        return NULL;
    } else {
        // まだ処理する文字がある
        TextSlice* t = next_slice;
        next_slice = TextSlice::allocate( MAX_CHAR_PER_INPUT_SLICE );
        char* p = t->end()+n;
        if( n==m ) {
            // 不完全な数を読む可能性がある    
            // その場合は不完全な数の文字数を次のスライスに転送
            while( p>t->begin() && isdigit(p[-1]) )
                --p;
            assert(p>t->begin(),"Number too large to fit in buffer.\n");
            next_slice->append( p, t->end()+n );
        }
        t->set_end(p);
        return t;
    }
}

構文は filter_t が構築されたときと、パイプラインを実行したときにコピーされるため、コピー・コンストラクターは必ず定義する必要があります。

parallel_pipeline は、インテル® TBB 3.0 で追加された新しい構文です。examples/pipeline/square ディレクトリーには、フィルターが継承で定義される古い低レベルの構文を使用した 2 乗サンプルの完全なコードが含まれています。両方の構文の説明は、『デベロッパー・リファレンス』を参照してください。

• 循環バッファーの使用
  
• パイプラインの処理能力
  
• 非線形パイプライン
  
• ループとパイプラインのまとめ