フリーズしないアプリケーションの作り方：特集：.NET開発者のための非同期入門（3/3 ページ）

» 2011年09月30日 00時00分公開

[岩永信之，著]

■非同期データフロー

　前述のデータ並列に対して、図 10に示すように、タスク並列（task parallelism）という考え方もある。ソフトウェア的なプログラミング・モデルとしては、後述するように、タスク並列は、データフロー（data-flow）という考え方に基づくことになる。

図 10　データ並列とタスク並列

　タスク並列の利点は、2次元的に入り組んだ複雑なデータの受け渡し（＝データフロー）があっても並列化しやすいことである。恣意（しい）的な例ではあるが、図 11に示すような複雑なデータフローを持つ処理を考えてみよう。

図 11　入り組んだデータフローを持つ処理の例

　このデータフローを、並列化を特に考えず書くなら、リスト 8のようになる（2次元的な結線は、文字ベースのプログラミング言語の苦手とするところで、あまり読みやすくはならない）。

public static int Run(int a, int b, int c)
{
int p, q, r, s, t, u, v, x;

処理1(a, out p, out q);
処理4(p, out t, out u);
処理2(b, t, out r);
処理3(c, out s);
処理5(q, r, out v);
処理6(s, u, v, out x);

return x;
}

リスト 8　図 11のデータフローをそのままC#で表現

　この処理1～6をそれぞれ別タスクとして並列実行しようというのが、タスク並列の考え方である。

●タスク並列と非同期データフロー

　タスク並列をイメージしやすいのは、ソフトウェアよりもむしろハードウェアの場合である。ハードウェアでは処理内容を柔軟に切り替えるのが難しいため、図 12に示すように、処理ごとに1つの回路を作り、並列に動作させる。前段や後段の処理回路の状況に応じて回路がスリープすることになる*5。

*5　当然、ハードウェア設計においては、回路ごとの処理時間をどれだけ均等にできるかが重要である。

図 12　ハードウェア的なタスク並列

　一方、ソフトウェア的に考えると、スリープの必要はなくなる。前段の処理が完了したときに初めて後段のタスクを起動すればよく、後段が処理中であっても別途もう1タスク起動するなり、バッファにデータをためておくなりすればよい。要するに、ソフトウェア的には、並列処理というよりむしろ、図 13に示すような、イベント駆動による非同期のデータ受け渡しになる。

図 13　イベント駆動による非同期データ受け渡し

　このような考え方を、「データフローに基づいて非同期処理を開始する」という意味で、先ほどの「非同期制御フロー」に対して、「非同期データフロー」と呼ぶ。

　また、数学的計算モデルの分野では「アクター・モデル（actor model）」とも呼ばれる。図 13のようなバッファを介したデータの受け渡し（＝アクター・モデルでは「メッセージ・パッシング（message passing）」と呼ぶ）以外に接点を持たない処理主体（これを「アクター」と呼ぶ）がそれぞれ独立して並列に動作するため、「アクター・モデル」と呼ばれている。

●Tasks.Dataflow

　非同期データフローの肝となるのは、タスク間をつなぐ非同期バッファの部分である。問題は、この非同期バッファをどう作るかだ。

　幸い、.NET Frameworkでは、次期バージョンで、この非同期バッファに当たるものが標準搭載される。それが、現在、「TPL Dataflow」という名前でプレビュー版が公開されているライブラリである。

　例えば、リスト 8と同じデータフロー結線を、TPL Dataflowライブラリを使って行うと、リスト 9のようになる。

static ISourceBlock<int> Asign(ISourceBlock<int> a, ISourceBlock<int> b, ISourceBlock<int> c)
{
var p = new BufferBlock<int>();
var q = new BufferBlock<int>();
var r = new BufferBlock<int>();
var s = new BufferBlock<int>();
var t = new BufferBlock<int>();
var u = new BufferBlock<int>();
var v = new BufferBlock<int>();
var x = new BufferBlock<int>();

Asign(a, 処理1, p, q);
Asign(b, t, 処理2, r);
Asign(c, 処理3, s);
Asign(p, 処理4, t, u);
Asign(q, r, 処理5, v);
Asign(s, u, v, 処理6, x);

return x;
}

static void Asign<X1, X2, Y1, Y2>(ISourceBlock<X1> x1, ISourceBlock<X2> x2, Func<Tuple<X1, X2>, Tuple<Y1, Y2>> f, ITargetBlock<Y1> y1, ITargetBlock<Y2> y2)
{
var x = Join(x1, x2);
var y = Broadcast<Y1, Y2>(y1, y2);
Asign(x, f, y);
}

……中略…… // 似たようなメソッドがいくつか

static void Asign<X, Y>(ISourceBlock<X> x, Func<X, Y> f, ITargetBlock<Y> y)
{
var t = new TransformBlock<X, Y>(f);
x.LinkTo(t);
t.LinkTo(y);
}

private static ITargetBlock<Tuple<Y1, Y2>> Broadcast<Y1, Y2>(ITargetBlock<Y1> y1, ITargetBlock<Y2> y2)
{
var t1 = new TransformBlock<Tuple<Y1, Y2>, Y1>(_ => _.Item1);
var t2 = new TransformBlock<Tuple<Y1, Y2>, Y2>(_ => _.Item2);
var y = new BroadcastBlock<Tuple<Y1, Y2>>(i => i);
y.LinkTo(t1);
y.LinkTo(t2);
t1.LinkTo(y1);
t2.LinkTo(y2);
return y;
}

private static ISourceBlock<Tuple<X1, X2>> Join<X1, X2>(ISourceBlock<X1> x1, ISourceBlock<X2> x2)
{
var x = new JoinBlock<X1, X2>();
x1.LinkTo(x.Target1);
x2.LinkTo(x.Target2);
return x;
}

リスト 9　TPL Dataflowライブラリを使った非同期データフローの例

　このコード中に出てきたインターフェイスやクラスに関する説明を表 1に示す。TPL Dataflowライブラリでは、データの処理主体に対して「ブロック」という名前を付けている。


インターフェイス	説明
ISourceBlock	データを送ってくる元のブロックを表す
ITargetBlock	データを送る先のブロックを表す
クラス	説明
BufferBlock	データをバッファして素通りさせるだけの処理ブロック
TransformBlock	データに対して何らかの変換処理を通す
BroadcastBlock	複数の処理ブロックに対して同じ値（のコピー）をブロードキャストする
JoinBlock	複数の処理ブロックから来た値を1つにまとめる
表1　TPL Dataflowライブラリ中のクラス（抜粋）