ActionBlock：または別のデッドロックに関する短編小説を勉強します

ほぼすべての実際のプロジェクトは、 生産者と消費者のキューの実装の何らかの形を使用していると思います 。 問題の背後にある考え方は非常に簡単です。 アプリケーションは、処理からデータの生成を切り離す必要があります。 たとえば、CLRのスレッドプールを考えてみましょう。ThreadPool.QueueUserWorkItemを呼び出すことで処理する要素を追加します。スレッドプール自体は、最適なワークフローの数を理解し、必要な並列度で要素を処理するメソッドを呼び出します。

ただし、標準のスレッドプールを使用することは、常に可能または合理的ではありません。 スレッドの最小数と最大数を指定する可能性にもかかわらず、この構成はグローバルであり、必要な部分ではなくアプリケーション全体に影響します。 消費者サプライヤーの問題を解決する方法は他にもたくさんあります。 アプリケーションロジックがマルチスレッド、キュー、および同期の側面と混在している場合、「正面」のソリューションになる可能性があります。 これは、ワークフローまたはタスクの数を手動で制御するBlockingCollectionのラッパーにすることができます。 または、TPL DataFlowのActionBlock <T>などの完全なソリューションに基づいたソリューションにすることもできます。

今日は、 ActionBlockクラスの内部構造を調べ 、その作成者が行った設計上の決定について議論し、使用時に問題を回避するためにこれらすべてを知る必要がある理由を見つけます。 準備はいい？ それでは行こう！

私の現在のプロジェクトでは、サプライヤーと消費者の問題を解決する必要があるいくつかのケースがあります。 その1つは次のようになります。TypeScriptに非常によく似た言語用のカスタムパーサーとインタープリターがあります。 詳細に深く入り込むことなく、一連のファイルを解析し、すべての依存関係のいわゆる「推移的クロージャー」を取得する必要があると言えます。 次に、実行に適したプレゼンテーションに変換して実行する必要があります。

解析ロジックは次のようになります。

1. Parsimファイル。
2. その内容を分析し、その依存関係を探します（すべての「import * from」、「require」、および同様の構成を分析します）。
3. 依存関係を計算します（つまり、現在のファイルが通常の操作に必要なファイルのセットを見つけます）。
4. 受信した依存関係ファイルを解析用のリストに追加します。

かなり簡単ですね。 そうです。 TPL DataflowおよびActionBlock <T>クラスに基づいたわずかに単純化された実装は次のようになります。

private static Task<ParsedFile> ParseFileAsync(string path) { Console.WriteLine($"Parsing '{path}'. {{0}}", $"Thread Id - {Thread.CurrentThread.ManagedThreadId}"); Thread.Sleep(10); return Task.FromResult( new ParsedFile() { FileName = path, Dependencies = GetFileDependencies(path), }); } static void Main(string[] args) { long numberOfProcessedFiles = 0; ActionBlock<string> actionBlock = null; Func<string, Task> processFile = async path => { Interlocked.Increment(ref numberOfProcessedFiles); ParsedFile parsedFile = await ParseFileAsync(path); foreach (var dependency in parsedFile.Dependencies) { Console.WriteLine($"Sending '{dependency}' to the queue... {{0}}", $"Thread Id - {Thread.CurrentThread.ManagedThreadId}"); await actionBlock.SendAsync(dependency); } if (actionBlock.InputCount == 0) { // This is a marker that this is a last file and there // is nothing to process actionBlock.Complete(); } }; actionBlock = new ActionBlock<string>(processFile); actionBlock.SendAsync("FooBar.ts").GetAwaiter().GetResult(); Console.WriteLine("Waiting for an action block to finish..."); actionBlock.Completion.GetAwaiter().GetResult(); Console.WriteLine($"Done. Processed {numberOfProcessedFiles}"); Console.ReadLine(); } 

ここで何が起こるか見てみましょう。 簡単にするために、すべてのコアロジックはMainメソッドにあります。 変数numberOfProcessedFilesは 、ロジックが正しいことを確認するために使用され、処理されたファイルの総数が含まれます。 主な作業はprocessFileデリゲートで行われ、デリゲートはActionBlockコンストラクターに渡されます。 このデリゲートは、「コンシューマ」と「プロバイダー」の両方の役割を果たします。path引数を介してファイルパスを取得し、ファイルを解析して、依存関係を見つけ、 actionBlock.SendAsyncメソッドを呼び出してキューに新しいファイルを送信します。 次に、処理キュー内の要素数のチェックがあり、新しい要素がない場合は、 actionBlock.Complete（） （*）を呼び出すことで操作全体が完了します。 次に、 MainメソッドはActionBlockインスタンスを作成し、最初のファイルの処理を開始して、プロセス全体の終了を待ちます。

ParseFileAsyncメソッドは、ファイル解析プロセスをエミュレートし、次のプリミティブロジックを使用して依存関係を計算します。ファイル「foo.ts」は「fo.ts」に依存し、「fo.ts」は「f.ts」に依存します。 つまり 各ファイルは、短い名前のファイルに依存しています。 これは非現実的なロジックですが、ファイルの推移的な閉包を計算する基本的な考え方を示すことができます。

ActionBlockクラスは、並行性を管理します。 確かに 、デフォルトの並列度は1であることを考慮する必要があり、これを変更するには、 ActionBlockコンストラクターでExecutionDataflowBlockOptionsクラスのインスタンスを渡す必要があります。 MaxDegreeOfParallelismプロパティが1より大きい場合、 ActionBlockは異なるスレッド（実際には異なるタスク）からコールバックデリゲートを呼び出して、キュー要素を並列処理します。

投稿と SendAsync：何をいつ使用するか

供給者と消費者の問題を少なくとも一度独立して解決しようとした人は誰でも問題に直面しました。入力データストリームが処理中の消費者の能力を超えた場合はどうすればよいでしょうか 入力ストリームを「絞る」方法は？ すべての入力要素をメモリに保存するだけですか？ 例外を投げますか？ アイテムの追加メソッドでfalseを返しますか？ 循環バッファを使用して古い要素を破棄しますか？ または、キューに場所が表示されるまでこのメソッドの実行をブロックしますか？

この問題を解決するために、 ActionBlockの作成者は、次の一般的に受け入れられているアプローチを使用することにしました。

1. クライアントは、 ActionBlockオブジェクトを作成するときにキューのサイズを指定できます。
2. キューがいっぱいの場合、 Postメソッドはfalseを返し、 SendAsync拡張メソッドは、キューに空き領域があるときに完了するタスクを返します。

前の例では、キューのサイズを設定しませんでした。 また、これは、新しい要素が処理されるよりも速く追加されると、アプリケーションが遅かれ早かれOutOfMemoryExceptionでクラッシュすることを意味します。 しかし、この状況を修正してみましょう。 そして、キューを非常に小さなサイズ、たとえば1要素に設定します。

 actionBlock = new ActionBlock<string>(processFile, new ExecutionDataflowBlockOptions() {BoundedCapacity = 1}); 

さて、このコードを実行すると、...デッドロックが発生します！

デッドロック

デザインの観点から消費者と供給者の問題を考えてみましょう。 要素を処理するためのコールバックメソッドを受け入れる独自のキューを作成します。 要素数の制限をサポートするかどうかを決定する必要があります。 「制限された」キューが必要な場合、おそらくActionBlockクラスの設計に非常によく似た設計になります。キューがいっぱいの場合にfalseを返す要素を追加する同期メソッドと、タスクを返す非同期メソッドを追加します。 キューがいっぱいの場合、クラスのクライアントは何をすべきかを決定する機会があります。要素を追加する同期バージョンを呼び出すことで独自に「オーバーフロー」を処理するか、非同期バージョンを使用してキュー内の空きスペースの出現を「待機」します。

次に、コールバックメソッドをいつ呼び出すかを決定する必要があります。 その結果、次のロジックに到達できます。キューが空でない場合、最初の要素が取得され、コールバックメソッドが呼び出され、処理が完了すると予想され、その後要素がキューから削除されます。 （実際の実装は、あらゆる種類の人種を考慮に入れる必要があるため、見かけよりもはるかに複雑になります）。 キューは、コールバックメソッドを呼び出す前にアイテムを削除することを決定できますが、すぐにわかるように、これはデッドロックを受け取る可能性に影響しません。

シンプルでエレガントなデザインを思いつきましたが、簡単に問題につながる可能性があります。 キューがいっぱいで、要素の1つを処理するためにコールバックが呼び出されているとします。 しかし、管理キューをすばやく「返す」代わりに、ハンドラーがawait SendAsyncを呼び出して別の要素を追加しようとするとどうなりますか。


コールバックメソッドがまだ完了していないため、キューがいっぱいで新しいアイテムを受け入れることができません。 しかし、このメソッドも、 SendAsyncの待機が完了するのを待って スタックし 、場所がキューイングされるまで移動できませんでした。 古典的なデッドロック！

コールバックメソッドが*完了した後、 ActionBlockがキューからアイテムを削除するため、デッドロックが発生します。 しかし、別のシナリオを見てみましょう： ActionBlockがコールバックメソッドを呼び出す前に*要素を削除するとどうなりますか？ 実際、何も変わりません。 デッドロックは引き続き可能です。

キューサイズが1で、並列度が2であると想像してください。

• スレッドT1は、アイテムをキューに追加します。 ActionBlockは 、キューから要素を削除し（キュー内の要素の数を0に減らします）、コールバックメソッドを呼び出します。
• スレッドT2は、アイテムをキューに追加します。 ActionBlockは 、キューから要素を削除し（キュー内の要素の数を0に減らします）、コールバックメソッドを呼び出します。
• スレッドT1は、アイテムをキューに追加します。 並列処理のレベルは2であり、すでに2つのハンドラーがあるため、ActionBlockは新しい要素のハンドラーを呼び出すことができません。 キューがいっぱいです。
• 処理中の最初のハンドラーは、キューに新しい要素を追加しようとしますが、キューがいっぱいであるため、「 await SendAsync 」の呼び出しに固執します。
• 処理中の2番目のハンドラーは、新しい要素をキューに追加しようとしますが、キューがいっぱいであるため、 ' await SendAsync 'の呼び出しに固執します。

処理前にキューからアイテムを削除しても効果がないことがわかります。 さらに、デッドロックの確率が大幅に低下するため、これは問題を悪化させるだけです（並列度がNに等しい場合、すべてのNコールバックメソッドが新しい要素を同時にキューに追加しようとする必要があります）。

別の欠点はそれほど明白ではありません。 ActionBlockはまだ汎用ソリューションではありません。 このクラスはITargetSourceインターフェイスを実装し、複雑なデータフロースクリプトの要素を処理するために使用できます。 たとえば、要素の並列処理のためのいくつかの「ターゲット」ブロックを持つBufferBlockがあります。 現在の実装では、バランシングハンドラーは簡単な方法で実装されています。 レシーバー（この場合はActionBlock ）がいっぱいになると、入力用の新しい要素の受け入れを停止します。 これにより、チェーン内の他のブロックが代わりに要素を処理できます。

要素が処理された後にのみ削除された場合、ActionBlockは貪欲になり、現在処理できるよりも多くの要素を受け入れます。 この場合、各ブロックの制限容量は「BoundedCapaciy」+「MaxDegreeOfParallelism」に等しくなります。

デッドロックの問題を解決する方法は？

私は何も恐れていない。 同時にキュー内の要素の数を制限する必要があり、コールバックメソッドが新しい要素を追加できる場合 、 ActionBlockを放棄する必要があります 。 別の方法として、 BlockingCollectionとワークフローの数の「手動」制御に基づくソリューションがあります。たとえば、タスクプールまたはParallel.Invokeを使用します。

並列度

TPLのプリミティブとは異なり、TPL Dataflowのすべてのブロックはデフォルトでシングルスレッドです。 つまり ActionBlock 、 TransformerBlockなどは、コールバックメソッドを1つずつ呼び出します。 TPL Dataflowの作成者は、パフォーマンスの向上よりもシンプルさが重要だと考えました。 データフローグラフについて考えるのは非常に難しく、すべてのブロックでデータを並列処理すると、このプロセスはさらに難しくなります。

並列度を変更するには、ブロックはExecutionDataflowBlockOptionsを渡し、 MaxDegreeOfParallelismプロパティを 1より大きい値に設定する必要があります。ちなみに、このプロパティが-1に設定されている場合、すべての着信要素は新しいタスクによって処理され、並列処理は使用されるタスクスケジューラ（ TaskSchedulerオブジェクト）の機能によってのみ制限されますExecutionDataflowBlockOptionsを介して渡すこともできます。

おわりに

使いやすいコンポーネントの設計は困難です。 並行性の問題を解決する使いやすいコンポーネントの設計は、二重に難しくなります。 これらのコンポーネントを正しく使用するには、それらの実装方法と、開発者が念頭に置いていた制限を知る必要があります。

ActionBlock <T>クラスは素晴らしいもので、サプライヤとコンシューマのパターンの実装を大幅に簡素化します。 ただし、この場合でも、並列度やオーバーフローの場合のブロックの動作など、TPL Dataflowのいくつかの側面について知っておく必要があります。

-（*）この例はスレッドセーフではないため、完全な実装ではactionBlock.InputCountを使用しないでください 。 問題がありますか？

（**） Postメソッドは、2つのケースのいずれかでfalseを返します。キューがいっぱいであるか、すでに完了しています（ Completeメソッドが呼び出されます）。 これらの2つのケースを区別できないため、この側面はこの方法の使用を複雑にする可能性があります。 一方、 SendAsyncメソッドの動作は多少異なります。このメソッドは、キューがいっぱいのときに未完了状態になるTask <bool>オブジェクトを返しますが、キューが既に完了しており、新しい要素を受け入れることができない場合、 task.Resultはfalseになります 。