# 機能設計書 48-Channel

## 概要

Julia Base ライブラリにおける Channel（タスク間通信キュー）の作成・送受信・ライフサイクル管理を提供する機能の設計書である。

### 本機能の処理概要

**業務上の目的・背景**：並行処理において、タスク間でデータを安全に受け渡す仕組みが必要である。Channel はスレッドセーフなキューとして機能し、プロデューサー・コンシューマーパターンや、複数タスク間のデータパイプラインの構築を可能にする。バッファリングの有無を選択でき、同期的・非同期的なデータ交換の両方に対応する。

**機能の利用シーン**：プロデューサー・コンシューマーパターンの実装、データパイプラインの構築、walkdir 等の遅延イテレータの基盤、@threads :greedy スケジューリングのアイテム分配、非同期 I/O の結果受け渡しなど。

**主要な処理内容**：
1. `Channel{T}(size)` によるバッファ付き/バッファなしチャネルの作成
2. `put!(c, v)` によるデータの送信（バッファフルでブロック）
3. `take!(c)` によるデータの受信（バッファ空でブロック）
4. `fetch(c)` によるデータの覗き見（バッファ付きのみ、削除しない）
5. `close(c)` によるチャネルのクローズ
6. `bind(c, task)` によるタスクとのライフサイクル連結
7. `isopen(c)` / `isready(c)` / `isfull(c)` / `isempty(c)` による状態確認
8. `Channel(func, size)` によるタスクバインド付きチャネル作成
9. `iterate(c)` による for ループでのイテレーション

**関連システム・外部連携**：Julia ランタイムのタスクスケジューラと連携する。Threads.Condition を内部的に使用してタスク間の待ち合わせを実現する。

**権限による制御**：Channel の操作に権限制御はない。ただし Channel はプロセス内のタスク間通信であり、プロセス間通信には使用できない。

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| - | CLI / REPL | 主画面 | Channel の対話的実行 |

## 機能種別

並行処理基盤 / タスク間通信

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| T | Type | No | チャネルの要素型（デフォルト: Any） | - |
| size | Integer | No | バッファサイズ（デフォルト: 0 = バッファなし） | size >= 0 |
| func | Function | No | チャネルにバインドするタスク関数 | Channel を引数に取ること |
| spawn | Bool | No | タスクを別スレッドで実行するか（デフォルト: false） | - |
| threadpool | Symbol | No | タスクのスレッドプール（デフォルト: :default） | - |
| v | T | Yes (put!) | 送信するデータ | チャネルの型に変換可能 |

### 入力データソース

ユーザーコードから直接引数として渡される。

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| channel | Channel{T} | 作成されたチャネルオブジェクト |
| value | T | take!(c) / fetch(c) で取得するデータ |
| status | Bool | isopen/isready/isfull/isempty の結果 |

### 出力先

関数の戻り値としてユーザーコードに返される。

## 処理フロー

### 処理シーケンス

```
1. チャネルの作成
   └─ Channel{T}(size) でバッファサイズを指定して生成
   └─ 内部に cond_take, cond_put, cond_wait の Condition を保持
2. データの送信 (put!)
   └─ バッファ付き: バッファに空きがあれば追加、なければブロック
   └─ バッファなし: take! 待ちのタスクがあれば直接渡す、なければブロック
3. データの受信 (take!)
   └─ バッファ付き: バッファにデータがあれば取得、なければブロック
   └─ バッファなし: put! 待ちのタスクからデータを受け取る
4. チャネルのクローズ
   └─ state を :closed に変更
   └─ 全待機タスクにエラーを通知
```

### フローチャート

```mermaid
flowchart TD
    A[Channel 作成] --> B{操作}
    B -->|put!| C{バッファ付き?}
    C -->|Yes| D{バッファ空きあり?}
    D -->|Yes| E[data に push]
    D -->|No| F[cond_put で wait]
    F --> D
    C -->|No| G{take! 待ちあり?}
    G -->|Yes| H[直接 schedule で渡す]
    G -->|No| I[cond_put で wait]
    I --> G
    B -->|take!| J{バッファ付き?}
    J -->|Yes| K{データあり?}
    K -->|Yes| L[data から popfirst!]
    K -->|No| M[cond_take で wait]
    M --> K
    J -->|No| N[cond_put に notify]
    N --> O[cond_take で wait]
    B -->|close| P[state = :closed]
    P --> Q[notify_error 全 Condition]
    E --> R[cond_take に notify]
    L --> S[cond_put に notify]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-48-01 | バッファサイズ 0 | size=0 のチャネルはバッファなし。put! と take! が同時に必要 | Channel(0) |
| BR-48-02 | 型変換 | put! 時に値がチャネルの型 T に convert される | put! 呼出時 |
| BR-48-03 | クローズ後の put! | クローズ済みチャネルへの put! は InvalidStateException | close 後 |
| BR-48-04 | クローズ後の take! | クローズ済みで空のチャネルへの take! は InvalidStateException | close 後かつ空 |
| BR-48-05 | bind のライフサイクル | bind されたタスクの終了時にチャネルが自動クローズ | bind 使用時 |
| BR-48-06 | bind の例外伝播 | タスクが例外で終了すると、チャネル待機者に TaskFailedException が伝播 | bind + タスク失敗時 |
| BR-48-07 | fetch は削除しない | fetch はバッファ先頭を返すが削除しない（バッファ付きのみ） | fetch 呼出時 |
| BR-48-08 | fetch はバッファなし非対応 | size=0 のチャネルで fetch は ErrorException | fetch + size=0 |
| BR-48-09 | iterate の自動終了 | for ループでの iterate はチャネルのクローズで終了する | for-in 使用時 |

### 計算ロジック

`n_avail_items` はロックフリーのアトミック変数で、`isready` / `isempty` / `isfull` の高速判定に使用される。バッファ付きチャネルではバッファ内のデータ数を、バッファなしチャネルでは put! 待ちのタスク数を表す。

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

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

該当なし。

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| InvalidStateException | Exception | クローズ済みチャネルへの put! / 空クローズ済みからの take! | isopen で事前チェック |
| ErrorException | Exception | バッファなしチャネルで fetch を呼び出し | バッファ付きチャネルを使用 |
| TaskFailedException | Exception | bind されたタスクが失敗 | タスク内で例外をハンドル |
| ArgumentError | ArgumentError | size < 0 | 0 以上の値を指定 |

### リトライ仕様

自動リトライは行われない。InvalidStateException はチャネルのライフサイクル終了を示すため、リトライではなく処理の終了が適切。

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

Channel の put! / take! はそれぞれアトミック操作として ReentrantLock で保護されている。put! と take! の対は論理的なトランザクション単位を形成するが、アプリケーションレベルでのトランザクション保証はない。

## パフォーマンス要件

- put! / take! はロック競合がない場合、マイクロ秒オーダー
- `n_avail_items` のアトミック読み取りにより、isready / isempty / isfull はロックフリーで高速
- バッファサイズを適切に設定することで、プロデューサーとコンシューマーの速度差を吸収できる

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

- Channel はプロセス内のタスク間通信であり、プロセス間のセキュリティ境界は提供しない
- Channel に格納されたデータは参照型の場合、複数タスクからの同時アクセスが可能であり、データ競合に注意

## 備考

- Channel は AbstractChannel の具象型であり、push! / popfirst! でも操作可能（AbstractChannel インターフェース）
- Channel(Inf) は typemax(Int) サイズのバッファ付きチャネルを作成する
- IteratorSize は SizeUnknown() であり、length は事前に不明

---

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

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

### 推奨読解順序

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

Channel の内部構造を理解することが最も重要である。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | channels.jl | `base/channels.jl` | **7-8行目**: `AbstractChannel{T}` 抽象型の定義 |
| 1-2 | channels.jl | `base/channels.jl` | **32-52行目**: `Channel{T}` 構造体の定義。`cond_take`, `cond_wait`, `cond_put`（Threads.Condition）、`@atomic state`（:open/:closed）、`data`（Vector{T}）、`@atomic n_avail_items`、`sz_max` の各フィールド |

**読解のコツ**: Channel は 3 つの Condition（cond_take, cond_wait, cond_put）を持ち、同一の ReentrantLock を共有する。バッファなし（size=0）では cond_wait は cond_take と区別されるが、バッファ付きでは同一。`@atomic` アノテーションはスレッドセーフなアクセスを示す。

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | channels.jl | `base/channels.jl` | **139-157行目**: `Channel{T}(func, size)` - タスクバインド付きコンストラクタ。Channel 作成 → Task 作成 → bind → schedule |
| 2-2 | channels.jl | `base/channels.jl` | **395-399行目**: `put!(c, v)` - データ送信のエントリーポイント。型変換後に buffered/unbuffered で分岐 |

**主要処理フロー**:
1. **395-399行目**: `put!` は `check_channel_state` で開状態を確認し、`isbuffered` で分岐
2. **411-434行目**: `put_buffered` - ロック取得 → n_avail インクリメント → バッファ空き待ち → push! → cond_take に notify
3. **436-455行目**: `put_unbuffered` - ロック取得 → take! 待ちタスク待ち → popfirst! で直接渡す → schedule + yield
4. **527-541行目**: `take_buffered` - ロック取得 → データ待ち → popfirst! → n_avail デクリメント → cond_put に notify
5. **544-553行目**: `take_unbuffered` - ロック取得 → cond_put に notify → cond_take で wait

#### Step 3: ライフサイクル管理を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | channels.jl | `base/channels.jl` | **199-212行目**: `close(c)` - state を :closed に設定し、全 Condition にエラー通知 |
| 3-2 | channels.jl | `base/channels.jl` | **253-259行目**: `isopen(c)` - @atomic :acquire で state を確認 |
| 3-3 | channels.jl | `base/channels.jl` | **329-334行目**: `bind(c, task)` - タスク終了時に close_chnl_on_taskdone を実行する Task を作成 |
| 3-4 | channels.jl | `base/channels.jl` | **362-376行目**: `close_chnl_on_taskdone` - タスク成功時は close、失敗時は TaskFailedException でclose |

#### Step 4: イテレーションを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | channels.jl | `base/channels.jl` | **705-728行目**: `iterate(c)` - isopen || isready なら take! を試行、InvalidStateException(:closed) で終了 |
| 4-2 | channels.jl | `base/channels.jl` | **593-598行目**: `isready(c)` / `n_avail(c)` - @atomic :monotonic で n_avail_items を読み取り |

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

```
Channel{T}(func, size)
    ├─ Channel{T}(size)         ... チャネル作成
    ├─ Task(() -> func(chnl))   ... タスク作成
    ├─ bind(chnl, task)         ... ライフサイクル連結
    │      └─ close_chnl_on_taskdone(task, c)
    └─ schedule(task) / yield(task)

put!(c, v)
    ├─ check_channel_state(c)   ... 開状態確認
    ├─ convert(T, v)            ... 型変換
    └─ isbuffered(c) ?
        ├─ put_buffered(c, v)
        │      ├─ lock(c)
        │      ├─ _increment_n_avail(c, 1)
        │      ├─ wait(c.cond_put) [バッファフル時]
        │      ├─ push!(c.data, v)
        │      ├─ notify(c.cond_take)
        │      └─ unlock(c)
        └─ put_unbuffered(c, v)
               ├─ lock(c)
               ├─ wait(c.cond_put) [taker なし時]
               ├─ popfirst!(c.cond_take.waitq)
               ├─ schedule(taker, v)
               └─ unlock(c) + yield()

take!(c)
    └─ isbuffered(c) ?
        ├─ take_buffered(c)
        │      ├─ lock(c)
        │      ├─ wait(c.cond_take) [データなし時]
        │      ├─ popfirst!(c.data)
        │      ├─ _increment_n_avail(c, -1)
        │      ├─ notify(c.cond_put)
        │      └─ unlock(c)
        └─ take_unbuffered(c)
               ├─ lock(c)
               ├─ notify(c.cond_put)
               └─ wait(c.cond_take)
```

### データフロー図

```
[Producer]              [Channel]                    [Consumer]

put!(c, v) ──────▶ [Buffer: Vector{T}]  ──────▶ take!(c) ──▶ value
                   sz_max で容量制限
                   n_avail_items で状態追跡

[バッファなしの場合]
put!(c, v) ──────▶ [直接渡し: schedule] ──────▶ take!(c) ──▶ value
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| channels.jl | `base/channels.jl` | ソース | Channel 構造体と全操作関数の定義 |
| condition.jl | `base/condition.jl` | ソース | GenericCondition（Channel の内部同期機構） |
| lock.jl | `base/lock.jl` | ソース | ReentrantLock（Channel の内部ロック） / Threads.Condition |
| task.jl | `base/task.jl` | ソース | Task（Channel にバインドされるタスク） |
| threadingconstructs.jl | `base/threadingconstructs.jl` | ソース | @spawn（spawn=true 時のスレッド割当） |
