# 機能設計書 22-System.Threading.Tasks.Dataflow

## 概要

本ドキュメントは、.NETランタイムのSystem.Threading.Tasks.Dataflow名前空間が提供するデータフロー機能について、その設計仕様と実装の詳細を記述する。TPL Dataflowは、並列データ処理パイプラインを構築するためのブロックベースのプログラミングモデルを提供する。

### 本機能の処理概要

System.Threading.Tasks.Dataflowは、アクターベースのプログラミングモデルを.NETに提供するライブラリである。データフローブロックを接続してパイプラインを構築し、メッセージ駆動型の並列処理を実現する。

**業務上の目的・背景**：複雑な並列処理やパイプライン処理を手動で実装することは困難であり、デッドロックやレース条件などの並行処理特有の問題が発生しやすい。TPL Dataflowは、ブロック単位で処理を分離し、ブロック間のデータの受け渡しを自動的に管理することで、これらの問題を軽減する。特に、ETLパイプライン、イベント処理、リアルタイムデータ処理などのシナリオで威力を発揮する。

**機能の利用シーン**：
- データ変換パイプラインの構築（ETL処理）
- 画像・動画処理のステージ分割
- ログ集約と分析
- リアルタイムイベントストリーム処理
- ワークフローエンジンの実装
- バッチ処理とストリーム処理の統合

**主要な処理内容**：
1. データフローブロックの生成（BufferBlock、ActionBlock、TransformBlock等）
2. ブロック間のリンク設定（LinkTo）
3. メッセージの送受信（Post、SendAsync、Receive）
4. 並列度・バウンディング・順序保証の制御
5. ブロックの完了と例外処理
6. グループ化処理（BatchBlock、JoinBlock）

**関連システム・外部連携**：System.Threading.Tasks、System.Threading.Channels、IAsyncEnumerableと統合され、他の非同期プログラミングパターンとシームレスに連携する。

**権限による制御**：特別な権限制御は不要。TaskSchedulerによりカスタムスケジューリングが可能。

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| - | - | - | 本機能はUIを持たないライブラリ機能である |

## 機能種別

データ連携 / 並列処理 / パイプライン処理

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| action | Action<T> / Func<T,Task> | Yes（ActionBlockの場合） | 処理デリゲート | nullでないこと |
| transform | Func<TInput,TOutput> / Func<TInput,Task<TOutput>> | Yes（TransformBlockの場合） | 変換デリゲート | nullでないこと |
| dataflowBlockOptions | DataflowBlockOptions | No | ブロックオプション | 各プロパティの範囲チェック |
| linkOptions | DataflowLinkOptions | No | リンクオプション | 整合性チェック |
| batchSize | int | Yes（BatchBlockの場合） | バッチサイズ | 1以上 |

### 入力データソース

- プログラムコードからのデータ投入（Post、SendAsync）
- 上流ブロックからのメッセージ転送
- IAsyncEnumerable<T>からのデータストリーム

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| OutputMessage | TOutput | 処理結果のメッセージ |
| Completion | Task | ブロック完了を表すTask |
| InputCount | int | 入力キューの現在のメッセージ数 |
| OutputCount | int | 出力キューの現在のメッセージ数 |

### 出力先

- 下流のリンク先ブロック
- プログラムコード（Receive、TryReceive）
- IAsyncEnumerable<T>経由での非同期イテレーション

## 処理フロー

### 処理シーケンス

```
1. ブロック作成
   └─ new ActionBlock<T>(action, options) など

2. パイプライン構築
   └─ sourceBlock.LinkTo(targetBlock, linkOptions)
   └─ 複数のブロックをチェーンで接続

3. データ投入
   └─ Post(item): 同期的な投入（即座にtrue/falseを返す）
   └─ SendAsync(item): 非同期投入（バックプレッシャー対応）

4. データ処理
   └─ 各ブロックが内部キューからメッセージを取得
   └─ ユーザー定義のデリゲートを実行
   └─ 結果を下流ブロックに転送

5. 完了処理
   └─ Complete(): 入力の完了を通知
   └─ Completion.Wait(): 処理完了を待機
   └─ 例外はCompletion.Exceptionで取得
```

### フローチャート

```mermaid
flowchart TD
    A[ブロック作成] --> B[LinkTo でパイプライン構築]
    B --> C{データ投入}
    C -->|Post| D[同期投入]
    C -->|SendAsync| E[非同期投入]
    D --> F{InputQueue}
    E --> F
    F --> G[デリゲート実行]
    G --> H{TransformBlock?}
    H -->|Yes| I[OutputQueueへ]
    H -->|No| J[処理完了]
    I --> K{リンク先あり?}
    K -->|Yes| L[下流ブロックへ転送]
    K -->|No| M[OutputQueueで待機]
    L --> F
    J --> N{すべて処理完了?}
    N -->|No| F
    N -->|Yes| O[Complete伝播]
    O --> P[Completion Task完了]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-01 | メッセージ順序保証 | EnsureOrdered=trueの場合、入力順序と出力順序を維持 | デフォルト有効 |
| BR-02 | バックプレッシャー | BoundedCapacity設定時、キュー満杯でPostはfalse、SendAsyncは待機 | BoundedCapacity > 0 |
| BR-03 | 並列度制御 | MaxDegreeOfParallelismで同時実行数を制限 | 常時 |
| BR-04 | 完了伝播 | PropagateCompletion=trueでリンク先に完了を自動伝播 | LinkOptions設定時 |
| BR-05 | グリーディモード | Greedy=trueで利用可能なメッセージを積極的に受け入れ | GroupingBlock |
| BR-06 | 例外処理 | ブロック内例外はCompletionタスクに伝播 | 例外発生時 |

### 計算ロジック

MaxDegreeOfParallelismによる並列度:
- 1（デフォルト）: シーケンシャル実行
- N > 1: 最大N個のタスクが並列実行
- Unbounded (-1): 制限なし

## データベース操作仕様

### 操作別データベース影響一覧

本機能はデータベースを使用しない。インメモリデータ構造のみを使用。

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| InvalidOperationException | 例外 | 完了済みブロックへの操作 | Completionを確認 |
| ArgumentNullException | 例外 | 必須パラメータがnull | 有効な値を指定 |
| ArgumentOutOfRangeException | 例外 | オプション値が範囲外 | 有効な範囲の値を指定 |
| AggregateException | 例外 | デリゲート内で例外発生 | Completion.Exceptionで詳細確認 |
| TaskCanceledException | 例外 | CancellationTokenがキャンセル | キャンセル状態を確認 |

### リトライ仕様

- Post失敗時は呼び出し側でリトライロジックを実装
- SendAsync失敗時は例外として伝播
- デリゲート内例外は協調キャンセル以外ブロックを停止

## トランザクション仕様

本機能はトランザクションを使用しない。各メッセージは独立して処理される。

## パフォーマンス要件

- TargetCore/SourceCoreによる効率的なメッセージバッファリング
- ReorderingBufferによる順序保証（並列実行時）
- SingleProducerConstrainedオプションによる単一プロデューサー最適化
- SpscTargetCoreによるシングルプロデューサー・シングルコンシューマー最適化

## セキュリティ考慮事項

- デリゲート内のコードはユーザー責任
- TaskSchedulerのカスタマイズによるセキュリティ境界の考慮
- メッセージ内の機密データの適切な管理

## 備考

- .NET Framework 4.5以降、.NET Core/.NET 5+で利用可能
- NuGetパッケージ: System.Threading.Tasks.Dataflow
- System.Threading.Channelsはより軽量な代替（単純なプロデューサー・コンシューマー向け）

---

## コードリーディングガイド

本機能を理解するために参照すべきファイルと、推奨する読み解き順序を以下に示す。

### 推奨読解順序

#### Step 1: データ構造を理解する

ブロックオプションとインターフェースを理解することが重要。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | IDataflowBlock.cs | `src/libraries/System.Threading.Tasks.Dataflow/src/Base/IDataflowBlock.cs` | 全ブロック共通インターフェース（Completion、Complete、Fault） |
| 1-2 | ITargetBlock.cs | `src/libraries/System.Threading.Tasks.Dataflow/src/Base/ITargetBlock.cs` | メッセージ受信側インターフェース（OfferMessage） |
| 1-3 | ISourceBlock.cs | `src/libraries/System.Threading.Tasks.Dataflow/src/Base/ISourceBlock.cs` | メッセージ送信側インターフェース（LinkTo、ConsumeMessage） |
| 1-4 | IPropagatorBlock.cs | `src/libraries/System.Threading.Tasks.Dataflow/src/Base/IPropagatorBlock.cs` | 双方向ブロックインターフェース |
| 1-5 | DataflowBlockOptions.cs | `src/libraries/System.Threading.Tasks.Dataflow/src/Base/DataflowBlockOptions.cs` | ブロックオプション設定 |

**読解のコツ**: IDataflowBlockが全ブロックの基盤であり、ITargetBlock<TInput>とISourceBlock<TOutput>の組み合わせでデータフローの方向性が決まる。

#### Step 2: 基本ブロックを理解する

最も基本的なブロックから理解を進める。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | BufferBlock.cs | `src/libraries/System.Threading.Tasks.Dataflow/src/Blocks/BufferBlock.cs` | 最もシンプルなバッファブロック |
| 2-2 | ActionBlock.cs | `src/libraries/System.Threading.Tasks.Dataflow/src/Blocks/ActionBlock.cs` | 処理実行ブロック（ターゲットのみ） |

**主要処理フロー（BufferBlock.cs）**:
- **28-29行目**: SourceCoreとBoundingStateを保持
- **88-141行目**: OfferMessageでメッセージ受け入れ判定
- **183行目**: LinkToでターゲットとの接続
- **192行目**: Countプロパティでバッファ内メッセージ数を取得

**主要処理フロー（ActionBlock.cs）**:
- **28-30行目**: TargetCoreまたはSpscTargetCoreを保持
- **69-123行目**: コンストラクタでデリゲートとオプションに基づき実装選択
- **128-146行目**: ProcessMessageで同期デリゲート実行
- **151-215行目**: ProcessMessageWithTaskで非同期デリゲート実行

#### Step 3: 変換ブロックを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | TransformBlock.cs | `src/libraries/System.Threading.Tasks.Dataflow/src/Blocks/TransformBlock.cs` | 1対1変換ブロック |
| 3-2 | TransformManyBlock.cs | `src/libraries/System.Threading.Tasks.Dataflow/src/Blocks/TransformManyBlock.cs` | 1対多変換ブロック |

**主要処理フロー（TransformBlock.cs）**:
- **28-32行目**: TargetCore、ReorderingBuffer、SourceCoreを保持
- **115-118行目**: 並列実行時の順序保証用ReorderingBuffer初期化
- **171-218行目**: ProcessMessageで同期変換実行
- **223-263行目**: ProcessMessageWithTaskで非同期変換実行
- **140-145行目**: ターゲット完了時にソースへ伝播

#### Step 4: グループ化ブロックを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | BatchBlock.cs | `src/libraries/System.Threading.Tasks.Dataflow/src/Blocks/BatchBlock.cs` | バッチ処理ブロック |
| 4-2 | JoinBlock.cs | `src/libraries/System.Threading.Tasks.Dataflow/src/Blocks/JoinBlock.cs` | 複数入力の結合ブロック |
| 4-3 | BroadcastBlock.cs | `src/libraries/System.Threading.Tasks.Dataflow/src/Blocks/BroadcastBlock.cs` | ブロードキャストブロック |
| 4-4 | WriteOnceBlock.cs | `src/libraries/System.Threading.Tasks.Dataflow/src/Blocks/WriteOnceBlock.cs` | 一回書き込みブロック |

#### Step 5: 内部コンポーネントを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 5-1 | TargetCore.cs | `src/libraries/System.Threading.Tasks.Dataflow/src/Internal/TargetCore.cs` | ターゲット側の共通実装 |
| 5-2 | SourceCore.cs | `src/libraries/System.Threading.Tasks.Dataflow/src/Internal/SourceCore.cs` | ソース側の共通実装 |
| 5-3 | ReorderingBuffer.cs | `src/libraries/System.Threading.Tasks.Dataflow/src/Internal/ReorderingBuffer.cs` | 順序保証バッファ |

### プログラム呼び出し階層図

```
DataflowBlock.Post<T>(target, item) / target.Post(item)
    │
    └─ ITargetBlock<T>.OfferMessage(header, item, source, consumeToAccept)
           │
           ├─ ActionBlock<T>
           │      └─ TargetCore<T>.OfferMessage()
           │             └─ ProcessMessage() / ProcessMessageWithTask()
           │
           ├─ TransformBlock<TInput,TOutput>
           │      └─ TargetCore<TInput>.OfferMessage()
           │             └─ ProcessMessage() / ProcessMessageWithTask()
           │                    └─ transform(input)
           │                    └─ ReorderingBuffer.AddItem() / SourceCore.AddMessage()
           │
           └─ BufferBlock<T>
                  └─ SourceCore<T>.AddMessage()
                         └─ TargetRegistry.PropagateMessage() [LinkTo先に転送]
```

### データフロー図

```
[入力]               [処理ブロック]           [出力]

Producer                                      Consumer
   │                                             │
   ├─ Post() ────▶ BufferBlock<T> ─────────────▶│
   │                    │                        │
   │              LinkTo│                        │
   │                    ▼                        │
   │              TransformBlock                 │
   │              ┌─────┴─────┐                  │
   │              │InputQueue │                  │
   │              │transform()│                  │
   │              │OutputQueue│                  │
   │              └─────┬─────┘                  │
   │              LinkTo│                        │
   │                    ▼                        │
   │              ActionBlock ──────────────────▶│
   │              ┌─────┴─────┐                  │
   │              │InputQueue │                  │
   │              │ action()  │                  │
   │              └───────────┘                  │
   │                                             │
   └─ Complete() ─────────────────────▶ Completion
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| IDataflowBlock.cs | `src/libraries/System.Threading.Tasks.Dataflow/src/Base/IDataflowBlock.cs` | ソース | 基本インターフェース |
| ITargetBlock.cs | `src/libraries/System.Threading.Tasks.Dataflow/src/Base/ITargetBlock.cs` | ソース | ターゲットインターフェース |
| ISourceBlock.cs | `src/libraries/System.Threading.Tasks.Dataflow/src/Base/ISourceBlock.cs` | ソース | ソースインターフェース |
| IPropagatorBlock.cs | `src/libraries/System.Threading.Tasks.Dataflow/src/Base/IPropagatorBlock.cs` | ソース | プロパゲーターインターフェース |
| DataflowBlockOptions.cs | `src/libraries/System.Threading.Tasks.Dataflow/src/Base/DataflowBlockOptions.cs` | ソース | オプション定義 |
| DataflowBlock.cs | `src/libraries/System.Threading.Tasks.Dataflow/src/Base/DataflowBlock.cs` | ソース | 拡張メソッド（Post、SendAsync等） |
| DataflowLinkOptions.cs | `src/libraries/System.Threading.Tasks.Dataflow/src/Base/DataflowLinkOptions.cs` | ソース | リンクオプション |
| BufferBlock.cs | `src/libraries/System.Threading.Tasks.Dataflow/src/Blocks/BufferBlock.cs` | ソース | バッファブロック |
| ActionBlock.cs | `src/libraries/System.Threading.Tasks.Dataflow/src/Blocks/ActionBlock.cs` | ソース | アクションブロック |
| TransformBlock.cs | `src/libraries/System.Threading.Tasks.Dataflow/src/Blocks/TransformBlock.cs` | ソース | 変換ブロック |
| TransformManyBlock.cs | `src/libraries/System.Threading.Tasks.Dataflow/src/Blocks/TransformManyBlock.cs` | ソース | 1対多変換ブロック |
| BatchBlock.cs | `src/libraries/System.Threading.Tasks.Dataflow/src/Blocks/BatchBlock.cs` | ソース | バッチブロック |
| JoinBlock.cs | `src/libraries/System.Threading.Tasks.Dataflow/src/Blocks/JoinBlock.cs` | ソース | ジョインブロック |
| BroadcastBlock.cs | `src/libraries/System.Threading.Tasks.Dataflow/src/Blocks/BroadcastBlock.cs` | ソース | ブロードキャストブロック |
| WriteOnceBlock.cs | `src/libraries/System.Threading.Tasks.Dataflow/src/Blocks/WriteOnceBlock.cs` | ソース | 一回書き込みブロック |
| TargetCore.cs | `src/libraries/System.Threading.Tasks.Dataflow/src/Internal/TargetCore.cs` | ソース | ターゲット共通実装 |
| SourceCore.cs | `src/libraries/System.Threading.Tasks.Dataflow/src/Internal/SourceCore.cs` | ソース | ソース共通実装 |
| SpscTargetCore.cs | `src/libraries/System.Threading.Tasks.Dataflow/src/Internal/SpscTargetCore.cs` | ソース | SPSC最適化実装 |
| ReorderingBuffer.cs | `src/libraries/System.Threading.Tasks.Dataflow/src/Internal/ReorderingBuffer.cs` | ソース | 順序保証バッファ |
| Common.cs | `src/libraries/System.Threading.Tasks.Dataflow/src/Internal/Common.cs` | ソース | 共通ユーティリティ |
