# 機能設計書 51-非同期マッピング

## 概要

本ドキュメントは、Julia の `asyncmap` / `asyncmap!` 関数および関連するイテレータ型（`AsyncCollector`, `AsyncGenerator`）による非同期並列マッピング処理の機能設計を記述する。

### 本機能の処理概要

本機能は、コレクションの各要素に対して関数を非同期的に適用し、複数のタスクを用いて並列にマッピング処理を実行する仕組みを提供する。通常の `map` 関数が逐次的に処理を行うのに対し、`asyncmap` は I/O 待ちやネットワーク通信などのブロッキング処理を含む操作において、複数のタスクを同時に実行することでスループットを向上させる。

**業務上の目的・背景**: I/O バウンドな処理（ファイル読み込み、ネットワークリクエスト、外部プロセス呼び出しなど）において、逐次処理では待ち時間が累積し全体の処理時間が長くなる。非同期マッピングを使用することで、一方のタスクが I/O を待っている間に他のタスクが処理を進め、全体のスループットを大幅に改善できる。Julia のグリーンスレッド（Task）ベースの協調的マルチタスクを活用し、ユーザーが明示的なタスク管理を行うことなく並列マッピングを実現する。

**機能の利用シーン**: 複数の URL からデータを並列ダウンロードする場合、複数のファイルを同時に処理する場合、外部 API への並列リクエスト、バッチ処理で多数のデータ項目を並列変換する場合に利用される。

**主要な処理内容**:
1. `asyncmap(f, c...; ntasks, batch_size)` によるコレクション要素への非同期関数適用
2. `asyncmap!(f, results, c...)` による結果の既存配列への直接書き込み
3. `AsyncCollector` イテレータによる非同期収集の段階的実行
4. `AsyncGenerator` イテレータによる遅延評価型の非同期マッピング
5. `ntasks` パラメータによるタスク並列度の制御（固定値または動的関数指定）
6. `batch_size` パラメータによるバッチモード処理の制御

**関連システム・外部連携**: 本機能は `Base.Task`（コルーチン）、`Base.Channel`（タスク間通信）を内部的に使用する。外部システムとの直接的な連携はないが、I/O 操作を含む関数の並列実行に広く活用される。

**権限による制御**: 特に権限制御は行われない。全てのユーザーコードから利用可能。

## 関連画面

本機能は GUI 画面を持たないライブラリ機能であり、直接的に関連する画面はない。REPL から対話的に呼び出して使用する。

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| - | - | - | REPL から asyncmap 関数を呼び出して使用 |

## 機能種別

計算処理（並列データ変換）

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| f | Function | Yes | コレクション要素に適用するマッピング関数 | batch_size 指定時は Vector を受け取り Vector を返す関数であること |
| c | コレクション（可変長引数） | Yes | マッピング対象のコレクション（1つ以上） | 複数指定時は同じ長さであること |
| ntasks | Union{Int, Function} | No | 並列タスク数。デフォルト 0（自動）。0-arg 関数も指定可。 | 0 以上の整数または 0 引数関数であること |
| batch_size | Union{Int, Nothing} | No | バッチサイズ。nothing でバッチモード無効。 | 1 以上の正の整数であること |
| results | AbstractArray | asyncmap! のみ必須 | 結果を格納する配列 | 出力を格納可能なサイズであること |

### 入力データソース

ユーザーコードから直接引数として渡されるコレクション（Array, Range, その他イテラブル）。

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| 結果配列 | Vector / BitArray / String | 入力コレクションの各要素に f を適用した結果。入力の順序が保持される |
| AsyncCollector | Iterator | 非同期収集を段階的に実行するイテレータ（iterate で nothing を返す） |
| AsyncGenerator | Iterator | 遅延評価型の非同期マッピングイテレータ（iterate で結果値を返す） |

### 出力先

関数の戻り値として呼び出し元に返却。`asyncmap!` の場合は引数 `results` に直接書き込み。

## 処理フロー

### 処理シーケンス

```
1. パラメータ検証
   └─ verify_ntasks: ntasks を検証・正規化（0 の場合は min(100, length(c)) に設定）
   └─ verify_batch_size: batch_size を検証（正の整数であること）
2. 実行関数の構築
   └─ batch_size 指定あり: バッチ処理関数を構築
   └─ batch_size 指定なし: 単要素処理関数を構築
3. Channel とワーカータスクのセットアップ
   └─ setup_chnl_and_tasks: 無バッファ Channel を生成、ntasks 分のワーカータスクを起動
4. ドライバー処理（2パスマッピング）
   └─ 第1パス: map で各要素を Ref オブジェクトに包み、Channel に送信
   └─ ワーカータスクが Channel から取り出し、f を適用して Ref に結果を書き込み
5. 完了待機と結果収集
   └─ Channel を close し、全ワーカータスクの完了を待機
   └─ 第2パス: Ref から結果値を抽出し、結果配列を構築
6. エラー処理
   └─ ワーカータスクで例外が発生した場合、Channel を close し例外を伝播
```

### フローチャート

```mermaid
flowchart TD
    A[asyncmap 呼び出し] --> B[パラメータ検証]
    B --> C{batch_size 指定?}
    C -->|Yes| D[バッチ処理関数構築]
    C -->|No| E[単要素処理関数構築]
    D --> F[Channel + ワーカータスク起動]
    E --> F
    F --> G[第1パス: map で Ref + Channel 送信]
    G --> H[ワーカータスクが並列処理]
    H --> I[Channel close]
    I --> J[全ワーカータスク完了待機]
    J --> K{例外発生?}
    K -->|Yes| L[例外を throw]
    K -->|No| M[第2パス: Ref から結果抽出]
    M --> N[結果配列を返却]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-51-1 | 出力順序保証 | 出力は入力コレクションの要素順序と同一であることが保証される | 常時 |
| BR-51-2 | デフォルトタスク数 | ntasks=0 の場合、min(100, length(c)) タスクが使用される | ntasks 未指定時 |
| BR-51-3 | 動的タスク数 | ntasks が関数の場合、各要素処理前にタスク数をチェックし必要に応じて追加起動 | ntasks が Function 型の場合 |
| BR-51-4 | バッチモード | batch_size 指定時、f は引数タプルの Vector を受け取り結果 Vector を返す必要がある | batch_size が正の整数の場合 |

### 計算ロジック

特別な計算ロジックは存在しない。マッピング関数 f の適用結果がそのまま出力となる。

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

該当なし。本機能はデータベース操作を行わない。

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| - | ArgumentError | ntasks が負の数値、または正の整数でも 0 引数関数でもない | 適切な ntasks 値を指定 |
| - | ArgumentError | batch_size が正の整数でない | 正の整数を指定、または nothing を使用 |
| - | InvalidStateException | ワーカータスクの例外により Channel が close された場合の put! 失敗 | ワーカータスクの例外を確認・修正 |
| - | 任意の例外 | マッピング関数 f の実行中に例外が発生 | f の実装を修正 |

### リトライ仕様

組み込みのリトライ機構は提供されない。リトライが必要な場合はマッピング関数 f 内で実装する。

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

該当なし。トランザクション管理は行わない。

## パフォーマンス要件

- タスク数のデフォルト上限は 100
- Channel は無バッファ（サイズ 0）で生成され、バックプレッシャーが自然に発生する
- バッチモードを使用することでタスク間通信のオーバーヘッドを削減可能

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

特になし。マッピング関数が外部リソースにアクセスする場合は、呼び出し元の責任で適切なセキュリティ対策を講じること。

## 備考

- `asyncmap` は CPU バウンドな処理には不向き。CPU 集約的な並列処理には `Threads.@threads` や `Distributed.pmap` の使用を推奨
- BitArray 入力の場合、マッピング関数が Bool を返すときは BitArray として結果が返される
- String 入力の場合、内部的に Vector{Char} に変換して処理し、最後に String に再構築される

---

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

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

### 推奨読解順序

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

まず、非同期マッピングで使用されるデータ構造を理解することが重要。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | asyncmap.jl | `base/asyncmap.jl` | AsyncCollector 構造体（258-267行目）: マッピング関数、結果格納先、列挙子、タスク数、バッチサイズを保持する mutable struct |
| 1-2 | asyncmap.jl | `base/asyncmap.jl` | AsyncCollectorState 構造体（293-299行目）: Channel、ワーカータスクリスト、列挙子状態を保持 |
| 1-3 | asyncmap.jl | `base/asyncmap.jl` | AsyncGenerator 構造体（349-351行目）: AsyncCollector をラップした遅延評価型イテレータ |
| 1-4 | asyncmap.jl | `base/asyncmap.jl` | AsyncGeneratorState 構造体（357-362行目）: 現在インデックス、完了フラグ、コレクター状態を保持 |

**読解のコツ**: Julia の `mutable struct` はフィールドが変更可能な参照型。`const` 修飾のないフィールドは実行中に更新される。`Ref{Any}` は結果の受け渡しに使われるボックス型。

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

処理の起点となる関数を特定する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | asyncmap.jl | `base/asyncmap.jl` | asyncmap 関数（76-78行目）: メインエントリーポイント。async_usemap に委譲 |
| 2-2 | asyncmap.jl | `base/asyncmap.jl` | asyncmap! 関数（402-405行目）: in-place 版。AsyncCollector を foreach で消費 |

**主要処理フロー**:
1. **76-78行目**: `asyncmap` は `async_usemap` を呼び出す
2. **80-100行目**: `async_usemap` でパラメータ検証、実行関数構築、Channel セットアップ、2パス実行

#### Step 3: パラメータ検証を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | asyncmap.jl | `base/asyncmap.jl` | verify_ntasks（116-130行目）: ntasks の検証と自動設定ロジック |
| 3-2 | asyncmap.jl | `base/asyncmap.jl` | verify_batch_size（103-113行目）: batch_size の検証ロジック |

**主要処理フロー**:
- **122-128行目**: ntasks=0 の場合、haslength なら `max(1, min(100, length(iterable)))` に、そうでなければ 100 に設定

#### Step 4: Channel とワーカータスクのセットアップを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | asyncmap.jl | `base/asyncmap.jl` | setup_chnl_and_tasks（187-206行目）: 無バッファ Channel 生成とワーカータスク起動 |
| 4-2 | asyncmap.jl | `base/asyncmap.jl` | start_worker_task!（208-240行目）: 個々のワーカータスクの生成と起動ロジック |

**主要処理フロー**:
- **201行目**: `Channel(0)` で無バッファ Channel を生成（バックプレッシャー制御）
- **209行目**: `@async` でワーカータスクを起動
- **213-227行目**: バッチモード時はバッチ収集後に exec_func を呼び出し
- **228-231行目**: 非バッチモード時は Channel から 1 要素ずつ取り出して exec_func を呼び出し

#### Step 5: 2パスマッピングを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 5-1 | asyncmap.jl | `base/asyncmap.jl` | wrap_n_exec_twice（132-150行目）: ドライバーロジックと動的タスク数チェック |
| 5-2 | asyncmap.jl | `base/asyncmap.jl` | maptwice（152-185行目）: 第1パス（Ref 生成 + Channel 送信）と第2パス（結果抽出） |

**主要処理フロー**:
- **135行目**: push_arg_to_channel: Ref を生成し、(Ref, args) を Channel に送信
- **157行目**: 第1パス: `map(wrapped_f, c...)` で全要素を Channel に送信
- **170行目**: Channel を close して全ワーカーの完了を待機
- **183行目**: 第2パス: `map(ref->ref.x, asyncrun)` で Ref から結果値を抽出

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

```
asyncmap(f, c...; ntasks, batch_size)
    │
    ├─ async_usemap(f, c...; ntasks, batch_size)
    │      ├─ verify_ntasks(c[1], ntasks)
    │      ├─ verify_batch_size(batch_size)
    │      ├─ setup_chnl_and_tasks(exec_func, ntasks, batch_size)
    │      │      └─ start_worker_task!(worker_tasks, exec_func, chnl, batch_size) [ntasks 回]
    │      │             └─ @async ワーカーループ
    │      │                    └─ exec_func(data...)  [= f(args...) or f(batch)]
    │      └─ wrap_n_exec_twice(chnl, worker_tasks, ntasks, exec_func, c...)
    │             └─ maptwice(map_f, chnl, worker_tasks, c...)
    │                    ├─ map(wrapped_f, c...)     [第1パス: Ref 生成]
    │                    ├─ close(chnl)
    │                    ├─ foreach(fetch, worker_tasks) [完了待機]
    │                    └─ map(ref->ref.x, asyncrun)  [第2パス: 結果抽出]
    │
    ├─ asyncmap(f, s::AbstractString; kwargs...)  [String 特殊化]
    │      └─ asyncmap!(f, s2, s; kwargs...)
    │
    └─ asyncmap(f, b::BitArray; kwargs...)         [BitArray 特殊化]
           └─ async_usemap(f, b; kwargs...)

asyncmap!(f, r, c...; ntasks, batch_size)
    └─ foreach(identity, AsyncCollector(f, r, c...; ntasks, batch_size))
           └─ iterate(::AsyncCollector) / iterate(::AsyncCollector, ::AsyncCollectorState)
```

### データフロー図

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

コレクション c ──────▶ map(wrapped_f, c...)  ─────▶ Vector{Ref{Any}}
                         │                              │
                         │ put!(chnl, (ref, args))      │
                         ▼                              │
                    Channel(0)                          │
                         │                              │
                         │ take! (ワーカータスク)         │
                         ▼                              │
                    exec_func(ref, args)                │
                    ref.x = f(args...)                  │
                         │                              │
                         ▼                              ▼
                    close(chnl)           map(ref->ref.x, refs) ──▶ 結果配列
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| asyncmap.jl | `base/asyncmap.jl` | ソース | 非同期マッピングの全実装（asyncmap, asyncmap!, AsyncCollector, AsyncGenerator） |
| channels.jl | `base/channels.jl` | ソース | Channel 型の実装。タスク間通信の基盤 |
| task.jl | `base/task.jl` | ソース | Task 型と @async マクロの実装。グリーンスレッドの基盤 |
| iterators.jl | `base/iterators.jl` | ソース | Enumerate イテレータの実装。asyncmap 内部で使用 |
| essentials.jl | `base/essentials.jl` | ソース | Ref 型の定義。結果受け渡しのボックスとして使用 |
