# 機能設計書 21-System.Threading.Channels

## 概要

本ドキュメントは、.NETランタイムのSystem.Threading.Channels名前空間が提供するチャネル機能について、その設計仕様と実装の詳細を記述する。チャネルは、プロデューサー・コンシューマーパターンを実装するためのスレッドセーフなデータ構造を提供する。

### 本機能の処理概要

System.Threading.Channelsは、非同期プログラミングにおけるプロデューサー・コンシューマーパターンを効率的に実装するためのライブラリである。複数のスレッド間でデータを安全に受け渡すための高性能なメカニズムを提供する。

**業務上の目的・背景**：現代のアプリケーションでは、非同期処理やマルチスレッド処理が一般的になっている。しかし、スレッド間でデータを安全に受け渡すことは複雑でエラーが発生しやすい。System.Threading.Channelsは、この課題を解決するために設計された。特に、バックプレッシャー制御、キャンセレーション対応、非同期イテレーションなど、現代的な非同期プログラミングの要件を満たす機能を提供する。

**機能の利用シーン**：
- ストリーミングデータの処理（ログ集約、イベント処理）
- パイプライン処理の構築（ETLパイプライン、データ変換）
- バックグラウンドワーカーへのタスク配布
- WebSocket/SignalRなどのリアルタイム通信における受信メッセージのキューイング
- レート制限が必要なAPI呼び出しのバッファリング

**主要な処理内容**：
1. チャネルの作成（Unbounded/Bounded/Rendezvous）
2. 非同期書き込み操作（TryWrite、WriteAsync、WaitToWriteAsync）
3. 非同期読み取り操作（TryRead、ReadAsync、WaitToReadAsync、ReadAllAsync）
4. チャネルの完了管理（Complete、Completion）
5. キャンセレーショントークンによる操作の中断
6. バックプレッシャー制御（BoundedChannelFullMode）

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

**権限による制御**：特別な権限制御は不要。ただし、チャネルオプションによりSingleReader/SingleWriterの制約を設定可能。

## 関連画面

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

## 機能種別

データ連携 / バッファリング / 非同期通信

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| capacity | int | No（Boundedの場合Yes） | チャネルの最大容量 | 0以上の整数（0はRendezvousチャネル） |
| options | ChannelOptions | No | チャネルの動作オプション | nullでない場合、各プロパティの整合性チェック |
| item | T | Yes（書き込み時） | チャネルに書き込むデータ | 型パラメータTの制約に従う |
| cancellationToken | CancellationToken | No | 操作のキャンセルトークン | キャンセル済みの場合は即座にキャンセル |

### 入力データソース

- プログラムコードからのデータ投入（Channel.Writer.WriteAsync等）
- 外部システムからの非同期データストリーム

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| Channel<T> | Channel<T> | 生成されたチャネルインスタンス |
| ReadResult | T | チャネルから読み取られたデータ |
| WriteResult | bool | 書き込み成功の可否 |
| WaitResult | bool | 読み取り/書き込み可能になったかどうか |
| Completion | Task | チャネル完了を表すTask |

### 出力先

- プログラムコード（Channel.Reader経由でのデータ取得）
- IAsyncEnumerable<T>経由での非同期イテレーション

## 処理フロー

### 処理シーケンス

```
1. チャネル作成
   └─ Channel.CreateUnbounded<T>() または Channel.CreateBounded<T>(capacity) を呼び出し
   └─ オプションに基づき適切なチャネル実装を選択（UnboundedChannel/BoundedChannel/SingleConsumerUnboundedChannel/RendezvousChannel）

2. データ書き込み（Producer側）
   └─ TryWrite(item): 同期的な書き込み試行
   └─ WriteAsync(item, ct): 非同期書き込み（空きができるまで待機）
   └─ WaitToWriteAsync(ct): 書き込み可能になるまで待機

3. データ読み取り（Consumer側）
   └─ TryRead(out item): 同期的な読み取り試行
   └─ ReadAsync(ct): 非同期読み取り（データが来るまで待機）
   └─ WaitToReadAsync(ct): 読み取り可能になるまで待機
   └─ ReadAllAsync(ct): IAsyncEnumerable<T>としてすべてのデータを列挙

4. チャネル完了
   └─ Writer.Complete(): これ以上データを書き込まないことを宣言
   └─ Reader.Completion: 完了を待機するTask
```

### フローチャート

```mermaid
flowchart TD
    A[チャネル作成] --> B{Bounded?}
    B -->|Yes| C[BoundedChannel生成]
    B -->|No| D[UnboundedChannel生成]
    C --> E[Writer/Reader取得]
    D --> E
    E --> F{Producer処理}
    F --> G[TryWrite/WriteAsync]
    G --> H{書き込み成功?}
    H -->|Yes| I[次のデータ]
    H -->|No, Bounded| J{FullMode?}
    J -->|Wait| K[空きを待機]
    J -->|DropNewest| L[最新を破棄]
    J -->|DropOldest| M[最古を破棄]
    J -->|DropWrite| N[書き込みを破棄]
    K --> G
    I --> F
    E --> O{Consumer処理}
    O --> P[TryRead/ReadAsync]
    P --> Q{読み取り成功?}
    Q -->|Yes| R[データ処理]
    Q -->|No| S[データを待機]
    R --> O
    S --> P
    F --> T[Writer.Complete]
    T --> U[Reader.Completion完了]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-01 | スレッドセーフティ | 複数スレッドからの同時アクセスが安全 | 常時 |
| BR-02 | FIFO順序保証 | 書き込み順序と読み取り順序が保証される | 常時（DropNewest/DropOldest除く） |
| BR-03 | バックプレッシャー | Boundedチャネルで容量超過時に制御 | BoundedChannel使用時 |
| BR-04 | 完了伝播 | Writer完了後、すべてのデータ読み取り後にReader完了 | Complete呼び出し時 |
| BR-05 | キャンセレーション | CancellationToken指定時に操作中断可能 | キャンセルトークン指定時 |

### 計算ロジック

バックプレッシャー制御（BoundedChannelFullMode）:
- Wait: 空きができるまで書き込みをブロック
- DropNewest: 最新のアイテムを破棄して新しいアイテムを追加
- DropOldest: 最古のアイテムを破棄して新しいアイテムを追加
- DropWrite: 書き込み自体を破棄（成功を返す）

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

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

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

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| ChannelClosedException | 例外 | 完了済みチャネルからの読み取り | チャネル完了を確認してから読み取り |
| InvalidOperationException | 例外 | 既に完了済みのチャネルにComplete呼び出し | TryCompleteを使用 |
| OperationCanceledException | 例外 | CancellationTokenがキャンセル済み | キャンセル状態を確認 |
| ArgumentNullException | 例外 | optionsがnull | 有効なオプションを指定 |
| ArgumentOutOfRangeException | 例外 | capacityが負数 | 0以上の値を指定 |

### リトライ仕様

- TryWrite/TryReadは即座にfalseを返すため、呼び出し側でリトライロジックを実装
- WriteAsync/ReadAsyncはキャンセルまたは完了まで自動的にリトライ

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

本機能はトランザクションを使用しない。各操作は独立してアトミックに実行される。

## パフォーマンス要件

- UnboundedChannel: ConcurrentQueueベースで高スループット
- BoundedChannel: Dequeベースでメモリ使用量を制限
- SingleConsumerUnboundedChannel: 単一リーダー向けに最適化
- 非同期操作のオーバーヘッドを最小化するためのValueTask使用
- シングルトンオペレーションの再利用によるアロケーション削減

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

- チャネル自体はスレッドセーフだが、格納されるデータの安全性は呼び出し側の責任
- 機密データをチャネル経由で受け渡す場合は、適切な暗号化を検討
- メモリダンプ時にチャネル内のデータが露出する可能性あり

## 備考

- .NET Core 3.0以降で利用可能
- IAsyncEnumerable<T>との統合により、C# 8.0のawait foreachで直接利用可能
- System.Threading.Tasks.Dataflowの代替として、より軽量なシナリオに適している

---

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

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

### 推奨読解順序

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

チャネルの基本構造とオプション設定を理解することが重要。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | Channel_1.cs | `src/libraries/System.Threading.Channels/src/System/Threading/Channels/Channel_1.cs` | Channel<T>の基本構造（Reader/Writerプロパティ） |
| 1-2 | Channel_2.cs | `src/libraries/System.Threading.Channels/src/System/Threading/Channels/Channel_2.cs` | Channel<TWrite, TRead>の構造 |
| 1-3 | ChannelOptions.cs | `src/libraries/System.Threading.Channels/src/System/Threading/Channels/ChannelOptions.cs` | SingleWriter/SingleReader/AllowSynchronousContinuationsオプション |
| 1-4 | BoundedChannelFullMode.cs | `src/libraries/System.Threading.Channels/src/System/Threading/Channels/BoundedChannelFullMode.cs` | バックプレッシャー制御モード（Wait/DropNewest/DropOldest/DropWrite） |

**読解のコツ**: ChannelOptionsクラスの各プロパティがチャネル実装の選択と動作にどう影響するかを理解する。

#### Step 2: エントリーポイントを理解する

チャネル作成の起点となるファクトリメソッドを確認。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | Channel.cs | `src/libraries/System.Threading.Channels/src/System/Threading/Channels/Channel.cs` | CreateUnbounded/CreateBoundedファクトリメソッド |

**主要処理フロー**:
1. **11-12行目**: CreateUnbounded<T>()でUnboundedChannelを生成
2. **19-29行目**: CreateUnbounded<T>(options)でオプションに基づきSingleConsumerUnboundedChannelまたはUnboundedChannelを選択
3. **40-43行目**: CreateBounded<T>(capacity)でcapacityに基づきBoundedChannelまたはRendezvousChannelを生成
4. **59-66行目**: CreateBounded<T>(options, itemDropped)で詳細オプション指定

#### Step 3: Reader/Writer抽象クラスを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | ChannelReader.cs | `src/libraries/System.Threading.Channels/src/System/Threading/Channels/ChannelReader.cs` | 読み取りAPIの抽象定義 |
| 3-2 | ChannelWriter.cs | `src/libraries/System.Threading.Channels/src/System/Threading/Channels/ChannelWriter.cs` | 書き込みAPIの抽象定義 |

**主要処理フロー（ChannelReader.cs）**:
- **36行目**: TryRead抽象メソッド（同期読み取り）
- **53行目**: WaitToReadAsync抽象メソッド（読み取り可能待機）
- **58-94行目**: ReadAsync仮想メソッド（TryRead + WaitToReadAsyncの組み合わせ）
- **103-112行目**: ReadAllAsync仮想メソッド（IAsyncEnumerable<T>生成）

**主要処理フロー（ChannelWriter.cs）**:
- **25行目**: TryWrite抽象メソッド（同期書き込み）
- **33行目**: WaitToWriteAsync抽象メソッド（書き込み可能待機）
- **39-52行目**: WriteAsync仮想メソッド（TryWrite + WaitToWriteAsyncの組み合わせ）
- **70-76行目**: Completeメソッド（チャネル完了）

#### Step 4: 具体的なチャネル実装を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | UnboundedChannel.cs | `src/libraries/System.Threading.Channels/src/System/Threading/Channels/UnboundedChannel.cs` | 無制限チャネルの実装 |
| 4-2 | BoundedChannel.cs | `src/libraries/System.Threading.Channels/src/System/Threading/Channels/BoundedChannel.cs` | 容量制限チャネルの実装 |
| 4-3 | SingleConsumerUnboundedChannel.cs | `src/libraries/System.Threading.Channels/src/System/Threading/Channels/SingleConsumerUnboundedChannel.cs` | 単一コンシューマー向け最適化 |
| 4-4 | RendezvousChannel.cs | `src/libraries/System.Threading.Channels/src/System/Threading/Channels/RendezvousChannel.cs` | 容量0のランデブーチャネル |

**主要処理フロー（UnboundedChannel.cs）**:
- **21行目**: ConcurrentQueue<T>をバッファとして使用
- **77-81行目**: TryDequeueでロックなしの高速読み取り
- **254-262行目**: TryWriteで直接キューイングまたはブロック中リーダーへの転送

**主要処理フロー（BoundedChannel.cs）**:
- **29行目**: Deque<T>をバッファとして使用（両端キュー）
- **404-442行目**: TryWriteでのFullMode処理（Wait/DropNewest/DropOldest/DropWrite）
- **239-284行目**: DequeueItemAndPostProcessでブロック中ライターの復帰処理

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

```
Channel.CreateUnbounded<T>() / Channel.CreateBounded<T>()
    │
    ├─ UnboundedChannel<T> / BoundedChannel<T> コンストラクタ
    │      ├─ Reader = new UnboundedChannelReader(this) / BoundedChannelReader(this)
    │      └─ Writer = new UnboundedChannelWriter(this) / BoundedChannelWriter(this)
    │
    ├─ ChannelWriter<T>.WriteAsync(item)
    │      ├─ TryWrite(item) [ロック内で処理]
    │      │      ├─ ブロック中Readerがいる場合: 直接転送
    │      │      └─ キューに追加 + WaitingReadersを通知
    │      └─ WriteAsyncCore (TryWrite失敗時)
    │             └─ WaitToWriteAsync + TryWrite ループ
    │
    └─ ChannelReader<T>.ReadAsync()
           ├─ TryRead(out item) [ロック内で処理]
           │      ├─ キューから取得
           │      └─ BoundedChannel: ブロック中Writerを復帰
           └─ ReadAsyncCore (TryRead失敗時)
                  └─ WaitToReadAsync + TryRead ループ
```

### データフロー図

```
[入力]               [処理]                    [出力]

Producer            Channel                   Consumer
   │                   │                         │
   ├─ WriteAsync() ───▶│◀─────────────────────── ReadAsync() ─┤
   │                   │                         │
   │              ┌────┴────┐                    │
   │              │ Buffer  │                    │
   │              │(Queue)  │                    │
   │              └────┬────┘                    │
   │                   │                         │
   └─ Complete() ─────▶│───▶ Completion Task ───▶│
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| Channel.cs | `src/libraries/System.Threading.Channels/src/System/Threading/Channels/Channel.cs` | ソース | ファクトリメソッド |
| Channel_1.cs | `src/libraries/System.Threading.Channels/src/System/Threading/Channels/Channel_1.cs` | ソース | Channel<T>型定義 |
| Channel_2.cs | `src/libraries/System.Threading.Channels/src/System/Threading/Channels/Channel_2.cs` | ソース | Channel<TWrite,TRead>型定義 |
| ChannelReader.cs | `src/libraries/System.Threading.Channels/src/System/Threading/Channels/ChannelReader.cs` | ソース | 読み取り抽象クラス |
| ChannelWriter.cs | `src/libraries/System.Threading.Channels/src/System/Threading/Channels/ChannelWriter.cs` | ソース | 書き込み抽象クラス |
| ChannelOptions.cs | `src/libraries/System.Threading.Channels/src/System/Threading/Channels/ChannelOptions.cs` | ソース | オプション設定 |
| BoundedChannelFullMode.cs | `src/libraries/System.Threading.Channels/src/System/Threading/Channels/BoundedChannelFullMode.cs` | ソース | バックプレッシャーモード列挙 |
| UnboundedChannel.cs | `src/libraries/System.Threading.Channels/src/System/Threading/Channels/UnboundedChannel.cs` | ソース | 無制限チャネル実装 |
| BoundedChannel.cs | `src/libraries/System.Threading.Channels/src/System/Threading/Channels/BoundedChannel.cs` | ソース | 容量制限チャネル実装 |
| SingleConsumerUnboundedChannel.cs | `src/libraries/System.Threading.Channels/src/System/Threading/Channels/SingleConsumerUnboundedChannel.cs` | ソース | 単一コンシューマー最適化 |
| RendezvousChannel.cs | `src/libraries/System.Threading.Channels/src/System/Threading/Channels/RendezvousChannel.cs` | ソース | ランデブーチャネル実装 |
| AsyncOperation.cs | `src/libraries/System.Threading.Channels/src/System/Threading/Channels/AsyncOperation.cs` | ソース | 非同期操作の基盤クラス |
| ChannelUtilities.cs | `src/libraries/System.Threading.Channels/src/System/Threading/Channels/ChannelUtilities.cs` | ソース | 共通ユーティリティ |
| ChannelClosedException.cs | `src/libraries/System.Threading.Channels/src/System/Threading/Channels/ChannelClosedException.cs` | ソース | 例外クラス |
